diff --git a/README.md b/README.md index e571461d3..33d67586c 100644 --- a/README.md +++ b/README.md @@ -187,7 +187,7 @@ The following devices have been tested and confirmed as working. If your device ### Supported Kasa devices - **Plugs**: EP10, EP25[^2], HS100[^2], HS103, HS105, HS110, KP100, KP105, KP115, KP125, KP125M[^1], KP401 -- **Power Strips**: EP40, EP40M[^1], HS107, HS300, KP200, KP303, KP400 +- **Power Strips**: EP40, EP40M[^1], HS107, HS300[^2], KP200, KP303, KP400 - **Wall Switches**: ES20M, HS200[^2], HS210, HS220[^2], KP405, KS200, KS200M, KS205[^1], KS220, KS220M, KS225[^1], KS230, KS240[^1] - **Bulbs**: KL110, KL110B, KL120, KL125, KL130, KL135, KL50, KL60, LB100, LB110, LB130 - **Light Strips**: KL400L10, KL400L5, KL420L5, KL430 diff --git a/SUPPORTED.md b/SUPPORTED.md index 1d38ee0e7..d0c96646a 100644 --- a/SUPPORTED.md +++ b/SUPPORTED.md @@ -68,6 +68,7 @@ Some newer Kasa devices require authentication. These are marked with [^1] in th - Hardware: 1.0 (US) / Firmware: 1.0.21 - Hardware: 2.0 (US) / Firmware: 1.0.12 - Hardware: 2.0 (US) / Firmware: 1.0.3 + - Hardware: 2.0 (US) / Firmware: 1.1.2[^1] - **KP200** - Hardware: 3.0 (US) / Firmware: 1.0.3 - **KP303** diff --git a/devtools/dump_devinfo.py b/devtools/dump_devinfo.py index fff57370f..10dd2cd39 100644 --- a/devtools/dump_devinfo.py +++ b/devtools/dump_devinfo.py @@ -303,6 +303,7 @@ def capture_raw(discovered: DiscoveredRaw): login_version=dr.mgt_encrypt_schm.lv, https=dr.mgt_encrypt_schm.is_support_https, http_port=dr.mgt_encrypt_schm.http_port, + new_klap=dr.mgt_encrypt_schm.new_klap, ) dc = DeviceConfig( host=host, @@ -320,11 +321,16 @@ def capture_raw(discovered: DiscoveredRaw): batch_size=batch_size, ) elif device_family and encrypt_type: + new_klap = None + if encrypt_type == DeviceEncryptionType.Klapv2.value: + encrypt_type = DeviceEncryptionType.Klap.value + new_klap = True ctype = DeviceConnectionParameters( DeviceFamily(device_family), DeviceEncryptionType(encrypt_type), login_version, https, + new_klap=new_klap, ) config = DeviceConfig( host=host, diff --git a/kasa/cli/discover.py b/kasa/cli/discover.py index af367e32b..82eee7eab 100644 --- a/kasa/cli/discover.py +++ b/kasa/cli/discover.py @@ -15,6 +15,7 @@ Discover, UnsupportedDeviceError, ) +from kasa.deviceconfig import DeviceConnectionParameters, DeviceEncryptionType from kasa.discover import ( NEW_DISCOVERY_REDACTORS, ConnectAttempt, @@ -59,6 +60,13 @@ def _discover_is_root_cmd(ctx: click.Context) -> bool: ) +def _get_cli_encrypt_type(cparams: DeviceConnectionParameters) -> str: + """Return the encryption type value needed to recreate the connection.""" + if cparams.new_klap: + return DeviceEncryptionType.Klapv2.value + return cparams.encryption_type.value + + @discover.command() @click.pass_context async def detail(ctx: click.Context) -> DeviceDict: @@ -275,10 +283,11 @@ def on_attempt(connect_attempt: ConnectAttempt, success: bool) -> None: ) if dev: cparams = dev.config.connection_type + encrypt_type = _get_cli_encrypt_type(cparams) echo("Managed to connect, cli options to connect are:") echo( f"--device-family {cparams.device_family.value} " - f"--encrypt-type {cparams.encryption_type.value} " + f"--encrypt-type {encrypt_type} " f"{'--https' if cparams.https else '--no-https'}" ) return {host: dev} diff --git a/kasa/cli/main.py b/kasa/cli/main.py index 3e32c1f45..36a3ddf74 100755 --- a/kasa/cli/main.py +++ b/kasa/cli/main.py @@ -325,11 +325,17 @@ async def cli( if not encrypt_type: encrypt_type = "KLAP" + new_klap = None + if encrypt_type and encrypt_type == DeviceEncryptionType.Klapv2.value: + encrypt_type = DeviceEncryptionType.Klap.value + new_klap = True + ctype = DeviceConnectionParameters( DeviceFamily(device_family), DeviceEncryptionType(encrypt_type), login_version, https, + new_klap=new_klap, ) config = DeviceConfig( host=host, diff --git a/kasa/device_factory.py b/kasa/device_factory.py index ecb0d0a13..8170f2aa9 100644 --- a/kasa/device_factory.py +++ b/kasa/device_factory.py @@ -104,21 +104,20 @@ 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 - ): - info = await protocol.query(GET_SYSINFO_QUERY) - _perf_log(True, "get_sysinfo") - device_class = get_device_class_from_sys_info(info) - device = device_class(config.host, protocol=protocol) - device.update_from_discover_info(info) - await device.update() - _perf_log(True, "update") - return device - elif device_class := get_device_class_from_family( + if device_class := get_device_class_from_family( config.connection_type.device_family.value, https=config.connection_type.https ): - device = device_class(host=config.host, protocol=protocol) + if issubclass(device_class, IotDevice): + # IoT devices report a generic device family (e.g. IOT.SMARTPLUGSWITCH) + # that does not distinguish plug/strip/switch/etc. Query sysinfo for + # the precise subclass regardless of which transport is in use. + info = await protocol.query(GET_SYSINFO_QUERY) + _perf_log(True, "get_sysinfo") + device_class = get_device_class_from_sys_info(info) + device = device_class(config.host, protocol=protocol) + device.update_from_discover_info(info) + else: + device = device_class(host=config.host, protocol=protocol) await device.update() _perf_log(True, "update") return device @@ -146,7 +145,10 @@ def get_device_class_from_sys_info(sysinfo: dict[str, Any]) -> type[IotDevice]: def get_device_class_from_family( - device_type: str, *, https: bool, require_exact: bool = False + device_type: str, + *, + https: bool, + require_exact: bool = False, ) -> type[Device] | None: """Return the device class from the type name.""" supported_device_types: dict[str, type[Device]] = { @@ -221,6 +223,7 @@ def get_protocol(config: DeviceConfig, *, strict: bool = False) -> BaseProtocol + "." + ctype.encryption_type.value + (".HTTPS" if ctype.https else "") + + (".NEW_KLAP" if ctype.new_klap else "") ) _LOGGER.debug("Finding transport for %s", protocol_transport_key) @@ -229,6 +232,7 @@ def get_protocol(config: DeviceConfig, *, strict: bool = False) -> BaseProtocol ] = { "IOT.XOR": (IotProtocol, XorTransport), "IOT.KLAP": (IotProtocol, KlapTransport), + "IOT.KLAP.NEW_KLAP": (IotProtocol, KlapTransportV2), "SMART.AES": (SmartProtocol, AesTransport), "SMART.KLAP": (SmartProtocol, KlapTransportV2), "SMART.KLAP.HTTPS": (SmartProtocol, KlapTransportV2), diff --git a/kasa/deviceconfig.py b/kasa/deviceconfig.py index ff0fbf8fe..816021592 100644 --- a/kasa/deviceconfig.py +++ b/kasa/deviceconfig.py @@ -62,6 +62,7 @@ class DeviceEncryptionType(Enum): Klap = "KLAP" Aes = "AES" Xor = "XOR" + Klapv2 = "KLAPV2" class DeviceFamily(Enum): @@ -101,6 +102,7 @@ class DeviceConnectionParameters(_DeviceConfigBaseMixin): login_version: int | None = None https: bool = False http_port: int | None = None + new_klap: bool | None = None @staticmethod def from_values( @@ -110,6 +112,7 @@ def from_values( login_version: int | None = None, https: bool | None = None, http_port: int | None = None, + new_klap: bool | None = None, ) -> DeviceConnectionParameters: """Return connection parameters from string values.""" try: @@ -121,6 +124,7 @@ def from_values( login_version, https, http_port=http_port, + new_klap=new_klap, ) except (ValueError, TypeError) as ex: raise KasaException( diff --git a/kasa/discover.py b/kasa/discover.py index 911c005c9..eee353770 100755 --- a/kasa/discover.py +++ b/kasa/discover.py @@ -111,6 +111,7 @@ from kasa import Device from kasa.credentials import Credentials from kasa.device_factory import ( + GET_SYSINFO_QUERY, get_device_class_from_family, get_device_class_from_sys_info, get_protocol, @@ -125,7 +126,7 @@ TimeoutError, UnsupportedDeviceError, ) -from kasa.iot.iotdevice import IotDevice, _extract_sys_info +from kasa.iot.iotdevice import _extract_sys_info from kasa.json import DataClassJSONMixin from kasa.json import dumps as json_dumps from kasa.json import loads as json_loads @@ -361,22 +362,33 @@ def datagram_received( config.timeout = self.timeout try: if port == self.discovery_port: - json_func = Discover._get_discovery_json_legacy - device_func = Discover._get_device_instance_legacy + info = Discover._get_discovery_json_legacy(data, ip) + if self.on_discovered_raw is not None: + self.on_discovered_raw( + { + "discovery_response": info, + "meta": {"ip": ip, "port": port}, + } + ) + device = Discover._get_device_instance_legacy(info, config) elif port in (Discover.DISCOVERY_PORT_2, Discover.DISCOVERY_PORT_3): - json_func = Discover._get_discovery_json - device_func = Discover._get_device_instance + info = Discover._get_discovery_json(data, ip) + if self.on_discovered_raw is not None: + self.on_discovered_raw( + { + "discovery_response": info, + "meta": {"ip": ip, "port": port}, + } + ) + # IoT devices with new_klap firmware use KlapTransportV2 and need + # an async sysinfo query to determine the correct device subclass; + # spawn a callback task so the datagram handler is not blocked. + if self._is_new_klap_iot(info): + self._run_callback_task(self._process_new_klap_device(info, config)) + return + device = Discover._get_device_instance(info, config) else: return - info = json_func(data, ip) - if self.on_discovered_raw is not None: - self.on_discovered_raw( - { - "discovery_response": info, - "meta": {"ip": ip, "port": port}, - } - ) - device = device_func(info, config) except UnsupportedDeviceError as udex: _LOGGER.debug("Unsupported device found at %s << %s", ip, udex) self.unsupported_device_exceptions[ip] = udex @@ -404,6 +416,77 @@ def _handle_discovered_event(self) -> None: if self.discover_task: self.discover_task.cancel() + @staticmethod + def _is_new_klap_iot(info: dict) -> bool: + """Return True if the port 20002 response is from an IoT new_klap device. + + IoT devices with new_klap firmware advertise KLAP encryption but use + KlapTransportV2. The discovery response alone does not carry enough + structural information to determine the precise IoT subclass, so an + async sysinfo query is required after connection. + """ + result = info.get("result", {}) + schm = result.get("mgt_encrypt_schm", {}) + return bool(schm.get("new_klap")) and result.get("device_type", "").startswith( + "IOT." + ) + + async def _process_new_klap_device(self, info: dict, config: DeviceConfig) -> None: + """Process an IoT new_klap device discovered via port 20002. + + These devices use KlapTransportV2 but require a sysinfo query to + determine the correct IoT subclass (IotStrip, IotBulb, etc.) because + the 20002 discovery response only carries the generic device family. + """ + ip = config.host + try: + discovery_result = DiscoveryResult.from_dict(info["result"]) + conn_params = Discover._get_connection_parameters(discovery_result) + config.connection_type = conn_params + + if (protocol := get_protocol(config)) is None: + raise UnsupportedDeviceError( + f"Unsupported encryption scheme {ip}: " + + f"{config.connection_type.to_dict()}: {info}", + discovery_result=discovery_result.to_dict(), + host=ip, + ) + + sysinfo = await protocol.query(GET_SYSINFO_QUERY) + device_class = get_device_class_from_sys_info(sysinfo) + device: Device = device_class(ip, protocol=protocol) + + di = discovery_result.to_dict() + di["model"], _, _ = discovery_result.device_model.partition("(") + device.update_from_discover_info(di) + + if _LOGGER.isEnabledFor(logging.DEBUG): + data = ( + redact_data(info, NEW_DISCOVERY_REDACTORS) + if Discover._redact_data + else info + ) + _LOGGER.debug("[DISCOVERY] %s << %s", ip, pf(data)) + except UnsupportedDeviceError as udex: + _LOGGER.debug("Unsupported device found at %s << %s", ip, udex) + self.unsupported_device_exceptions[ip] = udex + if self.on_unsupported is not None: + await self.on_unsupported(udex) + self._handle_discovered_event() + return + except KasaException as ex: + _LOGGER.debug("[DISCOVERY] Unable to find device type for %s: %s", ip, ex) + self.invalid_device_exceptions[ip] = ex + self._handle_discovered_event() + return + + self.discovered_devices[ip] = device + + if self.on_discovered is not None: + await self.on_discovered(device) + + self._handle_discovered_event() + def error_received(self, ex: Exception) -> None: """Handle asyncio.Protocol errors.""" _LOGGER.error("Got error: %s", ex) @@ -653,12 +736,14 @@ async def try_connect_all( for device_family in main_device_families for https in (True, False) for login_version in (None, 2) + for new_klap in (True, None) if ( conn_params := DeviceConnectionParameters( device_family=device_family, encryption_type=encrypt, login_version=login_version, https=https, + new_klap=new_klap, ) ) and ( @@ -713,7 +798,8 @@ def _get_device_class(info: dict) -> type[Device]: else False ) dev_class = get_device_class_from_family( - discovery_result.device_type, https=https + discovery_result.device_type, + https=https, ) if not dev_class: raise UnsupportedDeviceError( @@ -742,7 +828,7 @@ def _get_device_instance_legacy(info: dict, config: DeviceConfig) -> Device: data = redact_data(info, IOT_REDACTORS) if Discover._redact_data else info _LOGGER.debug("[DISCOVERY] %s << %s", config.host, pf(data)) - device_class = cast(type[IotDevice], Discover._get_device_class(info)) + device_class = cast(type[Device], Discover._get_device_class(info)) device = device_class(config.host, config=config) sys_info = _extract_sys_info(info) device_type = sys_info.get("mic_type", sys_info.get("type")) @@ -844,6 +930,7 @@ def _get_connection_parameters( login_version=login_version, https=encrypt_schm.is_support_https, http_port=encrypt_schm.http_port, + new_klap=encrypt_schm.new_klap, ) @staticmethod @@ -851,7 +938,7 @@ def _get_device_instance( info: dict, config: DeviceConfig, ) -> Device: - """Get SmartDevice from the new 20002 response.""" + """Get Device for the port 20002 response.""" debug_enabled = _LOGGER.isEnabledFor(logging.DEBUG) try: @@ -899,16 +986,6 @@ def _get_device_instance( host=config.host, ) from ex - if ( - device_class := get_device_class_from_family(type_, https=conn_params.https) - ) is None: - _LOGGER.debug("Got unsupported device type: %s", type_) - raise UnsupportedDeviceError( - f"Unsupported device {config.host} of type {type_}: {info}", - discovery_result=discovery_result.to_dict(), - host=config.host, - ) - if (protocol := get_protocol(config)) is None: _LOGGER.debug( "Got unsupported connection type: %s", config.connection_type.to_dict() @@ -920,6 +997,19 @@ def _get_device_instance( host=config.host, ) + if ( + device_class := cast( + type[Device], + Discover._get_device_class(info), + ) + ) is None: + _LOGGER.debug("Got unsupported device type: %s", type_) + raise UnsupportedDeviceError( + f"Unsupported device {config.host} of type {type_}: {info}", + discovery_result=discovery_result.to_dict(), + host=config.host, + ) + if debug_enabled: data = ( redact_data(info, NEW_DISCOVERY_REDACTORS) @@ -955,6 +1045,7 @@ class EncryptionScheme(_DiscoveryBaseMixin): encrypt_type: str | None = None http_port: int | None = None lv: int | None = None + new_klap: bool | None = None @dataclass diff --git a/tests/device_fixtures.py b/tests/device_fixtures.py index e8d131dec..26796db15 100644 --- a/tests/device_fixtures.py +++ b/tests/device_fixtures.py @@ -66,7 +66,6 @@ BULBS_VARIABLE_TEMP = {*BULBS_SMART_VARIABLE_TEMP, *BULBS_IOT_VARIABLE_TEMP} BULBS_COLOR = {*BULBS_SMART_COLOR, *BULBS_IOT_COLOR} - LIGHT_STRIPS = {*BULBS_SMART_LIGHT_STRIP, *BULBS_IOT_LIGHT_STRIP} BULBS = { *BULBS_IOT, diff --git a/tests/discovery_fixtures.py b/tests/discovery_fixtures.py index 3cf726f48..be0ff6d49 100644 --- a/tests/discovery_fixtures.py +++ b/tests/discovery_fixtures.py @@ -8,6 +8,7 @@ from typing import Any, TypedDict import pytest +from pytest_mock import MockerFixture from kasa.transports.xortransport import XorEncryption @@ -48,8 +49,8 @@ class DiscoveryResponse(TypedDict): def _make_unsupported( - device_family, - encrypt_type, + device_family: str, + encrypt_type: str, *, https: bool = False, omit_keys: dict[str, Any] | None = None, @@ -112,8 +113,12 @@ def _make_unsupported( def parametrize_discovery( - desc, *, data_root_filter=None, protocol_filter=None, model_filter=None -): + desc: str, + *, + data_root_filter: str | None = None, + protocol_filter: set[str] | None = None, + model_filter: set[str] | None = None, +) -> pytest.MarkDecorator: filtered_fixtures = filter_fixtures( desc, data_root_filter=data_root_filter, @@ -141,14 +146,14 @@ def parametrize_discovery( ), ids=idgenerator, ) -async def discovery_mock(request, mocker): +async def discovery_mock(request: pytest.FixtureRequest, mocker: MockerFixture) -> Any: """Mock discovery and patch protocol queries to use Fake protocols.""" fi: FixtureInfo = request.param fixture_info = FixtureInfo(fi.name, fi.protocol, copy.deepcopy(fi.data)) return patch_discovery({DISCOVERY_MOCK_IP: fixture_info}, mocker) -def create_discovery_mock(ip: str, fixture_data: dict): +def create_discovery_mock(ip: str, fixture_data: dict[str, Any]) -> Any: """Mock discovery and patch protocol queries to use Fake protocols.""" @dataclass @@ -164,6 +169,7 @@ class _DiscoveryMock: login_version: int | None = None port_override: int | None = None http_port: int | None = None + new_klap: bool | None = None @property def model(self) -> str: @@ -205,6 +211,7 @@ def _datagram(self) -> bytes: default_port = 443 if https else 80 else: default_port = http_port + new_klap = discovery_result["mgt_encrypt_schm"].get("new_klap", None) dm = _DiscoveryMock( ip, default_port, @@ -216,6 +223,7 @@ def _datagram(self) -> bytes: https, login_version, http_port=http_port, + new_klap=new_klap, ) else: sys_info = fixture_data["system"]["get_sysinfo"] @@ -233,12 +241,15 @@ def _datagram(self) -> bytes: encrypt_type, False, login_version, + new_klap=None, ) return dm -def patch_discovery(fixture_infos: dict[str, FixtureInfo], mocker): +def patch_discovery( + fixture_infos: dict[str, FixtureInfo], mocker: MockerFixture +) -> Any: """Mock discovery and patch protocol queries to use Fake protocols.""" discovery_mocks = { ip: create_discovery_mock(ip, fixture_info.data) @@ -271,7 +282,7 @@ async def process_callback_queue(finished_event: asyncio.Event) -> None: await exception_queue.put(None) callback_queue.task_done() - async def wait_for_coro(): + async def wait_for_coro() -> None: await callback_queue.join() if ex := exception_queue.get_nowait(): raise ex @@ -286,7 +297,7 @@ def _run_callback_task(self, coro: Coroutine) -> None: ) # do_discover_mock - async def mock_discover(self): + async def mock_discover(self: Any) -> None: """Call datagram_received for all mock fixtures. Handles test cases modifying the ip and hostname of the first fixture @@ -312,6 +323,12 @@ async def mock_discover(self): if fixture_info.protocol in {"SMARTCAM", "SMARTCAM.CHILD"} else FakeIotProtocol(fixture_info.data, fixture_info.name) ) + # Also register under the device IP when the host key is a hostname + # (e.g. discover_single with a hostname). _process_new_klap_device + # creates a protocol with config.host set to the resolved IP, so + # the _query mock must be able to look it up by IP as well. + if host != dm.ip: + protos[dm.ip] = protos[host] port = ( dm.port_override if dm.port_override and dm.discovery_port != 20002 @@ -327,13 +344,15 @@ async def mock_discover(self): mocker.patch("kasa.discover._DiscoverProtocol.do_discover", mock_discover) # query_mock - async def _query(self, request, retry_count: int = 3): + async def _query( + self: Any, request: dict[str, Any], retry_count: int = 3 + ) -> dict[str, Any]: return await protos[self._host].query(request) mocker.patch("kasa.IotProtocol.query", _query) mocker.patch("kasa.SmartProtocol.query", _query) - def _getaddrinfo(host, *_, **__): + def _getaddrinfo(host: str, *_: Any, **__: Any) -> list[tuple[Any, ...]]: nonlocal first_host, first_ip first_host = host # Store the hostname used by discover single first_ip = list(discovery_mocks.values())[ @@ -358,7 +377,9 @@ def _getaddrinfo(host, *_, **__): ), ids=idgenerator, ) -def discovery_data(request, mocker): +def discovery_data( + request: pytest.FixtureRequest, mocker: MockerFixture +) -> dict[str, Any] | DiscoveryResponse: """Return raw discovery file contents as JSON. Used for discovery tests.""" fixture_info = request.param fixture_data = copy.deepcopy(fixture_info.data) @@ -383,12 +404,14 @@ def discovery_data(request, mocker): @pytest.fixture( params=UNSUPPORTED_DEVICES.values(), ids=list(UNSUPPORTED_DEVICES.keys()) ) -def unsupported_device_info(request, mocker): +def unsupported_device_info( + request: pytest.FixtureRequest, mocker: MockerFixture +) -> DiscoveryResponse: """Return unsupported devices for cli and discovery tests.""" discovery_data = request.param host = "127.0.0.1" - async def mock_discover(self): + async def mock_discover(self: Any) -> None: if discovery_data: data = ( b"\x02\x00\x00\x01\x01[\x00\x00\x00\x00\x00\x00W\xcev\xf8" diff --git a/tests/fixtures/iot/HS300(US)_2.0_1.1.2.json b/tests/fixtures/iot/HS300(US)_2.0_1.1.2.json new file mode 100644 index 000000000..0d1c68df2 --- /dev/null +++ b/tests/fixtures/iot/HS300(US)_2.0_1.1.2.json @@ -0,0 +1,146 @@ +{ + "cnCloud": { + "get_info": { + "binded": 1, + "cld_connection": 1, + "err_code": 0, + "fwDlPage": "", + "fwNotifyType": -1, + "illegalType": 0, + "server": "n-devs.tplinkcloud.com", + "stopConnect": 0, + "tcspInfo": "", + "tcspStatus": 1, + "username": "user@example.com" + }, + "get_intl_fw_list": { + "err_code": 0, + "fw_list": [] + } + }, + "discovery_result": { + "error_code": 0, + "result": { + "device_id": "00000000000000000000000000000000", + "device_model": "HS300(US)", + "device_type": "IOT.SMARTPLUGSWITCH", + "factory_default": false, + "hw_ver": "2.0", + "ip": "127.0.0.123", + "mac": "F0-A7-31-00-00-00", + "mgt_encrypt_schm": { + "ANS": false, + "encrypt_type": "KLAP", + "http_port": 80, + "is_support_https": false, + "lv": 2, + "new_klap": 1 + }, + "obd_src": "tplink", + "owner": "00000000000000000000000000000000", + "protocol_version": 1 + } + }, + "emeter": { + "get_realtime": { + "current_ma": 119, + "err_code": 0, + "power_mw": 9021, + "slot_id": 0, + "total_wh": 4013, + "voltage_mv": 121937 + } + }, + "schedule": { + "get_next_action": { + "err_code": 0, + "type": -1 + }, + "get_rules": { + "enable": 1, + "err_code": 0, + "rule_list": [], + "version": 2 + } + }, + "system": { + "get_sysinfo": { + "alias": "#MASKED_NAME#", + "child_num": 6, + "children": [ + { + "alias": "#MASKED_NAME# 1", + "id": "SCRUBBED_CHILD_DEVICE_ID_1", + "next_action": { + "type": -1 + }, + "on_time": 1760, + "state": 1 + }, + { + "alias": "#MASKED_NAME# 2", + "id": "SCRUBBED_CHILD_DEVICE_ID_2", + "next_action": { + "type": -1 + }, + "on_time": 1759, + "state": 1 + }, + { + "alias": "#MASKED_NAME# 3", + "id": "SCRUBBED_CHILD_DEVICE_ID_3", + "next_action": { + "type": -1 + }, + "on_time": 1758, + "state": 1 + }, + { + "alias": "#MASKED_NAME# 4", + "id": "SCRUBBED_CHILD_DEVICE_ID_4", + "next_action": { + "type": -1 + }, + "on_time": 1758, + "state": 1 + }, + { + "alias": "#MASKED_NAME# 5", + "id": "SCRUBBED_CHILD_DEVICE_ID_5", + "next_action": { + "type": -1 + }, + "on_time": 0, + "state": 0 + }, + { + "alias": "#MASKED_NAME# 6", + "id": "SCRUBBED_CHILD_DEVICE_ID_6", + "next_action": { + "type": -1 + }, + "on_time": 0, + "state": 0 + } + ], + "deviceId": "0000000000000000000000000000000000000000", + "err_code": 0, + "feature": "TIM:ENE", + "hwId": "00000000000000000000000000000000", + "hw_ver": "2.0", + "latitude_i": 0, + "led_off": 0, + "lnk_on": 0, + "longitude_i": 0, + "mac": "F0:A7:31:00:00:00", + "mic_type": "IOT.SMARTPLUGSWITCH", + "model": "HS300(US)", + "obd_src": "tplink", + "oemId": "00000000000000000000000000000000", + "rssi": -45, + "status": "new", + "sw_ver": "1.1.2 Build 241220 Rel.171333", + "updating": 0 + } + } +} diff --git a/tests/fixtures/serialization/deviceconfig_plug-new-klap.json b/tests/fixtures/serialization/deviceconfig_plug-new-klap.json new file mode 100644 index 000000000..b7a9a596f --- /dev/null +++ b/tests/fixtures/serialization/deviceconfig_plug-new-klap.json @@ -0,0 +1,11 @@ +{ + "connection_type": { + "device_family": "IOT.SMARTPLUGSWITCH", + "encryption_type": "KLAP", + "https": false, + "login_version": 2, + "new_klap": true + }, + "host": "127.0.0.1", + "timeout": 5 +} diff --git a/tests/test_cli.py b/tests/test_cli.py index dd68031d1..92eba7c92 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -42,6 +42,8 @@ from kasa.cli.time import time from kasa.cli.usage import energy from kasa.cli.wifi import wifi +from kasa.device_factory import get_device_class_from_sys_info +from kasa.deviceconfig import DeviceEncryptionType from kasa.discover import Discover, DiscoveryResult, redact_data from kasa.iot import IotDevice from kasa.json import dumps as json_dumps @@ -75,6 +77,7 @@ async def test_help(runner): [ pytest.param(None, None, id="No connect params"), pytest.param("SMART.TAPOPLUG", None, id="Only device_family"), + pytest.param("SMART.TAPOPLUG", "KLAPV2", id="KLAPV2 encrypt type"), ], ) async def test_update_called_by_cli(dev, mocker, runner, device_family, encrypt_type): @@ -86,6 +89,13 @@ async def test_update_called_by_cli(dev, mocker, runner, device_family, encrypt_ mocker.patch("kasa.iot.iotdevice.IotDevice.features", return_value={}) mocker.patch("kasa.discover.Discover.discover_single", return_value=dev) + connect_mock = None + if encrypt_type: + connect_mock = mocker.patch( + "kasa.device.Device.connect", + new_callable=mocker.AsyncMock, + return_value=dev, + ) res = await runner.invoke( cli, @@ -104,7 +114,17 @@ async def test_update_called_by_cli(dev, mocker, runner, device_family, encrypt_ catch_exceptions=False, ) assert res.exit_code == 0 - update.assert_called() + if encrypt_type: + update.assert_not_called() + else: + update.assert_called() + if connect_mock: + connect_mock.assert_called() + if encrypt_type == "KLAPV2": + called_kwargs = connect_mock.call_args.kwargs + cfg = called_kwargs.get("config") + assert cfg is not None + assert getattr(cfg.connection_type, "new_klap", None) is True async def test_list_devices(discovery_mock, runner): @@ -144,7 +164,7 @@ async def test_discover_raw(discovery_mock, runner, mocker): } assert res.output == json_dumps(expected, indent=True) + "\n" - redact_spy.assert_not_called() + calls_before = redact_spy.call_count res = await runner.invoke( cli, @@ -153,7 +173,7 @@ async def test_discover_raw(discovery_mock, runner, mocker): ) assert res.exit_code == 0 - redact_spy.assert_called() + assert redact_spy.call_count > calls_before @pytest.mark.parametrize( @@ -172,6 +192,8 @@ async def test_discover_raw(discovery_mock, runner, mocker): async def test_list_update_failed(discovery_mock, mocker, runner, exception, expected): """Test that device update is called on main.""" device_class = Discover._get_device_class(discovery_mock.discovery_data) + if discovery_mock.new_klap and discovery_mock.device_type.startswith("IOT."): + device_class = get_device_class_from_sys_info(discovery_mock.query_data) mocker.patch.object( device_class, "update", @@ -280,8 +302,7 @@ async def test_raw_command(dev, mocker, runner): res = await runner.invoke(raw_command, params, obj=dev) # Make sure that update was not called for wifi - with pytest.raises(AssertionError): - update.assert_called() + update.assert_not_called() assert res.exit_code == 0 assert dev.model in res.output @@ -405,8 +426,7 @@ async def test_wifi_join_smartcam(dev, mocker, runner): ) # Make sure that update was not called for wifi - with pytest.raises(AssertionError): - update.assert_called() + update.assert_not_called() assert res.exit_code == 0 assert "Asking the device to connect to FOOBAR" in res.output @@ -550,14 +570,14 @@ async def test_emeter(dev: Device, mocker, runner): res = await runner.invoke(cli, [*base_cmd, "--index", "0"], obj=dev) assert "Voltage: 122.066 V" in res.output - child_status.assert_called() - assert child_status.call_count == 1 + child_status.assert_called_once() + child_status.reset_mock() res = await runner.invoke( cli, [*base_cmd, "--name", dev.children[0].alias], obj=dev ) assert "Voltage: 122.066 V" in res.output - assert child_status.call_count == 2 + child_status.assert_called_once() if isinstance(dev, IotDevice): monthly = mocker.patch.object(energy, "get_monthly_stats") @@ -775,6 +795,9 @@ async def _state(dev: Device): mocker.patch("kasa.cli.device.state", new=_state) dr = DiscoveryResult.from_dict(discovery_mock.discovery_data["result"]) + encrypt_type = dr.mgt_encrypt_schm.encrypt_type + if dr.mgt_encrypt_schm.new_klap: + encrypt_type = DeviceEncryptionType.Klapv2.value res = await runner.invoke( cli, [ @@ -787,7 +810,7 @@ async def _state(dev: Device): "--device-family", dr.device_type, "--encrypt-type", - dr.mgt_encrypt_schm.encrypt_type, + encrypt_type, "--login-version", dr.mgt_encrypt_schm.lv or 1, ], @@ -958,6 +981,9 @@ async def test_discover_auth_failed(discovery_mock, mocker, runner): host = "127.0.0.1" discovery_mock.ip = host device_class = Discover._get_device_class(discovery_mock.discovery_data) + # For new_klap IoT fixtures the precise subclass comes from sysinfo. + if discovery_mock.new_klap and discovery_mock.device_type.startswith("IOT."): + device_class = get_device_class_from_sys_info(discovery_mock.query_data) mocker.patch.object( device_class, "update", @@ -988,6 +1014,8 @@ async def test_host_auth_failed(discovery_mock, mocker, runner): host = "127.0.0.1" discovery_mock.ip = host device_class = Discover._get_device_class(discovery_mock.discovery_data) + if discovery_mock.new_klap and discovery_mock.device_type.startswith("IOT."): + device_class = get_device_class_from_sys_info(discovery_mock.query_data) mocker.patch.object( device_class, "update", @@ -1438,15 +1466,21 @@ async def test_discover_config(dev: Device, mocker, runner): ) assert res.exit_code == 0 cparam = dev.config.connection_type - expected = f"--device-family {cparam.device_family.value} --encrypt-type {cparam.encryption_type.value} {'--https' if cparam.https else '--no-https'}" - assert expected in res.output - assert re.search( - r"Attempt to connect to 127\.0\.0\.1 with \w+ \+ \w+ \+ \w+ \+ \w+ failed", - res.output.replace("\n", ""), + encrypt_type = ( + DeviceEncryptionType.Klapv2.value + if cparam.new_klap + else cparam.encryption_type.value ) - assert re.search( - r"Attempt to connect to 127\.0\.0\.1 with \w+ \+ \w+ \+ \w+ \+ \w+ succeeded", - res.output.replace("\n", ""), + expected = f"--device-family {cparam.device_family.value} --encrypt-type {encrypt_type} {'--https' if cparam.https else '--no-https'}" + assert expected in res.output + normalized = " ".join(res.output.split()) + attempt_segs = normalized.split(f"Attempt to connect to {host} with")[1:] + assert attempt_segs, f"No connection attempt lines found in output:\n{res.output}" + assert any(" succeeded" in seg.lower() for seg in attempt_segs), ( + f"No succeeded attempt line found in:\n{res.output}" + ) + assert any(" failed" in seg.lower() for seg in attempt_segs), ( + f"No failed attempt line found in:\n{res.output}" ) diff --git a/tests/test_device_factory.py b/tests/test_device_factory.py index 19ccfb73d..4ee33e548 100644 --- a/tests/test_device_factory.py +++ b/tests/test_device_factory.py @@ -7,10 +7,11 @@ """ import logging -from typing import cast +from typing import Any, cast import aiohttp import pytest # type: ignore # https://github.com/pytest-dev/pytest/issues/3342 +from pytest_mock import MockerFixture from kasa import ( BaseProtocol, @@ -22,12 +23,14 @@ SmartProtocol, ) from kasa.device_factory import ( + GET_SYSINFO_QUERY, Device, IotDevice, SmartCamDevice, SmartDevice, connect, get_device_class_from_family, + get_device_class_from_sys_info, get_protocol, ) from kasa.deviceconfig import ( @@ -37,6 +40,8 @@ DeviceFamily, ) from kasa.discover import DiscoveryResult +from kasa.exceptions import UnsupportedDeviceError +from kasa.iot import IotBulb from kasa.transports import ( AesTransport, BaseTransport, @@ -55,12 +60,23 @@ pytestmark = [pytest.mark.requires_dummy] -def _get_connection_type_device_class(discovery_info): +def _get_connection_type_device_class( + discovery_info: dict[str, Any], +) -> tuple[DeviceConnectionParameters, type[Device]]: + device_class: type[Device] if "result" in discovery_info: - device_class = Discover._get_device_class(discovery_info) dr = DiscoveryResult.from_dict(discovery_info["result"]) - connection_type = Discover._get_connection_parameters(dr) + # For IoT device families, the precise subclass is determined by sysinfo + # at connect time (not from the discovery response alone); use the base + # IotDevice class for isinstance checks in tests. + if connection_type.device_family in ( + DeviceFamily.IotSmartPlugSwitch, + DeviceFamily.IotSmartBulb, + ): + device_class = IotDevice + else: + device_class = Discover._get_device_class(discovery_info) else: connection_type = DeviceConnectionParameters.from_values( DeviceFamily.IotSmartPlugSwitch.value, DeviceEncryptionType.Xor.value @@ -72,8 +88,8 @@ def _get_connection_type_device_class(discovery_info): async def test_connect( discovery_mock, - mocker, -): + mocker: MockerFixture, +) -> None: """Test that if the protocol is passed in it gets set correctly.""" host = DISCOVERY_MOCK_IP ctype, device_class = _get_connection_type_device_class( @@ -259,6 +275,12 @@ async def test_device_class_from_unknown_family(caplog): KlapTransport, id="iot-klap", ), + pytest.param( + CP(DF.IotSmartPlugSwitch, ET.Klap, https=False, new_klap=True), + IotProtocol, + KlapTransportV2, + id="iot-new-klap", + ), pytest.param( CP(DF.IotSmartPlugSwitch, ET.Xor, https=False), IotProtocol, @@ -289,9 +311,106 @@ async def test_get_protocol( conn_params: DeviceConnectionParameters, expected_protocol: type[BaseProtocol], expected_transport: type[BaseTransport], -): +) -> None: """Test get_protocol returns the right protocol.""" config = DeviceConfig("127.0.0.1", connection_type=conn_params) protocol = get_protocol(config) assert isinstance(protocol, expected_protocol) assert isinstance(protocol._transport, expected_transport) + + +async def test_connect_with_host_and_config_raises(): + """Cover branch raising when both host and config are provided.""" + dummy = DeviceConfig(host="127.0.0.1") + with pytest.raises(KasaException): + await connect(host="127.0.0.1", config=dummy) + + +async def test_connect_protocol_none_raises(): + """Cover branch when get_protocol returns None.""" + params = DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, + encryption_type=DeviceEncryptionType.Xor, # No SMART.XOR mapping + ) + cfg = DeviceConfig(host="127.0.0.15", connection_type=params) + with pytest.raises(UnsupportedDeviceError): + await connect(config=cfg) + + +async def test__connect_iot_ipcamera_unsupported_family(): + """Cover else branch in _connect raising UnsupportedDeviceError.""" + # Use a valid DeviceFamily that is intentionally unsupported by get_device_class_from_family. + params = DeviceConnectionParameters( + device_family=DeviceFamily.IotIpCamera, + encryption_type=DeviceEncryptionType.Aes, + https=True, + ) + cfg = DeviceConfig(host="127.0.0.16", connection_type=params) + protocol = get_protocol(cfg) + assert protocol is not None + # Ensure first XOR path is not taken. + assert not isinstance(protocol._transport, XorTransport) + with pytest.raises(UnsupportedDeviceError): + await connect(config=cfg) + + +def test_get_protocol_returns_none_for_unsupported_combo() -> None: + """Cover protocol_transport_key miss returning None.""" + p = DeviceConnectionParameters( + device_family=DeviceFamily.SmartTapoPlug, + encryption_type=DeviceEncryptionType.Xor, + ) + proto = get_protocol(DeviceConfig("127.0.0.99", connection_type=p)) + assert proto is None + + +def test_get_protocol_strict_encryption_mismatch() -> None: + """Cover strict=True mismatch branch for camera.""" + p = DeviceConnectionParameters( + device_family=DeviceFamily.SmartIpCamera, + encryption_type=DeviceEncryptionType.Klap, + ) + proto = get_protocol(DeviceConfig("127.0.0.55", connection_type=p), strict=True) + assert proto is None + + +def test_get_device_class_from_sys_info_mapping() -> None: + """Cover get_device_class_from_sys_info mapping.""" + info = {"system": {"get_sysinfo": {"type": "IOT.SMARTBULB"}}} + cls = get_device_class_from_sys_info(info) + assert cls is IotBulb + + +async def test_connect_with_host_only_branch_raises_unsupported( + mocker: MockerFixture, +) -> None: + """Exercise connect(host=..., config=None) to hit 'if host:' branch.""" + mocker.patch("kasa.device_factory.get_protocol", return_value=None) + with pytest.raises(UnsupportedDeviceError): + await connect(host="127.0.0.66", config=None) # type: ignore[arg-type] + + +async def test_connect_iot_family_resolves_subclass_from_sysinfo( + mocker: MockerFixture, +) -> None: + """IoT discovery families are narrowed to the concrete subclass via sysinfo.""" + query_mock = mocker.patch( + "kasa.IotProtocol.query", + new_callable=mocker.AsyncMock, + return_value={"system": {"get_sysinfo": {"type": "IOT.SMARTBULB"}}}, + ) + update_mock = mocker.patch.object(IotBulb, "update", new_callable=mocker.AsyncMock) + config = DeviceConfig( + host="127.0.0.1", + connection_type=DeviceConnectionParameters( + DeviceFamily.IotSmartPlugSwitch, + DeviceEncryptionType.Klap, + new_klap=True, + ), + ) + + dev = await connect(config=config) + + assert isinstance(dev, IotBulb) + query_mock.assert_awaited_once_with(GET_SYSINFO_QUERY) + update_mock.assert_awaited_once() diff --git a/tests/test_deviceconfig.py b/tests/test_deviceconfig.py index aebdd3a61..cc38e8e85 100644 --- a/tests/test_deviceconfig.py +++ b/tests/test_deviceconfig.py @@ -24,6 +24,15 @@ DeviceFamily.SmartTapoPlug, DeviceEncryptionType.Klap, login_version=2 ), ) +PLUG_NEW_KLAP_CONFIG = DeviceConfig( + host="127.0.0.1", + connection_type=DeviceConnectionParameters( + DeviceFamily.IotSmartPlugSwitch, + DeviceEncryptionType.Klap, + login_version=2, + new_klap=True, + ), +) CAMERA_AES_CONFIG = DeviceConfig( host="127.0.0.1", connection_type=DeviceConnectionParameters( @@ -48,6 +57,7 @@ async def test_serialization(): [ ("deviceconfig_plug-xor.json", PLUG_XOR_CONFIG), ("deviceconfig_plug-klap.json", PLUG_KLAP_CONFIG), + ("deviceconfig_plug-new-klap.json", PLUG_NEW_KLAP_CONFIG), ("deviceconfig_camera-aes-https.json", CAMERA_AES_CONFIG), ], ids=lambda arg: arg.split("_")[-1] if isinstance(arg, str) else "", @@ -79,6 +89,7 @@ async def test_conn_param_no_https(): } param = DeviceConnectionParameters.from_dict(dict_val) assert param.https is False + assert param.new_klap is None assert param.to_dict() == {**dict_val, "https": False} diff --git a/tests/test_discovery.py b/tests/test_discovery.py index e2b5042a6..39521c2ee 100644 --- a/tests/test_discovery.py +++ b/tests/test_discovery.py @@ -8,12 +8,14 @@ import re import socket from asyncio import timeout as asyncio_timeout +from typing import NotRequired, TypedDict from unittest.mock import MagicMock import aiohttp import pytest # type: ignore # https://github.com/pytest-dev/pytest/issues/3342 from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.asymmetric import padding as asymmetric_padding +from pytest_mock import MockerFixture from kasa import ( Credentials, @@ -56,7 +58,74 @@ # A physical device has to respond to discovery for the tests to work. pytestmark = [pytest.mark.requires_dummy] -UNSUPPORTED = { + +class DiscoveryEncryptionScheme(TypedDict, total=False): + is_support_https: bool + encrypt_type: str + http_port: int + lv: int + new_klap: int + + +class DiscoveryEncryptInfo(TypedDict): + sym_schm: str + key: str + data: str + + +class DiscoveryResponseResult(TypedDict, total=False): + device_id: str + owner: str + device_type: str + device_model: str + ip: str + mac: str + is_support_iot_cloud: bool + obd_src: str + factory_default: bool + mgt_encrypt_schm: DiscoveryEncryptionScheme + encrypt_info: DiscoveryEncryptInfo + encrypt_type: list[str] + + +class DiscoveryResponseInfo(TypedDict): + result: DiscoveryResponseResult + error_code: NotRequired[int] + + +class LegacySysInfo(TypedDict, total=False): + alias: str + dev_name: str + deviceId: str + err_code: int + hwId: str + hw_ver: str + mac: str + mic_type: str + model: str + sw_ver: str + type: str + updating: int + + +class LegacySystem(TypedDict): + get_sysinfo: LegacySysInfo + + +class LegacyDiscoveryInfo(TypedDict): + system: LegacySystem + + +class NewKlapCheckResult(TypedDict): + device_type: str + mgt_encrypt_schm: DiscoveryEncryptionScheme + + +class NewKlapCheckInfo(TypedDict): + result: NewKlapCheckResult + + +UNSUPPORTED: DiscoveryResponseInfo = { "result": { "device_id": "xx", "owner": "xx", @@ -77,6 +146,42 @@ "error_code": 0, } +NEW_KLAP_INFO: DiscoveryResponseInfo = { + "result": { + "device_type": "IOT.SMARTPLUGSWITCH", + "device_model": "HS100(UK)", + "device_id": "id", + "ip": "127.0.0.1", + "mac": "00-00-00-00-00-00", + "mgt_encrypt_schm": { + "is_support_https": False, + "encrypt_type": "KLAP", + "http_port": 9999, + "lv": 1, + "new_klap": 1, + }, + # Force decrypt attempt to run and be logged (will fail harmlessly and be caught/logged) + "encrypt_info": {"sym_schm": "AES", "key": "", "data": ""}, + } +} + +IPCAMERA_INFO: DiscoveryResponseInfo = { + "result": { + "device_type": "SMART.IPCAMERA", + "device_model": "C100(US)", + "device_id": "id", + "ip": "127.0.0.2", + "mac": "00-00-00-00-00-01", + "mgt_encrypt_schm": { + "is_support_https": True, + "encrypt_type": "AES", + "http_port": 443, + "lv": 3, + }, + "encrypt_type": ["3"], + } +} + @wallswitch_iot async def test_type_detection_switch(dev: Device): @@ -166,6 +271,7 @@ async def test_discover_single(discovery_mock, custom_port, mocker): login_version=discovery_mock.login_version, https=discovery_mock.https, http_port=discovery_mock.http_port, + new_klap=discovery_mock.new_klap, ) config = DeviceConfig( host=host, @@ -196,8 +302,8 @@ async def test_discover_single_hostname(discovery_mock, mocker): x = await Discover.discover_single(host, credentials=Credentials()) -async def test_discover_credentials(mocker): - """Make sure that discover gives credentials precedence over un and pw.""" +async def test_credentials_precedence_discover(mocker): + """Ensure credentials precedence logic for Discover.discover().""" host = "127.0.0.1" async def mock_discover(self, *_, **__): @@ -211,21 +317,21 @@ async def mock_discover(self, *_, **__): # Only credentials passed await Discover.discover(credentials=Credentials(), timeout=0) assert dp.mock_calls[0].kwargs["credentials"] == Credentials() - # Credentials and un/pw passed + await Discover.discover( credentials=Credentials(), username="Foo", password="Bar", timeout=0 ) assert dp.mock_calls[1].kwargs["credentials"] == Credentials() - # Only un/pw passed + await Discover.discover(username="Foo", password="Bar", timeout=0) assert dp.mock_calls[2].kwargs["credentials"] == Credentials("Foo", "Bar") - # Only un passed, credentials should be None + await Discover.discover(username="Foo", timeout=0) assert dp.mock_calls[3].kwargs["credentials"] is None -async def test_discover_single_credentials(mocker): - """Make sure that discover_single gives credentials precedence over un and pw.""" +async def test_credentials_precedence_discover_single(mocker): + """Ensure credentials precedence logic for Discover.discover_single().""" host = "127.0.0.1" async def mock_discover(self, *_, **__): @@ -239,15 +345,16 @@ async def mock_discover(self, *_, **__): # Only credentials passed await Discover.discover_single(host, credentials=Credentials(), timeout=0) assert dp.mock_calls[0].kwargs["credentials"] == Credentials() + # Credentials and un/pw passed await Discover.discover_single( host, credentials=Credentials(), username="Foo", password="Bar", timeout=0 ) assert dp.mock_calls[1].kwargs["credentials"] == Credentials() - # Only un/pw passed + await Discover.discover_single(host, username="Foo", password="Bar", timeout=0) assert dp.mock_calls[2].kwargs["credentials"] == Credentials("Foo", "Bar") - # Only un passed, credentials should be None + await Discover.discover_single(host, username="Foo", timeout=0) assert dp.mock_calls[3].kwargs["credentials"] is None @@ -326,6 +433,10 @@ async def test_discover_datagram_received(mocker, discovery_data): mocker.patch("kasa.discover.json_loads", return_value=UNSUPPORTED) proto.datagram_received("", (addr2, 20002)) + # Wait for async processing of the discovery callbacks to finish + if proto.callback_tasks: + await asyncio.gather(*proto.callback_tasks, return_exceptions=True) + # Check that device in discovered_devices is initialized correctly assert len(proto.discovered_devices) == 1 # Check that unsupported device is 1 @@ -347,7 +458,7 @@ async def test_discover_invalid_responses(msg, data, mocker): assert len(proto.discovered_devices) == 0 -AUTHENTICATION_DATA_KLAP = { +AUTHENTICATION_DATA_KLAP: DiscoveryResponseInfo = { "result": { "device_id": "xx", "owner": "xx", @@ -374,6 +485,10 @@ async def test_discover_single_authentication(discovery_mock, mocker): host = "127.0.0.1" discovery_mock.ip = host device_class = Discover._get_device_class(discovery_mock.discovery_data) + # For new_klap IoT fixtures the exact subclass comes from sysinfo, not the + # family-level class returned by _get_device_class. + if discovery_mock.new_klap and discovery_mock.device_type.startswith("IOT."): + device_class = get_device_class_from_sys_info(discovery_mock.query_data) mocker.patch.object( device_class, "update", @@ -418,41 +533,31 @@ async def test_device_update_from_new_discovery_info(discovery_mock): assert device.modules -async def test_discover_single_http_client(discovery_mock, mocker): - """Make sure that discover_single returns an initialized SmartDevice instance.""" +@pytest.mark.parametrize("entrypoint", ["discover_single", "discover"]) +async def test_http_client_passthrough(discovery_mock, mocker, entrypoint): + """Ensure HTTP client handling is consistent for discover and discover_single.""" host = "127.0.0.1" discovery_mock.ip = host http_client = aiohttp.ClientSession() + try: + if entrypoint == "discover_single": + dev: Device = await Discover.discover_single(host) + else: + devices = await Discover.discover(discovery_timeout=0) + dev: Device = devices[host] - x: Device = await Discover.discover_single(host) - - assert x.config.uses_http == (discovery_mock.default_port != 9999) - - if discovery_mock.default_port != 9999: - assert x.protocol._transport._http_client.client != http_client - x.config.http_client = http_client - assert x.protocol._transport._http_client.client == http_client - - -async def test_discover_http_client(discovery_mock, mocker): - """Make sure that discover returns an initialized SmartDevice instance.""" - host = "127.0.0.1" - discovery_mock.ip = host - - http_client = aiohttp.ClientSession() + assert dev.config.uses_http == (discovery_mock.default_port != 9999) - devices = await Discover.discover(discovery_timeout=0) - x: Device = devices[host] - assert x.config.uses_http == (discovery_mock.default_port != 9999) + if discovery_mock.default_port != 9999: + assert dev.protocol._transport._http_client.client != http_client + dev.config.http_client = http_client + assert dev.protocol._transport._http_client.client == http_client + finally: + await http_client.close() - if discovery_mock.default_port != 9999: - assert x.protocol._transport._http_client.client != http_client - x.config.http_client = http_client - assert x.protocol._transport._http_client.client == http_client - -LEGACY_DISCOVER_DATA = { +LEGACY_DISCOVER_DATA: LegacyDiscoveryInfo = { "system": { "get_sysinfo": { "alias": "#MASKED_NAME#", @@ -656,12 +761,15 @@ async def test_discovery_decryption(): data = json.dumps(data_dict) encypted_data = encryption_session.encrypt(data.encode()) - encrypt_info = { + encrypt_info: DiscoveryEncryptInfo = { "data": encypted_data.decode(), "key": encrypted_key_iv_b4.decode(), "sym_schm": "AES", } - info = {**UNSUPPORTED["result"], "encrypt_info": encrypt_info} + info: DiscoveryResponseResult = { + **UNSUPPORTED["result"], + "encrypt_info": encrypt_info, + } dr = DiscoveryResult.from_dict(info) Discover._decrypt_discovery_data(dr) assert dr.decrypted_data == data_dict @@ -679,10 +787,14 @@ async def test_discover_try_connect_all(discovery_mock, mocker): login_version=discovery_mock.login_version, https=discovery_mock.https, http_port=discovery_mock.http_port, + new_klap=discovery_mock.new_klap, ) protocol = get_protocol( DeviceConfig(discovery_mock.ip, connection_type=cparams) ) + if issubclass(dev_class, IotDevice): + info = await protocol.query({"system": {"get_sysinfo": {}}}) + dev_class = get_device_class_from_sys_info(info) protocol_class = protocol.__class__ transport_class = protocol._transport.__class__ else: @@ -749,3 +861,265 @@ async def test_discovery_device_repr(discovery_mock, mocker): assert "update() needed" not in repr_ else: assert "update() needed" in repr_ + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("info", "host"), + [ + (NEW_KLAP_INFO, "127.0.0.1"), + (IPCAMERA_INFO, "127.0.0.2"), + ], + ids=["new_klap_unsupported", "non_iot_unsupported"], +) +async def test_get_device_instance_unsupported_logs( + mocker: MockerFixture, + caplog: pytest.LogCaptureFixture, + info: DiscoveryResponseInfo, + host: str, +) -> None: + """_get_device_instance raises UnsupportedDeviceError and logs when device class cannot be resolved.""" + # Also validates the debug log is emitted. + caplog.set_level(logging.DEBUG) + + class DummyProt: + def __init__(self) -> None: + self._transport = MagicMock() + + async def close(self) -> None: + return None + + mocker.patch("kasa.discover.get_protocol", return_value=DummyProt()) + # Force device_class resolution to fail to trigger the debug log + mocker.patch("kasa.discover.Discover._get_device_class", return_value=None) + + with pytest.raises(UnsupportedDeviceError): + await Discover._get_device_instance(info, DeviceConfig(host=host)) + + assert "Got unsupported device type" in caplog.text + + +def test_datagram_received_logs_for_exceptions( + caplog: pytest.LogCaptureFixture, mocker: MockerFixture +) -> None: + proto = _DiscoverProtocol() + caplog.set_level(logging.DEBUG) + + ip1 = "127.0.0.10" + ip2 = "127.0.0.11" + + # First call raises UnsupportedDeviceError -> goes to unsupported_device_exceptions and logs + mocker.patch( + "kasa.discover.Discover._get_discovery_json", + side_effect=[UnsupportedDeviceError("boom"), KasaException("bad")], + ) + + proto.datagram_received(b"x", (ip1, Discover.DISCOVERY_PORT_2)) + proto.datagram_received(b"x", (ip2, Discover.DISCOVERY_PORT_2)) + + # Bookkeeping recorded correctly + assert ip1 in proto.unsupported_device_exceptions + assert ip2 in proto.invalid_device_exceptions + + # Logs were written + assert f"Unsupported device found at {ip1} << boom" in caplog.text + assert f"[DISCOVERY] Unable to find device type for {ip2}: bad" in caplog.text + + +def test_handle_discovered_event_without_discover_task() -> None: + proto = _DiscoverProtocol(target="10.0.0.5") + proto.seen_hosts.add("10.0.0.5") + assert proto.discover_task is None + proto._handle_discovered_event() + assert proto.target_discovered is True + assert proto.discover_task is None + + +def test_error_received_logs(caplog: pytest.LogCaptureFixture) -> None: + caplog.set_level(logging.ERROR) + proto = _DiscoverProtocol() + proto.error_received(RuntimeError("boom")) + assert "Got error: boom" in caplog.text + + +def test_get_device_class_unknown_result_type() -> None: + info: DiscoveryResponseInfo = { + "result": { + "device_type": "FOO.BAR", + "device_model": "X100(US)", + "device_id": "id", + "ip": "127.0.0.1", + "mac": "AA-BB-CC-DD-EE-FF", + } + } + with pytest.raises(UnsupportedDeviceError) as ex: + Discover._get_device_class(info) + assert "Unknown device type: FOO.BAR" in str(ex.value) + + +def test_get_discovery_json_legacy_error(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr( + XorEncryption, "decrypt", lambda _: (_ for _ in ()).throw(ValueError("bad")) + ) + with pytest.raises(KasaException) as ex: + Discover._get_discovery_json_legacy(b"\x00\x01", "127.0.0.2") + assert "Unable to read response from device: 127.0.0.2" in str(ex.value) + + +def test_get_device_instance_legacy_no_debug_logging( + caplog: pytest.LogCaptureFixture, +) -> None: + caplog.set_level(logging.INFO) + info: LegacyDiscoveryInfo = { + "system": { + "get_sysinfo": { + "alias": "Plug", + "deviceId": "id", + "mac": "00:11:22:33:44:55", + "mic_type": "IOT.SMARTPLUGSWITCH", + "model": "HS100(UK)", + } + } + } + config = DeviceConfig(host="127.0.0.3") + dev = Discover._get_device_instance_legacy(info, config) + assert dev.host == "127.0.0.3" + assert "[DISCOVERY]" not in caplog.text + + +def test_get_discovery_json_error( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + caplog.set_level(logging.DEBUG) + monkeypatch.setattr( + "kasa.discover.json_loads", + lambda *_: (_ for _ in ()).throw(ValueError("bad json")), + ) + with pytest.raises(KasaException) as ex: + Discover._get_discovery_json(b"\x00" * 32, "127.0.0.9") + assert "Unable to read response from device: 127.0.0.9" in str(ex.value) + assert "Got invalid response from device 127.0.0.9" in caplog.text + + +def test_get_discovery_json_legacy_success(monkeypatch: pytest.MonkeyPatch) -> None: + sample: LegacyDiscoveryInfo = { + "system": {"get_sysinfo": {"type": "IOT.SMARTPLUGSWITCH"}} + } + monkeypatch.setattr( + XorEncryption, + "decrypt", + lambda _: b'{"system":{"get_sysinfo":{"type":"IOT.SMARTPLUGSWITCH"}}}', + ) + info = Discover._get_discovery_json_legacy(b"\x00\x01", "127.0.0.4") + assert info == sample + + +def test_is_new_klap_iot_true() -> None: + """IOT device with new_klap=1 should return True.""" + info: NewKlapCheckInfo = { + "result": { + "device_type": "IOT.SMARTPLUGSWITCH", + "mgt_encrypt_schm": {"new_klap": 1}, + } + } + assert _DiscoverProtocol._is_new_klap_iot(info) is True + + +def test_is_new_klap_iot_false_not_iot() -> None: + """Non-IOT device with new_klap should return False.""" + info: NewKlapCheckInfo = { + "result": { + "device_type": "SMART.TAPOPLUG", + "mgt_encrypt_schm": {"new_klap": 1}, + } + } + assert _DiscoverProtocol._is_new_klap_iot(info) is False + + +def test_is_new_klap_iot_false_no_flag() -> None: + """IOT device without the new_klap flag should return False.""" + info: NewKlapCheckInfo = { + "result": { + "device_type": "IOT.SMARTPLUGSWITCH", + "mgt_encrypt_schm": {"encrypt_type": "KLAP"}, + } + } + assert _DiscoverProtocol._is_new_klap_iot(info) is False + + +async def test_process_new_klap_device_protocol_none( + mocker: MockerFixture, +) -> None: + """UnsupportedDeviceError is stored when get_protocol returns None.""" + proto = _DiscoverProtocol() + mocker.patch("kasa.discover.get_protocol", return_value=None) + + await proto._process_new_klap_device(NEW_KLAP_INFO, DeviceConfig(host="127.0.0.1")) + + assert "127.0.0.1" in proto.unsupported_device_exceptions + assert "127.0.0.1" not in proto.discovered_devices + + +async def test_process_new_klap_device_on_unsupported_callback( + mocker: MockerFixture, +) -> None: + """on_unsupported callback is awaited in the UnsupportedDeviceError path.""" + on_unsupported = mocker.AsyncMock() + proto = _DiscoverProtocol(on_unsupported=on_unsupported) + mocker.patch("kasa.discover.get_protocol", return_value=None) + + await proto._process_new_klap_device(NEW_KLAP_INFO, DeviceConfig(host="127.0.0.1")) + + assert "127.0.0.1" in proto.unsupported_device_exceptions + on_unsupported.assert_awaited_once() + + +async def test_process_new_klap_device_kasa_exception( + mocker: MockerFixture, +) -> None: + """KasaException from sysinfo query is stored in invalid_device_exceptions.""" + proto = _DiscoverProtocol() + mock_protocol = MagicMock() + mock_protocol.query = mocker.AsyncMock(side_effect=KasaException("query failed")) + mocker.patch("kasa.discover.get_protocol", return_value=mock_protocol) + + await proto._process_new_klap_device(NEW_KLAP_INFO, DeviceConfig(host="127.0.0.1")) + + assert "127.0.0.1" in proto.invalid_device_exceptions + assert "127.0.0.1" not in proto.discovered_devices + + +async def test_process_new_klap_device_success_with_on_discovered( + mocker: MockerFixture, +) -> None: + """Success path fires on_discovered and adds device to discovered_devices.""" + on_discovered = mocker.AsyncMock() + proto = _DiscoverProtocol(on_discovered=on_discovered) + mock_protocol = MagicMock() + mock_protocol.query = mocker.AsyncMock( + return_value={"system": {"get_sysinfo": {"type": "IOT.SMARTPLUGSWITCH"}}}, + ) + mocker.patch("kasa.discover.get_protocol", return_value=mock_protocol) + + await proto._process_new_klap_device(NEW_KLAP_INFO, DeviceConfig(host="127.0.0.1")) + + assert "127.0.0.1" in proto.discovered_devices + on_discovered.assert_awaited_once() + + +@pytest.mark.xdist_group(name="caplog") +async def test_process_new_klap_device_debug_logging( + mocker: MockerFixture, caplog: pytest.LogCaptureFixture +) -> None: + """Debug logging branch in _process_new_klap_device emits [DISCOVERY].""" + caplog.set_level(logging.DEBUG) + proto = _DiscoverProtocol() + mock_protocol = MagicMock() + mock_protocol.query = mocker.AsyncMock( + return_value={"system": {"get_sysinfo": {"type": "IOT.SMARTPLUGSWITCH"}}}, + ) + mocker.patch("kasa.discover.get_protocol", return_value=mock_protocol) + + await proto._process_new_klap_device(NEW_KLAP_INFO, DeviceConfig(host="127.0.0.1")) + + assert "[DISCOVERY]" in caplog.text