chore: add local api test coverage (#284)

Author: allenporter

tests/conftest.py

@@ -1,6 +1,8 @@
+import asyncio
 import io
 import logging
 import re
+from asyncio import Protocol
 from collections.abc import Callable, Generator
 from queue import Queue
 from typing import Any
@@ -11,8 +13,9 @@
 
 from roborock import HomeData, UserData
 from roborock.containers import DeviceData
+from roborock.version_1_apis.roborock_local_client_v1 import RoborockLocalClientV1
 from roborock.version_1_apis.roborock_mqtt_client_v1 import RoborockMqttClientV1
-from tests.mock_data import HOME_DATA_RAW, USER_DATA
+from tests.mock_data import HOME_DATA_RAW, TEST_LOCAL_API_HOST, USER_DATA
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -191,3 +194,43 @@ def mock_rest() -> aioresponses:
             payload={"api": None, "code": 200, "result": HOME_DATA_RAW, "status": "ok", "success": True},
         )
         yield mocked
+
+
+@pytest.fixture(name="mock_create_local_connection")
+def create_local_connection_fixture(request_handler: RequestHandler) -> Generator[None, None, None]:
+    """Fixture that overrides the transport creation to wire it up to the mock socket."""
+
+    async def create_connection(protocol_factory: Callable[[], Protocol], *args) -> tuple[Any, Any]:
+        protocol = protocol_factory()
+
+        def handle_write(data: bytes) -> None:
+            _LOGGER.debug("Received: %s", data)
+            response = request_handler(data)
+            if response is not None:
+                _LOGGER.debug("Replying with %s", response)
+                loop = asyncio.get_running_loop()
+                loop.call_soon(protocol.data_received, response)
+
+        closed = asyncio.Event()
+
+        mock_transport = Mock()
+        mock_transport.write = handle_write
+        mock_transport.close = closed.set
+        mock_transport.is_reading = lambda: not closed.is_set()
+
+        return (mock_transport, "proto")
+
+    with patch("roborock.api.get_running_loop_or_create_one") as mock_loop:
+        mock_loop.return_value.create_connection.side_effect = create_connection
+        yield
+
+
+@pytest.fixture(name="local_client")
+def local_client_fixture(mock_create_local_connection: None) -> Generator[RoborockLocalClientV1, None, None]:
+    home_data = HomeData.from_dict(HOME_DATA_RAW)
+    device_info = DeviceData(
+        device=home_data.devices[0],
+        model=home_data.products[0].model,
+        host=TEST_LOCAL_API_HOST,
+    )
+    yield RoborockLocalClientV1(device_info)

tests/mock_data.py

@@ -347,3 +347,4 @@
 GET_CODE_RESPONSE = {"code": 200, "msg": "success", "data": None}
 HASHED_USER = hashlib.md5((USER_ID + ":" + K_VALUE).encode()).hexdigest()[2:10]
 MQTT_PUBLISH_TOPIC = f"rr/m/o/{USER_ID}/{HASHED_USER}/{PRODUCT_ID}"
+TEST_LOCAL_API_HOST = "1.1.1.1"

tests/test_local_api_v1.py

@@ -0,0 +1,37 @@
+"""Tests for the Roborock Local Client V1."""
+
+from queue import Queue
+
+from roborock.protocol import MessageParser
+from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
+from roborock.version_1_apis import RoborockLocalClientV1
+
+from .mock_data import LOCAL_KEY
+
+
+def build_rpc_response(protocol: RoborockMessageProtocol, seq: int) -> bytes:
+    """Build an encoded RPC response message."""
+    message = RoborockMessage(
+        protocol=protocol,
+        random=23,
+        seq=seq,
+        payload=b"ignored",
+    )
+    return MessageParser.build(message, local_key=LOCAL_KEY)
+
+
+async def test_async_connect(
+    local_client: RoborockLocalClientV1,
+    received_requests: Queue,
+    response_queue: Queue,
+):
+    """Test that we can connect to the Roborock device."""
+    response_queue.put(build_rpc_response(RoborockMessageProtocol.HELLO_RESPONSE, 1))
+    response_queue.put(build_rpc_response(RoborockMessageProtocol.PING_RESPONSE, 2))
+
+    await local_client.async_connect()
+    assert local_client.is_connected()
+    assert received_requests.qsize() == 2
+
+    await local_client.async_disconnect()
+    assert not local_client.is_connected()