chore: add snapshot tests for parsing device wire formats (#457)

* chore: add snapshot tests for parsing device wire formats

* chore: sort imports

Author: allenporter

poetry.lock

@@ -49,6 +49,7 @@ pyshark = "^0.6"
 aioresponses = "^0.7.7"
 freezegun = "^1.5.1"
 pytest-timeout = "^2.3.1"
+syrupy = "^4.9.1"
 
 [tool.semantic_release]
 branch = "main"

pyproject.toml

@@ -0,0 +1,48 @@
+# serializer version: 1
+# name: test_decode_rpc_payload[get_dnd]
+  20002
+# ---
+# name: test_decode_rpc_payload[get_dnd].1
+  '''
+  {
+    "start_hour": 22,
+    "start_minute": 0,
+    "end_hour": 8,
+    "end_minute": 0,
+    "enabled": 1
+  }
+  '''
+# ---
+# name: test_decode_rpc_payload[get_status]
+  20001
+# ---
+# name: test_decode_rpc_payload[get_status].1
+  '''
+  {
+    "msg_ver": 2,
+    "msg_seq": 515,
+    "state": 8,
+    "battery": 100,
+    "clean_time": 5405,
+    "clean_area": 91287500,
+    "error_code": 0,
+    "map_present": 1,
+    "in_cleaning": 0,
+    "in_returning": 0,
+    "in_fresh_state": 1,
+    "lab_status": 1,
+    "water_box_status": 0,
+    "fan_power": 106,
+    "dnd_enabled": 1,
+    "map_status": 3,
+    "is_locating": 0,
+    "lock_status": 0,
+    "water_box_mode": 204,
+    "distance_off": 0,
+    "water_box_carriage_status": 0,
+    "mop_forbidden_enable": 0,
+    "unsave_map_reason": 4,
+    "unsave_map_flag": 0
+  }
+  '''
+# ---

tests/protocols/__snapshots__/test_v1_protocol.ambr

@@ -1,10 +1,13 @@
 """Tests for the v1 protocol message encoding and decoding."""
 
+import json
+import pathlib
 from collections.abc import Generator
 from unittest.mock import patch
 
 import pytest
 from freezegun import freeze_time
+from syrupy import SnapshotAssertion
 
 from roborock.containers import RoborockBase, UserData
 from roborock.exceptions import RoborockException
@@ -30,6 +33,11 @@
 )
 
 
+TESTDATA_PATH = pathlib.Path("tests/protocols/testdata/v1_protocol/")
+TESTDATA_FILES = list(TESTDATA_PATH.glob("*.json"))
+TESTDATA_IDS = [x.stem for x in TESTDATA_FILES]
+
+
 @pytest.fixture(autouse=True)
 def fixed_time_fixture() -> Generator[None, None, None]:
     """Fixture to freeze time for predictable request IDs."""
@@ -132,6 +140,25 @@ def test_decode_rpc_response(payload: bytes, expected: RoborockBase) -> None:
     assert decoded_message.data == expected
 
 
+@pytest.mark.parametrize("filename", TESTDATA_FILES, ids=TESTDATA_IDS)
+def test_decode_rpc_payload(filename: str, snapshot: SnapshotAssertion) -> None:
+    """Test decoding a v1 RPC response protocol message."""
+    with open(filename, "rb") as f:
+        payload = f.read()
+    # The values other than the payload are arbitrary
+    message = RoborockMessage(
+        protocol=RoborockMessageProtocol.GENERAL_RESPONSE,
+        payload=payload,
+        seq=12750,
+        version=b"1.0",
+        random=97431,
+        timestamp=1652547161,
+    )
+    decoded_message = decode_rpc_response(message)
+    assert decoded_message.request_id == snapshot
+    assert json.dumps(decoded_message.data, indent=2) == snapshot
+
+
 def test_create_map_response_decoder():
     """Test creating and using a map response decoder."""
     test_data = b"some map\n"

tests/protocols/test_v1_protocol.py

@@ -0,0 +1 @@
+{"t": 1755785801, "dps": {"102": "{\"id\":20002,\"result\":[{\"start_hour\":22,\"start_minute\":0,\"end_hour\":8,\"end_minute\":0,\"enabled\":1}]}"}}

tests/protocols/testdata/v1_protocol/get_dnd.json

@@ -0,0 +1 @@
+{"t": 1755785773, "dps": {"102": "{\"id\":20001,\"result\":[{\"msg_ver\":2,\"msg_seq\":515,\"state\":8,\"battery\":100,\"clean_time\":5405,\"clean_area\":91287500,\"error_code\":0,\"map_present\":1,\"in_cleaning\":0,\"in_returning\":0,\"in_fresh_state\":1,\"lab_status\":1,\"water_box_status\":0,\"fan_power\":106,\"dnd_enabled\":1,\"map_status\":3,\"is_locating\":0,\"lock_status\":0,\"water_box_mode\":204,\"distance_off\":0,\"water_box_carriage_status\":0,\"mop_forbidden_enable\":0,\"unsave_map_reason\":4,\"unsave_map_flag\":0}]}"}}