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
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
|
import pytest
from anthemav.protocol import LOOKUP, MODEL_X20, ALM_NUMBER_x20
from anthemav import AVR
from unittest.mock import call, patch
@pytest.mark.asyncio
class TestProtocol:
def test_default_alm_list(self):
avr = AVR()
avr.set_model_command("MRX 1120")
almList = avr.audio_listening_mode_list
assert almList is not None
assert len(almList) == 15
async def test_restricted_alm_list(self):
avr = AVR()
await avr._parse_message("IDMMRX 520")
almList = avr.audio_listening_mode_list
assert almList is not None
assert len(almList) == 8
def test_set_alm_text(self):
avr = AVR()
avr.set_model_command("MRX 520")
with patch.object(avr, "command") as mock:
avr.audio_listening_mode_text = "Neo Music"
mock.assert_called_once_with("Z1ALM06")
def test_all_alm_matchnumber(self):
for alm in list(LOOKUP["Z1ALM"].values())[1:]:
assert alm in ALM_NUMBER_x20
async def test_power_on_force_refresh(self):
avr = AVR()
with patch.object(avr, "query") as mock, patch.object(avr, "_loop"):
await avr._parse_message("Z1AIC;")
mock.assert_called_once_with("Z1POW")
def test_mute_force_refresh(self):
avr = AVR()
with patch.object(avr, "query") as mock:
avr.zones[1].mute = True
mock.assert_called_once_with("Z1MUT")
def test_populate_input_x20(self):
avr = AVR()
with patch.object(avr, "query") as mock:
avr._populate_inputs(2)
mock.assert_any_call("ISN01")
mock.assert_called_with("ISN02")
def test_populate_input_x40(self):
avr = AVR()
with patch.object(avr, "query") as mock:
avr.set_model_command("MRX 1140")
avr._populate_inputs(2)
mock.assert_any_call("IS1IN")
mock.assert_any_call("IS1ARC")
mock.assert_any_call("IS2IN")
mock.assert_any_call("IS2ARC")
async def test_parse_input_x40(self):
avr = AVR()
with patch.object(avr, "query"):
await avr._parse_message("IS3INName")
assert avr._input_names.get(3, "") == "Name"
async def test_parse_arc_x40(self):
avr = AVR()
with patch.object(avr, "query"), patch.object(avr, "_loop"):
avr.set_model_command("MRX 1140")
avr.zones[1].input_number = 2
await avr._parse_message("Z1INP2")
await avr._parse_message("IS2ARC1")
await avr._parse_message("IS1ARC0")
assert avr.arc is True
async def test_parse_arc_x20(self):
avr = AVR()
with patch.object(avr, "query"):
avr.set_model_command("MRX 520")
await avr._parse_message("Z1ARC1")
assert avr.arc is True
async def test_set_arc_x20(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "command") as mock:
avr._model_series = MODEL_X20
avr.arc = True
mock.assert_called_once_with("Z1ARC1")
async def test_set_arc_x40(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "query"), patch.object(
avr, "command"
) as mock, patch.object(avr, "_loop"):
avr.set_model_command("MRX 1140")
await avr._parse_message("Z1INP3")
avr.arc = True
mock.assert_called_once_with("IS3ARC1")
@pytest.mark.parametrize(
"model,zone",
[
("MRX 520", 2),
("MRX 1140", 2),
("MDX8", 4),
("MDX16", 8),
("MDA8", 4),
("MDA16", 8),
],
)
async def test_zone_created(self, model: str, zone: int):
avr = AVR()
with patch.object(avr, "query"):
await avr._parse_message(f"IDM{model}")
assert len(avr.zones) == zone
async def test_power_refreshed_MDX16(self):
avr = AVR()
with patch.object(avr, "query") as mock:
await avr._parse_message("IDMMDX16")
for zone in range(1, 9):
mock.assert_any_call(f"Z{zone}POW")
assert call("Z9POW") not in mock.mock_calls
async def test_input_name_queried_for_MDX16(self):
avr = AVR()
with patch.object(avr, "query") as mock, patch.object(avr, "transport"):
await avr._parse_message("IDMMDX-16")
await avr.refresh_all()
for input_number in range(1, 13):
mock.assert_any_call(f"ISN{input_number:02d}")
async def test_input_name_queried_for_MDX8(self):
avr = AVR()
with patch.object(avr, "query") as mock, patch.object(avr, "transport"):
await avr._parse_message("IDMMDX-8")
await avr.refresh_all()
for input_number in range(1, 13):
if input_number in [1, 2, 3, 4, 9]:
mock.assert_any_call(f"ISN{input_number:02d}")
else:
assert call(f"ISN{input_number:02d}") not in mock.mock_calls
async def test_pvol_x40(self):
avr = AVR()
with patch.object(avr, "query"):
await avr._parse_message("IDMMRX 740")
await avr._parse_message("Z2PVOL51")
assert avr.zones[2].volume == 51
async def test_zone2_power(self):
avr = AVR()
with patch.object(avr, "refresh_zone") as refreshmock, patch.object(
avr, "_loop"
):
await avr._parse_message("IDMMRX 740")
assert avr.zones[2].power is False
await avr._parse_message("Z2POW1")
refreshmock.assert_called_with(2)
assert avr.zones[2].power is True
assert avr._device_power is True
async def test_attenuation(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "query"):
await avr._parse_message("IDMMRX 740")
assert avr.zones[1].attenuation == -90
await avr._parse_message("Z1VOL-42")
assert avr.zones[1].attenuation == -42
async def test_volume_x20(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "query"):
await avr._parse_message("IDMMRX 1120")
assert avr.zones[1].volume == 0
await avr._parse_message("Z1VOL-42")
assert avr.zones[1].volume == 53
async def test_zone_set_volume_as_percentage_x20(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "command") as mock:
avr._model_series = "x20"
assert avr.zones[1].volume_as_percentage == 0
avr.zones[1].volume_as_percentage = 0.53
mock.assert_any_call("Z1VOL-42")
async def test_set_volume_as_percentage_x20(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "command") as mock:
avr._model_series = "x20"
assert avr.volume_as_percentage == 0
avr.volume_as_percentage = 0.53
mock.assert_any_call("Z1VOL-42")
async def test_set_volume_as_percentage_x40(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "command") as mock:
avr._model_series = "x40"
avr.volume_as_percentage = 0.53
mock.assert_any_call("Z1PVOL53")
async def test_set_volume_as_percentage_mdx(self):
avr = AVR()
avr._device_power = True
with patch.object(avr, "command") as mock:
avr._model_series = "mdx"
avr.volume_as_percentage = 0.53
mock.assert_any_call("Z1VOL53")
async def test_refresh_zone_x40(self):
avr = AVR()
with patch.object(avr, "query") as mock, patch.object(avr, "transport"):
await avr._parse_message("IDMMRX 740")
await avr.refresh_zone(2)
mock.assert_any_call("Z2POW")
mock.assert_any_call("Z2INP")
mock.assert_any_call("Z2PVOL")
mock.assert_any_call("Z2VOL")
mock.assert_any_call("Z2MUT")
async def test_refresh_zone_x20(self):
avr = AVR()
with patch.object(avr, "query") as mock, patch.object(avr, "transport"):
await avr._parse_message("IDMMRX 720")
await avr.refresh_zone(2)
mock.assert_any_call("Z2POW")
mock.assert_any_call("Z2INP")
mock.assert_any_call("Z2VOL")
mock.assert_any_call("Z2MUT")
assert call("Z2PVOL") not in mock.mock_calls
async def test_device_power_off(self):
avr = AVR()
with patch.object(avr, "refresh_zone") as refreshmock, patch.object(
avr, "_loop"
):
await avr._parse_message("IDMMRX 740")
await avr._parse_message("Z2POW1")
refreshmock.assert_called_with(2)
assert avr._device_power is True
await avr._parse_message("Z1POW1")
assert avr._device_power is True
await avr._parse_message("Z1POW0")
assert avr._device_power is True
await avr._parse_message("Z2POW0")
assert avr._device_power is False
async def test_zone_input_format_mrx(self):
avr = AVR()
avr._model_series = "x40"
avr._device_power = True
await avr._parse_message("Z1VIR6")
await avr._parse_message("Z1AINDTS Master Audio")
assert avr.zones[1].input_format == "1080i50 DTS Master Audio"
def test_zone_input_format_mdx(self):
avr = AVR()
avr._model_series = "mdx"
avr._device_power = True
assert avr.zones[1].input_format == ""
async def test_input_format_mrx_zone2(self):
avr = AVR()
avr._model_series = "x40"
avr._device_power = True
avr.set_zones("MRX 540")
await avr._parse_message("Z1VIR6")
await avr._parse_message("Z1AINDTS Master Audio")
assert avr.zones[2].input_format == ""
@pytest.mark.parametrize(
"model,expected", [("MRX 520", True), ("MRX 740", True), ("MDX-8", False)]
)
async def test_support_zone1_MRX(self, model: str, expected: bool):
avr = AVR()
with patch.object(avr, "query"):
avr._device_power = True
avr.set_model_command(model)
avr.set_zones(model)
assert avr.support_audio_listening_mode == expected
assert avr.support_profile == expected
assert avr.support_arc == expected
@pytest.mark.parametrize(
"model,expected", [("MRX 520", True), ("MRX 1140", False), ("MDX8", False)]
)
async def test_support_attenuation(self, model: str, expected: bool):
avr = AVR()
with patch.object(avr, "query"):
avr._device_power = True
avr.set_model_command(model)
avr.set_zones(model)
assert avr.zones[1].support_attenuation == expected
assert avr.zones[2].support_attenuation == expected
|
