Mirrored_Repos/python-kasa
1
0
Fork 0
mirror of https://github.com/python-kasa/python-kasa.git synced 2025-08-09 20:24:02 +00:00
Code Issues Actions 1 Packages Projects Releases Wiki Activity

Make close behaviour consistent across new protocols and transports (#660)

Browse Source
This commit is contained in:
Steven B
2024-01-20 12:35:05 +00:00
committed by GitHub
parent e94cd118a4
commit 49cfef087c
4 changed files with 46 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
kasa/tests/test_klapprotocol.py
View File

@@ -16,7 +16,9 @@ from ..deviceconfig import DeviceConfig
from ..exceptions import (
AuthenticationException,
ConnectionException,
RetryableException,
SmartDeviceException,
TimeoutException,
)
from ..httpclient import HttpClient
from ..iotprotocol import IotProtocol
@@ -58,7 +60,7 @@ class _mock_response:
@pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport])
@pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol])
@pytest.mark.parametrize("retry_count", [1, 3, 5])
async def test_protocol_retries(
async def test_protocol_retries_via_client_session(
mocker, retry_count, protocol_class, transport_class, error, retry_expectation
):
host = "127.0.0.1"
@@ -74,6 +76,34 @@ async def test_protocol_retries(
assert conn.call_count == expected_count
@pytest.mark.parametrize(
"error, retry_expectation",
[
(SmartDeviceException("dummy exception"), False),
(RetryableException("dummy exception"), True),
(TimeoutException("dummy exception"), True),
],
ids=("SmartDeviceException", "RetryableException", "TimeoutException"),
)
@pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport])
@pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol])
@pytest.mark.parametrize("retry_count", [1, 3, 5])
async def test_protocol_retries_via_httpclient(
mocker, retry_count, protocol_class, transport_class, error, retry_expectation
):
host = "127.0.0.1"
conn = mocker.patch.object(HttpClient, "post", side_effect=error)
config = DeviceConfig(host)
with pytest.raises(SmartDeviceException):
await protocol_class(transport=transport_class(config=config)).query(
DUMMY_QUERY, retry_count=retry_count
)
expected_count = retry_count + 1 if retry_expectation else 1
assert conn.call_count == expected_count
@pytest.mark.parametrize("transport_class", [AesTransport, KlapTransport])
@pytest.mark.parametrize("protocol_class", [IotProtocol, SmartProtocol])
async def test_protocol_no_retry_on_connection_error(