python-kasa/tests/fakeprotocol_smart.py at patch · python-kasa/python-kasa

580 lines (528 loc) · 24.5 KB
import copy
from json import loads as json_loads
from warnings import warn
import pytest
from kasa import Credentials, DeviceConfig, SmartProtocol
from kasa.exceptions import SmartErrorCode
from kasa.smart import SmartChildDevice
from kasa.transports.basetransport import BaseTransport
class FakeSmartProtocol(SmartProtocol):
    def __init__(self, info, fixture_name, *, is_child=False, verbatim=False):
        super().__init__(
            transport=FakeSmartTransport(
                info, fixture_name, is_child=is_child, verbatim=verbatim
    async def query(self, request, retry_count: int = 3):
        """Implement query here so can still patch SmartProtocol.query."""
        resp_dict = await self._query(request, retry_count)
        return resp_dict
class FakeSmartTransport(BaseTransport):
    def __init__(
        self,
        info,
        fixture_name,
        list_return_size=10,
        component_nego_not_included=False,
        warn_fixture_missing_methods=True,
        fix_incomplete_fixture_lists=True,
        is_child=False,
        get_child_fixtures=True,
        verbatim=False,
        super().__init__(
            config=DeviceConfig(
                credentials=Credentials(
                    username="dummy_user",
                    password="dummy_password",  # noqa: S106
        self.fixture_name = fixture_name
        # Don't copy the dict if the device is a child so that updates on the
        # child are then still reflected on the parent's lis of child device in
        if not is_child:
            self.info = copy.deepcopy(info)
            if get_child_fixtures:
                self.child_protocols = self._get_child_protocols(
                    self.info, self.fixture_name, "get_child_device_list"
        else:
            self.info = info
        if not component_nego_not_included:
            self.components = {
                comp["id"]: comp["ver_code"]
                for comp in self.info["component_nego"]["component_list"]
        self.list_return_size = list_return_size
        self.warn_fixture_missing_methods = warn_fixture_missing_methods
        self.fix_incomplete_fixture_lists = fix_incomplete_fixture_lists
        # When True verbatim will bypass any extra processing of missing
        # methods and is used to test the fixture creation itself.
        self.verbatim = verbatim
        if verbatim:
            self.warn_fixture_missing_methods = False
            self.fix_incomplete_fixture_lists = False
    @property
    def default_port(self):
        """Default port for the transport."""
        return 80
    @property
    def credentials_hash(self):
        """The hashed credentials used by the transport."""
        return self._credentials.username + self._credentials.password + "hash"
    FIXTURE_MISSING_MAP = {
        "get_wireless_scan_info": ("wireless", {"ap_list": [], "wep_supported": False}),
        "get_auto_off_config": ("auto_off", {"delay_min": 10, "enable": False}),
        "get_led_info": (
            "led",
                "led_rule": "never",
                "led_status": False,
                "night_mode": {
                    "end_time": 420,
                    "night_mode_type": "sunrise_sunset",
                    "start_time": 1140,
                    "sunrise_offset": 0,
                    "sunset_offset": 0,
        "get_latest_fw": (
            "firmware",
                "fw_size": 0,
                "fw_ver": "1.0.5 Build 230801 Rel.095702",
                "hw_id": "",
                "need_to_upgrade": False,
                "oem_id": "",
                "release_date": "",
                "release_note": "",
                "type": 0,
        "get_auto_update_info": (
            "firmware",
            {"enable": True, "random_range": 120, "time": 180},
        "get_alarm_configure": (
            "alarm",
                "get_alarm_configure": {
                    "duration": 10,
                    "type": "Doorbell Ring 2",
                    "volume": "low",
        "get_support_alarm_type_list": (
            "alarm",
                "alarm_type_list": [
                    "Doorbell Ring 1",
        "get_device_usage": ("device", {}),
        "get_connect_cloud_state": ("cloud_connect", {"status": 0}),
        "get_emeter_data": (
            "energy_monitoring",
                "current_ma": 33,
                "energy_wh": 971,
                "power_mw": 1003,
                "voltage_mv": 121215,
        "get_emeter_vgain_igain": (
            "energy_monitoring",
            {"igain": 10861, "vgain": 118657},
    async def send(self, request: str):
        request_dict = json_loads(request)
        method = request_dict["method"]
        if method == "multipleRequest":
            params = request_dict["params"]
            responses = []
            for request in params["requests"]:
                response = await self._send_request(request)  # type: ignore[arg-type]
                # Devices do not continue after error
                if response["error_code"] != 0:
                response["method"] = request["method"]  # type: ignore[index]
                responses.append(response)
            return {"result": {"responses": responses}, "error_code": 0}
        else:
            return await self._send_request(request_dict)
    @staticmethod
    def _get_child_protocols(
        parent_fixture_info, parent_fixture_name, child_devices_key
        child_infos = parent_fixture_info.get(child_devices_key, {}).get(
            "child_device_list", []
        if not child_infos:
            return
        found_child_fixture_infos = []
        child_protocols = {}
        # imported here to avoid circular import
        from .conftest import filter_fixtures
        def try_get_child_fixture_info(child_dev_info):
            hw_version = child_dev_info["hw_ver"]
            sw_version = child_dev_info["fw_ver"]
            sw_version = sw_version.split(" ")[0]
            model = child_dev_info["model"]
            region = child_dev_info.get("specs", "XX")
            child_fixture_name = f"{model}({region})_{hw_version}_{sw_version}"
            child_fixtures = filter_fixtures(
                "Child fixture",
                protocol_filter={"SMART.CHILD"},
                model_filter={child_fixture_name},
            if child_fixtures:
                return next(iter(child_fixtures))
            return None
        for child_info in child_infos:
            if (  # Is SMART protocol
                (device_id := child_info.get("device_id"))
                and (category := child_info.get("category"))
                and category in SmartChildDevice.CHILD_DEVICE_TYPE_MAP
                if fixture_info_tuple := try_get_child_fixture_info(child_info):
                    child_fixture = copy.deepcopy(fixture_info_tuple.data)
                    child_fixture["get_device_info"]["device_id"] = device_id
                    found_child_fixture_infos.append(child_fixture["get_device_info"])
                    child_protocols[device_id] = FakeSmartProtocol(
                        child_fixture, fixture_info_tuple.name, is_child=True
                # Look for fixture inline
                elif (child_fixtures := parent_fixture_info.get("child_devices")) and (
                    child_fixture := child_fixtures.get(device_id)
                    found_child_fixture_infos.append(child_fixture["get_device_info"])
                    child_protocols[device_id] = FakeSmartProtocol(
                        child_fixture,
                        f"{parent_fixture_name}-{device_id}",
                        is_child=True,
                else:
                    pytest.fixtures_missing_methods.setdefault(  # type: ignore[attr-defined]
                        parent_fixture_name, set()
                    ).add("child_devices")
            else:
                warn(
                    f"Child is a cameraprotocol which needs to be implemented {child_info}",
                    stacklevel=2,
        # Replace parent child infos with the infos from the child fixtures so
        # that updates update both
        if child_infos and found_child_fixture_infos:
            parent_fixture_info[child_devices_key]["child_device_list"] = (
                found_child_fixture_infos
        return child_protocols
    async def _handle_control_child(self, params: dict):
        """Handle control_child command."""
        device_id = params.get("device_id")
        if device_id not in self.child_protocols:
            # no need to warn as the warning was raised during protocol init
            return self._handle_control_child_missing(params)
        child_protocol: SmartProtocol = self.child_protocols[device_id]
        request_data = params.get("requestData", {})
        child_method = request_data.get("method")
        child_params = request_data.get("params")  # noqa: F841
        resp = await child_protocol.query({child_method: child_params})
        resp["error_code"] = 0
        for val in resp.values():
            return {
                "result": {"responseData": {"result": val, "error_code": 0}},
                "error_code": 0,
    def _handle_control_child_missing(self, params: dict):
        """Handle control_child command.
        Used for older fixtures where child info wasn't stored in the fixture.
        TODO: Should be removed somehow for future maintanability.
        """
        device_id = params.get("device_id")
        request_data = params.get("requestData", {})
        child_method = request_data.get("method")
        child_params = request_data.get("params")
        info = self.info
        children = info["get_child_device_list"]["child_device_list"]
        for child in children:
            if child["device_id"] == device_id:
                info = child
                break
        # Create the child_devices fixture section for fixtures generated before it was added
        if "child_devices" not in self.info:
            self.info["child_devices"] = {}
        # Get the method calls made directly on the child devices
        child_device_calls = self.info["child_devices"].setdefault(device_id, {})
        # We only support get & set device info in this method for missing.
        if child_method == "get_device_info":
            result = copy.deepcopy(info)
            return {"result": result, "error_code": 0}
        elif child_method == "set_device_info":
            info.update(child_params)
            return {"error_code": 0}
        elif child_method == "set_preset_rules":
            return self._set_child_preset_rules(info, child_params)
        elif child_method == "set_on_off_gradually_info":
            return self._set_on_off_gradually_info(info, child_params)
        elif child_method in child_device_calls:
            result = copy.deepcopy(child_device_calls[child_method])
            return {"result": result, "error_code": 0}
        elif (
            # FIXTURE_MISSING is for service calls not in place when
            # SMART fixtures started to be generated
            missing_result := self.FIXTURE_MISSING_MAP.get(child_method)
        ) and missing_result[0] in self.components:
            # Copy to info so it will work with update methods
            child_device_calls[child_method] = copy.deepcopy(missing_result[1])
            result = copy.deepcopy(info[child_method])
            retval = {"result": result, "error_code": 0}
            return retval
        elif child_method[:4] == "set_":
            target_method = f"get_{child_method[4:]}"
            if target_method not in child_device_calls:
                raise RuntimeError(
                    f"No {target_method} in child info, calling set before get not supported."
            child_device_calls[target_method].update(child_params)
            return {"error_code": 0}
        else:
            # PARAMS error returned for KS240 when get_device_usage called
            # on parent device.  Could be any error code though.
            # TODO: Try to figure out if there's a way to prevent the KS240 smartdevice
            # calling the unsupported device in the first place.
            retval = {
                "error_code": SmartErrorCode.PARAMS_ERROR.value,
                "method": child_method,
            return retval
        raise NotImplementedError(f"Method {child_method} not implemented for children")
    def _get_on_off_gradually_info(self, info, params):
        if self.components["on_off_gradually"] == 1:
            info["get_on_off_gradually_info"] = {"enable": True}
        else:
            info["get_on_off_gradually_info"] = {
                "off_state": {"duration": 5, "enable": False, "max_duration": 60},
                "on_state": {"duration": 5, "enable": False, "max_duration": 60},
        return copy.deepcopy(info["get_on_off_gradually_info"])
    def _set_on_off_gradually_info(self, info, params):
        # Child devices can have the required properties directly in info
        # the _handle_control_child_missing directly passes in get_device_info
        sys_info = info.get("get_device_info", info)
        if self.components["on_off_gradually"] == 1:
            info["get_on_off_gradually_info"] = {"enable": params["enable"]}
        elif on_state := params.get("on_state"):
            if "fade_on_time" in sys_info and "gradually_on_mode" in sys_info:
                sys_info["gradually_on_mode"] = 1 if on_state["enable"] else 0
                if "duration" in on_state:
                    sys_info["fade_on_time"] = on_state["duration"]
            if "get_on_off_gradually_info" in info:
                info["get_on_off_gradually_info"]["on_state"]["enable"] = on_state[
                    "enable"
                if "duration" in on_state:
                    info["get_on_off_gradually_info"]["on_state"]["duration"] = (
                        on_state["duration"]
        elif off_state := params.get("off_state"):
            if "fade_off_time" in sys_info and "gradually_off_mode" in sys_info:
                sys_info["gradually_off_mode"] = 1 if off_state["enable"] else 0
                if "duration" in off_state:
                    sys_info["fade_off_time"] = off_state["duration"]
            if "get_on_off_gradually_info" in info:
                info["get_on_off_gradually_info"]["off_state"]["enable"] = off_state[
                    "enable"
                if "duration" in off_state:
                    info["get_on_off_gradually_info"]["off_state"]["duration"] = (
                        off_state["duration"]
        return {"error_code": 0}
    def _set_dynamic_light_effect(self, info, params):
        """Set or remove values as per the device behaviour."""
        info["get_device_info"]["dynamic_light_effect_enable"] = params["enable"]
        info["get_dynamic_light_effect_rules"]["enable"] = params["enable"]
        if params["enable"]:
            info["get_device_info"]["dynamic_light_effect_id"] = params["id"]
            info["get_dynamic_light_effect_rules"]["current_rule_id"] = params["id"]
        else:
            if "dynamic_light_effect_id" in info["get_device_info"]:
                del info["get_device_info"]["dynamic_light_effect_id"]
            if "current_rule_id" in info["get_dynamic_light_effect_rules"]:
                del info["get_dynamic_light_effect_rules"]["current_rule_id"]
    def _set_edit_dynamic_light_effect_rule(self, info, params):
        """Edit dynamic light effect rule."""
        rules = info["get_dynamic_light_effect_rules"]["rule_list"]
        for rule in rules:
            if rule["id"] == params["id"]:
                rule.update(params)
                return
        raise Exception("Unable to find rule with id")
    def _set_light_strip_effect(self, info, params):
        """Set or remove values as per the device behaviour."""
        # Brightness is not always available
        if (brightness := params.get("brightness")) is not None:
            info["get_device_info"]["lighting_effect"]["brightness"] = brightness
        if "enable" in params:
            info["get_device_info"]["lighting_effect"]["enable"] = params["enable"]
            info["get_device_info"]["lighting_effect"]["name"] = params["name"]
            info["get_device_info"]["lighting_effect"]["id"] = params["id"]
            info["get_lighting_effect"] = copy.deepcopy(params)
    def _set_led_info(self, info, params):
        """Set or remove values as per the device behaviour."""
        info["get_led_info"]["led_status"] = params["led_rule"] != "never"
        info["get_led_info"]["led_rule"] = params["led_rule"]
    def _set_preset_rules(self, info, params):
        """Set or remove values as per the device behaviour."""
        if "brightness" not in info["get_preset_rules"]:
            return {"error_code": SmartErrorCode.PARAMS_ERROR}
        info["get_preset_rules"]["brightness"] = params["brightness"]
        # So far the only child device with light preset (KS240) also has the
        # data available to read in the device_info.
        device_info = info["get_device_info"]
        if "preset_state" in device_info:
            device_info["preset_state"] = [
                {"brightness": b} for b in params["brightness"]
        return {"error_code": 0}
    def _set_child_preset_rules(self, info, params):
        """Set or remove values as per the device behaviour."""
        # So far the only child device with light preset (KS240) has the
        # data available to read in the device_info.  If a child device
        # appears that doesn't have this this will need to be extended.
        if "preset_state" not in info:
            return {"error_code": SmartErrorCode.PARAMS_ERROR}
        info["preset_state"] = [{"brightness": b} for b in params["brightness"]]
        return {"error_code": 0}
    def _edit_preset_rules(self, info, params):
        """Set or remove values as per the device behaviour."""
        if "states" not in info["get_preset_rules"] is None:
            return {"error_code": SmartErrorCode.PARAMS_ERROR}
        info["get_preset_rules"]["states"][params["index"]] = params["state"]
        return {"error_code": 0}
    def _set_temperature_unit(self, info, params):
        """Set or remove values as per the device behaviour."""
        unit = params["temp_unit"]
        if unit not in {"celsius", "fahrenheit"}:
            raise ValueError(f"Invalid value for temperature unit {unit}")
        if "temp_unit" not in info["get_device_info"]:
            return {"error_code": SmartErrorCode.UNKNOWN_METHOD_ERROR}
        else:
            info["get_device_info"]["temp_unit"] = unit
            return {"error_code": 0}
    def _update_sysinfo_key(self, info: dict, key: str, value: str) -> dict:
        """Update a single key in the main system info.
        This is used to implement child device setters that change the main sysinfo state.
        """
        sys_info = info.get("get_device_info", info)
        sys_info[key] = value
        return {"error_code": 0}
    async def _send_request(self, request_dict: dict):
        method = request_dict["method"]
        info = self.info
        if method == "control_child":
            return await self._handle_control_child(request_dict["params"])
        params = request_dict.get("params", {})
        if method in {"component_nego", "qs_component_nego"} or method[:4] == "get_":
            if method in info:
                result = copy.deepcopy(info[method])
                if result and "start_index" in result and "sum" in result:
                    list_key = next(
                        iter([key for key in result if isinstance(result[key], list)])
                    start_index = (
                        start_index
                        if (params and (start_index := params.get("start_index")))
                    # Fixtures generated before _handle_response_lists was implemented
                    # could have incomplete lists.
                        len(result[list_key]) < result["sum"]
                        and self.fix_incomplete_fixture_lists
                        result["sum"] = len(result[list_key])
                        if self.warn_fixture_missing_methods:
                            pytest.fixtures_missing_methods.setdefault(  # type: ignore[attr-defined]
                                self.fixture_name, set()
                            ).add(f"{method} (incomplete '{list_key}' list)")
                    result[list_key] = result[list_key][
                        start_index : start_index + self.list_return_size
                return {"result": result, "error_code": 0}
            if self.verbatim:
                return {
                    "error_code": SmartErrorCode.PARAMS_ERROR.value,
                    "method": method,
                # FIXTURE_MISSING is for service calls not in place when
                # SMART fixtures started to be generated
                missing_result := self.FIXTURE_MISSING_MAP.get(method)
            ) and missing_result[0] in self.components:
                # Copy to info so it will work with update methods
                info[method] = copy.deepcopy(missing_result[1])
                result = copy.deepcopy(info[method])
                retval = {"result": result, "error_code": 0}
            elif (
                method == "get_on_off_gradually_info"
                and "on_off_gradually" in self.components
                # Need to call a method here to determine which version schema to return
                result = self._get_on_off_gradually_info(info, params)
                return {"result": result, "error_code": 0}
            else:
                # PARAMS error returned for KS240 when get_device_usage called
                # on parent device.  Could be any error code though.
                # TODO: Try to figure out if there's a way to prevent the KS240 smartdevice
                # calling the unsupported device in the first place.
                retval = {
                    "error_code": SmartErrorCode.PARAMS_ERROR.value,
                    "method": method,
            # Reduce warning spam by consolidating and reporting at the end of the run
            if self.warn_fixture_missing_methods:
                pytest.fixtures_missing_methods.setdefault(  # type: ignore[attr-defined]
                    self.fixture_name, set()
                ).add(method)
            return retval
        elif method in ["set_qs_info", "fw_download"]:
            return {"error_code": 0}
        elif method == "set_dynamic_light_effect_rule_enable":
            self._set_dynamic_light_effect(info, params)
            return {"error_code": 0}
        elif method == "edit_dynamic_light_effect_rule":
            self._set_edit_dynamic_light_effect_rule(info, params)
            return {"error_code": 0}
        elif method == "set_lighting_effect":
            self._set_light_strip_effect(info, params)
            return {"error_code": 0}
        elif method == "set_led_info":
            self._set_led_info(info, params)
            return {"error_code": 0}
        elif method == "set_preset_rules":
            return self._set_preset_rules(info, params)
        elif method == "edit_preset_rules":
            return self._edit_preset_rules(info, params)
        elif method == "set_temperature_unit":
            return self._set_temperature_unit(info, params)
        elif method == "set_on_off_gradually_info":
            return self._set_on_off_gradually_info(info, params)
        elif method == "set_child_protection":
            return self._update_sysinfo_key(info, "child_protection", params["enable"])
        elif method[:4] == "set_":
            target_method = f"get_{method[4:]}"
            info[target_method].update(params)
            return {"error_code": 0}
    async def close(self) -> None:
        pass
    async def reset(self) -> None:
        pass
Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Uh oh!

FilesExpand file tree

fakeprotocol_smart.py

Latest commit

History

fakeprotocol_smart.py

File metadata and controls
