1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
|
"""Tests for the Maps related functionality."""
from unittest.mock import AsyncMock
import pytest
from roborock.devices.device import RoborockDevice
from roborock.devices.traits.v1.maps import MapsTrait
from roborock.devices.traits.v1.status import StatusTrait
from roborock.roborock_typing import RoborockCommand
from tests import mock_data
UPDATED_STATUS = {
**mock_data.STATUS,
"map_status": 123 * 4 + 3, # Set current map to 123
}
MULTI_MAP_LIST_DATA = [
{
"max_multi_map": 1,
"max_bak_map": 1,
"multi_map_count": 1,
"map_info": [
{
"mapFlag": 0,
"add_time": 1747132930,
"length": 0,
"name": "Map 1",
"bak_maps": [{"mapFlag": 4, "add_time": 1747132936}],
},
{
"mapFlag": 123,
"add_time": 1747132930,
"length": 0,
"name": "Map 2",
"bak_maps": [{"mapFlag": 4, "add_time": 1747132936}],
},
],
}
]
@pytest.fixture
def status_trait(device: RoborockDevice) -> StatusTrait:
"""Create a MapsTrait instance with mocked dependencies."""
assert device.v1_properties
return device.v1_properties.status
@pytest.fixture
def maps_trait(device: RoborockDevice) -> MapsTrait:
"""Create a MapsTrait instance with mocked dependencies."""
assert device.v1_properties
return device.v1_properties.maps
async def test_refresh_maps_trait(
maps_trait: MapsTrait,
mock_rpc_channel: AsyncMock,
mock_mqtt_rpc_channel: AsyncMock,
status_trait: StatusTrait,
) -> None:
"""Test successfully getting multi maps list."""
# Setup mock to return the sample multi maps list
mock_rpc_channel.send_command.side_effect = [
mock_data.STATUS, # Initial status fetch
]
mock_mqtt_rpc_channel.send_command.side_effect = [
MULTI_MAP_LIST_DATA,
]
await status_trait.refresh()
assert status_trait.current_map == 0
# Populating the status information gives us the current map
# flag, but we have not loaded the rest of the information.
assert maps_trait.current_map == 0
assert maps_trait.current_map_info is None
# Load the maps information
await maps_trait.refresh()
assert maps_trait.max_multi_map == 1
assert maps_trait.max_bak_map == 1
assert maps_trait.multi_map_count == 1
assert maps_trait.map_info
assert len(maps_trait.map_info) == 2
map_infos = maps_trait.map_info
assert len(map_infos) == 2
assert map_infos[0].map_flag == 0
assert map_infos[0].name == "Map 1"
assert map_infos[0].add_time == 1747132930
assert map_infos[1].map_flag == 123
assert map_infos[1].name == "Map 2"
assert map_infos[1].add_time == 1747132930
assert maps_trait.current_map == 0
assert maps_trait.current_map_info is not None
assert maps_trait.current_map_info.map_flag == 0
assert maps_trait.current_map_info.name == "Map 1"
# Verify the RPC call was made correctly
assert mock_rpc_channel.send_command.call_count == 1
mock_rpc_channel.send_command.assert_any_call(RoborockCommand.GET_STATUS)
assert mock_mqtt_rpc_channel.send_command.call_count == 1
mock_mqtt_rpc_channel.send_command.assert_any_call(RoborockCommand.GET_MULTI_MAPS_LIST)
async def test_set_current_map(
status_trait: StatusTrait,
maps_trait: MapsTrait,
mock_rpc_channel: AsyncMock,
mock_mqtt_rpc_channel: AsyncMock,
) -> None:
"""Test successfully setting the current map."""
assert hasattr(maps_trait, "mqtt_rpc_channel")
mock_rpc_channel.send_command.side_effect = [
mock_data.STATUS, # Initial status fetch
UPDATED_STATUS, # Response for refreshing status
]
mock_mqtt_rpc_channel.send_command.side_effect = [
MULTI_MAP_LIST_DATA, # Response for LOAD_MULTI_MAP
{}, # Response for setting the current map
]
await status_trait.refresh()
# First refresh to populate initial state
await maps_trait.refresh()
# Verify current map
assert maps_trait.current_map == 0
assert maps_trait.current_map_info
assert maps_trait.current_map_info.map_flag == 0
assert maps_trait.current_map_info.name == "Map 1"
# Call the method to set current map
await maps_trait.set_current_map(123)
# Verify the current map is updated
assert maps_trait.current_map == 123
assert maps_trait.current_map_info
assert maps_trait.current_map_info.map_flag == 123
assert maps_trait.current_map_info.name == "Map 2"
# Verify the command sent are:
# 1. GET_STATUS to get initial status
# 2. GET_MULTI_MAPS_LIST to get the map list
# 3. LOAD_MULTI_MAP to set the map
# 4. GET_STATUS to refresh the current map in status
assert mock_rpc_channel.send_command.call_count == 2
mock_rpc_channel.send_command.assert_any_call(RoborockCommand.GET_STATUS)
assert mock_mqtt_rpc_channel.send_command.call_count == 2
mock_mqtt_rpc_channel.send_command.assert_any_call(RoborockCommand.GET_MULTI_MAPS_LIST)
mock_mqtt_rpc_channel.send_command.assert_any_call(RoborockCommand.LOAD_MULTI_MAP, params=[123])
|
