test_containers.py

"""Test cases for the containers module."""

from dataclasses import dataclass
from typing import Any

from syrupy import SnapshotAssertion

from roborock import CleanRecord, CleanSummary, Consumable, DnDTimer, HomeData, UserData
from roborock.b01_containers import (
    B01Fault,
    B01Props,
    SCWindMapping,
    WorkStatusMapping,
)
from roborock.code_mappings import (
    RoborockCategory,
    RoborockDockErrorCode,
    RoborockDockTypeCode,
    RoborockErrorCode,
    RoborockStateCode,
)
from roborock.containers import MultiMapsList, RoborockBase, get_custom_status

from .mock_data import (
    CLEAN_RECORD,
    CLEAN_SUMMARY,
    CONSUMABLE,
    DND_TIMER,
    HOME_DATA_RAW,
    K_VALUE,
    LOCAL_KEY,
    PRODUCT_ID,
    STATUS,
    USER_DATA,
)


@dataclass
class SimpleObject(RoborockBase):
    """Simple object for testing serialization."""

    name: str | None = None
    value: int | None = None


@dataclass
class ComplexObject(RoborockBase):
    """Complex object for testing serialization."""

    simple: SimpleObject | None = None
    items: list[str] | None = None
    value: int | None = None
    nested_dict: dict[str, SimpleObject] | None = None
    nested_list: list[SimpleObject] | None = None
    any: Any | None = None


def test_simple_object() -> None:
    """Test serialization and deserialization of a simple object."""

    obj = SimpleObject(name="Test", value=42)
    serialized = obj.as_dict()
    assert serialized == {"name": "Test", "value": 42}
    deserialized = SimpleObject.from_dict(serialized)
    assert deserialized.name == "Test"
    assert deserialized.value == 42


def test_complex_object() -> None:
    """Test serialization and deserialization of a complex object."""
    simple = SimpleObject(name="Nested", value=100)
    obj = ComplexObject(
        simple=simple,
        items=["item1", "item2"],
        value=200,
        nested_dict={
            "nested1": SimpleObject(name="Nested1", value=1),
            "nested2": SimpleObject(name="Nested2", value=2),
        },
        nested_list=[SimpleObject(name="Nested3", value=3), SimpleObject(name="Nested4", value=4)],
        any="This can be anything",
    )
    serialized = obj.as_dict()
    assert serialized == {
        "simple": {"name": "Nested", "value": 100},
        "items": ["item1", "item2"],
        "value": 200,
        "nestedDict": {
            "nested1": {"name": "Nested1", "value": 1},
            "nested2": {"name": "Nested2", "value": 2},
        },
        "nestedList": [
            {"name": "Nested3", "value": 3},
            {"name": "Nested4", "value": 4},
        ],
        "any": "This can be anything",
    }
    deserialized = ComplexObject.from_dict(serialized)
    assert deserialized.simple.name == "Nested"
    assert deserialized.simple.value == 100
    assert deserialized.items == ["item1", "item2"]
    assert deserialized.value == 200
    assert deserialized.nested_dict == {
        "nested1": SimpleObject(name="Nested1", value=1),
        "nested2": SimpleObject(name="Nested2", value=2),
    }
    assert deserialized.nested_list == [
        SimpleObject(name="Nested3", value=3),
        SimpleObject(name="Nested4", value=4),
    ]
    assert deserialized.any == "This can be anything"


def test_ignore_unknown_keys() -> None:
    """Test that we don't fail on unknown keys."""
    data = {
        "ignored_key": "This key should be ignored",
        "name": "named_object",
        "value": 42,
    }
    deserialized = SimpleObject.from_dict(data)
    assert deserialized.name == "named_object"
    assert deserialized.value == 42


def test_user_data():
    ud = UserData.from_dict(USER_DATA)
    assert ud.uid == 123456
    assert ud.tokentype == "token_type"
    assert ud.token == "abc123"
    assert ud.rruid == "abc123"
    assert ud.region == "us"
    assert ud.country == "US"
    assert ud.countrycode == "1"
    assert ud.nickname == "user_nickname"
    assert ud.rriot.u == "user123"
    assert ud.rriot.s == "pass123"
    assert ud.rriot.h == "unknown123"
    assert ud.rriot.k == K_VALUE
    assert ud.rriot.r.r == "US"
    assert ud.rriot.r.a == "https://api-us.roborock.com"
    assert ud.rriot.r.m == "tcp://mqtt-us.roborock.com:8883"
    assert ud.rriot.r.l == "https://wood-us.roborock.com"
    assert ud.tuya_device_state == 2
    assert ud.avatarurl == "https://files.roborock.com/iottest/default_avatar.png"


def test_home_data():
    hd = HomeData.from_dict(HOME_DATA_RAW)
    assert hd.id == 123456
    assert hd.name == "My Home"
    assert hd.lon is None
    assert hd.lat is None
    assert hd.geo_name is None
    product = hd.products[0]
    assert product.id == PRODUCT_ID
    assert product.name == "Roborock S7 MaxV"
    assert product.code == "a27"
    assert product.model == "roborock.vacuum.a27"
    assert product.icon_url is None
    assert product.attribute is None
    assert product.capability == 0
    assert product.category == RoborockCategory.VACUUM
    schema = product.schema
    assert schema[0].id == "101"
    assert schema[0].name == "rpc_request"
    assert schema[0].code == "rpc_request_code"
    assert schema[0].mode == "rw"
    assert schema[0].type == "RAW"
    assert schema[0].product_property is None
    assert schema[0].desc is None
    device = hd.devices[0]
    assert device.duid == "abc123"
    assert device.name == "Roborock S7 MaxV"
    assert device.attribute is None
    assert device.active_time == 1672364449
    assert device.local_key == LOCAL_KEY
    assert device.runtime_env is None
    assert device.time_zone_id == "America/Los_Angeles"
    assert device.icon_url == "no_url"
    assert device.product_id == "product-id-123"
    assert device.lon is None
    assert device.lat is None
    assert not device.share
    assert device.share_time is None
    assert device.online
    assert device.fv == "02.56.02"
    assert device.pv == "1.0"
    assert device.room_id == 2362003
    assert device.tuya_uuid is None
    assert not device.tuya_migrated
    assert device.extra == '{"RRPhotoPrivacyVersion": "1"}'
    assert device.sn == "abc123"
    assert device.feature_set == "2234201184108543"
    assert device.new_feature_set == "0000000000002041"
    # status = device.device_status
    # assert status.name ==
    assert device.silent_ota_switch
    assert hd.rooms[0].id == 2362048
    assert hd.rooms[0].name == "Example room 1"


def test_serialize_and_unserialize():
    ud = UserData.from_dict(USER_DATA)
    ud_dict = ud.as_dict()
    assert ud_dict == USER_DATA


def test_consumable():
    c = Consumable.from_dict(CONSUMABLE)
    assert c.main_brush_work_time == 74382
    assert c.side_brush_work_time == 74383
    assert c.filter_work_time == 74384
    assert c.filter_element_work_time == 0
    assert c.sensor_dirty_time == 74385
    assert c.strainer_work_times == 65
    assert c.dust_collection_work_times == 25
    assert c.cleaning_brush_work_times == 66


def test_status(s7_device_features):
    status = get_custom_status(
        s7_device_features,
        "US",
    )
    s = status.from_dict(STATUS)
    assert s.msg_ver == 2
    assert s.msg_seq == 458
    assert s.state == RoborockStateCode.charging
    assert s.battery == 100
    assert s.clean_time == 1176
    assert s.clean_area == 20965000
    assert s.square_meter_clean_area == 21.0
    assert s.error_code == RoborockErrorCode.none
    assert s.map_present == 1
    assert s.in_cleaning == 0
    assert s.in_returning == 0
    assert s.in_fresh_state == 1
    assert s.lab_status == 1
    assert s.water_box_status == 1
    assert s.back_type == -1
    assert s.wash_phase == 0
    assert s.wash_ready == 0
    assert s.fan_power == 102
    assert s.dnd_enabled == 0
    assert s.map_status == 3
    assert s.current_map == 0
    assert s.is_locating == 0
    assert s.lock_status == 0
    assert s.water_box_mode == 203
    assert s.water_box_carriage_status == 1
    assert s.mop_forbidden_enable == 1
    assert s.camera_status == 3457
    assert s.is_exploring == 0
    assert s.home_sec_status == 0
    assert s.home_sec_enable_password == 0
    assert s.adbumper_status == [0, 0, 0]
    assert s.water_shortage_status == 0
    assert s.dock_type == RoborockDockTypeCode.empty_wash_fill_dock
    assert s.dust_collection_status == 0
    assert s.auto_dust_collection == 1
    assert s.avoid_count == 19
    assert s.mop_mode == 300
    assert s.debug_mode == 0
    assert s.collision_avoid_status == 1
    assert s.switch_map_mode == 0
    assert s.dock_error_status == RoborockDockErrorCode.ok
    assert s.charge_status == 1
    assert s.unsave_map_reason == 0
    assert s.unsave_map_flag == 0
    assert s.fan_power == 102
    assert s.fan_speed == "BALANCED"
    assert s.mop_route == "STANDARD"
    assert s.water_mode == "INTENSE"


def test_current_map(s7_device_features) -> None:
    """Test the current map logic based on map status."""
    status = get_custom_status(
        s7_device_features,
        "US",
    )
    s = status.from_dict(STATUS)
    assert s.map_status == 3
    assert s.current_map == 0

    s.map_status = 7
    assert s.current_map == 1

    s.map_status = 11
    assert s.current_map == 2

    s.map_status = None
    assert not s.current_map


def test_dnd_timer():
    dnd = DnDTimer.from_dict(DND_TIMER)
    assert dnd.start_hour == 22
    assert dnd.start_minute == 0
    assert dnd.end_hour == 7
    assert dnd.end_minute == 0
    assert dnd.enabled == 1


def test_clean_summary():
    cs = CleanSummary.from_dict(CLEAN_SUMMARY)
    assert cs.clean_time == 74382
    assert cs.clean_area == 1159182500
    assert cs.square_meter_clean_area == 1159.2
    assert cs.clean_count == 31
    assert cs.dust_collection_count == 25
    assert cs.records
    assert len(cs.records) == 2
    assert cs.records[1] == 1672458041


def test_clean_record():
    cr = CleanRecord.from_dict(CLEAN_RECORD)
    assert cr.begin == 1672543330
    assert cr.end == 1672544638
    assert cr.duration == 1176
    assert cr.area == 20965000
    assert cr.square_meter_area == 21.0
    assert cr.error == 0
    assert cr.complete == 1
    assert cr.start_type == 2
    assert cr.clean_type == 3
    assert cr.finish_reason == 56
    assert cr.dust_collection_status == 1
    assert cr.avoid_count == 19
    assert cr.wash_count == 2
    assert cr.map_flag == 0


def test_no_value(s7_device_features):
    modified_status = STATUS.copy()
    modified_status["dock_type"] = 9999
    status = get_custom_status(
        s7_device_features,
        "US",
    )

    s = status.from_dict(modified_status)
    assert s.dock_type == RoborockDockTypeCode.unknown
    assert -9999 not in RoborockDockTypeCode.keys()
    assert "missing" not in RoborockDockTypeCode.values()


def test_b01props_deserialization():
    """Test that B01Props can be deserialized after its module is dynamically imported."""

    B01_PROPS_MOCK_DATA = {
        "status": 6,
        "fault": 510,
        "wind": 3,
        "water": 2,
        "mode": 1,
        "quantity": 1,
        "alarm": 0,
        "volume": 60,
        "hypa": 90,
        "mainBrush": 80,
        "sideBrush": 70,
        "mopLife": 60,
        "mainSensor": 50,
        "netStatus": {
            "rssi": "-60",
            "loss": 1,
            "ping": 20,
            "ip": "192.168.1.102",
            "mac": "BB:CC:DD:EE:FF:00",
            "ssid": "MyOtherWiFi",
            "frequency": 2.4,
            "bssid": "00:FF:EE:DD:CC:BB",
        },
        "repeatState": 1,
        "tankState": 0,
        "sweepType": 0,
        "cleanPathPreference": 1,
        "clothState": 1,
        "timeZone": -5,
        "timeZoneInfo": "America/New_York",
        "language": 2,
        "cleaningTime": 1500,
        "realCleanTime": 1400,
        "cleaningArea": 600000,
        "customType": 1,
        "sound": 0,
        "workMode": 3,
        "stationAct": 1,
        "chargeState": 0,
        "currentMapId": 2,
        "mapNum": 3,
        "dustAction": 0,
        "quietIsOpen": 1,
        "quietBeginTime": 23,
        "quietEndTime": 7,
        "cleanFinish": 0,
        "voiceType": 2,
        "voiceTypeVersion": 1,
        "orderTotal": {"total": 12, "enable": 0},
        "buildMap": 0,
        "privacy": {
            "aiRecognize": 1,
            "dirtRecognize": 1,
            "petRecognize": 1,
            "carpetTurbo": 1,
            "carpetAvoid": 1,
            "carpetShow": 1,
            "mapUploads": 1,
            "aiAgent": 1,
            "aiAvoidance": 1,
            "recordUploads": 1,
            "alongFloor": 1,
            "autoUpgrade": 1,
        },
        "dustAutoState": 0,
        "dustFrequency": 1,
        "childLock": 1,
        "multiFloor": 0,
        "mapSave": 0,
        "lightMode": 0,
        "greenLaser": 0,
        "dustBagUsed": 1,
        "orderSaveMode": 0,
        "manufacturer": "Roborock-Test",
        "backToWash": 0,
        "chargeStationType": 2,
        "pvCutCharge": 1,
        "pvCharging": {"status": 1, "beginTime": 10, "endTime": 18},
        "serialNumber": "987654321",
        "recommend": {"sill": 0, "wall": 0, "roomId": [4, 5, 6]},
        "addSweepStatus": 1,
    }

    deserialized = B01Props.from_dict(B01_PROPS_MOCK_DATA)
    assert isinstance(deserialized, B01Props)
    assert deserialized.fault == B01Fault.F_510
    assert deserialized.status == WorkStatusMapping.SWEEP_MOPING_2
    assert deserialized.wind == SCWindMapping.SUPER_STRONG
    assert deserialized.net_status.ip == "192.168.1.102"


def test_multi_maps_list_info(snapshot: SnapshotAssertion) -> None:
    """Test that MultiMapsListInfo can be deserialized correctly."""
    data = {
        "max_multi_map": 4,
        "max_bak_map": 1,
        "multi_map_count": 2,
        "map_info": [
            {
                "mapFlag": 0,
                "add_time": 1757636125,
                "length": 10,
                "name": "Downstairs",
                "bak_maps": [{"mapFlag": 4, "add_time": 1739205442}],
                "rooms": [
                    {"id": 16, "tag": 12, "iot_name_id": "6990322", "iot_name": "Room"},
                    {"id": 17, "tag": 15, "iot_name_id": "7140977", "iot_name": "Room"},
                    {"id": 18, "tag": 12, "iot_name_id": "6985623", "iot_name": "Room"},
                    {"id": 19, "tag": 14, "iot_name_id": "6990378", "iot_name": "Room"},
                    {"id": 20, "tag": 10, "iot_name_id": "7063728", "iot_name": "Room"},
                    {"id": 22, "tag": 12, "iot_name_id": "6995506", "iot_name": "Room"},
                    {"id": 23, "tag": 15, "iot_name_id": "7140979", "iot_name": "Room"},
                    {"id": 25, "tag": 13, "iot_name_id": "6990383", "iot_name": "Room"},
                    {"id": 24, "tag": -1, "iot_name_id": "-1", "iot_name": "Room"},
                ],
                "furnitures": [
                    {"id": 1, "type": 46, "subtype": 2},
                    {"id": 2, "type": 47, "subtype": 0},
                    {"id": 3, "type": 56, "subtype": 0},
                    {"id": 4, "type": 43, "subtype": 0},
                    {"id": 5, "type": 44, "subtype": 0},
                    {"id": 6, "type": 44, "subtype": 0},
                    {"id": 7, "type": 44, "subtype": 0},
                    {"id": 8, "type": 46, "subtype": 0},
                    {"id": 9, "type": 46, "subtype": 0},
                ],
            },
            {
                "mapFlag": 1,
                "add_time": 1734283706,
                "length": 5,
                "name": "Foyer",
                "bak_maps": [{"mapFlag": 5, "add_time": 1728184107}],
                "rooms": [],
                "furnitures": [],
            },
        ],
    }
    deserialized = MultiMapsList.from_dict(data)
    assert isinstance(deserialized, MultiMapsList)
    assert deserialized == snapshot


def test_accurate_map_flag(s7_device_features) -> None:
    """Test that we parse the map flag accurately."""
    status = get_custom_status(
        s7_device_features,
        "US",
    )
    s = status.from_dict(STATUS)
    assert s.current_map == 0
    s = status.from_dict(
        {
            **STATUS,
            "map_status": 252,  # Code for no map
        }
    )
    assert s.current_map is None