test(q7): simplify map parser follow-up fixtures

Author: arduano

roborock/map/b01_map_parser.py

@@ -29,7 +29,6 @@
 
 from .map_parser import ParsedMapData
 
-_B64_CHARS = set(b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=")
 _MAP_FILE_FORMAT = "PNG"
 
 
@@ -91,9 +90,6 @@ def _derive_map_key(serial: str, model: str) -> bytes:
 
 def _decode_base64_payload(raw_payload: bytes) -> bytes:
     blob = raw_payload.strip()
-    if len(blob) < 32 or any(b not in _B64_CHARS for b in blob):
-        raise RoborockException("Failed to decode B01 map payload")
-
     padded = blob + b"=" * (-len(blob) % 4)
     try:
         return base64.b64decode(padded, validate=True)
@@ -125,13 +121,6 @@ def _parse_proto(blob: bytes, message: Message, *, context: str) -> None:
         raise RoborockException(f"Failed to parse {context}") from err
 
 
-def _decode_map_data_bytes(value: bytes) -> bytes:
-    try:
-        return zlib.decompress(value)
-    except zlib.error:
-        return value
-
-
 def _parse_scmap_payload(payload: bytes) -> RobotMap:
     """Parse inflated SCMap bytes into a generated protobuf message."""
     parsed = RobotMap()
@@ -148,7 +137,7 @@ def _extract_grid(parsed: RobotMap) -> tuple[int, int, bytes]:
     if not size_x or not size_y or not parsed.mapData.HasField("mapData"):
         raise RoborockException("Failed to parse B01 map header/grid")
 
-    map_data = _decode_map_data_bytes(parsed.mapData.mapData)
+    map_data = parsed.mapData.mapData
     expected_len = size_x * size_y
     if len(map_data) < expected_len:
         raise RoborockException("B01 map data shorter than expected dimensions")

tests/map/test_b01_map_parser.py

@@ -2,7 +2,6 @@
 import gzip
 import hashlib
 import io
-import struct
 import zlib
 from pathlib import Path
 
@@ -13,6 +12,7 @@
 
 from roborock.exceptions import RoborockException
 from roborock.map.b01_map_parser import B01MapParser, _parse_scmap_payload
+from roborock.map.proto.b01_scmap_pb2 import RobotMap  # type: ignore[attr-defined]
 
 FIXTURE = Path(__file__).resolve().parent / "testdata" / "raw-mqtt-map301.bin.inflated.bin.gz"
 
@@ -26,30 +26,6 @@ def _derive_map_key(serial: str, model: str) -> bytes:
     return md5[8:24].encode()
 
 
-def _encode_varint(value: int) -> bytes:
-    encoded = bytearray()
-    while True:
-        to_write = value & 0x7F
-        value >>= 7
-        if value:
-            encoded.append(to_write | 0x80)
-        else:
-            encoded.append(to_write)
-            return bytes(encoded)
-
-
-def _field_varint(field_no: int, value: int) -> bytes:
-    return _encode_varint((field_no << 3) | 0) + _encode_varint(value)
-
-
-def _field_len(field_no: int, value: bytes) -> bytes:
-    return _encode_varint((field_no << 3) | 2) + _encode_varint(len(value)) + value
-
-
-def _field_float32(field_no: int, value: float) -> bytes:
-    return _encode_varint((field_no << 3) | 5) + struct.pack("<f", value)
-
-
 def test_b01_map_parser_decodes_and_renders_fixture() -> None:
     serial = "testsn012345"
     model = "roborock.vacuum.sc05"
@@ -87,62 +63,41 @@ def test_b01_map_parser_decodes_and_renders_fixture() -> None:
 
 
 def test_b01_scmap_parser_maps_observed_schema_fields() -> None:
-    room_name_post = _field_float32(1, 11.25) + _field_float32(2, 22.5)
-    room_one = b"".join(
-        [
-            _field_varint(1, 42),
-            _field_len(2, b"Kitchen"),
-            _field_varint(5, 1),
-            _field_len(8, room_name_post),
-            _field_varint(10, 7),
-            _field_varint(12, 9),
-        ]
-    )
-    room_two = b"".join([_field_varint(1, 99), _field_varint(5, 0)])
-
-    boundary_info = b"".join(
-        [
-            _field_len(1, b"md5"),
-            _field_varint(2, 10),
-            _field_varint(3, 20),
-            _field_varint(4, 30),
-            _field_varint(5, 40),
-        ]
-    )
-    map_ext_info = b"".join(
-        [
-            _field_varint(1, 100),
-            _field_varint(2, 200),
-            _field_varint(3, 1),
-            _field_varint(8, 3),
-            _field_len(7, boundary_info),
-        ]
-    )
-    map_head = b"".join(
-        [
-            _field_varint(1, 7),
-            _field_varint(2, 2),
-            _field_varint(3, 2),
-            _field_float32(4, 1.5),
-            _field_float32(5, 2.5),
-            _field_float32(6, 3.5),
-            _field_float32(7, 4.5),
-            _field_float32(8, 0.05),
-        ]
-    )
-    map_data = _field_len(1, zlib.compress(bytes([0, 127, 128, 128])))
-    payload = b"".join(
-        [
-            _field_varint(1, 1),
-            _field_len(2, map_ext_info),
-            _field_len(3, map_head),
-            _field_len(4, map_data),
-            _field_len(12, room_one),
-            _field_len(12, room_two),
-        ]
-    )
-
-    parsed = _parse_scmap_payload(payload)
+    payload = RobotMap()
+    payload.mapType = 1
+    payload.mapExtInfo.taskBeginDate = 100
+    payload.mapExtInfo.mapUploadDate = 200
+    payload.mapExtInfo.mapValid = 1
+    payload.mapExtInfo.mapVersion = 3
+    payload.mapExtInfo.boudaryInfo.mapMd5 = "md5"
+    payload.mapExtInfo.boudaryInfo.vMinX = 10
+    payload.mapExtInfo.boudaryInfo.vMaxX = 20
+    payload.mapExtInfo.boudaryInfo.vMinY = 30
+    payload.mapExtInfo.boudaryInfo.vMaxY = 40
+    payload.mapHead.mapHeadId = 7
+    payload.mapHead.sizeX = 2
+    payload.mapHead.sizeY = 2
+    payload.mapHead.minX = 1.5
+    payload.mapHead.minY = 2.5
+    payload.mapHead.maxX = 3.5
+    payload.mapHead.maxY = 4.5
+    payload.mapHead.resolution = 0.05
+    payload.mapData.mapData = bytes([0, 127, 128, 128])
+
+    room_one = payload.roomDataInfo.add()
+    room_one.roomId = 42
+    room_one.roomName = "Kitchen"
+    room_one.cleanState = 1
+    room_one.roomNamePost.x = 11.25
+    room_one.roomNamePost.y = 22.5
+    room_one.colorId = 7
+    room_one.global_seq = 9
+
+    room_two = payload.roomDataInfo.add()
+    room_two.roomId = 99
+    room_two.cleanState = 0
+
+    parsed = _parse_scmap_payload(payload.SerializeToString())
 
     assert parsed.mapType == 1
     assert parsed.HasField("mapExtInfo")
@@ -157,7 +112,7 @@ def test_b01_scmap_parser_maps_observed_schema_fields() -> None:
     assert parsed.mapHead.resolution == pytest.approx(0.05)
     assert parsed.HasField("mapData")
     assert parsed.mapData.HasField("mapData")
-    assert zlib.decompress(parsed.mapData.mapData) == bytes([0, 127, 128, 128])
+    assert parsed.mapData.mapData == bytes([0, 127, 128, 128])
     assert parsed.roomDataInfo[0].roomId == 42
     assert parsed.roomDataInfo[0].roomName == "Kitchen"
     assert parsed.roomDataInfo[0].HasField("roomNamePost")