From cab12962747598d43b2035a4f5a29d449864d6a8 Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Fri, 30 Oct 2026 20:27:15 -0400 Subject: [PATCH 1/2] feat(testing): serve v1 maps from the simulator Teach V1VacuumSimulator to answer get_map_v1 the way a device does: an encrypted, gzipped binary frame on the map response protocol rather than a JSON result. The simulated map carries a small image, a path that tests grow with advance_path(), and the robot pose, and parses through the real map decoder and renderer. The simulator's security data was never usable for this: the endpoint did not fit the eight byte header field and the nonce was not a valid sixteen byte AES key. Replace both with values that decode. --- roborock/testing/v1_simulator.py | 87 +++++++++++++++++++++++++++++- tests/testing/test_v1_simulator.py | 28 ++++++++++ 2 files changed, 114 insertions(+), 1 deletion(-) diff --git a/roborock/testing/v1_simulator.py b/roborock/testing/v1_simulator.py index 8dfd36091..dcd5f00c3 100644 --- a/roborock/testing/v1_simulator.py +++ b/roborock/testing/v1_simulator.py @@ -4,8 +4,10 @@ machine and JSON RPC commands for V1 vacuum cleaners. """ +import gzip import json import logging +import struct import time from collections.abc import Callable from dataclasses import asdict, replace @@ -37,6 +39,7 @@ ) from roborock.devices.cache import DeviceCache, InMemoryCache from roborock.devices.rpc.v1_channel import V1Channel +from roborock.protocol import Utils from roborock.protocols.v1_protocol import SecurityData from roborock.roborock_message import RoborockDataProtocol, RoborockMessage, RoborockMessageProtocol from roborock.testing.channel import FakeChannel @@ -165,6 +168,31 @@ def _serialize_dataclass(obj: Any) -> dict[str, Any]: ) +# Eight character endpoint, matching the width of the map response header field. +MAP_ENDPOINT = "fakeendp" + +# Side length in cells of the simulated map image. +MAP_IMAGE_SIZE = 32 + +_MAP_FILE_HEADER_LEN = 20 +_IMAGE_BLOCK_TYPE = 2 +_PATH_BLOCK_TYPE = 3 +_POSE_BLOCK_TYPE = 8 +_MOP_BLOCK_TYPE = 18 +_DIGEST_BLOCK_TYPE = 1024 +_MAP_INSIDE = 0xFF + + +def _path_point(index: int) -> bytes: + """A deterministic path point, so tests can assert on exact bytes.""" + return struct.pack(" bytes: + """Frame one map block: type, header length, payload length, header, payload.""" + return struct.pack(" list[dict[str, Any]]: def _handle_get_network_info(self, params: Any) -> dict[str, Any]: return _serialize_dataclass(self.network_info) + # --- Simulated map ------------------------------------------------- + + def advance_path(self, count: int) -> None: + """Extend the current path, as the robot does while it cleans.""" + self.path_length += count + + def path_payload(self, start: int, count: int) -> bytes: + """The point records for a range of the current path.""" + return b"".join(_path_point(i) for i in range(start, start + count)) + + def mop_payload(self, start: int, count: int) -> bytes: + """The mop flags for a range of the current path, alternating per point.""" + return bytes(i % 2 for i in range(start, start + count)) + + def _pose_block(self) -> bytes: + return _block(_POSE_BLOCK_TYPE, struct.pack(" bytes: + """Blocks 3 and 18 for a range of the path, as both a map and a page carry them.""" + return _block(_PATH_BLOCK_TYPE, self.path_payload(start, count), struct.pack(" bytes: + """Build a decompressed map file reflecting the current robot state.""" + image = _block( + _IMAGE_BLOCK_TYPE, + bytes([_MAP_INSIDE]) * (MAP_IMAGE_SIZE * MAP_IMAGE_SIZE), + struct.pack(" RoborockMessage: + """Frame the map as the encrypted binary response the device sends.""" + body = Utils.encrypt_cbc(gzip.compress(self.build_map_bytes()), self.security_data.nonce) + header = struct.pack("<8s8sH6s", self.security_data.endpoint.encode(), bytes(8), request_id & 0xFFFF, bytes(6)) + return RoborockMessage(protocol=RoborockMessageProtocol.MAP_RESPONSE, payload=header + body) + async def _handle_publish(self, message: RoborockMessage, channel: FakeChannel) -> None: if not message.payload: return @@ -350,6 +430,11 @@ async def _handle_publish(self, message: RoborockMessage, channel: FakeChannel) _LOGGER.debug("Failed to parse plaintext JSON RPC payload: %s", e, exc_info=True) return + if method == "get_map_v1": + # The map arrives as an encrypted binary frame rather than a JSON result. + channel.notify_subscribers(self._map_response_message(msg_id)) + return + result = None error = None diff --git a/tests/testing/test_v1_simulator.py b/tests/testing/test_v1_simulator.py index 8883047c6..d48f2e675 100644 --- a/tests/testing/test_v1_simulator.py +++ b/tests/testing/test_v1_simulator.py @@ -274,3 +274,31 @@ async def test_multiple_devices_network_info_override(): await device2.v1_properties.network_info.refresh() assert device2.v1_properties.network_info.ip == "192.168.1.50" assert device2.v1_properties.network_info.ssid == "custom_wifi" + + +# --- Map content --------------------------------------------------------------- + + +def _path_points(device) -> list[tuple[float, float]]: + """The rendered path, as the map card would receive it.""" + map_data = device.v1_properties.map_content.map_data + assert map_data is not None and map_data.path is not None + return [(point.x, point.y) for point in map_data.path.path[0]] + + +async def test_full_map_refresh_from_simulator(): + """The simulated map decodes through the real map channel and parser.""" + cloud = FakeRoborockCloud() + simulator = V1VacuumSimulator(duid="s7_map") + simulator.advance_path(40) + simulator.robot_position = (31000, 32000, 180) + device = await _create_connected_device(cloud, simulator) + + await device.v1_properties.map_content.refresh() + + map_content = device.v1_properties.map_content + assert map_content.image_content is not None + assert map_content.image_content.startswith(b"\x89PNG") + assert len(_path_points(device)) == 40 + assert map_content.map_data is not None and map_content.map_data.vacuum_position is not None + assert (map_content.map_data.vacuum_position.x, map_content.map_data.vacuum_position.y) == (31000, 32000) From cab602e9f97e0af19f0b2d2b60b54dd49c38292e Mon Sep 17 00:00:00 2001 From: Luke Lashley Date: Fri, 30 Oct 2026 21:27:06 -0400 Subject: [PATCH 2/2] fix(testing): generate valid V1 maps and honor custom handlers --- roborock/testing/v1_simulator.py | 45 ++++++++++------- tests/testing/test_v1_simulator.py | 77 ++++++++++++++++++++++++++++++ 2 files changed, 104 insertions(+), 18 deletions(-) diff --git a/roborock/testing/v1_simulator.py b/roborock/testing/v1_simulator.py index dcd5f00c3..3e6b6e43b 100644 --- a/roborock/testing/v1_simulator.py +++ b/roborock/testing/v1_simulator.py @@ -5,6 +5,7 @@ """ import gzip +import hashlib import json import logging import struct @@ -171,10 +172,12 @@ def _serialize_dataclass(obj: Any) -> dict[str, Any]: # Eight character endpoint, matching the width of the map response header field. MAP_ENDPOINT = "fakeendp" -# Side length in cells of the simulated map image. +# Minimum side length in cells of the simulated map image. MAP_IMAGE_SIZE = 32 _MAP_FILE_HEADER_LEN = 20 +_MAP_CELL_SIZE_MM = 50 +_MAP_IMAGE_PADDING = 2 _IMAGE_BLOCK_TYPE = 2 _PATH_BLOCK_TYPE = 3 _POSE_BLOCK_TYPE = 8 @@ -247,6 +250,7 @@ def __init__( "reset_consumable": self._handle_reset_consumable, "app_get_init_status": self._handle_app_get_init_status, "get_network_info": self._handle_get_network_info, + "get_map_v1": lambda params: self.build_map_bytes(), } # The simulated robot's own view of its map. Tests extend the path @@ -393,23 +397,28 @@ def _path_blocks(self, start: int, count: int) -> bytes: def build_map_bytes(self) -> bytes: """Build a decompressed map file reflecting the current robot state.""" + # Full-map image offsets are cells; path and robot coordinates are millimetres. + points = [self.robot_position[:2], *struct.iter_unpack(" RoborockMessage: + def _map_response_message(self, request_id: int, map_bytes: bytes) -> RoborockMessage: """Frame the map as the encrypted binary response the device sends.""" - body = Utils.encrypt_cbc(gzip.compress(self.build_map_bytes()), self.security_data.nonce) + body = Utils.encrypt_cbc(gzip.compress(map_bytes), self.security_data.nonce) header = struct.pack("<8s8sH6s", self.security_data.endpoint.encode(), bytes(8), request_id & 0xFFFF, bytes(6)) return RoborockMessage(protocol=RoborockMessageProtocol.MAP_RESPONSE, payload=header + body) @@ -430,11 +439,6 @@ async def _handle_publish(self, message: RoborockMessage, channel: FakeChannel) _LOGGER.debug("Failed to parse plaintext JSON RPC payload: %s", e, exc_info=True) return - if method == "get_map_v1": - # The map arrives as an encrypted binary frame rather than a JSON result. - channel.notify_subscribers(self._map_response_message(msg_id)) - return - result = None error = None @@ -449,6 +453,11 @@ async def _handle_publish(self, message: RoborockMessage, channel: FakeChannel) else: result = "ok" + if method == "get_map_v1" and error is None and isinstance(result, bytes): + # Custom handlers return decompressed map bytes, just like the default. + channel.notify_subscribers(self._map_response_message(msg_id, result)) + return + response_data = { "dps": {"102": json.dumps({"id": msg_id, "result": result, "error": error})}, "t": int(time.time()), diff --git a/tests/testing/test_v1_simulator.py b/tests/testing/test_v1_simulator.py index d48f2e675..fa5f011e9 100644 --- a/tests/testing/test_v1_simulator.py +++ b/tests/testing/test_v1_simulator.py @@ -1,6 +1,9 @@ +import hashlib +import struct from dataclasses import replace import pytest +from vacuum_map_parser_base.config.drawable import Drawable from roborock.data import UserData from roborock.data.v1 import RoborockStateCode @@ -9,6 +12,7 @@ from roborock.devices.device_manager import UserParams, create_device_manager from roborock.devices.traits.v1.consumeable import ConsumableAttribute from roborock.exceptions import RoborockException +from roborock.map.map_parser import MapParser, MapParserConfig from roborock.testing import ( DEFAULT_NETWORK_INFO, DEFAULT_STATUS, @@ -302,3 +306,76 @@ async def test_full_map_refresh_from_simulator(): assert len(_path_points(device)) == 40 assert map_content.map_data is not None and map_content.map_data.vacuum_position is not None assert (map_content.map_data.vacuum_position.x, map_content.map_data.vacuum_position.y) == (31000, 32000) + + +@pytest.mark.parametrize("count", [0, 40, 400]) +def test_simulated_map_length_and_digest(count: int) -> None: + """Synthetic maps satisfy the app's full-file length and SHA-1 checks.""" + simulator = V1VacuumSimulator() + simulator.advance_path(count) + raw = simulator.build_map_bytes() + magic, header_length, payload_length, major, minor = struct.unpack_from("<2sHIHH", raw) + assert magic == b"rr" + assert (major, minor) == (1, 0) + assert header_length + payload_length == len(raw) + assert struct.unpack_from(" None: + """Map geometry keeps all simulated overlays inside the rendered image.""" + simulator = V1VacuumSimulator() + simulator.robot_position = position + simulator.advance_path(count) + parsed = MapParser(MapParserConfig()).parse(simulator.build_map_bytes()) + assert parsed is not None and parsed.map_data is not None + data = parsed.map_data + assert data.image is not None and data.vacuum_position is not None and data.path is not None + width, height = data.image.data.size + for point in [data.vacuum_position, *data.path.path[0]]: + pixel = point.to_img(data.image.dimensions) + assert 0 <= pixel.x < width + assert 0 <= pixel.y < height + + +@pytest.mark.parametrize("drawable", [Drawable.PATH, Drawable.VACUUM_POSITION]) +def test_simulated_map_renders_overlays(drawable: Drawable) -> None: + """Each overlay changes the image rather than silently being clipped away.""" + simulator = V1VacuumSimulator() + simulator.advance_path(40) + raw = simulator.build_map_bytes() + background = MapParser(MapParserConfig(drawables=[])).parse(raw) + rendered = MapParser(MapParserConfig(drawables=[drawable])).parse(raw) + assert background is not None and rendered is not None + assert rendered.image_content != background.image_content + + +async def test_map_custom_handler_override() -> None: + """A custom map travels through the same binary response path as the default.""" + custom_simulator = V1VacuumSimulator() + custom_simulator.advance_path(7) + custom_simulator.robot_position = (31000, 32000, 180) + custom_map = custom_simulator.build_map_bytes() + + def get_map(params): + assert params == [] + return custom_map + + simulator = V1VacuumSimulator(custom_handlers={"get_map_v1": get_map}) + device = await _create_connected_device(FakeRoborockCloud(), simulator) + await device.v1_properties.map_content.refresh() + assert device.v1_properties.map_content.raw_api_response == custom_map + assert len(_path_points(device)) == 7 + + +async def test_map_custom_handler_error() -> None: + """A failing override emits an RPC error instead of a successful default map.""" + + def get_map(params): + raise RoborockException("map unavailable") + + simulator = V1VacuumSimulator(custom_handlers={"get_map_v1": get_map}) + with pytest.raises(RoborockException, match="map unavailable"): + await simulator.v1_channel.mqtt_rpc_channel.send_command("get_map_v1")