diff --git a/roborock/testing/__init__.py b/roborock/testing/__init__.py index 4eed0b52..12e86e56 100644 --- a/roborock/testing/__init__.py +++ b/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", ] diff --git a/roborock/testing/b01_q10_simulator.py b/roborock/testing/b01_q10_simulator.py new file mode 100644 index 00000000..859619dc --- /dev/null +++ b/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) diff --git a/roborock/testing/channel.py b/roborock/testing/channel.py index 7ffccc2d..022d449b 100644 --- a/roborock/testing/channel.py +++ b/roborock/testing/channel.py @@ -5,7 +5,8 @@ in-memory replacement for `MqttChannel` and `LocalChannel` during testing. """ -from collections.abc import Awaitable, Callable +import asyncio +from collections.abc import AsyncGenerator, Awaitable, Callable from typing import Any from unittest.mock import AsyncMock, MagicMock @@ -53,7 +54,7 @@ def __init__(self, is_local: bool = False): self._is_local = is_local # A callback to intercept published messages (e.g., bound simulator handler). - # Can be either synchronous or asynchronous: Callable[[RoborockMessage], Awaitable[Any]]. + # 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 @@ -146,3 +147,17 @@ def clear_error(self) -> None: 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() diff --git a/roborock/testing/cloud.py b/roborock/testing/cloud.py index 9fee2bb2..61a0c543 100644 --- a/roborock/testing/cloud.py +++ b/roborock/testing/cloud.py @@ -16,6 +16,7 @@ from roborock.data import HomeData, Reference, RRiot, UserData from roborock.devices.rpc.v1_channel import create_v1_channel as original_create_v1_channel from roborock.devices.transport.mqtt_channel import create_mqtt_channel as original_create_mqtt_channel +from roborock.testing.b01_q10_simulator import Q10VacuumSimulator from roborock.testing.simulator import RoborockDeviceSimulator from roborock.testing.v1_simulator import V1VacuumSimulator @@ -242,11 +243,6 @@ def patch_device_manager(self): # Wrapper function for create_v1_channel def mock_create_v1_channel(user_data, mqtt_params, mqtt_session, device, device_cache): - if device.pv in ("A01", "B01"): - raise NotImplementedError( - f"Simulating protocol {device.pv} is not yet supported. " - "TODO: Implement stateful simulators for B01 (Q7/Q10) and A01 (Zeo/Dyad) devices." - ) server = self.simulated_devices.get(device.duid) if server is not None: if not isinstance(server, V1VacuumSimulator): @@ -255,18 +251,29 @@ def mock_create_v1_channel(user_data, mqtt_params, mqtt_session, device, device_ f"simulator, but create_v1_channel requires a V1VacuumSimulator." ) return server.v1_channel + if device.pv in ("A01", "B01"): + raise NotImplementedError( + f"Simulating protocol {device.pv} is not yet supported. " + "TODO: Implement stateful simulators for B01 (Q7) and A01 (Zeo/Dyad) devices." + ) return original_create_v1_channel(user_data, mqtt_params, mqtt_session, device, device_cache) # Wrapper function for create_mqtt_channel def mock_create_mqtt_channel(user_data, mqtt_params, mqtt_session, device): + server = self.simulated_devices.get(device.duid) + if server: + if device.pv == "B01" and not isinstance(server, Q10VacuumSimulator): + raise NotImplementedError( + f"Simulating protocol {device.pv} is not yet supported. " + "Only Q10VacuumSimulator is supported for B01 protocols currently." + ) + server.mqtt_channel._is_connected = True + return server.mqtt_channel if device.pv in ("A01", "B01"): raise NotImplementedError( f"Simulating protocol {device.pv} is not yet supported. " - "TODO: Implement stateful simulators for B01 (Q7/Q10) and A01 (Zeo/Dyad) devices." + "TODO: Implement stateful simulators for B01 (Q7) and A01 (Zeo/Dyad) devices." ) - server = self.simulated_devices.get(device.duid) - if server: - return server.mqtt_channel return original_create_mqtt_channel(user_data, mqtt_params, mqtt_session, device) # Route Web requests using the dynamic FakeWebApiClient diff --git a/tests/devices/test_device_manager.py b/tests/devices/test_device_manager.py index d5851935..734c2065 100644 --- a/tests/devices/test_device_manager.py +++ b/tests/devices/test_device_manager.py @@ -15,7 +15,7 @@ from roborock.devices.device import RoborockDevice from roborock.devices.device_manager import UserParams, create_device_manager, create_web_api_wrapper from roborock.exceptions import RoborockException, RoborockInvalidCredentials -from roborock.testing import FakeRoborockCloud, V1VacuumSimulator +from roborock.testing import FakeRoborockCloud, Q10VacuumSimulator, V1VacuumSimulator from tests import mock_data USER_DATA = UserData.from_dict(mock_data.USER_DATA) @@ -56,6 +56,83 @@ async def test_no_devices(cloud: FakeRoborockCloud, patch_device_manager: None) assert devices == [] +async def test_with_q10_device(cloud: FakeRoborockCloud, patch_device_manager: None) -> None: + """Test DeviceManager with a Q10 device simulator registered.""" + product = HomeDataProduct( + id="product-id-q10", + name="Roborock Q10", + model="roborock.vacuum.ss07", + category=RoborockCategory.VACUUM, + ) + device_info = HomeDataDevice( + duid="q10_duid", + name="My Q10", + local_key="key123key123key1", + product_id=product.id, + sn="q10_serial", + pv="B01", + ) + + home_data = cloud.web_api.get_default_home_data() + home_data.devices.append(device_info) + home_data.products.append(product) + cloud.web_api.set_homes_response(home_data) + + q10_sim = Q10VacuumSimulator( + duid="q10_duid", + device_info=device_info, + product=product, + ) + cloud.add_device(q10_sim) + + device_manager = await create_device_manager(USER_PARAMS) + devices = await device_manager.get_devices() + + # The setup includes fake_device (V1) by default because of the fake_device fixture + # which is not requested here (we only request cloud and patch_device_manager), + # but the mock EAPI returns the full home_data layout containing our Q10 device. + assert len(devices) == 1 + + q10_device = await device_manager.get_device("q10_duid") + assert q10_device is not None + assert q10_device.name == "My Q10" + assert q10_device.product.model == "roborock.vacuum.ss07" + + # Wait for background connect to establish + for _ in range(20): + if q10_device.is_connected: + break + await asyncio.sleep(0.05) + assert q10_device.is_connected + + assert q10_device.b01_q10_properties is not None + assert q10_device.b01_q10_properties.status.status is None + + await q10_device.b01_q10_properties.refresh() + + for _ in range(20): + if q10_device.b01_q10_properties.status.battery == 100: + break + await asyncio.sleep(0.05) + + assert q10_device.b01_q10_properties.status.battery == 100 + from roborock.data.b01_q10.b01_q10_code_mappings import YXDeviceState + + assert q10_device.b01_q10_properties.status.status == YXDeviceState.CHARGING + + await q10_device.b01_q10_properties.vacuum.start_clean() + + for _ in range(20): + if q10_device.b01_q10_properties.status.status == YXDeviceState.CLEANING: + break + await asyncio.sleep(0.05) + + assert q10_device.b01_q10_properties.status.status == YXDeviceState.CLEANING + assert q10_sim.status[121] == 5 + + await device_manager.close() + + async def test_with_device( cloud: FakeRoborockCloud, fake_device: V1VacuumSimulator, patch_device_manager: None ) -> None: diff --git a/tests/testing/test_q10_simulator.py b/tests/testing/test_q10_simulator.py new file mode 100644 index 00000000..ed6e8416 --- /dev/null +++ b/tests/testing/test_q10_simulator.py @@ -0,0 +1,162 @@ +"""Unit tests for Q10VacuumSimulator.""" + +import json +from typing import Any + +import pytest + +from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol +from roborock.testing import DEFAULT_Q10_STATUS, Q10VacuumSimulator + +B01_VERSION = b"B01" + + +@pytest.fixture(name="q10_sim") +def q10_sim_fixture() -> Q10VacuumSimulator: + """Fixture to create a Q10VacuumSimulator instance.""" + return Q10VacuumSimulator(duid="test_q10_duid") + + +def test_default_status(q10_sim: Q10VacuumSimulator) -> None: + """Test that the simulator initializes with default Q10 status.""" + assert q10_sim.status == DEFAULT_Q10_STATUS + assert q10_sim.status[121] == 8 # Charging + assert q10_sim.status[122] == 100 # Battery + assert isinstance(q10_sim.status[101], dict) + assert q10_sim.status[101][26] == 74 # Volume + + +async def test_request_dps(q10_sim: Q10VacuumSimulator) -> None: + """Test that the simulator responds to REQUEST_DPS with a full status push.""" + notified_messages = [] + + def mock_notify(message: RoborockMessage) -> None: + notified_messages.append(message) + + await q10_sim.mqtt_channel.subscribe(mock_notify) + + # Send REQUEST_DPS command (102) + req_payload: dict[str, Any] = {"dps": {"102": {}}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + + assert len(notified_messages) == 1 + resp_msg = notified_messages[0] + assert resp_msg.protocol == RoborockMessageProtocol.RPC_RESPONSE + assert resp_msg.version == B01_VERSION + assert resp_msg.payload is not None + + resp_payload = json.loads(resp_msg.payload.decode()) + assert "dps" in resp_payload + # The response is keyed by strings representing DP integer codes + assert resp_payload["dps"]["121"] == 8 + assert resp_payload["dps"]["122"] == 100 + # Nested dictionary 101 must also have string keys + assert resp_payload["dps"]["101"]["26"] == 74 + + +async def test_clean_pause_resume_stop(q10_sim: Q10VacuumSimulator) -> None: + """Test clean, pause, resume, and stop command sequences on Q10 simulator.""" + notified_messages = [] + + def mock_notify(message: RoborockMessage) -> None: + notified_messages.append(message) + + await q10_sim.mqtt_channel.subscribe(mock_notify) + + # 1. Start Clean + req_payload: dict[str, Any] = {"dps": {"201": 1}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert q10_sim.status[121] == 5 # Cleaning + assert q10_sim.status[138] == 1 # Standard Clean + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"121": 5, "138": 1} + + # 2. Pause + req_payload = {"dps": {"204": 0}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert q10_sim.status[121] == 10 # Paused + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"121": 10} + + # 3. Resume + req_payload = {"dps": {"205": 0}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert q10_sim.status[121] == 5 # Cleaning + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"121": 5} + + # 4. Stop + req_payload = {"dps": {"206": 0}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert q10_sim.status[121] == 3 # Idle + assert q10_sim.status[138] == 0 # Task type idle + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"121": 3, "138": 0} + + +async def test_setters(q10_sim: Q10VacuumSimulator) -> None: + """Test setting fan speed, water level, and sound volume.""" + notified_messages = [] + + def mock_notify(message: RoborockMessage) -> None: + notified_messages.append(message) + + await q10_sim.mqtt_channel.subscribe(mock_notify) + + # Set Fan level (123) to TURBO (3) + req_payload: dict[str, Any] = {"dps": {"123": 3}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert q10_sim.status[123] == 3 + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"123": 3} + + # Set Volume (26) to 50 + req_payload = {"dps": {"26": 50}} + msg = RoborockMessage( + protocol=RoborockMessageProtocol.RPC_REQUEST, + version=B01_VERSION, + payload=json.dumps(req_payload).encode("utf-8"), + ) + await q10_sim._handle_publish(msg, q10_sim.mqtt_channel) + assert isinstance(q10_sim.status[101], dict) + assert q10_sim.status[101][26] == 50 + # The output payload should map nested 101 dict keys to string + last_msg = notified_messages[-1] + assert last_msg.payload is not None + assert json.loads(last_msg.payload.decode())["dps"] == {"101": {"26": 50}}