diff --git a/kasa/device_factory.py b/kasa/device_factory.py index ecb0d0a13..c4f977274 100644 --- a/kasa/device_factory.py +++ b/kasa/device_factory.py @@ -104,8 +104,8 @@ def _perf_log(has_params: bool, perf_type: str) -> None: device_class: type[Device] | None device: Device | None = None - if isinstance(protocol, IotProtocol) and isinstance( - protocol._transport, XorTransport + if isinstance(protocol, IotProtocol) and not isinstance( + protocol._transport, LinkieTransportV2 ): info = await protocol.query(GET_SYSINFO_QUERY) _perf_log(True, "get_sysinfo") @@ -228,7 +228,6 @@ def get_protocol(config: DeviceConfig, *, strict: bool = False) -> BaseProtocol str, tuple[type[BaseProtocol], type[BaseTransport]] ] = { "IOT.XOR": (IotProtocol, XorTransport), - "IOT.KLAP": (IotProtocol, KlapTransport), "SMART.AES": (SmartProtocol, AesTransport), "SMART.KLAP": (SmartProtocol, KlapTransportV2), "SMART.KLAP.HTTPS": (SmartProtocol, KlapTransportV2), @@ -236,6 +235,12 @@ def get_protocol(config: DeviceConfig, *, strict: bool = False) -> BaseProtocol # https to distuingish from SmartProtocol devices "SMART.AES.HTTPS": (SmartCamProtocol, SslAesTransport), } + if protocol_transport_key == "IOT.KLAP": + transport_cls: type[BaseTransport] = ( + KlapTransportV2 if ctype.login_version == 2 else KlapTransport + ) + return IotProtocol(transport=transport_cls(config=config)) + if not (prot_tran_cls := supported_device_protocols.get(protocol_transport_key)): return None protocol_cls, transport_cls = prot_tran_cls diff --git a/tests/test_device_factory.py b/tests/test_device_factory.py index 19ccfb73d..164e71c1a 100644 --- a/tests/test_device_factory.py +++ b/tests/test_device_factory.py @@ -259,6 +259,12 @@ async def test_device_class_from_unknown_family(caplog): KlapTransport, id="iot-klap", ), + pytest.param( + CP(DF.IotSmartPlugSwitch, ET.Klap, login_version=2, https=False), + IotProtocol, + KlapTransportV2, + id="iot-klap-lv2", + ), pytest.param( CP(DF.IotSmartPlugSwitch, ET.Xor, https=False), IotProtocol, diff --git a/tests/test_hs300_klap_lv2.py b/tests/test_hs300_klap_lv2.py new file mode 100644 index 000000000..3b7102b73 --- /dev/null +++ b/tests/test_hs300_klap_lv2.py @@ -0,0 +1,103 @@ +from __future__ import annotations + +import json +import os + +import pytest + +from kasa import Credentials, IotProtocol +from kasa.device_factory import connect +from kasa.deviceconfig import ( + DeviceConfig, + DeviceConnectionParameters, + DeviceEncryptionType, + DeviceFamily, +) +from kasa.iot import IotStrip +from kasa.transports import KlapTransportV2 + +from .conftest import load_fixture +from .fakeprotocol_iot import FakeIotTransport + + +def _get_credentials_from_request(request) -> Credentials: + username = request.config.getoption("--username") or os.environ.get("KASA_USERNAME") + password = request.config.getoption("--password") or os.environ.get("KASA_PASSWORD") + + if not username or not password: + pytest.skip("requires --username/--password or KASA_USERNAME/KASA_PASSWORD") + + return Credentials(username=username, password=password) + + +@pytest.mark.requires_dummy +async def test_hs300_iot_klap_lv2_connect_uses_strip_fixture(mocker): + """Verify direct connect uses KLAP v2 and creates an IotStrip for HS300.""" + fixture_data = FakeIotTransport._build_fake_proto( + json.loads(load_fixture("iot", "HS300(US)_2.0_1.0.12.json")) + ) + config = DeviceConfig( + host="127.0.0.123", + credentials=Credentials("dummy_user", "dummy_password"), + connection_type=DeviceConnectionParameters( + device_family=DeviceFamily.IotSmartPlugSwitch, + encryption_type=DeviceEncryptionType.Klap, + login_version=2, + http_port=80, + ), + ) + + async def _query(self, *_args, **_kwargs): + assert isinstance(self, IotProtocol) + assert isinstance(self._transport, KlapTransportV2) + return fixture_data + + async def _update(self, *_args, **_kwargs): + return None + + mocker.patch("kasa.IotProtocol.query", new=_query) + mocker.patch.object(IotStrip, "update", new=_update) + + dev = await connect(config=config) + try: + assert isinstance(dev, IotStrip) + assert isinstance(dev.protocol, IotProtocol) + assert isinstance(dev.protocol._transport, KlapTransportV2) + assert dev.model == "HS300" + assert dev.port == 80 + assert dev.sys_info["child_num"] == 6 + assert dev.config.connection_type.login_version == 2 + finally: + await dev.disconnect() + + +async def test_hs300_iot_klap_lv2_direct_connect_real_device(request): + """Verify the explicit HS300 KLAP lv2 config reconnects correctly.""" + ip = request.config.getoption("--ip") + if not ip: + pytest.skip("requires --ip to run against a real device") + + credentials = _get_credentials_from_request(request) + config = DeviceConfig( + host=ip, + credentials=credentials, + timeout=10, + connection_type=DeviceConnectionParameters( + device_family=DeviceFamily.IotSmartPlugSwitch, + encryption_type=DeviceEncryptionType.Klap, + login_version=2, + http_port=80, + ), + ) + + dev = await connect(config=config) + try: + assert isinstance(dev, IotStrip) + assert isinstance(dev.protocol, IotProtocol) + assert isinstance(dev.protocol._transport, KlapTransportV2) + assert dev.model == "HS300" + assert dev.port == 80 + assert len(dev.children) == 6 + assert dev.config.connection_type.login_version == 2 + finally: + await dev.disconnect()