mirror of
https://github.com/python-kasa/python-kasa.git
synced 2025-08-09 20:24:02 +00:00
Handle smartcam partial list responses (#1411)
This commit is contained in:
@@ -33,6 +33,7 @@ class FakeSmartCamTransport(BaseTransport):
|
||||
*,
|
||||
list_return_size=10,
|
||||
is_child=False,
|
||||
get_child_fixtures=True,
|
||||
verbatim=False,
|
||||
components_not_included=False,
|
||||
):
|
||||
@@ -52,9 +53,12 @@ class FakeSmartCamTransport(BaseTransport):
|
||||
self.verbatim = verbatim
|
||||
if not is_child:
|
||||
self.info = copy.deepcopy(info)
|
||||
self.child_protocols = FakeSmartTransport._get_child_protocols(
|
||||
self.info, self.fixture_name, "getChildDeviceList"
|
||||
)
|
||||
# We don't need to get the child fixtures if testing things like
|
||||
# lists
|
||||
if get_child_fixtures:
|
||||
self.child_protocols = FakeSmartTransport._get_child_protocols(
|
||||
self.info, self.fixture_name, "getChildDeviceList"
|
||||
)
|
||||
else:
|
||||
self.info = info
|
||||
# self.child_protocols = self._get_child_protocols()
|
||||
@@ -229,9 +233,16 @@ class FakeSmartCamTransport(BaseTransport):
|
||||
list_key = next(
|
||||
iter([key for key in result if isinstance(result[key], list)])
|
||||
)
|
||||
assert isinstance(params, dict)
|
||||
module_name = next(iter(params))
|
||||
|
||||
start_index = (
|
||||
start_index
|
||||
if (params and (start_index := params.get("start_index")))
|
||||
if (
|
||||
params
|
||||
and module_name
|
||||
and (start_index := params[module_name].get("start_index"))
|
||||
)
|
||||
else 0
|
||||
)
|
||||
|
||||
|
@@ -10,6 +10,7 @@ from kasa.exceptions import (
|
||||
KasaException,
|
||||
SmartErrorCode,
|
||||
)
|
||||
from kasa.protocols.smartcamprotocol import SmartCamProtocol
|
||||
from kasa.protocols.smartprotocol import SmartProtocol, _ChildProtocolWrapper
|
||||
from kasa.smart import SmartDevice
|
||||
|
||||
@@ -373,6 +374,46 @@ async def test_smart_protocol_lists_multiple_request(mocker, list_sum, batch_siz
|
||||
assert resp == response
|
||||
|
||||
|
||||
@pytest.mark.parametrize("list_sum", [5, 10, 30])
|
||||
@pytest.mark.parametrize("batch_size", [1, 2, 3, 50])
|
||||
async def test_smartcam_protocol_list_request(mocker, list_sum, batch_size):
|
||||
"""Test smartcam protocol list handling for lists."""
|
||||
child_list = [{"foo": i} for i in range(list_sum)]
|
||||
|
||||
response = {
|
||||
"getChildDeviceList": {
|
||||
"child_device_list": child_list,
|
||||
"start_index": 0,
|
||||
"sum": list_sum,
|
||||
},
|
||||
"getChildDeviceComponentList": {
|
||||
"child_component_list": child_list,
|
||||
"start_index": 0,
|
||||
"sum": list_sum,
|
||||
},
|
||||
}
|
||||
request = {
|
||||
"getChildDeviceList": {"childControl": {"start_index": 0}},
|
||||
"getChildDeviceComponentList": {"childControl": {"start_index": 0}},
|
||||
}
|
||||
|
||||
ft = FakeSmartCamTransport(
|
||||
response,
|
||||
"foobar",
|
||||
list_return_size=batch_size,
|
||||
components_not_included=True,
|
||||
get_child_fixtures=False,
|
||||
)
|
||||
protocol = SmartCamProtocol(transport=ft)
|
||||
query_spy = mocker.spy(protocol, "_execute_query")
|
||||
resp = await protocol.query(request)
|
||||
expected_count = 1 + 2 * (
|
||||
int(list_sum / batch_size) + (0 if list_sum % batch_size else -1)
|
||||
)
|
||||
assert query_spy.call_count == expected_count
|
||||
assert resp == response
|
||||
|
||||
|
||||
async def test_incomplete_list(mocker, caplog):
|
||||
"""Test for handling incomplete lists returned from queries."""
|
||||
info = {
|
||||
|
Reference in New Issue
Block a user