mirror of
https://github.com/python-kasa/python-kasa.git
synced 2026-10-26 22:52:55 +00:00
Add Device.supports_module() and stop querying HomeKit for iot devices (#1808)
Some legacy iot firmware stops responding when an unsupported module follows `system` in a request, which made every update time out once the HomeKit query was added. This stops querying HomeKit in `update()` for IOT devices and adds `Device.supports_module()` API to allow checking if some module is supported by the device.
This commit is contained in:
@@ -8,13 +8,23 @@ from kasa.iot.modules.homekit import HomeKit
|
||||
|
||||
from ...device_fixtures import device_iot
|
||||
|
||||
HOMEKIT = "smartlife.iot.homekit"
|
||||
|
||||
|
||||
async def _homekit_module(dev: IotDevice) -> HomeKit:
|
||||
"""Create the module with the data from the device, skip if not supported."""
|
||||
if not await dev.supports_module(Module.IotHomeKit):
|
||||
pytest.skip("HomeKit module not present on this device")
|
||||
homekit = HomeKit(dev, HOMEKIT)
|
||||
dev._last_update.update(await dev.protocol.query(homekit.query()))
|
||||
homekit._initialize_features()
|
||||
return homekit
|
||||
|
||||
|
||||
@device_iot
|
||||
def test_homekit_getters(dev: IotDevice) -> None:
|
||||
async def test_homekit_getters(dev: IotDevice) -> None:
|
||||
# HomeKit can be present on any IOT device
|
||||
if Module.IotHomeKit not in dev.modules:
|
||||
pytest.skip("HomeKit module not present on this device")
|
||||
homekit: HomeKit = dev.modules[Module.IotHomeKit]
|
||||
homekit = await _homekit_module(dev)
|
||||
info = homekit.info
|
||||
if not info:
|
||||
pytest.skip("No HomeKit data present for this fixture")
|
||||
@@ -31,10 +41,8 @@ def test_homekit_getters(dev: IotDevice) -> None:
|
||||
|
||||
|
||||
@device_iot
|
||||
def test_homekit_feature(dev: IotDevice) -> None:
|
||||
if Module.IotHomeKit not in dev.modules:
|
||||
pytest.skip("HomeKit module not present on this device")
|
||||
homekit: HomeKit = dev.modules[Module.IotHomeKit]
|
||||
async def test_homekit_feature(dev: IotDevice) -> None:
|
||||
homekit = await _homekit_module(dev)
|
||||
if not homekit.info:
|
||||
pytest.skip("No HomeKit data present for this device")
|
||||
feature = homekit._all_features.get("homekit_setup_code")
|
||||
@@ -46,9 +54,7 @@ def test_homekit_feature(dev: IotDevice) -> None:
|
||||
|
||||
@device_iot
|
||||
def test_initialize_features_skips_when_no_data(dev: IotDevice) -> None:
|
||||
if Module.IotHomeKit not in dev.modules:
|
||||
pytest.skip("HomeKit module not present on this device")
|
||||
homekit: HomeKit = dev.modules[Module.IotHomeKit]
|
||||
homekit = HomeKit(dev, HOMEKIT)
|
||||
if "homekit_setup_code" in homekit._all_features:
|
||||
pytest.skip("HomeKit feature already present on this device")
|
||||
# Patch .data so it looks like no homekit data is present
|
||||
@@ -57,3 +63,4 @@ def test_initialize_features_skips_when_no_data(dev: IotDevice) -> None:
|
||||
homekit._initialize_features()
|
||||
# Since there was no data, no features should be added
|
||||
assert "homekit_setup_code" not in homekit._all_features
|
||||
assert homekit.info == {}
|
||||
|
||||
@@ -18,12 +18,15 @@ from voluptuous import (
|
||||
)
|
||||
|
||||
from kasa import DeviceType, KasaException, Module
|
||||
from kasa.exceptions import AuthenticationError, TimeoutError
|
||||
from kasa.iot import IotDevice
|
||||
from kasa.iot.iotmodule import _merge_dict
|
||||
from tests.conftest import get_device_for_fixture_protocol, handle_turn_on, turn_on
|
||||
from tests.device_fixtures import device_iot, has_emeter_iot, no_emeter_iot
|
||||
from tests.fakeprotocol_iot import FakeIotProtocol
|
||||
|
||||
HOMEKIT = "smartlife.iot.homekit"
|
||||
|
||||
TZ_SCHEMA = Schema(
|
||||
{"zone_str": str, "dst_offset": int, "index": All(int, Range(min=0)), "tz_str": str}
|
||||
)
|
||||
@@ -310,3 +313,66 @@ def test_merge_dict() -> None:
|
||||
"get_daystat": {"month": 8, "year": 2024},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.requires_dummy
|
||||
@device_iot
|
||||
async def test_update_does_not_query_homekit(dev, mocker: MockerFixture) -> None:
|
||||
spy = mocker.spy(dev.protocol, "query")
|
||||
await dev.update()
|
||||
assert not [call for call in spy.call_args_list if HOMEKIT in call.args[0]]
|
||||
|
||||
|
||||
@pytest.mark.requires_dummy
|
||||
@pytest.mark.parametrize(
|
||||
("fixture", "supported"),
|
||||
[("EP25(US)_1.0_1.0.14.json", True), ("HS100(US)_2.0_1.5.6.json", False)],
|
||||
)
|
||||
async def test_supports_module(
|
||||
mocker: MockerFixture, fixture: str, supported: bool
|
||||
) -> None:
|
||||
dev = await get_device_for_fixture_protocol(fixture, "IOT")
|
||||
spy = mocker.spy(dev.protocol, "query")
|
||||
|
||||
assert await dev.supports_module(Module.IotHomeKit) is supported
|
||||
|
||||
# The answer is remembered, so the second call does not query the device
|
||||
assert await dev.supports_module(Module.IotHomeKit) is supported
|
||||
assert spy.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.requires_dummy
|
||||
async def test_timeout_is_not_supported_and_not_remembered(
|
||||
mocker: MockerFixture,
|
||||
) -> None:
|
||||
dev = await get_device_for_fixture_protocol("EP25(US)_1.0_1.0.14.json", "IOT")
|
||||
query = dev.protocol.query
|
||||
|
||||
mocker.patch.object(dev.protocol, "query", side_effect=TimeoutError("timeout"))
|
||||
assert await dev.supports_module(Module.IotHomeKit) is False
|
||||
|
||||
mocker.patch.object(dev.protocol, "query", side_effect=query)
|
||||
assert await dev.supports_module(Module.IotHomeKit) is True
|
||||
|
||||
|
||||
@pytest.mark.requires_dummy
|
||||
async def test_unsupported_answer_is_remembered(mocker: MockerFixture) -> None:
|
||||
dev = await get_device_for_fixture_protocol("HS100(US)_2.0_1.5.6.json", "IOT")
|
||||
query = mocker.patch.object(
|
||||
dev.protocol,
|
||||
"query",
|
||||
return_value={HOMEKIT: {"err_code": -1, "err_msg": "module not support"}},
|
||||
)
|
||||
|
||||
assert await dev.supports_module(Module.IotHomeKit) is False
|
||||
assert await dev.supports_module(Module.IotHomeKit) is False
|
||||
assert query.call_count == 1
|
||||
|
||||
|
||||
@pytest.mark.requires_dummy
|
||||
async def test_other_errors_propagate(mocker: MockerFixture) -> None:
|
||||
dev = await get_device_for_fixture_protocol("EP25(US)_1.0_1.0.14.json", "IOT")
|
||||
mocker.patch.object(dev.protocol, "query", side_effect=AuthenticationError("auth"))
|
||||
|
||||
with pytest.raises(AuthenticationError):
|
||||
await dev.supports_module(Module.IotHomeKit)
|
||||
|
||||
Reference in New Issue
Block a user