python-kasa/tests/fakeprotocol_smartcam.py

from __future__ import annotations

import copy
from json import loads as json_loads
from typing import Any

from kasa import Credentials, DeviceConfig, SmartProtocol
from kasa.protocols.smartcamprotocol import SmartCamProtocol
from kasa.transports.basetransport import BaseTransport

from .fakeprotocol_smart import FakeSmartTransport


class FakeSmartCamProtocol(SmartCamProtocol):
    def __init__(self, info, fixture_name, *, is_child=False, verbatim=False):
        super().__init__(
            transport=FakeSmartCamTransport(
                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 FakeSmartCamTransport(BaseTransport):
    def __init__(
        self,
        info,
        fixture_name,
        *,
        list_return_size=10,
        is_child=False,
        verbatim=False,
        components_not_included=False,
    ):
        super().__init__(
            config=DeviceConfig(
                "127.0.0.123",
                credentials=Credentials(
                    username="dummy_user",
                    password="dummy_password",  # noqa: S106
                ),
            ),
        )

        self.fixture_name = fixture_name
        # When True verbatim will bypass any extra processing of missing
        # methods and is used to test the fixture creation itself.
        self.verbatim = verbatim
        if not is_child:
            self.info = copy.deepcopy(info)
            self.child_protocols = FakeSmartTransport._get_child_protocols(
                self.info, self.fixture_name, "getChildDeviceList"
            )
        else:
            self.info = info
        # self.child_protocols = self._get_child_protocols()
        self.list_return_size = list_return_size

        # Setting this flag allows tests to create dummy transports without
        # full fixture info for testing specific cases like list handling etc
        self.components_not_included = (components_not_included,)
        if not components_not_included:
            self.components = {
                comp["name"]: comp["version"]
                for comp in self.info["getAppComponentList"]["app_component"][
                    "app_component_list"
                ]
            }

    @property
    def default_port(self):
        """Default port for the transport."""
        return 443

    @property
    def credentials_hash(self):
        """The hashed credentials used by the transport."""
        return self._credentials.username + self._credentials.password + "camerahash"

    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]
                response["method"] = request["method"]  # type: ignore[index]
                responses.append(response)
                # Devices do not continue after error
                if response["error_code"] != 0:
                    break
            return {"result": {"responses": responses}, "error_code": 0}
        else:
            return await self._send_request(request_dict)

    async def _handle_control_child(self, params: dict):
        """Handle control_child command."""
        device_id = params.get("device_id")
        assert device_id in self.child_protocols, "Fixture does not have child info"

        child_protocol: SmartProtocol = self.child_protocols[device_id]

        request_data = params.get("request_data", {})

        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": {"response_data": {"result": val, "error_code": 0}},
                "error_code": 0,
            }

    @staticmethod
    def _get_param_set_value(info: dict, set_keys: list[str], value):
        for key in set_keys[:-1]:
            info = info[key]
        info[set_keys[-1]] = value

    FIXTURE_MISSING_MAP = {
        "getMatterSetupInfo": (
            "matter",
            {
                "setup_code": "00000000000",
                "setup_payload": "00:0000000-0000.00.000",
            },
        )
    }
    # Setters for when there's not a simple mapping of setters to getters
    SETTERS = {
        ("system", "sys", "dev_alias"): [
            "getDeviceInfo",
            "device_info",
            "basic_info",
            "device_alias",
        ],
        # setTimezone maps to getClockStatus
        ("system", "clock_status", "seconds_from_1970"): [
            "getClockStatus",
            "system",
            "clock_status",
            "seconds_from_1970",
        ],
        # setTimezone maps to getClockStatus
        ("system", "clock_status", "local_time"): [
            "getClockStatus",
            "system",
            "clock_status",
            "local_time",
        ],
    }

    @staticmethod
    def _get_second_key(request_dict: dict[str, Any]) -> str:
        assert (
            len(request_dict) == 2
        ), f"Unexpected dict {request_dict}, should be length 2"
        it = iter(request_dict)
        next(it, None)
        return next(it)

    async def _send_request(self, request_dict: dict):
        method = request_dict["method"]

        info = self.info
        if method == "controlChild":
            return await self._handle_control_child(
                request_dict["params"]["childControl"]
            )

        if method[:3] == "set":
            get_method = "g" + method[1:]
            for key, val in request_dict.items():
                if key == "method":
                    continue
                # key is params for multi request and the actual params
                # for single requests
                if key == "params":
                    module = next(iter(val))
                    val = val[module]
                else:
                    module = key
                section = next(iter(val))
                skey_val = val[section]
                if not isinstance(skey_val, dict):  # single level query
                    section_key = section
                    section_val = skey_val
                    if (get_info := info.get(get_method)) and section_key in get_info:
                        get_info[section_key] = section_val
                    else:
                        return {"error_code": -1}
                    break
                for skey, sval in skey_val.items():
                    section_key = skey
                    section_value = sval
                    if setter_keys := self.SETTERS.get((module, section, section_key)):
                        self._get_param_set_value(info, setter_keys, section_value)
                    elif (
                        section := info.get(get_method, {})
                        .get(module, {})
                        .get(section, {})
                    ) and section_key in section:
                        section[section_key] = section_value
                    else:
                        return {"error_code": -1}
                break
            return {"error_code": 0}
        elif method == "get":
            module = self._get_second_key(request_dict)
            get_method = f"get_{module}"
            if get_method in info:
                result = copy.deepcopy(info[get_method]["get"])
                return {**result, "error_code": 0}
            else:
                return {"error_code": -1}
        elif method[:3] == "get":
            params = request_dict.get("params")
            if method in info:
                result = copy.deepcopy(info[method])
                if "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")))
                        else 0
                    )

                    result[list_key] = result[list_key][
                        start_index : start_index + self.list_return_size
                    ]
                return {"result": result, "error_code": 0}
            if (
                # 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])
                return {"result": result, "error_code": 0}

            return {"error_code": -1}
        return {"error_code": -1}

    async def close(self) -> None:
        pass

    async def reset(self) -> None:
        pass