chore: testing and PR comments

Author: Lash-L

roborock/data/b01_q7/b01_q7_containers.py

@@ -1,3 +1,4 @@
+import datetime
 from dataclasses import dataclass, field
 
 from ..containers import RoborockBase
@@ -213,6 +214,7 @@ class CleanRecordDetail(RoborockBase):
     method: int | None = None
     record_use_time: int | None = None
     clean_count: int | None = None
+    # This is seemingly returned in meters (non-squared)
     record_clean_area: int | None = None
     record_clean_mode: int | None = None
     record_clean_way: int | None = None
@@ -222,24 +224,45 @@ class CleanRecordDetail(RoborockBase):
     clean_current_map: int | None = None
     record_map_url: str | None = None
 
+    @property
+    def start_datetime(self) -> datetime.datetime | None:
+        """Convert the start datetime into a datetime object."""
+        if self.record_start_time is not None:
+            return datetime.datetime.fromtimestamp(self.record_start_time).astimezone(datetime.UTC)
+        return None
+
+    @property
+    def square_meters_area_cleaned(self) -> float | None:
+        """Returns the area cleaned in square meters."""
+        if self.record_clean_area is not None:
+            return self.record_clean_area / 100
+        return None
+
 
 @dataclass
 class CleanRecordListItem(RoborockBase):
     """Represents an entry in the clean record list returned by `service.get_record_list`."""
 
     url: str | None = None
-    detail: str | dict | None = None
+    detail: str | None = None
 
 
 @dataclass
 class CleanRecordList(RoborockBase):
     """Represents the clean record list response from `service.get_record_list`."""
 
     total_area: int | None = None
-    total_time: int | None = None
+    total_time: int | None = None  # stored in seconds
     total_count: int | None = None
     record_list: list[CleanRecordListItem] = field(default_factory=list)
 
+    @property
+    def square_meters_area_cleaned(self) -> float | None:
+        """Returns the area cleaned in square meters."""
+        if self.total_area is not None:
+            return self.total_area / 100
+        return None
+
 
 @dataclass
 class CleanRecordSummary(RoborockBase):

roborock/devices/traits/b01/q7/clean_summary.py

@@ -35,16 +35,16 @@ def __init__(self, channel: MqttChannel) -> None:
 
     async def refresh(self) -> None:
         """Refresh totals and last record detail from the device."""
-        record_list = await self.get_record_list()
+        record_list = await self._get_record_list()
 
         self.total_time = record_list.total_time
         self.total_area = record_list.total_area
         self.total_count = record_list.total_count
 
-        details = await self.get_clean_record_details(record_list=record_list)
+        details = await self._get_clean_record_details(record_list=record_list)
         self.last_record_detail = details[0] if details else None
 
-    async def get_record_list(self) -> CleanRecordList:
+    async def _get_record_list(self) -> CleanRecordList:
         """Fetch the raw device clean record list (`service.get_record_list`)."""
         result = await send_decoded_command(
             self._channel,
@@ -55,33 +55,21 @@ async def get_record_list(self) -> CleanRecordList:
             raise TypeError(f"Unexpected response type for GET_RECORD_LIST: {type(result).__name__}: {result!r}")
         return CleanRecordList.from_dict(result)
 
-    @staticmethod
-    def _parse_record_detail(detail: dict | str | None) -> CleanRecordDetail | None:
-        if detail is None:
-            return None
-        if isinstance(detail, str):
-            try:
-                parsed = json.loads(detail)
-            except json.JSONDecodeError as ex:
-                raise RoborockException(f"Invalid B01 record detail JSON: {detail!r}") from ex
-            if not isinstance(parsed, dict):
-                raise RoborockException(f"Unexpected B01 record detail type: {type(parsed).__name__}: {parsed!r}")
-            return CleanRecordDetail.from_dict(parsed)
-        if isinstance(detail, dict):
-            return CleanRecordDetail.from_dict(detail)
-        raise RoborockException(f"Unexpected B01 record detail type: {type(detail).__name__}: {detail!r}")
-
-    async def get_clean_record_details(self, *, record_list: CleanRecordList | None = None) -> list[CleanRecordDetail]:
+    async def _get_clean_record_details(self, *, record_list: CleanRecordList) -> list[CleanRecordDetail]:
         """Return parsed record detail objects (newest-first)."""
-        if record_list is None:
-            record_list = await self.get_record_list()
-
         details: list[CleanRecordDetail] = []
         for item in record_list.record_list:
-            parsed = self._parse_record_detail(item.detail)
+            if item.detail is None:
+                continue
+            try:
+                parsed = json.loads(item.detail)
+            except json.JSONDecodeError as ex:
+                raise RoborockException(f"Invalid B01 record detail JSON: {item.detail!r}") from ex
+            parsed = CleanRecordDetail.from_dict(parsed)
+
             if parsed is not None:
                 details.append(parsed)
 
-        # The app returns the newest record at the end of record_list; reverse so newest is first (index 0).
+        # The server returns the newest record at the end of record_list; reverse so newest is first (index 0).
         details.reverse()
         return details

tests/data/b01_q7/test_b01_q7_containers.py

@@ -10,7 +10,6 @@
     SCWindMapping,
     WorkStatusMapping,
 )
-from roborock.devices.traits.b01.q7.clean_summary import CleanSummaryTrait
 
 
 def test_b01props_deserialization():
@@ -141,8 +140,15 @@ def test_b01_q7_clean_record_list_parses_detail_fields():
     assert isinstance(parsed, CleanRecordList)
     assert parsed.record_list[0].url == "/userdata/record_map/1766368207_1766368283_0_clean_map.bin"
 
-    detail = CleanSummaryTrait._parse_record_detail(parsed.record_list[0].detail)
+    detail_dict = json.loads(parsed.record_list[0].detail or "{}")
+    detail = CleanRecordDetail.from_dict(detail_dict)
     assert isinstance(detail, CleanRecordDetail)
+    assert detail.record_start_time == 1766368207
+    assert detail.record_use_time == 60
+    assert detail.record_clean_area == 85
+    assert detail.record_clean_mode == 0
+    assert detail.record_task_status == 20
+    assert detail.record_map_url == "/userdata/record_map/1766368207_1766368283_0_clean_map.bin"
     assert detail.method == 0
     assert detail.clean_count == 1
     assert detail.record_clean_way == 0

tests/devices/traits/b01/q7/__init__.py

@@ -0,0 +1,41 @@
+import json
+from typing import Any
+
+from Crypto.Cipher import AES
+from Crypto.Util.Padding import pad
+
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi
+from roborock.roborock_message import RoborockMessage, RoborockMessageProtocol
+from tests.fixtures.channel_fixtures import FakeChannel
+
+
+class B01MessageBuilder:
+    """Helper class to build B01 RPC response messages for tests."""
+
+    def __init__(self) -> None:
+        self.msg_id = 123456789
+        self.seq = 2020
+
+    def build(self, data: dict[str, Any] | str, code: int | None = None) -> RoborockMessage:
+        """Build an encoded B01 RPC response message."""
+        message: dict[str, Any] = {
+            "msgId": str(self.msg_id),
+            "data": data,
+        }
+        if code is not None:
+            message["code"] = code
+        return self._build_dps(message)
+
+    def _build_dps(self, message: dict[str, Any] | str) -> RoborockMessage:
+        """Build an encoded B01 RPC response message."""
+        dps_payload = {"dps": {"10000": json.dumps(message)}}
+        self.seq += 1
+        return RoborockMessage(
+            protocol=RoborockMessageProtocol.RPC_RESPONSE,
+            payload=pad(
+                json.dumps(dps_payload).encode(),
+                AES.block_size,
+            ),
+            version=b"B01",
+            seq=self.seq,
+        )

tests/devices/traits/b01/q7/conftest.py

@@ -0,0 +1,42 @@
+import math
+import time
+from collections.abc import Generator
+from unittest.mock import patch
+
+import pytest
+
+from roborock.devices.traits.b01.q7 import Q7PropertiesApi
+from tests.fixtures.channel_fixtures import FakeChannel
+
+from . import B01MessageBuilder
+
+
+@pytest.fixture(name="fake_channel")
+def fake_channel_fixture() -> FakeChannel:
+    return FakeChannel()
+
+
+@pytest.fixture(name="q7_api")
+def q7_api_fixture(fake_channel: FakeChannel) -> Q7PropertiesApi:
+    return Q7PropertiesApi(fake_channel)  # type: ignore[arg-type]
+
+
+@pytest.fixture(name="expected_msg_id", autouse=True)
+def next_message_id_fixture() -> Generator[int, None, None]:
+    """Fixture to patch get_next_int to return the expected message ID.
+
+    We pick an arbitrary number, but just need it to ensure we can craft a fake
+    response with the message id matched to the outgoing RPC.
+    """
+    expected_msg_id = math.floor(time.time())
+
+    # Patch get_next_int to return our expected msg_id so the channel waits for it
+    with patch("roborock.protocols.b01_q7_protocol.get_next_int", return_value=expected_msg_id):
+        yield expected_msg_id
+
+
+@pytest.fixture(name="message_builder")
+def message_builder_fixture(expected_msg_id: int) -> B01MessageBuilder:
+    builder = B01MessageBuilder()
+    builder.msg_id = expected_msg_id
+    return builder