feat: implement stateful Q10 device simulator and integration tests

roborock/testing/__init__.py

@@ -63,6 +63,10 @@ async def test_start_vacuum_service():
 ```
 """
 
+from roborock.testing.b01_q10_simulator import (
+    DEFAULT_Q10_STATUS,
+    Q10VacuumSimulator,
+)
 from roborock.testing.channel import FakeChannel
 from roborock.testing.cloud import FakeRoborockCloud, FakeWebApiClient
 from roborock.testing.simulator import (
@@ -92,10 +96,12 @@ async def test_start_vacuum_service():
     "DEFAULT_LOCAL_KEY",
     "DEFAULT_NETWORK_INFO",
     "DEFAULT_PRODUCT_ID",
+    "DEFAULT_Q10_STATUS",
     "DEFAULT_STATUS",
     "FakeChannel",
     "FakeRoborockCloud",
     "FakeWebApiClient",
+    "Q10VacuumSimulator",
     "RoborockDeviceSimulator",
     "V1VacuumSimulator",
 ]

roborock/testing/b01_q10_simulator.py

@@ -0,0 +1,213 @@
+"""Stateful simulator for Roborock Q10 (B01) devices."""
+
+import copy
+import json
+import logging
+from typing import Any
+
+from roborock.data import HomeDataDevice, HomeDataProduct, RoborockCategory
+from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
+from roborock.testing.simulator import RoborockDeviceSimulator
+
+_LOGGER = logging.getLogger(__name__)
+
+B01_VERSION = b"B01"
+DEFAULT_PRODUCT_ID = "product-id-q10"
+
+DEFAULT_Q10_STATUS = {
+    101: {
+        104: 0,
+        105: False,
+        109: "us",
+        207: 0,
+        25: 1,
+        26: 74,
+        29: 0,
+        30: 0,
+        31: 0,
+        37: 1,
+        40: 1,
+        45: 0,
+        47: 0,
+        50: 0,
+        51: True,
+        53: False,
+        6: 0,
+        60: 1,
+        67: 0,
+        7: 0,
+        76: 0,
+        78: 0,
+        79: {"timeZoneCity": "America/Los_Angeles", "timeZoneSec": -28800},
+        80: 0,
+        81: {"ipAdress": "1.1.1.2", "mac": "99:AA:88:BB:77:CC", "signal": -50, "wifiName": "wifi-network-name"},
+        83: 1,
+        86: 1,
+        87: 100,
+        88: 0,
+        90: 0,
+        92: {"disturb_dust_enable": 1, "disturb_light": 1, "disturb_resume_clean": 1, "disturb_voice": 1},
+        93: 1,
+        96: 0,
+    },
+    121: 8,
+    122: 100,
+    123: 2,
+    124: 1,
+    125: 0,
+    126: 0,
+    127: 0,
+    136: 1,
+    137: 1,
+    138: 0,
+    139: 5,
+}
+
+
+class Q10VacuumSimulator(RoborockDeviceSimulator):
+    """Stateful firmware simulator for Roborock Q10 (B01 ss07) devices."""
+
+    def __init__(
+        self,
+        duid: str,
+        status: dict[int, Any] | None = None,
+        device_info: HomeDataDevice | None = None,
+        product: HomeDataProduct | None = None,
+    ):
+        product = product or HomeDataProduct(
+            id=DEFAULT_PRODUCT_ID,
+            name="Roborock Q10",
+            model="roborock.vacuum.ss07",
+            category=RoborockCategory.VACUUM,
+        )
+        device_info = device_info or HomeDataDevice(
+            duid=duid,
+            name="Roborock Q10",
+            local_key="fake_localkey_16bytes",
+            product_id=product.id,
+            sn="fake_serial_number",
+            pv="B01",
+        )
+        super().__init__(duid, device_info, product, has_local_channel=False)
+        self.status = copy.deepcopy(status or DEFAULT_Q10_STATUS)
+
+    async def _handle_publish(self, message: RoborockMessage, channel: Any) -> None:
+        """Process incoming Q10 RPC command payload."""
+        if not message.payload:
+            return
+
+        try:
+            payload = json.loads(message.payload.decode())
+        except (json.JSONDecodeError, UnicodeDecodeError) as ex:
+            _LOGGER.warning("Simulator failed to parse incoming B01 payload: %s", ex)
+            return
+
+        datapoints = payload.get("dps", {})
+        if not isinstance(datapoints, dict):
+            return
+
+        updated_dps: dict[int, Any] = {}
+
+        for code_str, params in datapoints.items():
+            try:
+                code = int(code_str)
+            except ValueError:
+                continue
+
+            # Command: REQUEST_DPS (102) -> trigger status push
+            if code == 102:
+                # REQUEST_DPS triggers a dump of all status properties
+                # (both root properties and nested properties under 101)
+                self.trigger_push_update()
+                continue
+
+            # Command: START_CLEAN (201)
+            elif code == 201:
+                # 1 = sweep/mop, cmd inside dict = segment clean
+                self.status[121] = 5  # YXDeviceState.CLEANING
+                if isinstance(params, dict) and "cmd" in params:
+                    self.status[138] = params.get("cmd", 0)  # clean_task_type
+                else:
+                    self.status[138] = 1  # CLEANING
+                updated_dps[121] = self.status[121]
+                updated_dps[138] = self.status[138]
+
+            # Command: START_BACK (202)
+            elif code == 202:
+                # 5 = returning to charge
+                self.status[121] = 6  # YXDeviceState.RETURNING_HOME
+                self.status[139] = 5  # BACK_CHARGING
+                updated_dps[121] = self.status[121]
+                updated_dps[139] = self.status[139]
+
+            # Command: START_DOCK_TASK (203)
+            elif code == 203:
+                # Empty dustbin
+                self.status[121] = 22  # EMPTYING_THE_BIN
+                updated_dps[121] = self.status[121]
+
+            # Command: PAUSE (204)
+            elif code == 204:
+                self.status[121] = 10  # YXDeviceState.PAUSED
+                updated_dps[121] = self.status[121]
+
+            # Command: RESUME (205)
+            elif code == 205:
+                self.status[121] = 5  # YXDeviceState.CLEANING
+                updated_dps[121] = self.status[121]
+
+            # Command: STOP (206)
+            elif code == 206:
+                self.status[121] = 3  # YXDeviceState.IDLE
+                self.status[138] = 0  # YXDeviceCleanTask.UNKNOWN/IDLE
+                updated_dps[121] = self.status[121]
+                updated_dps[138] = self.status[138]
+
+            # Command: FAN_LEVEL (123)
+            elif code == 123:
+                self.status[123] = params
+                updated_dps[123] = params
+
+            # Command: WATER_LEVEL (124)
+            elif code == 124:
+                self.status[124] = params
+                updated_dps[124] = params
+
+            # Command: CLEAN_MODE (137)
+            elif code == 137:
+                self.status[137] = params
+                updated_dps[137] = params
+
+            # Custom settings properties routed to sub-traits (DND, child lock, volume, dust collections)
+            elif code in (26, 47, 25, 37, 50):
+                # These live in the nested dpCommon (101) dict
+                common_dict = self.status[101]
+                if isinstance(common_dict, dict):
+                    common_dict[code] = params
+                # We return them nested under 101
+                if 101 not in updated_dps:
+                    updated_dps[101] = {}
+                updated_dps[101][code] = params
+
+        if updated_dps:
+            # Send the changed datapoints back to the client
+            self.push_dps(updated_dps)
+
+    def push_dps(self, dps_updates: dict[int, Any]) -> None:
+        """Push a set of status datapoint changes to the client."""
+        payload = {
+            "dps": {
+                str(k): ({str(sub_k): sub_v for sub_k, sub_v in v.items()} if isinstance(v, dict) and k == 101 else v)
+                for k, v in dps_updates.items()
+            }
+        }
+        message = RoborockMessage(
+            protocol=RoborockMessageProtocol.RPC_RESPONSE,
+            version=B01_VERSION,
+            payload=json.dumps(payload).encode("utf-8"),
+        )
+        self.mqtt_channel.notify_subscribers(message)
+
+    def trigger_push_update(self) -> None:
+        """Send the full status dump to the client."""
+        self.push_dps(self.status)

roborock/testing/channel.py

@@ -5,7 +5,8 @@
 in-memory replacement for `MqttChannel` and `LocalChannel` during testing.
 """
 
-from collections.abc import Callable
+import asyncio
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock
 
@@ -37,6 +38,9 @@ class FakeChannel(Channel):
       publish (useful for low-level RPC request/response testing).
     - **Push unsolicited messages**: Call ``channel.notify_subscribers(msg)``
       to simulate the device broadcasting a state change.
+    - **Intercept published messages**: Register a handler/callback via
+      ``channel.publish_handler = my_handler`` (e.g. stateful simulator)
+      to reactively process commands.
     """
 
     subscribe: Any
@@ -49,6 +53,11 @@ def __init__(self, is_local: bool = False):
         self._is_connected = False
         self._is_local = is_local
 
+        # A callback to intercept published messages (e.g., bound simulator handler).
+        # Must be asynchronous: Callable[[RoborockMessage], Awaitable[Any]].
+        # By default, routes to self._default_publish_handler to handle the response_queue.
+        self.publish_handler: Callable[[RoborockMessage], Awaitable[Any]] | None = self._default_publish_handler
+
         # Set this to an exception instance to make the next publish raise it.
         # This is a convenience shortcut; callers can also replace
         # ``publish.side_effect`` directly for more control.
@@ -96,14 +105,16 @@ def is_local_connected(self) -> bool:
     async def _publish(self, message: RoborockMessage) -> None:
         """Default publish implementation.
 
-        Records the message in ``published_messages`` and, if
-        ``response_queue`` is non-empty, pops the first response and
-        delivers it to all current subscribers (simulating a
-        request/response round-trip).
+        Records the message in ``published_messages`` and executes ``publish_handler``.
         """
         self.published_messages.append(message)
         if self.publish_side_effect:
             raise self.publish_side_effect
+        if self.publish_handler:
+            await self.publish_handler(message)
+
+    async def _default_publish_handler(self, message: RoborockMessage) -> None:
+        """Default handler that pops canned responses from response_queue."""
         if self.response_queue:
             response = self.response_queue.pop(0)
             self.notify_subscribers(response)
@@ -124,3 +135,29 @@ def notify_subscribers(self, message: RoborockMessage) -> None:
         """
         for subscriber in list(self.subscribers):
             subscriber(message)
+
+    def inject_error(self, exception: Exception) -> None:
+        """Inject a transient failure into all channel operations (publish, subscribe, connect)."""
+        self.publish.side_effect = exception
+        self.subscribe.side_effect = exception
+        self.connect.side_effect = exception
+
+    def clear_error(self) -> None:
+        """Restore default success behaviors on all channel operations."""
+        self.publish.side_effect = self._publish
+        self.subscribe.side_effect = self._subscribe
+        self.connect.side_effect = self._connect
+
+    async def subscribe_stream(self) -> AsyncGenerator[RoborockMessage, None]:
+        """Stream messages received via this channel."""
+        queue: asyncio.Queue[RoborockMessage] = asyncio.Queue()
+
+        def callback(message: RoborockMessage) -> None:
+            queue.put_nowait(message)
+
+        unsub = await self.subscribe(callback)
+        try:
+            while True:
+                yield await queue.get()
+        finally:
+            unsub()