Support smart child modules queries (#967)

Required for the P300 firmware update with `auto_off` module on child
devices. Will query child modules for parent devices that are not hubs.

Coverage will be fixed when the P300 fixture is added
https://github.com/python-kasa/python-kasa/pull/915
This commit is contained in:
Steven B
2024-06-10 15:47:00 +01:00
committed by GitHub
parent 927fe648ac
commit db6276d3fd
6 changed files with 64 additions and 17 deletions

View File

@@ -149,6 +149,11 @@ class FakeSmartTransport(BaseTransport):
if child["device_id"] == device_id:
info = child
break
# Create the child_devices fixture section for fixtures generated before it was added
if "child_devices" not in self.info:
self.info["child_devices"] = {}
# Get the method calls made directly on the child devices
child_device_calls = self.info["child_devices"].setdefault(device_id, {})
# We only support get & set device info for now.
if child_method == "get_device_info":
@@ -159,14 +164,27 @@ class FakeSmartTransport(BaseTransport):
return {"error_code": 0}
elif child_method == "set_preset_rules":
return self._set_child_preset_rules(info, child_params)
elif child_method in child_device_calls:
result = copy.deepcopy(child_device_calls[child_method])
return {"result": result, "error_code": 0}
elif (
# FIXTURE_MISSING is for service calls not in place when
# SMART fixtures started to be generated
missing_result := self.FIXTURE_MISSING_MAP.get(child_method)
) and missing_result[0] in self.components:
result = copy.deepcopy(missing_result[1])
# Copy to info so it will work with update methods
child_device_calls[child_method] = copy.deepcopy(missing_result[1])
result = copy.deepcopy(info[child_method])
retval = {"result": result, "error_code": 0}
return retval
elif child_method[:4] == "set_":
target_method = f"get_{child_method[4:]}"
if target_method not in child_device_calls:
raise RuntimeError(
f"No {target_method} in child info, calling set before get not supported."
)
child_device_calls[target_method].update(child_params)
return {"error_code": 0}
else:
# PARAMS error returned for KS240 when get_device_usage called
# on parent device. Could be any error code though.

View File

@@ -9,7 +9,7 @@ from pytest_mock import MockerFixture
from kasa import Module
from kasa.smart import SmartDevice
from kasa.tests.device_fixtures import parametrize
from kasa.tests.device_fixtures import get_parent_and_child_modules, parametrize
autooff = parametrize(
"has autooff", component_filter="auto_off", protocol_filter={"SMART"}
@@ -33,13 +33,13 @@ async def test_autooff_features(
dev: SmartDevice, feature: str, prop_name: str, type: type
):
"""Test that features are registered and work as expected."""
autooff = dev.modules.get(Module.AutoOff)
autooff = next(get_parent_and_child_modules(dev, Module.AutoOff))
assert autooff is not None
prop = getattr(autooff, prop_name)
assert isinstance(prop, type)
feat = dev.features[feature]
feat = autooff._device.features[feature]
assert feat.value == prop
assert isinstance(feat.value, type)
@@ -47,13 +47,13 @@ async def test_autooff_features(
@autooff
async def test_settings(dev: SmartDevice, mocker: MockerFixture):
"""Test autooff settings."""
autooff = dev.modules.get(Module.AutoOff)
autooff = next(get_parent_and_child_modules(dev, Module.AutoOff))
assert autooff
enabled = dev.features["auto_off_enabled"]
enabled = autooff._device.features["auto_off_enabled"]
assert autooff.enabled == enabled.value
delay = dev.features["auto_off_minutes"]
delay = autooff._device.features["auto_off_minutes"]
assert autooff.delay == delay.value
call = mocker.spy(autooff, "call")
@@ -86,10 +86,10 @@ async def test_auto_off_at(
dev: SmartDevice, mocker: MockerFixture, is_timer_active: bool
):
"""Test auto-off at sensor."""
autooff = dev.modules.get(Module.AutoOff)
autooff = next(get_parent_and_child_modules(dev, Module.AutoOff))
assert autooff
autooff_at = dev.features["auto_off_at"]
autooff_at = autooff._device.features["auto_off_at"]
mocker.patch.object(
type(autooff),

View File

@@ -9,7 +9,7 @@ from unittest.mock import patch
import pytest
from pytest_mock import MockerFixture
from kasa import KasaException, Module
from kasa import Device, KasaException, Module
from kasa.exceptions import SmartErrorCode
from kasa.smart import SmartDevice
@@ -112,6 +112,11 @@ async def test_update_module_queries(dev: SmartDevice, mocker: MockerFixture):
device_queries: dict[SmartDevice, dict[str, Any]] = {}
for mod in dev._modules.values():
device_queries.setdefault(mod._device, {}).update(mod.query())
# Hubs do not query child modules by default.
if dev.device_type != Device.Type.Hub:
for child in dev.children:
for mod in child.modules.values():
device_queries.setdefault(mod._device, {}).update(mod.query())
spies = {}
for device in device_queries:
@@ -120,7 +125,8 @@ async def test_update_module_queries(dev: SmartDevice, mocker: MockerFixture):
await dev.update()
for device in device_queries:
if device_queries[device]:
spies[device].assert_called_with(device_queries[device])
# Need assert any here because the child device updates use the parent's protocol
spies[device].assert_any_call(device_queries[device])
else:
spies[device].assert_not_called()