refactor: export and define all simulator state default constants at package level

Author: allenporter

roborock/testing/__init__.py

@@ -66,9 +66,25 @@ async def test_start_vacuum_service():
 from roborock.testing.channel import FakeChannel
 from roborock.testing.cloud import FakeRoborockCloud, FakeUserState, FakeWebApiClient
 from roborock.testing.simulator import RoborockDeviceSimulator
-from roborock.testing.v1_simulator import V1VacuumSimulator
+from roborock.testing.v1_simulator import (
+    DEFAULT_APP_INIT,
+    DEFAULT_CLEAN_SUMMARY,
+    DEFAULT_CONSUMABLE,
+    DEFAULT_DND_TIMER,
+    DEFAULT_LAST_CLEAN_RECORD,
+    DEFAULT_NETWORK_INFO,
+    DEFAULT_STATUS,
+    V1VacuumSimulator,
+)
 
 __all__ = [
+    "DEFAULT_APP_INIT",
+    "DEFAULT_CLEAN_SUMMARY",
+    "DEFAULT_CONSUMABLE",
+    "DEFAULT_DND_TIMER",
+    "DEFAULT_LAST_CLEAN_RECORD",
+    "DEFAULT_NETWORK_INFO",
+    "DEFAULT_STATUS",
     "FakeChannel",
     "FakeRoborockCloud",
     "FakeUserState",

roborock/testing/v1_simulator.py

@@ -113,6 +113,57 @@ def _serialize_dataclass(obj: Any) -> dict[str, Any]:
     new_feature_info_2=8192,
 )
 
+DEFAULT_NETWORK_INFO = NetworkInfo(
+    ip="1.1.1.1",
+    ssid="test_wifi",
+    mac="aa:bb:cc:dd:ee:ff",
+    bssid="aa:bb:cc:dd:ee:ff",
+    rssi=-50,
+)
+
+DEFAULT_CONSUMABLE = Consumable(
+    main_brush_work_time=74382,
+    side_brush_work_time=74383,
+    filter_work_time=74384,
+    filter_element_work_time=0,
+    sensor_dirty_time=74385,
+    strainer_work_times=65,
+    dust_collection_work_times=25,
+    cleaning_brush_work_times=66,
+)
+
+DEFAULT_DND_TIMER = DnDTimer(
+    start_hour=22,
+    start_minute=0,
+    end_hour=7,
+    end_minute=0,
+    enabled=1,
+)
+
+DEFAULT_CLEAN_SUMMARY = CleanSummary(
+    clean_time=74382,
+    clean_area=1159182500,
+    clean_count=31,
+    dust_collection_count=25,
+    records=[1672543330, 1672458041],
+)
+
+DEFAULT_LAST_CLEAN_RECORD = CleanRecord(
+    begin=1672543330,
+    end=1672544638,
+    duration=1176,
+    area=20965000,
+    error=0,
+    complete=1,
+    start_type=RoborockStartType.app,
+    clean_type=RoborockCleanType.select_zone,
+    finish_reason=RoborockFinishReason.finished_cleaning_4,
+    dust_collection_status=1,
+    avoid_count=19,
+    wash_count=2,
+    map_flag=0,
+)
+
 
 class V1VacuumSimulator(RoborockDeviceSimulator):
     """Firmware simulator for a V1/L01 vacuum device.
@@ -129,84 +180,29 @@ class V1VacuumSimulator(RoborockDeviceSimulator):
     def __init__(
         self,
         duid: str = "fake_duid",
-        battery: int = 100,
-        state: RoborockStateCode | int = RoborockStateCode.charging,
-        fan_power: int = 102,  # balanced
-        dnd_enabled: int = 0,
-        mop_mode: int = 300,
-        water_box_mode: int = 200,
+        status: StatusV2 | None = None,
+        app_init: AppInitStatus | None = None,
+        network_info: NetworkInfo | None = None,
+        consumables: Consumable | None = None,
+        dnd_timer: DnDTimer | None = None,
+        clean_summary: CleanSummary | None = None,
+        last_clean_record: CleanRecord | None = None,
         custom_handlers: dict[str, Callable[[list[Any]], Any]] | None = None,
         device_info: HomeDataDevice | None = None,
         product: HomeDataProduct | None = None,
-        dss: int = 169,
-        dock_type: RoborockDockTypeCode | int = 3,
     ):
         super().__init__(duid=duid, device_info=device_info, product=product)
-        self.status = replace(DEFAULT_STATUS)
-        self.app_init = replace(DEFAULT_APP_INIT)
-        self.app_init.local_info = replace(DEFAULT_APP_INIT.local_info)
-
-        self.status.battery = battery
-        self.status.state = RoborockStateCode(state)
-        self.status.fan_power = fan_power
-        self.status.dnd_enabled = dnd_enabled
-        self.status.mop_mode = mop_mode
-        self.status.water_box_mode = water_box_mode
-        self.status.dss = dss
-        self.status.dock_type = RoborockDockTypeCode(dock_type)
+        self.status = status or replace(DEFAULT_STATUS)
+        self.app_init = app_init or replace(DEFAULT_APP_INIT)
+        if app_init is None:
+            self.app_init.local_info = replace(DEFAULT_APP_INIT.local_info)
+        self.network_info = network_info or replace(DEFAULT_NETWORK_INFO)
+        self.consumables = consumables or replace(DEFAULT_CONSUMABLE)
+        self.dnd_timer = dnd_timer or replace(DEFAULT_DND_TIMER)
+        self.clean_summary = clean_summary or replace(DEFAULT_CLEAN_SUMMARY)
+        self.last_clean_record = last_clean_record or replace(DEFAULT_LAST_CLEAN_RECORD)
         self.custom_handlers = custom_handlers or {}
 
-        self.network_info = NetworkInfo(
-            ip="1.1.1.1",
-            ssid="test_wifi",
-            mac="aa:bb:cc:dd:ee:ff",
-            bssid="aa:bb:cc:dd:ee:ff",
-            rssi=-50,
-        )
-
-        self.consumables = Consumable(
-            main_brush_work_time=74382,
-            side_brush_work_time=74383,
-            filter_work_time=74384,
-            filter_element_work_time=0,
-            sensor_dirty_time=74385,
-            strainer_work_times=65,
-            dust_collection_work_times=25,
-            cleaning_brush_work_times=66,
-        )
-
-        self.dnd_timer = DnDTimer(
-            start_hour=22,
-            start_minute=0,
-            end_hour=7,
-            end_minute=0,
-            enabled=1,
-        )
-
-        self.clean_summary = CleanSummary(
-            clean_time=74382,
-            clean_area=1159182500,
-            clean_count=31,
-            dust_collection_count=25,
-            records=[1672543330, 1672458041],
-        )
-
-        self.last_clean_record = CleanRecord(
-            begin=1672543330,
-            end=1672544638,
-            duration=1176,
-            area=20965000,
-            error=0,
-            complete=1,
-            start_type=RoborockStartType.app,
-            clean_type=RoborockCleanType.select_zone,
-            finish_reason=RoborockFinishReason.finished_cleaning_4,
-            dust_collection_status=1,
-            avoid_count=19,
-            wash_count=2,
-            map_flag=0,
-        )
-
         # Set up default handlers dictionary
         self.default_handlers: dict[str, Callable[[Any], Any]] = {
             "get_status": lambda params: [self.get_status_dict()],

tests/testing/test_cloud.py

@@ -1,11 +1,13 @@
+from dataclasses import replace
+
 import pytest
 
 from roborock.data import HomeDataDevice, HomeDataProduct, RoborockCategory, UserData
 from roborock.data.v1 import RoborockStateCode
 from roborock.devices.cache import InMemoryCache
 from roborock.devices.device_manager import UserParams, create_device_manager
 from roborock.exceptions import RoborockException
-from roborock.testing import FakeRoborockCloud, V1VacuumSimulator
+from roborock.testing import DEFAULT_STATUS, FakeRoborockCloud, V1VacuumSimulator
 from roborock.web_api import RoborockApiClient
 from tests import mock_data
 
@@ -15,7 +17,10 @@
 async def test_fake_roborock_cloud():
     """Verify that FakeRoborockCloud can discover devices via fake HTTP requests and connect them."""
     cloud = FakeRoborockCloud()
-    fake_device = V1VacuumSimulator(duid="living_room_s7", battery=99, state=RoborockStateCode.charging)
+    fake_device = V1VacuumSimulator(
+        duid="living_room_s7",
+        status=replace(DEFAULT_STATUS, battery=99, state=RoborockStateCode.charging),
+    )
     cloud.add_device(fake_device)
 
     with cloud.patch_device_manager():
@@ -81,7 +86,10 @@ async def test_cloud_dynamic_device_addition():
     cloud = FakeRoborockCloud()
 
     with cloud.patch_device_manager():
-        fake_device = V1VacuumSimulator(duid="dynamic_s7", battery=42)
+        fake_device = V1VacuumSimulator(
+            duid="dynamic_s7",
+            status=replace(DEFAULT_STATUS, battery=42),
+        )
         cloud.add_device(fake_device)
 
         manager = await create_device_manager(

tests/testing/test_v1_simulator.py

@@ -1,3 +1,5 @@
+from dataclasses import replace
+
 import pytest
 
 from roborock.data import UserData
@@ -7,7 +9,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.testing import FakeRoborockCloud, V1VacuumSimulator
+from roborock.testing import DEFAULT_STATUS, FakeRoborockCloud, V1VacuumSimulator
 from tests import mock_data
 
 USER_DATA = UserData.from_dict(mock_data.USER_DATA)
@@ -71,7 +73,7 @@ async def test_trait_dnd_refresh():
 async def test_trait_fan_speed_change():
     """Verify that sending set_custom_mode updates the simulator fan speed and the trait reflects it."""
     cloud = FakeRoborockCloud()
-    fake_device = V1VacuumSimulator(duid="s7_fan", fan_power=102)
+    fake_device = V1VacuumSimulator(duid="s7_fan", status=replace(DEFAULT_STATUS, fan_power=102))
     device = await _create_connected_device(cloud, fake_device)
 
     await device.v1_properties.status.refresh()
@@ -100,7 +102,7 @@ async def test_trait_clean_summary_refresh():
 async def test_trait_multiple_state_transitions():
     """Verify a sequence of state transitions through trait commands."""
     cloud = FakeRoborockCloud()
-    fake_device = V1VacuumSimulator(duid="s7_transitions", state=RoborockStateCode.charging)
+    fake_device = V1VacuumSimulator(duid="s7_transitions")
     device = await _create_connected_device(cloud, fake_device)
 
     # Start cleaning
@@ -123,7 +125,10 @@ async def test_trait_multiple_state_transitions():
 async def test_trait_push_update_propagation():
     """Verify that unsolicited push updates propagate to client traits without refresh."""
     cloud = FakeRoborockCloud()
-    fake_device = V1VacuumSimulator(duid="s7_push", battery=99, state=RoborockStateCode.charging)
+    fake_device = V1VacuumSimulator(
+        duid="s7_push",
+        status=replace(DEFAULT_STATUS, battery=99, state=RoborockStateCode.charging),
+    )
     device = await _create_connected_device(cloud, fake_device)
 
     await device.v1_properties.status.refresh()
@@ -160,7 +165,15 @@ def custom_get_status(params):
 
 async def test_trait_properties_and_dss_config():
     """Verify that properties, dss config, and dock_type config are correctly exposed on the simulator."""
-    fake_device = V1VacuumSimulator(duid="s7_properties", state=RoborockStateCode.cleaning, dss=42, dock_type=5)
+    fake_device = V1VacuumSimulator(
+        duid="s7_properties",
+        status=replace(
+            DEFAULT_STATUS,
+            state=RoborockStateCode.cleaning,
+            dss=42,
+            dock_type=RoborockDockTypeCode(5),
+        ),
+    )
     assert fake_device.in_cleaning == RoborockInCleaning.global_clean_not_complete
     assert fake_device.in_returning == 0
     assert fake_device.charge_status == RoborockChargeStatus.charge_waiting