diff --git a/devtools/dump_devinfo.py b/devtools/dump_devinfo.py index c03e97d55..c7822d5f7 100644 --- a/devtools/dump_devinfo.py +++ b/devtools/dump_devinfo.py @@ -17,11 +17,14 @@ import asyncclick as click -from kasa import AuthenticationException, Credentials, Discover, SmartDevice +from kasa import AuthenticationException, Credentials, Discover from kasa.discover import DiscoveryResult +from kasa.exceptions import SmartErrorCode +from kasa.smartrequests import COMPONENT_REQUESTS, SmartRequest from kasa.tapo.tapodevice import TapoDevice Call = namedtuple("Call", "module method") +SmartCall = namedtuple("SmartCall", "module request should_succeed") def scrub(res): @@ -46,11 +49,19 @@ def scrub(res): "oem_id", "nickname", "alias", + "bssid", + "channel", ] for k, v in res.items(): if isinstance(v, collections.abc.Mapping): res[k] = scrub(res.get(k)) + elif ( + isinstance(v, list) + and len(v) > 0 + and isinstance(v[0], collections.abc.Mapping) + ): + res[k] = [scrub(vi) for vi in v] else: if k in keys_to_scrub: if k in ["latitude", "latitude_i", "longitude", "longitude_i"]: @@ -64,6 +75,8 @@ def scrub(res): v = base64.b64encode(b"#MASKED_NAME#").decode() elif k in ["alias"]: v = "#MASKED_NAME#" + elif isinstance(res[k], int): + v = 0 else: v = re.sub(r"\w", "0", v) @@ -200,33 +213,86 @@ async def get_legacy_fixture(device): return save_filename, copy_folder, final -async def get_smart_fixture(device: SmartDevice): +async def get_smart_fixture(device: TapoDevice): """Get fixture for new TAPO style protocol.""" - items = [ - Call(module="component_nego", method="component_nego"), - Call(module="device_info", method="get_device_info"), - Call(module="device_usage", method="get_device_usage"), - Call(module="device_time", method="get_device_time"), - Call(module="energy_usage", method="get_energy_usage"), - Call(module="current_power", method="get_current_power"), - Call(module="temp_humidity_records", method="get_temp_humidity_records"), - Call(module="child_device_list", method="get_child_device_list"), - Call( - module="trigger_logs", - method={"get_trigger_logs": {"page_size": 5, "start_id": 0}}, + extra_test_calls = [ + SmartCall( + module="temp_humidity_records", + request=SmartRequest.get_raw_request("get_temp_humidity_records"), + should_succeed=False, ), - Call( + SmartCall( + module="child_device_list", + request=SmartRequest.get_raw_request("get_child_device_list"), + should_succeed=False, + ), + SmartCall( module="child_device_component_list", - method="get_child_device_component_list", + request=SmartRequest.get_raw_request("get_child_device_component_list"), + should_succeed=False, + ), + SmartCall( + module="trigger_logs", + request=SmartRequest.get_raw_request( + "get_trigger_logs", SmartRequest.GetTriggerLogsParams(5, 0) + ), + should_succeed=False, ), ] successes = [] - for test_call in items: + try: + click.echo("Testing component_nego call ..", nl=False) + component_info_response = await device._smart_query_helper( + SmartRequest.component_nego() + ) + click.echo(click.style("OK", fg="green")) + successes.append( + SmartCall( + module="component_nego", + request=SmartRequest("component_nego"), + should_succeed=True, + ) + ) + except AuthenticationException as ex: + click.echo( + click.style( + f"Unable to query the device due to an authentication error: {ex}", + bold=True, + fg="red", + ) + ) + exit(1) + except Exception: + click.echo( + click.style("CRITICAL FAIL on component_nego call, exiting", fg="red") + ) + exit(1) + + test_calls = [] + should_succeed = [] + + for item in component_info_response["component_nego"]["component_list"]: + component_id = item["id"] + if requests := COMPONENT_REQUESTS.get(component_id): + component_test_calls = [ + SmartCall(module=component_id, request=request, should_succeed=True) + for request in requests + ] + test_calls.extend(component_test_calls) + should_succeed.extend(component_test_calls) + elif component_id not in COMPONENT_REQUESTS: + click.echo(f"Skipping {component_id}..", nl=False) + click.echo(click.style("UNSUPPORTED", fg="yellow")) + + test_calls.extend(extra_test_calls) + + for test_call in test_calls: + click.echo(f"Testing {test_call.module}..", nl=False) try: click.echo(f"Testing {test_call}..", nl=False) - response = await device.protocol.query(test_call.method) + response = await device._smart_query_helper(test_call.request) except AuthenticationException as ex: click.echo( click.style( @@ -237,22 +303,38 @@ async def get_smart_fixture(device: SmartDevice): ) exit(1) except Exception as ex: - click.echo(click.style(f"FAIL {ex}", fg="red")) + if ( + not test_call.should_succeed + and hasattr(ex, "error_code") + and ex.error_code == SmartErrorCode.UNKNOWN_METHOD_ERROR + ): + click.echo(click.style("FAIL - EXPECTED", fg="green")) + else: + click.echo(click.style(f"FAIL {ex}", fg="red")) else: if not response: - click.echo(click.style("FAIL not suported", fg="red")) + click.echo(click.style("FAIL no response", fg="red")) else: - click.echo(click.style("OK", fg="green")) + if not test_call.should_succeed: + click.echo(click.style("OK - EXPECTED FAIL", fg="red")) + else: + click.echo(click.style("OK", fg="green")) successes.append(test_call) requests = [] for succ in successes: - requests.append({"method": succ.method}) - - final_query = {"multipleRequest": {"requests": requests}} + requests.append(succ.request) + final = {} try: - responses = await device.protocol.query(final_query) + end = len(requests) + step = 10 # Break the requests down as there seems to be a size limit + for i in range(0, end, step): + x = i + requests_step = requests[x : x + step] + responses = await device._smart_query_helper(requests_step) + for method, result in responses.items(): + final[method] = result except AuthenticationException as ex: click.echo( click.style( @@ -269,9 +351,6 @@ async def get_smart_fixture(device: SmartDevice): ) ) exit(1) - final = {} - for method, result in responses.items(): - final[method] = result # Need to recreate a DiscoverResult here because we don't want the aliases # in the fixture, we want the actual field names as returned by the device. diff --git a/kasa/smartrequests.py b/kasa/smartrequests.py new file mode 100644 index 000000000..e933e0308 --- /dev/null +++ b/kasa/smartrequests.py @@ -0,0 +1,350 @@ +"""SmartRequest helper classes and functions for new SMART/TAPO devices. + +List of known requests with associated parameter classes. + +Other requests that are known but not currently implemented +or tested are: + +get_child_device_component_list +get_child_device_list +control_child +get_device_running_info - seems to be a subset of get_device_info + +get_tss_info +get_raw_dvi +get_homekit_info + +fw_download + +sync_env +account_sync + +device_reset +close_device_ble +heart_beat + +""" + +import logging +from dataclasses import asdict, dataclass +from typing import List, Optional, Union + +_LOGGER = logging.getLogger(__name__) +logging.getLogger("httpx").propagate = False + + +class SmartRequest: + """Class to represent a smart protocol request.""" + + def __init__(self, method_name: str, params: Optional["SmartRequestParams"] = None): + self.method_name = method_name + if params: + self.params = params.to_dict() + else: + self.params = None + + def __repr__(self): + return f"SmartRequest({self.method_name})" + + def to_dict(self): + """Return the request as a dict suitable for passing to query().""" + return {self.method_name: self.params} + + @dataclass + class SmartRequestParams: + """Base class for Smart request params. + + The to_dict() method of this class omits null values which + is required by the devices. + """ + + def to_dict(self): + """Return the params as a dict with values of None ommited.""" + return asdict( + self, dict_factory=lambda x: {k: v for (k, v) in x if v is not None} + ) + + @dataclass + class DeviceOnParams(SmartRequestParams): + """Get Rules Params.""" + + device_on: bool + + @dataclass + class GetRulesParams(SmartRequestParams): + """Get Rules Params.""" + + start_index: int = 0 + + @dataclass + class GetTriggerLogsParams(SmartRequestParams): + """Trigger Logs params.""" + + page_size: int = 5 + start_id: int = 0 + + @dataclass + class LedStatusParams(SmartRequestParams): + """LED Status params.""" + + led_rule: Optional[str] = None + + @staticmethod + def from_bool(state: bool): + """Set the led_rule from the state.""" + rule = "always" if state else "never" + return SmartRequest.LedStatusParams(led_rule=rule) + + @dataclass + class LightInfoParams(SmartRequestParams): + """LightInfo params.""" + + brightness: Optional[int] = None + color_temp: Optional[int] = None + hue: Optional[int] = None + saturation: Optional[int] = None + + @dataclass + class DynamicLightEffectParams(SmartRequestParams): + """LightInfo params.""" + + enable: bool + id: Optional[str] = None + + @staticmethod + def get_raw_request( + method: str, params: Optional[SmartRequestParams] = None + ) -> "SmartRequest": + """Send a raw request to the device.""" + return SmartRequest(method, params) + + @staticmethod + def component_nego() -> "SmartRequest": + """Get quick setup component info.""" + return SmartRequest("component_nego") + + @staticmethod + def get_device_info() -> "SmartRequest": + """Get device info.""" + return SmartRequest("get_device_info") + + @staticmethod + def get_device_usage() -> "SmartRequest": + """Get device usage.""" + return SmartRequest("get_device_usage") + + @staticmethod + def device_info_list() -> list["SmartRequest"]: + """Get device info list.""" + return [ + SmartRequest.get_device_info(), + SmartRequest.get_device_usage(), + ] + + @staticmethod + def get_auto_update_info() -> "SmartRequest": + """Get auto update info.""" + return SmartRequest("get_auto_update_info") + + @staticmethod + def firmware_info_list() -> List["SmartRequest"]: + """Get info list.""" + return [ + SmartRequest.get_auto_update_info(), + SmartRequest.get_raw_request("get_fw_download_state"), + SmartRequest.get_raw_request("get_latest_fw"), + ] + + @staticmethod + def qs_component_nego() -> "SmartRequest": + """Get quick setup component info.""" + return SmartRequest("qs_component_nego") + + @staticmethod + def get_device_time() -> "SmartRequest": + """Get device time.""" + return SmartRequest("get_device_time") + + @staticmethod + def get_wireless_scan_info() -> "SmartRequest": + """Get wireless scan info.""" + return SmartRequest("get_wireless_scan_info") + + @staticmethod + def get_schedule_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get schedule rules.""" + return SmartRequest( + "get_schedule_rules", params or SmartRequest.GetRulesParams() + ) + + @staticmethod + def get_next_event(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get next scheduled event.""" + return SmartRequest("get_next_event", params or SmartRequest.GetRulesParams()) + + @staticmethod + def schedule_info_list() -> List["SmartRequest"]: + """Get schedule info list.""" + return [ + SmartRequest.get_schedule_rules(), + SmartRequest.get_next_event(), + ] + + @staticmethod + def get_countdown_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get countdown rules.""" + return SmartRequest( + "get_countdown_rules", params or SmartRequest.GetRulesParams() + ) + + @staticmethod + def get_antitheft_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get antitheft rules.""" + return SmartRequest( + "get_antitheft_rules", params or SmartRequest.GetRulesParams() + ) + + @staticmethod + def get_led_info(params: Optional[LedStatusParams] = None) -> "SmartRequest": + """Get led info.""" + return SmartRequest("get_led_info", params or SmartRequest.LedStatusParams()) + + @staticmethod + def get_auto_off_config(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get auto off config.""" + return SmartRequest( + "get_auto_off_config", params or SmartRequest.GetRulesParams() + ) + + @staticmethod + def get_delay_action_info() -> "SmartRequest": + """Get delay action info.""" + return SmartRequest("get_delay_action_info") + + @staticmethod + def auto_off_list() -> List["SmartRequest"]: + """Get energy usage.""" + return [ + SmartRequest.get_auto_off_config(), + SmartRequest.get_delay_action_info(), # May not live here + ] + + @staticmethod + def get_energy_usage() -> "SmartRequest": + """Get energy usage.""" + return SmartRequest("get_energy_usage") + + @staticmethod + def energy_monitoring_list() -> List["SmartRequest"]: + """Get energy usage.""" + return [ + SmartRequest("get_energy_usage"), + SmartRequest.get_raw_request("get_electricity_price_config"), + ] + + @staticmethod + def get_current_power() -> "SmartRequest": + """Get current power.""" + return SmartRequest("get_current_power") + + @staticmethod + def power_protection_list() -> List["SmartRequest"]: + """Get power protection info list.""" + return [ + SmartRequest.get_current_power(), + SmartRequest.get_raw_request("get_max_power"), + SmartRequest.get_raw_request("get_protection_power"), + ] + + @staticmethod + def get_preset_rules(params: Optional[GetRulesParams] = None) -> "SmartRequest": + """Get preset rules.""" + return SmartRequest("get_preset_rules", params or SmartRequest.GetRulesParams()) + + @staticmethod + def get_auto_light_info() -> "SmartRequest": + """Get auto light info.""" + return SmartRequest("get_auto_light_info") + + @staticmethod + def get_dynamic_light_effect_rules( + params: Optional[GetRulesParams] = None + ) -> "SmartRequest": + """Get dynamic light effect rules.""" + return SmartRequest( + "get_dynamic_light_effect_rules", params or SmartRequest.GetRulesParams() + ) + + @staticmethod + def set_device_on(params: DeviceOnParams) -> "SmartRequest": + """Set device on state.""" + return SmartRequest("set_device_info", params) + + @staticmethod + def set_light_info(params: LightInfoParams) -> "SmartRequest": + """Set color temperature.""" + return SmartRequest("set_device_info", params) + + @staticmethod + def set_dynamic_light_effect_rule_enable( + params: DynamicLightEffectParams + ) -> "SmartRequest": + """Enable dynamic light effect rule.""" + return SmartRequest("set_dynamic_light_effect_rule_enable", params) + + @staticmethod + def get_component_info_requests(component_nego_response) -> List["SmartRequest"]: + """Get a list of requests based on the component info response.""" + request_list = [] + for component in component_nego_response["component_list"]: + if requests := COMPONENT_REQUESTS.get(component["id"]): + request_list.extend(requests) + return request_list + + @staticmethod + def _create_request_dict( + smart_request: Union["SmartRequest", list["SmartRequest"]] + ) -> dict: + """Create request dict to be passed to SmartProtocol.query().""" + if isinstance(smart_request, list): + request = {} + for sr in smart_request: + request[sr.method_name] = sr.params + else: + request = smart_request.to_dict() + return request + + +COMPONENT_REQUESTS = { + "device": SmartRequest.device_info_list(), + "firmware": SmartRequest.firmware_info_list(), + "quick_setup": [SmartRequest.qs_component_nego()], + "inherit": [SmartRequest.get_raw_request("get_inherit_info")], + "time": [SmartRequest.get_device_time()], + "wireless": [SmartRequest.get_wireless_scan_info()], + "schedule": SmartRequest.schedule_info_list(), + "countdown": [SmartRequest.get_countdown_rules()], + "antitheft": [SmartRequest.get_antitheft_rules()], + "account": None, + "synchronize": None, # sync_env + "sunrise_sunset": None, # for schedules + "led": [SmartRequest.get_led_info()], + "cloud_connect": [SmartRequest.get_raw_request("get_connect_cloud_state")], + "iot_cloud": None, + "device_local_time": None, + "default_states": None, # in device_info + "auto_off": [SmartRequest.get_auto_off_config()], + "localSmart": None, + "energy_monitoring": SmartRequest.energy_monitoring_list(), + "power_protection": SmartRequest.power_protection_list(), + "current_protection": None, # overcurrent in device_info + "matter": None, + "preset": [SmartRequest.get_preset_rules()], + "brightness": None, # in device_info + "color": None, # in device_info + "color_temperature": None, # in device_info + "auto_light": [SmartRequest.get_auto_light_info()], + "light_effect": [SmartRequest.get_dynamic_light_effect_rules()], + "bulb_quick_control": None, + "on_off_gradually": [SmartRequest.get_raw_request("get_on_off_gradually_info")], +} diff --git a/kasa/tapo/tapodevice.py b/kasa/tapo/tapodevice.py index 717de7ef4..292f7c418 100644 --- a/kasa/tapo/tapodevice.py +++ b/kasa/tapo/tapodevice.py @@ -2,7 +2,7 @@ import base64 import logging from datetime import datetime, timedelta, timezone -from typing import Any, Dict, Optional, Set, cast +from typing import Any, Dict, Optional, Set, Union, cast from ..aestransport import AesTransport from ..deviceconfig import DeviceConfig @@ -10,6 +10,7 @@ from ..protocol import TPLinkProtocol from ..smartdevice import SmartDevice from ..smartprotocol import SmartProtocol +from ..smartrequests import SmartRequest _LOGGER = logging.getLogger(__name__) @@ -38,15 +39,13 @@ async def update(self, update_children: bool = True): raise AuthenticationException("Tapo plug requires authentication.") if self._components is None: - resp = await self.protocol.query("component_nego") + resp = await self._smart_query_helper(SmartRequest.component_nego()) self._components = resp["component_nego"] - req = { - "get_device_info": None, - "get_device_usage": None, - "get_device_time": None, - } - resp = await self.protocol.query(req) + requests = SmartRequest.get_component_info_requests(self._components) + + resp = await self._smart_query_helper(requests) + self._info = resp["get_device_info"] self._usage = resp["get_device_usage"] self._time = resp["get_device_time"] @@ -139,10 +138,12 @@ def internal_state(self) -> Any: """Return all the internal state data.""" return self._data - async def _query_helper( - self, target: str, cmd: str, arg: Optional[Dict] = None, child_ids=None + async def _smart_query_helper( + self, smart_request: Union[SmartRequest, list[SmartRequest]] ) -> Any: - res = await self.protocol.query({cmd: arg}) + res = await self.protocol.query( + SmartRequest._create_request_dict(smart_request) + ) return res @@ -168,11 +169,15 @@ def is_on(self) -> bool: async def turn_on(self, **kwargs): """Turn on the device.""" - await self.protocol.query({"set_device_info": {"device_on": True}}) + await self._smart_query_helper( + SmartRequest.set_device_on(SmartRequest.DeviceOnParams(True)) + ) async def turn_off(self, **kwargs): """Turn off the device.""" - await self.protocol.query({"set_device_info": {"device_on": False}}) + await self._smart_query_helper( + SmartRequest.set_device_on(SmartRequest.DeviceOnParams(False)) + ) def update_from_discover_info(self, info): """Update state from info from the discover call.""" diff --git a/kasa/tests/conftest.py b/kasa/tests/conftest.py index 11efe6937..19d972199 100644 --- a/kasa/tests/conftest.py +++ b/kasa/tests/conftest.py @@ -2,6 +2,7 @@ import glob import json import os +import warnings from dataclasses import dataclass from json import dumps as json_dumps from os.path import basename @@ -43,7 +44,7 @@ SUPPORTED_DEVICES = SUPPORTED_IOT_DEVICES + SUPPORTED_SMART_DEVICES # Tapo bulbs -BULBS_SMART_VARIABLE_TEMP = {"L530E"} +BULBS_SMART_VARIABLE_TEMP = {"L530E", "L510B"} BULBS_SMART_COLOR = {"L530E"} BULBS_SMART_LIGHT_STRIP: Set[str] = set() BULBS_SMART_DIMMABLE: Set[str] = set() @@ -534,6 +535,21 @@ def mock_discover(self): yield discovery_data +def pytest_configure(): + pytest.fixtures_missing_methods = {} + + +def pytest_sessionfinish(session, exitstatus): + for fixture, methods in pytest.fixtures_missing_methods.items(): + method_list = "\n".join(methods) + warnings.warn( + UserWarning( + f"Fixture {fixture} missing expected methods and needs regenerating:\n{method_list}\n" + ), + stacklevel=1, + ) + + def pytest_addoption(parser): parser.addoption( "--ip", action="store", default=None, help="run against device on given ip" diff --git a/kasa/tests/fixtures/smart/L510B(EU)_3.0_1.0.5.json b/kasa/tests/fixtures/smart/L510B(EU)_3.0_1.0.5.json new file mode 100644 index 000000000..c8ff15fee --- /dev/null +++ b/kasa/tests/fixtures/smart/L510B(EU)_3.0_1.0.5.json @@ -0,0 +1,246 @@ +{ + "component_nego": { + "component_list": [ + { + "id": "device", + "ver_code": 2 + }, + { + "id": "firmware", + "ver_code": 2 + }, + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "time", + "ver_code": 1 + }, + { + "id": "wireless", + "ver_code": 1 + }, + { + "id": "schedule", + "ver_code": 2 + }, + { + "id": "countdown", + "ver_code": 2 + }, + { + "id": "antitheft", + "ver_code": 1 + }, + { + "id": "account", + "ver_code": 1 + }, + { + "id": "synchronize", + "ver_code": 1 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "brightness", + "ver_code": 1 + }, + { + "id": "cloud_connect", + "ver_code": 1 + }, + { + "id": "default_states", + "ver_code": 1 + }, + { + "id": "preset", + "ver_code": 1 + }, + { + "id": "device_local_time", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "bulb_quick_control", + "ver_code": 1 + } + ] + }, + "discovery_result": { + "device_id": "00000000000000000000000000000000", + "device_model": "L510B(EU)", + "device_type": "SMART.TAPOBULB", + "factory_default": false, + "ip": "127.0.0.123", + "is_support_iot_cloud": true, + "mac": "00-00-00-00-00-00", + "mgt_encrypt_schm": { + "encrypt_type": "AES", + "http_port": 80, + "is_support_https": false, + "lv": 2 + }, + "obd_src": "tplink", + "owner": "00000000000000000000000000000000" + }, + "get_antitheft_rules": { + "antitheft_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_auto_update_info": { + "enable": false, + "random_range": 120, + "time": 180 + }, + "get_connect_cloud_state": { + "status": 0 + }, + "get_countdown_rules": { + "countdown_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_device_info": { + "avatar": "bulb", + "brightness": 100, + "color_temp_range": [ + 2700, + 2700 + ], + "default_states": { + "brightness": { + "type": "last_states", + "value": 100 + }, + "re_power_type": "always_on" + }, + "device_id": "0000000000000000000000000000000000000000", + "device_on": true, + "fw_id": "00000000000000000000000000000000", + "fw_ver": "1.0.5 Build 230529 Rel.113426", + "has_set_location_info": true, + "hw_id": "00000000000000000000000000000000", + "hw_ver": "3.0", + "ip": "127.0.0.123", + "lang": "en_US", + "latitude": 0, + "longitude": 0, + "mac": "00-00-00-00-00-00", + "model": "L510", + "music_rhythm_enable": false, + "music_rhythm_mode": "single_lamp", + "nickname": "I01BU0tFRF9OQU1FIw==", + "oem_id": "00000000000000000000000000000000", + "overheated": false, + "region": "Europe/London", + "rssi": -59, + "signal_level": 2, + "specs": "", + "ssid": "I01BU0tFRF9TU0lEIw==", + "time_diff": 0, + "type": "SMART.TAPOBULB" + }, + "get_device_time": { + "region": "Europe/London", + "time_diff": 0, + "timestamp": 1703329221 + }, + "get_device_usage": { + "power_usage": { + "past30": 223, + "past7": 223, + "today": 0 + }, + "saved_power": { + "past30": 1023, + "past7": 1023, + "today": 1 + }, + "time_usage": { + "past30": 1246, + "past7": 1246, + "today": 1 + } + }, + "get_fw_download_state": { + "auto_upgrade": false, + "download_progress": 0, + "reboot_time": 5, + "status": 0, + "upgrade_time": 5 + }, + "get_inherit_info": null, + "get_latest_fw": { + "fw_size": 786432, + "fw_ver": "1.1.2 Build 231120 Rel.201048", + "hw_id": "00000000000000000000000000000000", + "need_to_upgrade": true, + "oem_id": "00000000000000000000000000000000", + "release_date": "2023-12-04", + "release_note": "Modifications and Bug Fixes:\n1. Added the support for setting the Fade In time manually.\n2. Optimized Wi-Fi connection stability\n3. Enhanced local communication security.\n4. Fixed some minor bugs.", + "type": 1 + }, + "get_next_event": {}, + "get_preset_rules": { + "brightness": [ + 1, + 25, + 50, + 75, + 100 + ] + }, + "get_schedule_rules": { + "enable": false, + "rule_list": [], + "schedule_rule_max_count": 32, + "start_index": 0, + "sum": 0 + }, + "get_wireless_scan_info": { + "ap_list": [], + "wep_supported": false + }, + "qs_component_nego": { + "component_list": [ + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "firmware", + "ver_code": 2 + } + ], + "extra_info": { + "device_model": "L510", + "device_type": "SMART.TAPOBULB" + } + } +} diff --git a/kasa/tests/fixtures/smart/L530E(EU)_3.0_1.1.0.json b/kasa/tests/fixtures/smart/L530E(EU)_3.0_1.1.0.json new file mode 100644 index 000000000..50c41c182 --- /dev/null +++ b/kasa/tests/fixtures/smart/L530E(EU)_3.0_1.1.0.json @@ -0,0 +1,440 @@ +{ + "component_nego": { + "component_list": [ + { + "id": "device", + "ver_code": 2 + }, + { + "id": "firmware", + "ver_code": 2 + }, + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "time", + "ver_code": 1 + }, + { + "id": "wireless", + "ver_code": 1 + }, + { + "id": "schedule", + "ver_code": 2 + }, + { + "id": "countdown", + "ver_code": 2 + }, + { + "id": "antitheft", + "ver_code": 1 + }, + { + "id": "account", + "ver_code": 1 + }, + { + "id": "synchronize", + "ver_code": 1 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "cloud_connect", + "ver_code": 1 + }, + { + "id": "default_states", + "ver_code": 1 + }, + { + "id": "preset", + "ver_code": 1 + }, + { + "id": "brightness", + "ver_code": 1 + }, + { + "id": "color", + "ver_code": 1 + }, + { + "id": "color_temperature", + "ver_code": 1 + }, + { + "id": "auto_light", + "ver_code": 1 + }, + { + "id": "on_off_gradually", + "ver_code": 1 + }, + { + "id": "device_local_time", + "ver_code": 1 + }, + { + "id": "light_effect", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "bulb_quick_control", + "ver_code": 1 + }, + { + "id": "localSmart", + "ver_code": 1 + } + ] + }, + "discovery_result": { + "device_id": "00000000000000000000000000000000", + "device_model": "L530E(EU)", + "device_type": "SMART.TAPOBULB", + "factory_default": false, + "ip": "127.0.0.123", + "is_support_iot_cloud": true, + "mac": "00-00-00-00-00-00", + "mgt_encrypt_schm": { + "encrypt_type": "KLAP", + "http_port": 80, + "is_support_https": false, + "lv": 2 + }, + "obd_src": "tplink", + "owner": "00000000000000000000000000000000" + }, + "get_antitheft_rules": { + "antitheft_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_auto_light_info": { + "enable": false, + "mode": "light_track" + }, + "get_auto_update_info": { + "enable": false, + "random_range": 120, + "time": 180 + }, + "get_connect_cloud_state": { + "status": 0 + }, + "get_countdown_rules": { + "countdown_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_device_info": { + "avatar": "bulb", + "brightness": 50, + "color_temp": 2700, + "color_temp_range": [ + 2500, + 6500 + ], + "default_states": { + "re_power_type": "last_states", + "state": { + "brightness": 50, + "color_temp": 2700, + "hue": 0, + "saturation": 100 + }, + "type": "last_states" + }, + "device_id": "0000000000000000000000000000000000000000", + "device_on": false, + "dynamic_light_effect_enable": false, + "fw_id": "00000000000000000000000000000000", + "fw_ver": "1.1.0 Build 230823 Rel.163903", + "has_set_location_info": true, + "hue": 0, + "hw_id": "00000000000000000000000000000000", + "hw_ver": "3.0", + "ip": "127.0.0.123", + "lang": "en_US", + "latitude": 0, + "longitude": 0, + "mac": "00-00-00-00-00-00", + "model": "L530", + "nickname": "I01BU0tFRF9OQU1FIw==", + "oem_id": "00000000000000000000000000000000", + "overheated": false, + "region": "Europe/London", + "rssi": -68, + "saturation": 100, + "signal_level": 2, + "specs": "", + "ssid": "I01BU0tFRF9TU0lEIw==", + "time_diff": 0, + "type": "SMART.TAPOBULB" + }, + "get_device_time": { + "region": "Europe/London", + "time_diff": 0, + "timestamp": 1703329217 + }, + "get_device_usage": { + "power_usage": { + "past30": 167, + "past7": 167, + "today": 0 + }, + "saved_power": { + "past30": 949, + "past7": 949, + "today": 0 + }, + "time_usage": { + "past30": 1116, + "past7": 1116, + "today": 0 + } + }, + "get_dynamic_light_effect_rules": { + "enable": false, + "max_count": 2, + "rule_list": [ + { + "change_mode": "direct", + "change_time": 3000, + "color_status_list": [ + [ + 100, + 0, + 0, + 2700 + ], + [ + 100, + 321, + 99, + 0 + ], + [ + 100, + 196, + 99, + 0 + ], + [ + 100, + 6, + 97, + 0 + ], + [ + 100, + 160, + 100, + 0 + ], + [ + 100, + 274, + 95, + 0 + ], + [ + 100, + 48, + 100, + 0 + ], + [ + 100, + 242, + 99, + 0 + ] + ], + "id": "L1", + "scene_name": "cGFydHky" + }, + { + "change_mode": "bln", + "change_time": 2000, + "color_status_list": [ + [ + 100, + 54, + 6, + 0 + ], + [ + 100, + 19, + 39, + 0 + ], + [ + 100, + 194, + 52, + 0 + ], + [ + 100, + 324, + 24, + 0 + ], + [ + 100, + 170, + 34, + 0 + ], + [ + 100, + 276, + 27, + 0 + ], + [ + 100, + 56, + 46, + 0 + ], + [ + 100, + 221, + 36, + 0 + ] + ], + "id": "L2", + "scene_name": "" + } + ], + "start_index": 0, + "sum": 2 + }, + "get_fw_download_state": { + "auto_upgrade": false, + "download_progress": 0, + "reboot_time": 5, + "status": 0, + "upgrade_time": 5 + }, + "get_inherit_info": null, + "get_latest_fw": { + "fw_size": 0, + "fw_ver": "1.1.0 Build 230823 Rel.163903", + "hw_id": "", + "need_to_upgrade": false, + "oem_id": "", + "release_date": "", + "release_note": "", + "type": 0 + }, + "get_next_event": {}, + "get_on_off_gradually_info": { + "enable": false + }, + "get_preset_rules": { + "states": [ + { + "brightness": 50, + "color_temp": 2700, + "hue": 0, + "saturation": 100 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 240, + "saturation": 100 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 0, + "saturation": 100 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 120, + "saturation": 100 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 277, + "saturation": 86 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 60, + "saturation": 100 + }, + { + "brightness": 100, + "color_temp": 0, + "hue": 300, + "saturation": 100 + } + ] + }, + "get_schedule_rules": { + "enable": false, + "rule_list": [], + "schedule_rule_max_count": 32, + "start_index": 0, + "sum": 0 + }, + "get_wireless_scan_info": { + "ap_list": [], + "wep_supported": false + }, + "qs_component_nego": { + "component_list": [ + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "firmware", + "ver_code": 2 + } + ], + "extra_info": { + "device_model": "L530", + "device_type": "SMART.TAPOBULB", + "is_klap": true + } + } +} diff --git a/kasa/tests/fixtures/smart/P110(UK)_1.0_1.3.0.json b/kasa/tests/fixtures/smart/P110(UK)_1.0_1.3.0.json index 9033e8002..f57dac10a 100644 --- a/kasa/tests/fixtures/smart/P110(UK)_1.0_1.3.0.json +++ b/kasa/tests/fixtures/smart/P110(UK)_1.0_1.3.0.json @@ -104,8 +104,38 @@ "obd_src": "tplink", "owner": "00000000000000000000000000000000" }, + "get_antitheft_rules": { + "antitheft_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, + "get_auto_off_config": { + "delay_min": 120, + "enable": false + }, + "get_auto_update_info": { + "enable": true, + "random_range": 120, + "time": 180 + }, + "get_connect_cloud_state": { + "status": 0 + }, + "get_countdown_rules": { + "countdown_rule_max_count": 1, + "enable": false, + "rule_list": [] + }, "get_current_power": { - "current_power": 0 + "current_power": 11 + }, + "get_delay_action_info": { + "delay_trigger": "single_click", + "enabled": false, + "state": { + "on": false + }, + "time": 0 }, "get_device_info": { "auto_off_remain_time": 0, @@ -130,12 +160,12 @@ "model": "P110", "nickname": "I01BU0tFRF9OQU1FIw==", "oem_id": "00000000000000000000000000000000", - "on_time": 119335, + "on_time": 5954, "overcurrent_status": "normal", "overheated": false, "power_protection_status": "normal", "region": "Europe/London", - "rssi": -57, + "rssi": -53, "signal_level": 2, "specs": "", "ssid": "I01BU0tFRF9TU0lEIw==", @@ -145,36 +175,271 @@ "get_device_time": { "region": "Europe/London", "time_diff": 0, - "timestamp": 1701370224 + "timestamp": 1703238391 }, "get_device_usage": { "power_usage": { - "past30": 75, - "past7": 69, - "today": 0 + "past30": 700, + "past7": 630, + "today": 18 }, "saved_power": { - "past30": 2029, - "past7": 1964, - "today": 1130 + "past30": 27622, + "past7": 6168, + "today": 81 }, "time_usage": { - "past30": 2104, - "past7": 2033, - "today": 1130 + "past30": 28322, + "past7": 6798, + "today": 99 } }, + "get_electricity_price_config": { + "constant_price": 0, + "time_of_use_config": { + "summer": { + "midpeak": 0, + "offpeak": 0, + "onpeak": 0, + "period": [ + 0, + 0, + 0, + 0 + ], + "weekday_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ], + "weekend_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ] + }, + "winter": { + "midpeak": 0, + "offpeak": 0, + "onpeak": 0, + "period": [ + 0, + 0, + 0, + 0 + ], + "weekday_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ], + "weekend_config": [ + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1, + 1 + ] + } + }, + "type": "constant" + }, "get_energy_usage": { - "current_power": 0, + "current_power": 11196, "electricity_charge": [ 0, 0, 0 ], - "local_time": "2023-11-30 18:50:24", - "month_energy": 75, - "month_runtime": 2104, - "today_energy": 0, - "today_runtime": 1130 + "local_time": "2023-12-22 09:46:31", + "month_energy": 630, + "month_runtime": 25961, + "today_energy": 18, + "today_runtime": 99 + }, + "get_fw_download_state": { + "auto_upgrade": false, + "download_progress": 0, + "reboot_time": 5, + "status": 0, + "upgrade_time": 5 + }, + "get_latest_fw": { + "fw_size": 0, + "fw_ver": "1.3.0 Build 230905 Rel.152200", + "hw_id": "", + "need_to_upgrade": false, + "oem_id": "", + "release_date": "", + "release_note": "", + "type": 0 + }, + "get_led_info": { + "led_rule": "always", + "led_status": true, + "night_mode": { + "end_time": 486, + "night_mode_type": "sunrise_sunset", + "start_time": 955, + "sunrise_offset": 0, + "sunset_offset": 0 + } + }, + "get_max_power": { + "max_power": 3342 + }, + "get_next_event": { + "desired_states": { + "on": true + }, + "e_time": 0, + "id": "S1", + "s_time": 1703245800, + "type": 1 + }, + "get_protection_power": { + "enabled": true, + "protection_power": 2960 + }, + "get_schedule_rules": { + "enable": true, + "rule_list": [ + { + "day": 22, + "desired_states": { + "on": true + }, + "e_action": "none", + "e_min": 0, + "e_type": "normal", + "enable": true, + "id": "S1", + "mode": "repeat", + "month": 12, + "s_min": 710, + "s_type": "normal", + "time_offset": 0, + "week_day": 127, + "year": 2023 + } + ], + "schedule_rule_max_count": 32, + "start_index": 0, + "sum": 1 + }, + "get_wireless_scan_info": { + "ap_list": [], + "wep_supported": false + }, + "qs_component_nego": { + "component_list": [ + { + "id": "quick_setup", + "ver_code": 3 + }, + { + "id": "sunrise_sunset", + "ver_code": 1 + }, + { + "id": "iot_cloud", + "ver_code": 1 + }, + { + "id": "inherit", + "ver_code": 1 + }, + { + "id": "firmware", + "ver_code": 2 + } + ], + "extra_info": { + "device_model": "P110", + "device_type": "SMART.TAPOPLUG", + "is_klap": true + } } } diff --git a/kasa/tests/newfakes.py b/kasa/tests/newfakes.py index 13d11d3d9..95bb9e3a3 100644 --- a/kasa/tests/newfakes.py +++ b/kasa/tests/newfakes.py @@ -1,9 +1,9 @@ import copy import logging import re -import warnings from json import loads as json_loads +import pytest from voluptuous import ( REMOVE_EXTRA, All, @@ -331,16 +331,21 @@ async def send(self, request: str): def _send_request(self, request_dict: dict): method = request_dict["method"] params = request_dict["params"] - if method == "component_nego" or method[:4] == "get_": + if method in ["component_nego", "qs_component_nego"] or method[:4] == "get_": if method in self.info: return {"result": self.info[method], "error_code": 0} else: - warnings.warn( - UserWarning( - f"Fixture missing expected method {method}, try to regenerate" - ), - stacklevel=1, + fixture_name = ( + self.info["discovery_result"]["device_model"] + + "_" + + self.info["get_device_info"]["hw_ver"] + + "_" + + self.info["get_device_info"]["fw_ver"].split(" ", maxsplit=1)[0] ) + if fixture_name not in pytest.fixtures_missing_methods: + pytest.fixtures_missing_methods[fixture_name] = set() + pytest.fixtures_missing_methods[fixture_name].add(method) + return {"result": {}, "error_code": 0} elif method[:4] == "set_": target_method = f"get_{method[4:]}"