diff --git a/roborock/data/b01_q10/b01_q10_code_mappings.py b/roborock/data/b01_q10/b01_q10_code_mappings.py index afbbff451..f0e5756ec 100644 --- a/roborock/data/b01_q10/b01_q10_code_mappings.py +++ b/roborock/data/b01_q10/b01_q10_code_mappings.py @@ -152,6 +152,7 @@ class YXWaterLevel(RoborockModeEnum): class YXCleanLine(RoborockModeEnum): + UNKNOWN = "unknown", -1 FAST = "fast", 0 DAILY = "daily", 1 FINE = "fine", 2 @@ -356,6 +357,35 @@ class RemoteCommand(IntEnum): EXIT = 5 +class Q10RoomFanLevel(RoborockModeEnum): + """Suction levels used inside a Q10 customized-room payload.""" + + UNKNOWN = "unknown", -1 + QUIET = "quiet", 1 + BALANCED = "balanced", 2 + TURBO = "turbo", 3 + MAX = "max", 4 + MAX_PLUS = "max_plus", 5 + + +class Q10CleanCount(RoborockModeEnum): + """Number of passes for a Q10 cleaning task.""" + + UNKNOWN = "unknown", -1 + ONCE = "once", 1 + TWICE = "twice", 2 + THREE_TIMES = "three_times", 3 + + +class Q10RoomCleanType(RoborockModeEnum): + """Work performed in one Q10 customized-room setting.""" + + UNKNOWN = "unknown", -1 + VAC_AND_MOP = "vac_and_mop", 1 + VACUUM = "vacuum", 2 + MOP = "mop", 3 + + class Q10RestrictionType(RoborockEnum): """Restriction types supported by the Q10 map editor.""" diff --git a/roborock/data/b01_q10/b01_q10_containers.py b/roborock/data/b01_q10/b01_q10_containers.py index 90b23fbdf..d101f7305 100644 --- a/roborock/data/b01_q10/b01_q10_containers.py +++ b/roborock/data/b01_q10/b01_q10_containers.py @@ -12,6 +12,9 @@ from ..containers import RoborockBase from .b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, YXAreaUnit, YXBackType, YXCarpetCleanType, @@ -288,6 +291,13 @@ def fault_name(self) -> str | None: """Returns the name of the current fault.""" return self.fault.value if self.fault is not None else None + @property + def clean_count_mode(self) -> Q10CleanCount | None: + """Return the typed cleaning pass count from the reported status.""" + if self.clean_count is None: + return None + return Q10CleanCount.from_code_optional(self.clean_count) or Q10CleanCount.UNKNOWN + @dataclass class SoundVolume(RoborockBase): @@ -344,3 +354,15 @@ class Q10NetworkInfo(RoborockBase): """ net_info: dpNetInfo | None = field(default=None, metadata={"dps": B01_Q10_DP.NET_INFO}) + + +@dataclass +class Q10RoomCleanSettings(RoborockBase): + """Writable cleaning settings for one Q10 room.""" + + room_id: int + fan_level: Q10RoomFanLevel + water_level: YXWaterLevel + clean_type: Q10RoomCleanType + clean_count: Q10CleanCount + clean_line: YXCleanLine diff --git a/roborock/devices/traits/b01/q10/__init__.py b/roborock/devices/traits/b01/q10/__init__.py index 194568ab9..e950fd8d5 100644 --- a/roborock/devices/traits/b01/q10/__init__.py +++ b/roborock/devices/traits/b01/q10/__init__.py @@ -28,6 +28,7 @@ from .maps import MapsTrait from .network_info import NetworkInfoTrait from .remote import RemoteTrait +from .room_cleaning import RoomCleaningTrait from .status import StatusTrait from .vacuum import VacuumTrait from .volume import SoundVolumeTrait @@ -43,6 +44,7 @@ "MapContentTrait", "MapsTrait", "NetworkInfoTrait", + "RoomCleaningTrait", "SoundVolumeTrait", "StatusTrait", ] @@ -89,6 +91,9 @@ class Q10PropertiesApi(Trait): map: MapContentTrait """Composed map image plus caller-facing map and trace data.""" + room_cleaning: RoomCleaningTrait + """Customized cleaning settings and room-cleaning controls.""" + maps: MapsTrait """Saved-map list metadata.""" @@ -123,6 +128,7 @@ def __init__( self.command, map_parser_config=map_parser_config, ) + self.room_cleaning = RoomCleaningTrait(self.command) self.clean_history = CleanHistoryTrait(self.command) self.vacuum = VacuumTrait(self.command, self.status, self.map) # Read-model traits updated from the device's DPS push stream. @@ -137,6 +143,7 @@ def __init__( self.clean_history, self._map_dps, self.maps, + self.room_cleaning, ] self._subscribe_task: asyncio.Task[None] | None = None diff --git a/roborock/devices/traits/b01/q10/room_cleaning.py b/roborock/devices/traits/b01/q10/room_cleaning.py new file mode 100644 index 000000000..aaf0af345 --- /dev/null +++ b/roborock/devices/traits/b01/q10/room_cleaning.py @@ -0,0 +1,105 @@ +"""Customized room-cleaning controls for Q10 devices.""" + +import logging +from collections.abc import Sequence +from dataclasses import dataclass, field +from typing import Any + +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoomCleanSettings +from roborock.data.containers import RoborockBase +from roborock.devices.traits.common import DpsDataConverter +from roborock.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import decode_room_clean_settings, encode_room_clean_settings + +from .command import CommandTrait +from .common import UpdatableTrait + +_LOGGER = logging.getLogger(__name__) + +_ROOM_SETTING_MODES = ( + ("fan_level", Q10RoomFanLevel, Q10RoomFanLevel.UNKNOWN.code), + ("water_level", YXWaterLevel, YXWaterLevel.UNKNOWN.code), + ("clean_type", Q10RoomCleanType, Q10RoomCleanType.UNKNOWN.code), + ("clean_count", Q10CleanCount, Q10CleanCount.UNKNOWN.code), + ("clean_line", YXCleanLine, YXCleanLine.UNKNOWN.code), +) + + +def _normalize_settings( + records: list[dict[str, int]], current: list[Q10RoomCleanSettings] | None, complete: bool +) -> list[dict[str, int]]: + """Normalize mode codes and merge compact records before model conversion.""" + settings = [] + for record in records: + room = {"room_id": record["room_id"]} + for field_name, mode_type, unknown_code in _ROOM_SETTING_MODES: + code = record[field_name] + room[field_name] = code if mode_type.from_code_optional(code) is not None else unknown_code + settings.append(room) + if complete: + return settings + + by_room = { + room.room_id: { + "room_id": room.room_id, + **{field_name: getattr(room, field_name).code for field_name, _, _ in _ROOM_SETTING_MODES}, + } + for room in (current or []) + } + by_room.update((room["room_id"], room) for room in settings) + return list(by_room.values()) + + +@dataclass +class RoomCleaning(RoborockBase): + """Current customized room-cleaning state.""" + + settings: list[Q10RoomCleanSettings] | None = field(default=None, metadata={"dps": B01_Q10_DP.CUSTOMER_CLEAN}) + + +class RoomCleaningTrait(RoomCleaning, UpdatableTrait): + """Read and control Q10 per-room cleaning settings.""" + + _CONVERTER = DpsDataConverter.from_dataclass(RoomCleaning) + + def __init__(self, command: CommandTrait) -> None: + RoomCleaning.__init__(self) + UpdatableTrait.__init__(self, command, _LOGGER) + self._command: CommandTrait = command + + def settings_for_room(self, room_id: int) -> Q10RoomCleanSettings | None: + """Return the latest reported settings for one room.""" + if self.settings is None: + return None + return next((room for room in self.settings if room.room_id == room_id), None) + + async def refresh(self) -> None: + """Request the complete customized-room settings.""" + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN_REQUEST.code): 0}) + + async def set_settings(self, settings: Sequence[Q10RoomCleanSettings]) -> None: + """Publish customized settings, then request a complete settings push.""" + payload = encode_room_clean_settings(settings) + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) + await self.refresh() + + def update_from_dps(self, decoded_dps: dict[B01_Q10_DP, Any]) -> None: + """Replace full settings or upsert the rooms in a compact push.""" + payload = decoded_dps.get(B01_Q10_DP.CUSTOMER_CLEAN) + if not isinstance(payload, str): + return + try: + records, complete = decode_room_clean_settings(payload) + except RoborockException: + _LOGGER.debug("Ignoring malformed Q10 customized-room settings", exc_info=True) + return + settings = _normalize_settings(records, self.settings, complete) + super().update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: settings}) diff --git a/roborock/devices/traits/b01/q10/vacuum.py b/roborock/devices/traits/b01/q10/vacuum.py index 4b846576b..5c655e089 100644 --- a/roborock/devices/traits/b01/q10/vacuum.py +++ b/roborock/devices/traits/b01/q10/vacuum.py @@ -2,18 +2,21 @@ import asyncio import logging -from collections.abc import Callable +from collections.abc import Callable, Sequence from math import hypot from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + YXCleanLine, YXCleanType, YXDeviceCleanTask, YXFanLevel, + YXWaterLevel, ) -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.exceptions import RoborockException -from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params +from roborock.protocols.b01_q10_protocol import CleanParams, encode_clean_params, encode_room_clean_settings from .command import CommandTrait from .goto import GotoAction, GotoActionCommand, GotoSnapshot @@ -167,6 +170,18 @@ async def clean_segments(self, segment_ids: list[int]) -> None: ) self.cancel_goto() + async def clean_rooms(self, settings: Sequence[Q10RoomCleanSettings]) -> None: + """Start a customized clean for the supplied rooms.""" + selected = tuple(settings) + payload = encode_room_clean_settings(selected) + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CUSTOMER_CLEAN.code): payload}) + await self._command.send(B01_Q10_DP.CLEAN_MODE, YXCleanType.CUSTOMIZED.code) + await self._command.send( + B01_Q10_DP.START_CLEAN, + {"cmd": YXDeviceCleanTask.ELECTORAL.code, "clean_paramters": [room.room_id for room in selected]}, + ) + self.cancel_goto() + async def clean_zone( self, first_corner: Q10RoborockPoint, @@ -278,6 +293,8 @@ async def empty_dustbin(self) -> None: async def set_clean_mode(self, mode: YXCleanType) -> None: """Set the cleaning mode (vacuum, mop, or both).""" + if mode is YXCleanType.UNKNOWN: + raise ValueError("mode must be a supported YXCleanType") await self._command.send( command=B01_Q10_DP.CLEAN_MODE, params=mode.code, @@ -285,7 +302,27 @@ async def set_clean_mode(self, mode: YXCleanType) -> None: async def set_fan_level(self, level: YXFanLevel) -> None: """Set the fan suction level.""" + if level is YXFanLevel.UNKNOWN: + raise ValueError("level must be a supported YXFanLevel") await self._command.send( command=B01_Q10_DP.FAN_LEVEL, params=level.code, ) + + async def set_water_level(self, level: YXWaterLevel) -> None: + """Set the mop water level.""" + if level is YXWaterLevel.UNKNOWN: + raise ValueError("level must be a supported YXWaterLevel") + await self._command.send(B01_Q10_DP.WATER_LEVEL, level.code) + + async def set_clean_count(self, count: Q10CleanCount) -> None: + """Set the number of passes for ordinary cleaning.""" + if count is Q10CleanCount.UNKNOWN: + raise ValueError("count must be a supported Q10CleanCount") + await self._command.send(B01_Q10_DP.CLEAN_COUNT, count.code) + + async def set_clean_line(self, line: YXCleanLine) -> None: + """Set the cleaning route preference.""" + if line is YXCleanLine.UNKNOWN: + raise ValueError("line must be a supported YXCleanLine") + await self._command.send(B01_Q10_DP.COMMON, {str(B01_Q10_DP.CLEAN_LINE.code): line.code}) diff --git a/roborock/protocols/b01_q10_protocol.py b/roborock/protocols/b01_q10_protocol.py index 4178cb1e0..30fdbbd6f 100644 --- a/roborock/protocols/b01_q10_protocol.py +++ b/roborock/protocols/b01_q10_protocol.py @@ -1,17 +1,25 @@ """Roborock B01 Protocol encoding and decoding.""" -import base64 +import binascii import json import logging import struct -from base64 import b64encode +from base64 import b64decode, b64encode from collections.abc import Sequence from dataclasses import dataclass from struct import pack from typing import Any -from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, Q10RestrictionType -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RestrictionType, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import ( Q10CleanRecordDetail, @@ -79,6 +87,112 @@ def encode_clean_params(params: CleanParams) -> str: return b64encode(payload).decode() +_Q10_CUSTOM_ROOM_COMPACT_SIZE = 6 +_Q10_CUSTOM_ROOM_PROPERTIES_SIZE = 26 +_Q10_CUSTOM_ROOM_NAME_SIZE = 20 + + +def encode_room_clean_settings(settings: Sequence[Q10RoomCleanSettings]) -> str: + """Encode the compact six-byte-per-room ``dpCustomerClean`` write.""" + selected = tuple(settings) + if not selected or len(selected) > 255: + raise ValueError("settings must contain between 1 and 255 rooms") + + payload = bytearray((len(selected),)) + room_ids: set[int] = set() + fields = ( + ("fan_level", Q10RoomFanLevel), + ("water_level", YXWaterLevel), + ("clean_type", Q10RoomCleanType), + ("clean_count", Q10CleanCount), + ("clean_line", YXCleanLine), + ) + for room in selected: + if not 0 <= room.room_id <= 255: + raise ValueError("room_id must be an integer between 0 and 255") + if room.room_id in room_ids: + raise ValueError(f"duplicate room_id: {room.room_id}") + room_ids.add(room.room_id) + values = [room.room_id] + for field_name, mode_type in fields: + value = getattr(room, field_name) + if value.code < 0: + raise ValueError(f"{field_name} must be a supported {mode_type.__name__}") + values.append(value.code) + payload.extend(values) + return b64encode(payload).decode() + + +def decode_room_clean_settings(payload: str) -> tuple[list[dict[str, int]], bool]: + """Decode room field dictionaries and whether a ``dpCustomerClean`` push is complete. + + Complete records contain 26 property bytes, a 20-byte name block, one + vertex-count byte, then four bytes per vertex. Relevant property offsets + are room id 0, count 5, type 7, fan 8, water 9, and route 11. + """ + try: + raw = b64decode(payload, validate=True) + except (ValueError, binascii.Error) as ex: + raise RoborockException("Invalid Q10 customized-room payload: malformed base64") from ex + if not raw: + raise RoborockException("Invalid Q10 customized-room payload: empty data") + + count = raw[0] + if count == 0 and len(raw) == 1: + return [], True + if len(raw) == 1 + count * _Q10_CUSTOM_ROOM_COMPACT_SIZE: + offset = 1 + rooms = [] + room_ids: set[int] = set() + for _ in range(count): + room_id = raw[offset] + if room_id in room_ids: + raise RoborockException("Invalid Q10 customized-room payload: duplicate room id") + room_ids.add(room_id) + rooms.append( + { + "room_id": room_id, + "fan_level": raw[offset + 1], + "water_level": raw[offset + 2], + "clean_type": raw[offset + 3], + "clean_count": raw[offset + 4], + "clean_line": raw[offset + 5], + } + ) + offset += _Q10_CUSTOM_ROOM_COMPACT_SIZE + return rooms, False + + offset = 1 + rooms = [] + room_ids = set() + for _ in range(count): + minimum_end = offset + _Q10_CUSTOM_ROOM_PROPERTIES_SIZE + _Q10_CUSTOM_ROOM_NAME_SIZE + 1 + if minimum_end > len(raw): + raise RoborockException("Invalid Q10 customized-room payload: truncated room record") + room_id = int.from_bytes(raw[offset : offset + 2], "big") + if room_id in room_ids: + raise RoborockException("Invalid Q10 customized-room payload: duplicate room id") + room_ids.add(room_id) + clean_count = int.from_bytes(raw[offset + 5 : offset + 7], "big") + rooms.append( + { + "room_id": room_id, + "fan_level": raw[offset + 8], + "water_level": raw[offset + 9], + "clean_type": raw[offset + 7], + "clean_count": clean_count, + "clean_line": raw[offset + 11], + } + ) + vertex_count = raw[offset + _Q10_CUSTOM_ROOM_PROPERTIES_SIZE + _Q10_CUSTOM_ROOM_NAME_SIZE] + offset = minimum_end + vertex_count * 4 + if offset > len(raw): + raise RoborockException("Invalid Q10 customized-room payload: truncated vertices") + if offset != len(raw): + raise RoborockException("Invalid Q10 customized-room payload: trailing data") + return rooms, True + + _Q10_ZONE_RECORD_SIZE = 38 _Q10_ROOM_NAME_FIELD_SIZE = 19 @@ -128,7 +242,7 @@ def encode_restricted_zones(zones: Sequence[Q10RestrictedZone]) -> str: if len(zones) > 255: raise ValueError("zones must be a sequence of at most 255 restrictions") if not zones: - return base64.b64encode(b"\x01\x00").decode() + return b64encode(b"\x01\x00").decode() payload = bytearray((1, len(zones))) for zone in zones: @@ -151,7 +265,7 @@ def encode_restricted_zones(zones: Sequence[Q10RestrictedZone]) -> str: ) record.extend(bytes(_Q10_ZONE_RECORD_SIZE - len(record))) payload.extend(record) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_virtual_walls(walls: Sequence[Q10VirtualWall]) -> str: @@ -172,7 +286,7 @@ def encode_virtual_walls(walls: Sequence[Q10VirtualWall]) -> str: y, ) ) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_room_name(room_id: int, name: str) -> str: @@ -188,7 +302,7 @@ def encode_room_name(room_id: int, name: str) -> str: payload = bytearray((1, room_id, 0, len(encoded_name))) payload.extend(encoded_name) payload.extend(bytes(_Q10_ROOM_NAME_FIELD_SIZE - len(encoded_name))) - return base64.b64encode(payload).decode() + return b64encode(payload).decode() def encode_mqtt_payload(command: B01_Q10_DP, params: ParamsType) -> RoborockMessage: diff --git a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr index 2604e87be..fb07403ca 100644 --- a/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr +++ b/tests/devices/traits/b01/q10/__snapshots__/test_status.ambr @@ -48,6 +48,8 @@ 'wifiName': 'wifi-network-name', }), }), + 'room_cleaning': dict({ + }), 'status': dict({ 'addCleanState': False, 'areaUnit': 'square_meter', @@ -435,6 +437,8 @@ 'wifiName': '**REDACTED**', }), }), + 'room_cleaning': dict({ + }), 'status': dict({ 'addCleanState': False, 'areaUnit': 'square_meter', @@ -512,6 +516,8 @@ }), 'network_info': dict({ }), + 'room_cleaning': dict({ + }), 'status': dict({ }), 'volume': dict({ diff --git a/tests/devices/traits/b01/q10/test_room_cleaning.py b/tests/devices/traits/b01/q10/test_room_cleaning.py new file mode 100644 index 000000000..94280b85c --- /dev/null +++ b/tests/devices/traits/b01/q10/test_room_cleaning.py @@ -0,0 +1,265 @@ +"""Tests for customized Q10 room-cleaning commands and pushed settings.""" + +import base64 +import pathlib +from dataclasses import replace + +import pytest + +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoomCleanSettings +from roborock.devices.traits.b01.q10 import Q10PropertiesApi +from roborock.exceptions import RoborockException +from roborock.protocols.b01_q10_protocol import encode_room_clean_settings + +from .conftest import FakeB01Q10Channel + + +def _settings(room_id: int = 3, *, fan: Q10RoomFanLevel = Q10RoomFanLevel.TURBO) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings( + room_id=room_id, + fan_level=fan, + water_level=YXWaterLevel.HIGH, + clean_type=Q10RoomCleanType.VAC_AND_MOP, + clean_count=Q10CleanCount.TWICE, + clean_line=YXCleanLine.FINE, + ) + + +def _complete_payload(settings: Q10RoomCleanSettings) -> str: + properties = bytearray(26) + properties[0:2] = settings.room_id.to_bytes(2, "big") + properties[3:5] = b"\xff\xff" + properties[5:7] = settings.clean_count.code.to_bytes(2, "big") + properties[7] = settings.clean_type.code + properties[8] = settings.fan_level.code + properties[9] = settings.water_level.code + properties[10] = 0xFF + properties[11] = settings.clean_line.code + name = b"Example" + name_block = bytes((len(name),)) + name + bytes(19 - len(name)) + return base64.b64encode(b"\x01" + properties + name_block + b"\x00").decode() + + +async def test_set_settings_publishes_then_requests_full_settings( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel +) -> None: + settings = (_settings(),) + await q10_api.room_cleaning.set_settings(settings) + assert fake_channel.published_commands == [ + (B01_Q10_DP.COMMON, {"62": "AQMDAwECAg=="}), + (B01_Q10_DP.COMMON, {"63": 0}), + ] + assert q10_api.room_cleaning.settings is None + + +async def test_set_settings_failure_does_not_request_refresh( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, monkeypatch: pytest.MonkeyPatch +) -> None: + async def send(command, params=None): + raise RoborockException("publish failed") + + monkeypatch.setattr(fake_channel, "send_command", send) + with pytest.raises(RoborockException, match="publish failed"): + await q10_api.room_cleaning.set_settings((_settings(),)) + assert fake_channel.published_commands == [] + + +def test_full_push_replaces_and_compact_push_upserts(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((_settings(9),))}) + assert trait.settings == [_settings(3), _settings(9)] + changed = _settings(3, fan=Q10RoomFanLevel.MAX) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: encode_room_clean_settings((changed,))}) + assert trait.settings == [changed, _settings(9)] + assert trait.settings_for_room(3) == changed + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(1))}) + assert trait.settings == [_settings(1)] + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "invalid"}) + assert trait.settings == [_settings(1)] + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == [] + assert len(notified) == 5 + + +async def test_refresh_requests_complete_settings(q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel) -> None: + await q10_api.room_cleaning.refresh() + assert fake_channel.published_commands == [(B01_Q10_DP.COMMON, {"63": 0})] + assert q10_api.room_cleaning.settings is None + + +def test_settings_for_room_unpopulated(q10_api: Q10PropertiesApi) -> None: + assert q10_api.room_cleaning.settings_for_room(3) is None + assert q10_api.room_cleaning.as_dict() == {} + assert q10_api.as_dict()["room_cleaning"] == {} + + +def test_settings_for_room_missing(q10_api: Q10PropertiesApi) -> None: + q10_api.room_cleaning.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + assert q10_api.room_cleaning.settings_for_room(9) is None + + +def test_repeated_pushes_do_not_notify(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "invalid"}) + assert trait.settings is None + assert not notified + compact = encode_room_clean_settings((_settings(3),)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: compact}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: compact}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: _complete_payload(_settings(3))}) + assert trait.settings == [_settings(3)] + assert len(notified) == 1 + assert isinstance(trait.as_dict()["settings"], list) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == [] + assert trait.as_dict() == {"settings": []} + assert trait.settings_for_room(3) is None + assert len(notified) == 2 + + +def test_first_empty_push_is_reported(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AA=="}) + assert trait.settings == [] + assert trait.as_dict() == {"settings": []} + assert notified == [True] + + +@pytest.mark.parametrize("complete", [False, True]) +@pytest.mark.parametrize( + ("field", "compact_offset", "full_offset", "width"), + [ + ("fan_level", 1, 8, 1), + ("water_level", 2, 9, 1), + ("clean_type", 3, 7, 1), + ("clean_count", 4, 5, 2), + ("clean_line", 5, 11, 1), + ], +) +def test_unknown_mode_codes_use_enum_fallback( + q10_api: Q10PropertiesApi, field: str, compact_offset: int, full_offset: int, width: int, complete: bool +) -> None: + trait = q10_api.room_cleaning + payload = _complete_payload(_settings()) if complete else "AQMDAwECAg==" + raw = bytearray(base64.b64decode(payload)) + offset = full_offset + 1 if complete else compact_offset + 1 + code_width = width if complete else 1 + raw[offset : offset + code_width] = (99).to_bytes(code_width, "big") + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: base64.b64encode(raw).decode()}) + expected = _settings() + setattr(expected, field, type(getattr(expected, field)).UNKNOWN) + assert trait.settings == [expected] + # A compact update must preserve the cached unknown code during conversion. + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AQkDAwECAg=="}) + assert trait.settings == [expected, _settings(9)] + + +@pytest.mark.parametrize("complete", [False, True]) +def test_zero_mode_codes_are_preserved(q10_api: Q10PropertiesApi, complete: bool) -> None: + settings = replace(_settings(), water_level=YXWaterLevel.OFF, clean_line=YXCleanLine.FAST) + payload = _complete_payload(settings) if complete else "AQMDAAECAA==" + q10_api.room_cleaning.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: payload}) + assert q10_api.room_cleaning.settings == [settings] + + +@pytest.mark.parametrize("populated", [False, True]) +@pytest.mark.parametrize("payload", [None, 1, {}, [], "invalid", "", "AQAA"]) +def test_invalid_push_does_not_change_settings(q10_api: Q10PropertiesApi, populated: bool, payload: object) -> None: + trait = q10_api.room_cleaning + if populated: + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AQMDAwECAg=="}) + before = trait.settings + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: payload}) + trait.update_from_dps({B01_Q10_DP.BATTERY: 100}) + assert trait.settings == before + assert not notified + + +def test_malformed_later_record_is_atomic(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AQMDAwECAg=="}) + notified: list[bool] = [] + trait.add_update_listener(lambda: notified.append(True)) + first = base64.b64decode(_complete_payload(_settings(9)))[1:] + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: base64.b64encode(b"\x02" + first + b"\x00").decode()}) + assert trait.settings == [_settings(3)] + assert not notified + + +def test_compact_update_preserves_existing_room_order(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AgMDAwECAgkDAwECAg=="}) + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AgkEAwECAgEBAwECAg=="}) + assert trait.settings == [ + _settings(3), + _settings(9, fan=Q10RoomFanLevel.MAX), + _settings(1, fan=Q10RoomFanLevel.QUIET), + ] + + +def test_sanitized_full_capture_builds_typed_models(q10_api: Q10PropertiesApi) -> None: + payload = ( + pathlib.Path("tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64").read_text().strip() + ) + q10_api.room_cleaning.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: payload}) + assert q10_api.room_cleaning.settings == [ + _settings(1, fan=Q10RoomFanLevel.QUIET), + _settings(2, fan=Q10RoomFanLevel.MAX_PLUS), + replace( + _settings(3, fan=Q10RoomFanLevel.MAX_PLUS), + water_level=YXWaterLevel.MEDIUM, + clean_type=Q10RoomCleanType.VACUUM, + ), + replace( + _settings(4, fan=Q10RoomFanLevel.MAX_PLUS), + water_level=YXWaterLevel.MEDIUM, + clean_type=Q10RoomCleanType.VACUUM, + ), + _settings(5, fan=Q10RoomFanLevel.QUIET), + _settings(6, fan=Q10RoomFanLevel.MAX_PLUS), + ] + + +@pytest.mark.parametrize("method", ["set_settings", "clean_rooms"]) +async def test_commands_leave_reported_settings_unchanged(q10_api: Q10PropertiesApi, method: str) -> None: + q10_api.room_cleaning.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AQMDAwECAg=="}) + target = q10_api.room_cleaning if method == "set_settings" else q10_api.vacuum + await getattr(target, method)((_settings(9),)) + assert q10_api.room_cleaning.settings == [_settings(3)] + + +def test_populated_diagnostics_use_model_field_names_and_enum_labels(q10_api: Q10PropertiesApi) -> None: + trait = q10_api.room_cleaning + trait.update_from_dps({B01_Q10_DP.CUSTOMER_CLEAN: "AQMDAwECAg=="}) + expected = { + "settings": [ + { + "roomId": 3, + "fanLevel": "turbo", + "waterLevel": "high", + "cleanType": "vac_and_mop", + "cleanCount": "twice", + "cleanLine": "fine", + } + ] + } + assert trait.as_dict() == expected + assert q10_api.as_dict()["room_cleaning"] == expected diff --git a/tests/devices/traits/b01/q10/test_status.py b/tests/devices/traits/b01/q10/test_status.py index 240c87f5a..339a6ccc6 100644 --- a/tests/devices/traits/b01/q10/test_status.py +++ b/tests/devices/traits/b01/q10/test_status.py @@ -11,6 +11,7 @@ from roborock.data import HomeDataDevice, HomeDataProduct from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, YXAreaUnit, YXCarpetCleanType, YXCleanLine, @@ -282,3 +283,19 @@ async def test_q10_properties_as_dict_and_diagnostic_data( # Verify diagnostic_data extracts the properties correctly diagnostics = device.diagnostic_data() assert diagnostics == snapshot + + +@pytest.mark.parametrize( + ("code", "expected"), + [ + (None, None), + (1, Q10CleanCount.ONCE), + (2, Q10CleanCount.TWICE), + (3, Q10CleanCount.THREE_TIMES), + (99, Q10CleanCount.UNKNOWN), + ], +) +def test_clean_count_mode(q10_api: Q10PropertiesApi, code: int | None, expected: Q10CleanCount | None) -> None: + if code is not None: + q10_api.status.update_from_dps({B01_Q10_DP.CLEAN_COUNT: code}) + assert q10_api.status.clean_count_mode is expected diff --git a/tests/devices/traits/b01/q10/test_vacuum.py b/tests/devices/traits/b01/q10/test_vacuum.py index 6b6c477ef..8c726f3be 100644 --- a/tests/devices/traits/b01/q10/test_vacuum.py +++ b/tests/devices/traits/b01/q10/test_vacuum.py @@ -8,12 +8,17 @@ from roborock.data.b01_q10.b01_q10_code_mappings import ( B01_Q10_DP, + Q10CleanCount, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, YXCleanType, YXDeviceCleanTask, YXDeviceState, YXFanLevel, + YXWaterLevel, ) -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.devices.traits.b01.q10 import Q10PropertiesApi from roborock.devices.traits.b01.q10 import vacuum as vacuum_module from roborock.devices.traits.b01.q10.vacuum import VacuumTrait @@ -236,3 +241,113 @@ async def test_goto_position_at_current_position_pauses_owned_zone( B01_Q10_DP.START_CLEAN, B01_Q10_DP.PAUSE, ] + + +@pytest.mark.parametrize( + ("method", "value", "command", "payload"), + [ + ("set_water_level", YXWaterLevel.HIGH, B01_Q10_DP.WATER_LEVEL, 3), + ("set_clean_count", Q10CleanCount.THREE_TIMES, B01_Q10_DP.CLEAN_COUNT, 3), + ("set_clean_line", YXCleanLine.FINE, B01_Q10_DP.COMMON, {"78": 2}), + ], +) +async def test_advanced_cleaning_controls(q10_api, fake_channel, method, value, command, payload) -> None: + await getattr(q10_api.vacuum, method)(value) + assert fake_channel.published_commands == [(command, payload)] + + +@pytest.mark.parametrize( + ("method", "value"), + [ + ("set_clean_mode", YXCleanType.UNKNOWN), + ("set_fan_level", YXFanLevel.UNKNOWN), + ("set_water_level", YXWaterLevel.UNKNOWN), + ("set_clean_count", Q10CleanCount.UNKNOWN), + ("set_clean_line", YXCleanLine.UNKNOWN), + ], +) +async def test_unknown_cleaning_modes_are_not_published(q10_api, fake_channel, method, value) -> None: + with pytest.raises(ValueError, match="supported"): + await getattr(q10_api.vacuum, method)(value) + assert fake_channel.published_commands == [] + + +def _room_settings(room_id: int = 3) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings( + room_id, + Q10RoomFanLevel.TURBO, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ) + + +async def test_clean_rooms_sends_settings_mode_and_start_in_order( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel +) -> None: + settings = (_room_settings(3), _room_settings(9)) + await q10_api.vacuum.clean_rooms(settings) + assert fake_channel.published_commands == [ + (B01_Q10_DP.COMMON, {"62": "AgMDAwECAgkDAwECAg=="}), + (B01_Q10_DP.CLEAN_MODE, 4), + (B01_Q10_DP.START_CLEAN, {"cmd": 2, "clean_paramters": [3, 9]}), + ] + + +@pytest.mark.parametrize("fail_on", [1, 2, 3]) +async def test_publish_failure_stops_remaining_clean_commands( + q10_api: Q10PropertiesApi, fake_channel: FakeB01Q10Channel, fail_on: int, monkeypatch: pytest.MonkeyPatch +) -> None: + original = fake_channel.send_command + calls = 0 + + async def send(command, params=None): + nonlocal calls + calls += 1 + if calls == fail_on: + raise RoborockException("publish failed") + await original(command, params) + + monkeypatch.setattr(fake_channel, "send_command", send) + with pytest.raises(RoborockException, match="publish failed"): + await q10_api.vacuum.clean_rooms((_room_settings(),)) + assert calls == fail_on + assert ( + fake_channel.published_commands + == [ + (B01_Q10_DP.COMMON, {"62": "AQMDAwECAg=="}), + (B01_Q10_DP.CLEAN_MODE, 4), + (B01_Q10_DP.START_CLEAN, {"cmd": 2, "clean_paramters": [3]}), + ][: fail_on - 1] + ) + + +async def test_clean_rooms_cancels_active_goto( + q10_api: Q10PropertiesApi, + fake_channel: FakeB01Q10Channel, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """An old goto cannot pause or stop a successful customized room clean.""" + monkeypatch.setattr(vacuum_module, "_GOTO_TIMEOUT", 0.01) + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(0, 0)], sequence=1)) + q10_api.status.update_from_dps( + { + B01_Q10_DP.CLEAN_TASK_TYPE: YXDeviceCleanTask.DIVIDE_AREAS.code, + B01_Q10_DP.STATUS: YXDeviceState.CLEANING.code, + } + ) + try: + await q10_api.vacuum.goto_position(Q10RoborockPoint(29900, 28650)) + await q10_api.vacuum.clean_rooms((_room_settings(),)) + # Keep the old zone status while its trace reaches the goto target. + q10_api.map.update_from_trace_packet(Q10TracePacket(points=[Q10Point(1760, 1260)], sequence=2)) + await asyncio.sleep(0.03) + assert [command for command, _ in fake_channel.published_commands] == [ + B01_Q10_DP.START_CLEAN, + B01_Q10_DP.COMMON, + B01_Q10_DP.CLEAN_MODE, + B01_Q10_DP.START_CLEAN, + ] + finally: + await q10_api.close() diff --git a/tests/protocols/test_b01_q10_protocol.py b/tests/protocols/test_b01_q10_protocol.py index ebd6b15bd..2f99d3821 100644 --- a/tests/protocols/test_b01_q10_protocol.py +++ b/tests/protocols/test_b01_q10_protocol.py @@ -12,8 +12,16 @@ from freezegun import freeze_time from syrupy import SnapshotAssertion -from roborock.data.b01_q10.b01_q10_code_mappings import B01_Q10_DP, Q10RestrictionType, YXWaterLevel -from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint +from roborock.data.b01_q10.b01_q10_code_mappings import ( + B01_Q10_DP, + Q10CleanCount, + Q10RestrictionType, + Q10RoomCleanType, + Q10RoomFanLevel, + YXCleanLine, + YXWaterLevel, +) +from roborock.data.b01_q10.b01_q10_containers import Q10RoborockPoint, Q10RoomCleanSettings from roborock.data.code_mappings import completed_warnings from roborock.exceptions import RoborockException from roborock.map.b01_q10_map_parser import Q10CleanRecordDetail, Q10MapPacket, Q10MapPacketKind, Q10TracePacket @@ -22,10 +30,12 @@ CleanParams, Q10DpsUpdate, decode_message, + decode_room_clean_settings, decode_rpc_response, encode_clean_params, encode_mqtt_payload, encode_restricted_zones, + encode_room_clean_settings, encode_room_name, encode_virtual_walls, ) @@ -296,6 +306,155 @@ def test_encode_mqtt_payload(command: B01_Q10_DP, params: dict[str, Any], snapsh json.loads(message.payload.decode()) +def _room_settings( + room_id: int = 3, + *, + fan: Q10RoomFanLevel = Q10RoomFanLevel.TURBO, + water: YXWaterLevel = YXWaterLevel.HIGH, + clean_type: Q10RoomCleanType = Q10RoomCleanType.VAC_AND_MOP, + count: Q10CleanCount = Q10CleanCount.TWICE, + line: YXCleanLine = YXCleanLine.FINE, +) -> Q10RoomCleanSettings: + return Q10RoomCleanSettings(room_id, fan, water, clean_type, count, line) + + +def _full_room_record(settings: Q10RoomCleanSettings, name: str = "Example") -> bytes: + properties = bytearray(26) + properties[0:2] = settings.room_id.to_bytes(2, "big") + properties[2] = 1 + properties[3:5] = b"\xff\xff" + properties[5:7] = settings.clean_count.code.to_bytes(2, "big") + properties[7] = settings.clean_type.code + properties[8] = settings.fan_level.code + properties[9] = settings.water_level.code + properties[10] = 0xFF + properties[11] = settings.clean_line.code + encoded_name = name.encode() + name_block = bytes((len(encoded_name),)) + encoded_name + bytes(19 - len(encoded_name)) + return bytes(properties) + name_block + b"\x00" + + +def test_encode_room_clean_settings_uses_compact_format() -> None: + settings = [_room_settings(), _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW)] + + encoded = encode_room_clean_settings(settings) + + assert base64.b64decode(encoded) == bytes((2, 3, 3, 3, 1, 2, 2, 9, 5, 1, 1, 2, 2)) + assert settings == [ + _room_settings(), + _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW), + ] + + +def test_decode_room_clean_settings_distinguishes_echo_and_complete_response() -> None: + first = _room_settings() + second = _room_settings(9, fan=Q10RoomFanLevel.MAX_PLUS, water=YXWaterLevel.LOW) + echo, echo_complete = decode_room_clean_settings("AgMDAwECAgkFAQECAg==") + full, full_complete = decode_room_clean_settings( + base64.b64encode(b"\x02" + _full_room_record(first) + _full_room_record(second)).decode() + ) + + expected = [ + {"room_id": 3, "fan_level": 3, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + {"room_id": 9, "fan_level": 5, "water_level": 1, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + ] + assert echo == expected + assert echo_complete is False + assert full == expected + assert full_complete is True + + +def test_decode_room_clean_settings_preserves_unknown_values() -> None: + record = bytearray(_full_room_record(_room_settings())) + record[8] = 42 + record[11] = 99 + + records, complete = decode_room_clean_settings(base64.b64encode(b"\x01" + record).decode()) + + assert records[0]["fan_level"] == 42 + assert records[0]["clean_line"] == 99 + assert complete is True + + +def test_decode_room_clean_settings_empty_complete_response() -> None: + assert decode_room_clean_settings("AA==") == ([], True) + + +def test_encode_room_clean_settings_rejects_more_than_wire_count() -> None: + with pytest.raises(ValueError, match="between 1 and 255"): + encode_room_clean_settings([_room_settings(room_id) for room_id in range(256)]) + + +def test_encode_room_clean_settings_matches_physical_echoes() -> None: + candidate = Q10RoomCleanSettings( + 1, + Q10RoomFanLevel.TURBO, + YXWaterLevel.MEDIUM, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.ONCE, + YXCleanLine.DAILY, + ) + restored = Q10RoomCleanSettings( + 1, + Q10RoomFanLevel.QUIET, + YXWaterLevel.HIGH, + Q10RoomCleanType.VAC_AND_MOP, + Q10CleanCount.TWICE, + YXCleanLine.FINE, + ) + + assert encode_room_clean_settings((candidate,)) == "AQEDAgEBAQ==" + assert encode_room_clean_settings((restored,)) == "AQEBAwECAg==" + assert decode_room_clean_settings("AQEDAgEBAQ==") == ( + [{"room_id": 1, "fan_level": 3, "water_level": 2, "clean_type": 1, "clean_count": 1, "clean_line": 1}], + False, + ) + assert decode_room_clean_settings("AQEBAwECAg==") == ( + [{"room_id": 1, "fan_level": 1, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}], + False, + ) + + +def test_encode_room_clean_settings_supports_three_pass_protocol_value() -> None: + settings = _room_settings(count=Q10CleanCount.THREE_TIMES) + + assert base64.b64decode(encode_room_clean_settings((settings,)))[5] == 3 + + +@pytest.mark.parametrize( + "payload", + [ + "not-base64", + "", + base64.b64encode(b"\x01\x00").decode(), + base64.b64encode(b"\x01" + _full_room_record(_room_settings())[:-1]).decode(), + base64.b64encode(b"\x01" + _full_room_record(_room_settings()) + b"extra").decode(), + base64.b64encode(b"\x02" + bytes((3, 3, 3, 1, 2, 2)) * 2).decode(), + base64.b64encode(b"\x02" + _full_room_record(_room_settings()) * 2).decode(), + ], +) +def test_decode_room_clean_settings_rejects_malformed_payload(payload: str) -> None: + with pytest.raises(RoborockException, match="customized-room"): + decode_room_clean_settings(payload) + + +def test_decode_sanitized_physical_six_room_capture() -> None: + payload = (TESTDATA_PATH / "customer_clean_full_sanitized.b64").read_text().strip() + + records, complete = decode_room_clean_settings(payload) + + assert complete is True + assert len(base64.b64decode(payload)) == 283 + assert records == [ + {"room_id": 1, "fan_level": 1, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + {"room_id": 2, "fan_level": 5, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + {"room_id": 3, "fan_level": 5, "water_level": 2, "clean_type": 2, "clean_count": 2, "clean_line": 2}, + {"room_id": 4, "fan_level": 5, "water_level": 2, "clean_type": 2, "clean_count": 2, "clean_line": 2}, + {"room_id": 5, "fan_level": 1, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + {"room_id": 6, "fan_level": 5, "water_level": 3, "clean_type": 1, "clean_count": 2, "clean_line": 2}, + ] + + def _point(x: int, y: int) -> Q10RoborockPoint: """Build a public-coordinate point from a captured Q10 wire vector.""" return Q10RoborockPoint.from_vector(x, y) @@ -476,3 +635,34 @@ def test_encode_room_name_rejects_invalid_room_id(room_id: int) -> None: def test_encode_room_name_rejects_invalid_name(name: str) -> None: with pytest.raises(ValueError, match="name"): encode_room_name(7, name) + + +@pytest.mark.parametrize("field", ["fan_level", "water_level", "clean_type", "clean_count", "clean_line"]) +def test_unknown_setting_cannot_be_written(field: str) -> None: + settings = _room_settings() + setattr(settings, field, type(getattr(settings, field)).UNKNOWN) + with pytest.raises(ValueError, match=field): + encode_room_clean_settings((settings,)) + + +@pytest.mark.parametrize( + "rooms", [(), (_room_settings(-1),), (_room_settings(256),), (_room_settings(), _room_settings())] +) +def test_semantic_room_guards(rooms: tuple[Q10RoomCleanSettings, ...]) -> None: + with pytest.raises(ValueError): + encode_room_clean_settings(rooms) + + +def test_decode_room_clean_settings_skips_vertices_between_records() -> None: + first = _full_room_record(_room_settings(3))[:-1] + b"\x02" + bytes(8) + second = _full_room_record(_room_settings(9)) + records, complete = decode_room_clean_settings(base64.b64encode(b"\x02" + first + second).decode()) + assert [room["room_id"] for room in records] == [3, 9] + assert complete is True + + +@pytest.mark.parametrize("vertex_bytes", range(4)) +def test_decode_room_clean_settings_rejects_truncated_vertices(vertex_bytes: int) -> None: + record = _full_room_record(_room_settings())[:-1] + b"\x01" + bytes(vertex_bytes) + with pytest.raises(RoborockException, match="truncated vertices"): + decode_room_clean_settings(base64.b64encode(b"\x01" + record).decode()) diff --git a/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 b/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 new file mode 100644 index 000000000..dbc00f660 --- /dev/null +++ b/tests/protocols/testdata/b01_q10_protocol/customer_clean_full_sanitized.b64 @@ -0,0 +1 @@ +BgABAP//AAIBAQP/AgAAAAAAAAAAAAAAAAAABlJvb20gMQAAAAAAAAAAAAAAAAAAAAIA//8AAgEFA/8CAAAAAAAAAAAAAAAAAAAGUm9vbSAyAAAAAAAAAAAAAAAAAAAAAwL//wACAgUC/wIAAAAAAAAAAAAAAAAAAAZSb29tIDMAAAAAAAAAAAAAAAAAAAAEAf//AAICBQL/AgAAAAAAAAAAAAAAAAAABlJvb20gNAAAAAAAAAAAAAAAAAAAAAUJ//8AAgEBA/8CAAAAAAAAAAAAAAAAAAAGUm9vbSA1AAAAAAAAAAAAAAAAAAAABgD//wACAQUD/wIAAAAAAAAAAAAAAAAAAAZSb29tIDYAAAAAAAAAAAAAAAAAAA==