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
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
|
"""Unit test for Fan objects."""
from unittest.mock import Mock, patch
from xknx import XKNX
from xknx.devices import Fan
from xknx.dpt import DPTArray, DPTBinary
from xknx.telegram import GroupAddress, Telegram
from xknx.telegram.apci import GroupValueRead, GroupValueWrite
class TestFan:
"""Class for testing Fan objects."""
#
# SYNC
#
async def test_sync(self) -> None:
"""Test sync function / sending group reads to KNX bus."""
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed_state="1/2/3")
await fan.sync()
assert xknx.telegrams.qsize() == 1
telegram1 = xknx.telegrams.get_nowait()
assert telegram1 == Telegram(
destination_address=GroupAddress("1/2/3"), payload=GroupValueRead()
)
async def test_sync_step(self) -> None:
"""Test sync function / sending group reads to KNX bus."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed_state="1/2/3",
)
await fan.sync()
assert xknx.telegrams.qsize() == 1
telegram1 = xknx.telegrams.get_nowait()
assert telegram1 == Telegram(
destination_address=GroupAddress("1/2/3"), payload=GroupValueRead()
)
#
# SYNC WITH STATE ADDRESS
#
async def test_sync_state_address(self) -> None:
"""Test sync function / sending group reads to KNX bus."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
group_address_speed_state="1/2/4",
)
await fan.sync()
assert xknx.telegrams.qsize() == 1
telegram1 = xknx.telegrams.get_nowait()
assert telegram1 == Telegram(
destination_address=GroupAddress("1/2/4"), payload=GroupValueRead()
)
#
# TEST SWITCH ON/OFF
#
async def test_switch_on_off(self) -> None:
"""Test switching on/off of a Fan."""
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed="1/2/3")
# Turn the fan on via speed GA. First try without providing a speed,
# which will set it to the default 50% percentage.
await fan.turn_on()
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
# 128 is 50% as byte (0...255)
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(128)),
)
# Try again, but this time with a speed provided
await fan.turn_on(55)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
# 140 is 55% as byte (0...255)
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
# Turn the fan off via the speed GA
await fan.turn_off()
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(0)),
)
fan_with_switch = Fan(
xknx,
name="TestFanSwitch",
group_address_speed="1/2/3",
group_address_switch="4/5/6",
)
# Turn the fan on via the switch GA, which should not adjust the speed
await fan_with_switch.turn_on()
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(1)),
)
# Turn the fan off via the switch GA
await fan_with_switch.turn_off()
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(0)),
)
# Turn the fan on again this time with a provided speed, which for a switch GA fan
# should result in separate telegrams to switch on the fan and then set the speed.
await fan_with_switch.turn_on(55)
assert xknx.telegrams.qsize() == 2
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(1)),
)
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
#
# TEST SET SPEED
#
async def test_set_speed(self) -> None:
"""Test setting the speed of a Fan."""
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed="1/2/3")
await fan.set_speed(55)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
# 140 is 55% as byte (0...255)
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
fan.process(telegram)
assert fan.is_on is True
# A speed of 0 will turn off the fan implicitly if there is no
# dedicated switch GA
await fan.set_speed(0)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
# 140 is 55% as byte (0...255)
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(0)),
)
fan.process(telegram)
assert fan.is_on is False
fan_with_switch = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
group_address_switch="4/5/6",
)
await fan_with_switch.turn_on()
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(1)),
)
fan_with_switch.process(telegram)
assert fan_with_switch.is_on is True
# A speed of 0 will not turn off the fan implicitly if there is a
# dedicated switch GA defined. So we only expect a speed change telegram,
# but no state switch one.
await fan_with_switch.set_speed(0)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(0)),
)
fan_with_switch.process(telegram)
assert fan_with_switch.is_on is True
#
# TEST SET SPEED STEP
#
async def test_set_speed_step(self) -> None:
"""Test setting the speed of a Fan in step mode."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
max_step=3,
)
await fan.set_speed(2)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(2)),
)
#
# TEST SET OSCILLATION
#
async def test_set_oscillation(self) -> None:
"""Test setting the oscillation of a Fan."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
group_address_oscillation="1/2/5",
)
await fan.set_oscillation(False)
assert xknx.telegrams.qsize() == 1
telegram = xknx.telegrams.get_nowait()
assert telegram == Telegram(
destination_address=GroupAddress("1/2/5"),
payload=GroupValueWrite(DPTBinary(0)),
)
#
# TEST PROCESS SPEED
#
async def test_process_speed(self) -> None:
"""Test process / reading telegrams from telegram queue. Test if speed is processed."""
xknx = XKNX()
fan = Fan(xknx, name="TestFan", group_address_speed="1/2/3")
assert fan.is_on is False
assert fan.current_speed is None
# 140 is 55% as byte (0...255)
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
fan.process(telegram)
# Setting a speed for a fan that has no dedicated switch GA,
# should turn on the fan.
assert fan.is_on is True
assert fan.current_speed == 55
# Now set a speed of zero which should turn off the fan.
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(0)),
)
fan.process(telegram)
assert fan.is_on is False
assert fan.current_speed == 0
async def test_process_speed_wrong_payload(self) -> None:
"""Test process wrong telegrams. (wrong payload type)."""
xknx = XKNX()
cb_mock = Mock()
fan = Fan(
xknx, name="TestFan", group_address_speed="1/2/3", device_updated_cb=cb_mock
)
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTBinary(1)),
)
with patch("logging.Logger.warning") as log_mock:
fan.process(telegram)
log_mock.assert_called_once()
cb_mock.assert_not_called()
#
# TEST PROCESS SWITCH
#
async def test_process_switch(self) -> None:
"""Test process / reading telegrams from telegram queue. Test if switch is handled correctly."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
group_address_switch="4/5/6",
)
assert fan.is_on is False
assert fan.current_speed is None
# 140 is 55% as byte (0...255)
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
fan.process(telegram)
# Setting a speed for a fan with dedicated switch GA,
# should not turn on the fan
assert fan.is_on is False
assert fan.current_speed == 55
# Now turn on the fan via its switch GA
telegram = Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(1)),
)
fan.process(telegram)
assert fan.is_on is True
assert fan.current_speed == 55
# Setting a speed of 0 should not turn off the fan
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(0)),
)
fan.process(telegram)
assert fan.is_on is True
assert fan.current_speed == 0
# Set the speed again so we can verify that switching off the fan does not
# modify the set speed
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(140)),
)
fan.process(telegram)
assert fan.is_on is True
assert fan.current_speed == 55
# Now turn off the fan via the dedicated switch GA
telegram = Telegram(
destination_address=GroupAddress("4/5/6"),
payload=GroupValueWrite(DPTBinary(0)),
)
fan.process(telegram)
assert fan.is_on is False
assert fan.current_speed == 55
#
# TEST PROCESS OSCILLATION
#
async def test_process_oscillation(self) -> None:
"""Test process / reading telegrams from telegram queue. Test if oscillation is processed."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
group_address_oscillation="1/2/5",
)
assert fan.current_oscillation is None
telegram = Telegram(
destination_address=GroupAddress("1/2/5"),
payload=GroupValueWrite(DPTBinary(1)),
)
fan.process(telegram)
assert fan.current_oscillation
async def test_process_fan_payload_invalid_length(self) -> None:
"""Test process wrong telegrams. (wrong payload length)."""
xknx = XKNX()
cb_mock = Mock()
fan = Fan(
xknx, name="TestFan", group_address_speed="1/2/3", device_updated_cb=cb_mock
)
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray((23, 24))),
)
with patch("logging.Logger.warning") as log_mock:
fan.process(telegram)
log_mock.assert_called_once()
cb_mock.assert_not_called()
#
# TEST PROCESS STEP MODE
#
async def test_process_speed_step(self) -> None:
"""Test process / reading telegrams from telegram queue. Test if speed is processed."""
xknx = XKNX()
fan = Fan(
xknx,
name="TestFan",
group_address_speed="1/2/3",
max_step=3,
)
assert fan.current_speed is None
telegram = Telegram(
destination_address=GroupAddress("1/2/3"),
payload=GroupValueWrite(DPTArray(2)),
)
fan.process(telegram)
assert fan.current_speed == 2
def test_has_group_address(self) -> None:
"""Test has_group_address."""
xknx = XKNX()
fan = Fan(
xknx,
"TestFan",
group_address_speed="1/7/1",
group_address_speed_state="1/7/2",
group_address_oscillation="1/6/1",
group_address_oscillation_state="1/6/2",
group_address_switch="1/5/1",
group_address_switch_state="1/5/2",
)
assert fan.has_group_address(GroupAddress("1/7/1"))
assert fan.has_group_address(GroupAddress("1/7/2"))
assert not fan.has_group_address(GroupAddress("1/7/3"))
assert fan.has_group_address(GroupAddress("1/6/1"))
assert fan.has_group_address(GroupAddress("1/6/2"))
assert fan.has_group_address(GroupAddress("1/5/1"))
assert fan.has_group_address(GroupAddress("1/5/2"))
|
