File: test_zdo_types.py

package info (click to toggle)
zigpy 0.80.1-2
  • links: PTS, VCS
  • area: main
  • in suites: trixie
  • size: 3,012 kB
  • sloc: python: 34,822; sql: 2,109; makefile: 7
file content (448 lines) | stat: -rw-r--r-- 14,492 bytes parent folder | download | duplicates (2)
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
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
import pytest

import zigpy.types as t
from zigpy.zdo import types


def test_multi_address_3():
    ma = types.MultiAddress()
    ma.addrmode = 3
    ma.ieee = t.EUI64(map(t.uint8_t, [0, 1, 2, 3, 4, 5, 6, 7]))
    ma.endpoint = 1
    ser = ma.serialize()

    assert "ieee" in repr(ma)
    assert "endpoint" in repr(ma)
    assert "nwk" not in repr(ma)

    ma2, data = types.MultiAddress.deserialize(ser)
    assert data == b""
    assert ma2.addrmode == ma.addrmode
    assert ma2.ieee == ma.ieee
    assert ma2.endpoint == ma.endpoint


def test_multi_address_1():
    ma = types.MultiAddress()
    ma.addrmode = 1
    ma.nwk = 123
    ser = ma.serialize()

    assert "ieee" not in repr(ma)
    assert "endpoint" not in repr(ma)
    assert "nwk" in repr(ma)

    ma2, data = types.MultiAddress.deserialize(ser)
    assert data == b""
    assert ma2.addrmode == ma.addrmode
    assert ma2.nwk == ma.nwk


def test_multi_address_invalid():
    ma = types.MultiAddress()
    ma.addrmode = 255
    with pytest.raises(ValueError):
        ma.serialize()

    with pytest.raises(ValueError):
        types.MultiAddress.deserialize(b"\xffnot read")


def test_channels():
    """Test Channels bitmap."""

    assert t.Channels.from_channel_list([]) == t.Channels.NO_CHANNELS
    assert t.Channels.from_channel_list(range(11, 26 + 1)) == t.Channels.ALL_CHANNELS
    assert (
        t.Channels.from_channel_list([11, 21])
        == t.Channels.CHANNEL_11 | t.Channels.CHANNEL_21
    )

    with pytest.raises(ValueError):
        t.Channels.from_channel_list([11, 13, 10])  # 10 is not a valid channel

    with pytest.raises(ValueError):
        t.Channels.from_channel_list([27, 13, 15, 18])  # 27 is not a valid channel

    assert list(t.Channels.from_channel_list([11, 13, 25])) == [11, 13, 25]
    assert list(t.Channels.ALL_CHANNELS) == list(range(11, 26 + 1))
    assert list(t.Channels.NO_CHANNELS) == []

    for expected, channel in zip(t.Channels.ALL_CHANNELS, range(11, 26 + 1)):
        assert expected == channel

    # t.Channels.from_channel_list(another_channel) should be idempotent
    channels = t.Channels.from_channel_list([11, 13, 25])
    assert channels == t.Channels.from_channel_list(channels)

    # Even though this is a "valid" channels bitmap, it has unknown channels
    invalid_channels = t.Channels(0xFFFFFFFF)

    with pytest.raises(ValueError):
        list(invalid_channels)


def test_node_descriptor():
    data = b"\x00\x01\x02\x03\x03\x04\x05\x05\x06\x06\x07\x07\x08\xff"
    nd, rest = types.NodeDescriptor.deserialize(data)

    assert rest == b"\xff"

    new_node_desc = types.NodeDescriptor(nd)
    assert new_node_desc.logical_type == 0
    assert new_node_desc.complex_descriptor_available == 0
    assert new_node_desc.user_descriptor_available == 0
    assert new_node_desc.reserved == 0
    assert new_node_desc.aps_flags == 1
    assert new_node_desc.frequency_band == 0
    assert new_node_desc.mac_capability_flags == 0x02
    assert new_node_desc.manufacturer_code == 0x0303
    assert new_node_desc.maximum_buffer_size == 0x04
    assert new_node_desc.maximum_incoming_transfer_size == 0x0505
    assert new_node_desc.server_mask == 0x0606
    assert new_node_desc.maximum_outgoing_transfer_size == 0x0707
    assert new_node_desc.descriptor_capability_field == 0x08

    nd2 = types.NodeDescriptor(0, 1, 2, 0x0303, 0x04, 0x0505, 0x0606, 0x0707, 0x08)
    assert nd2.serialize() == new_node_desc.serialize()


def test_node_descriptor_is_valid():
    for field in types.NodeDescriptor.fields:
        nd = types.NodeDescriptor(0, 1, 2, 0x0303, 0x04, 0x0505, 0x0606, 0x0707, 0x08)
        assert nd.is_valid is True
        setattr(nd, field.name, None)
        assert nd.is_valid is False


def test_node_descriptor_props():
    props = (
        "logical_type",
        "complex_descriptor_available",
        "user_descriptor_available",
        "is_alternate_pan_coordinator",
        "is_full_function_device",
        "is_mains_powered",
        "is_receiver_on_when_idle",
        "is_security_capable",
        "allocate_address",
    )

    empty_nd = types.NodeDescriptor()
    for prop in props:
        value = getattr(empty_nd, prop)
        assert value is None

    nd = types.NodeDescriptor(
        0b11111000, 0xFF, 0xFF, 0xFFFF, 0xFF, 0xFFFF, 0xFFFF, 0xFFFF, 0xFF
    )
    assert nd.logical_type is not None
    for prop in props:
        if prop == "logical_type":
            continue
        assert getattr(nd, prop)


def test_node_descriptor_logical_types():
    nd = types.NodeDescriptor()
    assert nd.is_coordinator is None
    assert nd.is_end_device is None
    assert nd.is_router is None

    nd = types.NodeDescriptor(
        0b11111000, 0xFF, 0xFF, 0xFFFF, 0xFF, 0xFFFF, 0xFFFF, 0xFFFF, 0xFF
    )
    assert nd.is_coordinator is True
    assert nd.is_end_device is False
    assert nd.is_router is False

    nd = types.NodeDescriptor(
        0b11111001, 0xFF, 0xFF, 0xFFFF, 0xFF, 0xFFFF, 0xFFFF, 0xFFFF, 0xFF
    )
    assert nd.is_coordinator is False
    assert nd.is_end_device is False
    assert nd.is_router is True

    nd = types.NodeDescriptor(
        0b11111010, 0xFF, 0xFF, 0xFFFF, 0xFF, 0xFFFF, 0xFFFF, 0xFFFF, 0xFF
    )
    assert nd.is_coordinator is False
    assert nd.is_end_device is True
    assert nd.is_router is False


def test_node_descriptor_repr():
    nd = types.NodeDescriptor(
        0b11111010, 0xFF, 0xFF, 0xFFFF, 0xFF, 0xFFFF, 0xFFFF, 0xFFFF, 0xFF
    )
    assert nd.is_coordinator is False
    assert "*is_coordinator=False" in repr(nd)

    assert nd.is_end_device is True
    assert "*is_end_device=True" in repr(nd)

    assert nd.is_router is False
    assert "*is_router=False" in repr(nd)


def test_size_prefixed_simple_descriptor():
    sd = types.SizePrefixedSimpleDescriptor()
    sd.endpoint = t.uint8_t(1)
    sd.profile = t.uint16_t(2)
    sd.device_type = t.uint16_t(3)
    sd.device_version = t.uint8_t(4)
    sd.input_clusters = t.LVList[t.uint16_t]([t.uint16_t(5), t.uint16_t(6)])
    sd.output_clusters = t.LVList[t.uint16_t]([t.uint16_t(7), t.uint16_t(8)])

    ser = sd.serialize()
    assert ser[0] == len(ser) - 1

    sd2, data = types.SizePrefixedSimpleDescriptor.deserialize(ser + b"extra")
    assert sd.input_clusters == sd2.input_clusters
    assert sd.output_clusters == sd2.output_clusters
    assert isinstance(sd2, types.SizePrefixedSimpleDescriptor)
    assert data == b"extra"


def test_empty_size_prefixed_simple_descriptor():
    r = types.SizePrefixedSimpleDescriptor.deserialize(b"\x00")
    assert r == (None, b"")


def test_invalid_size_prefixed_simple_descriptor():
    with pytest.raises(ValueError):
        types.SizePrefixedSimpleDescriptor.deserialize(b"\x01")


def test_status_undef():
    data = b"\xff"
    extra = b"extra"

    status, rest = types.Status.deserialize(data + extra)
    assert rest == extra
    assert status == 0xFF
    assert status.value == 0xFF
    assert status.name == "undefined_0xff"
    assert isinstance(status, types.Status)


def test_zdo_header():
    tsn = t.uint8_t(0xAA)
    cmd_id = 0x55
    data = tsn.serialize()
    extra = b"abcdefExtraDataHere"
    hdr, rest = types.ZDOHeader.deserialize(cmd_id, data + extra)
    assert rest == extra
    assert hdr.tsn == tsn
    assert hdr.command_id == cmd_id
    assert not hdr.is_reply

    hdr.command_id = types.ZDOCmd.Bind_rsp
    assert hdr.is_reply

    assert hdr.serialize() == data

    new_tsn = 0xBB
    hdr.tsn = new_tsn
    assert isinstance(hdr.tsn, t.uint8_t)
    assert hdr.tsn == new_tsn


def test_zdo_header_cmd_id():
    unk_cmd = 0x00FF
    assert unk_cmd not in list(types.ZDOCmd)
    hdr = types.ZDOHeader(unk_cmd, 0x55)
    assert isinstance(hdr.command_id, types.ZDOCmd)
    assert hdr.command_id == unk_cmd

    unk_cmd += 1
    assert unk_cmd not in list(types.ZDOCmd)
    hdr.command_id = unk_cmd
    assert isinstance(hdr.command_id, types.ZDOCmd)
    assert hdr.command_id == unk_cmd


def test_nwkupdate():
    """Test NwkUpdate class."""

    extra = b"extra data\xaa\x55"

    upd = types.NwkUpdate(t.Channels.ALL_CHANNELS, 0x05, 0x04, 0xAA, 0x1234)
    data = upd.serialize()
    assert data == b"\x00\xf8\xff\x07\x05\x04"

    new, rest = types.NwkUpdate.deserialize(data + extra)
    assert rest == extra
    assert new.ScanChannels == t.Channels.ALL_CHANNELS
    assert new.ScanDuration == 0x05
    assert new.ScanCount == 0x04
    assert new.nwkUpdateId is None
    assert new.nwkManagerAddr is None


def test_nwkupdate_nwk_update_id():
    """Test NwkUpdate class."""

    extra = b"extra data\xaa\x55"

    upd = types.NwkUpdate(t.Channels.ALL_CHANNELS, 0xFE, 0x04, 0xAA, 0x1234)
    data = upd.serialize()
    assert data == b"\x00\xf8\xff\x07\xfe\xaa"

    new, rest = types.NwkUpdate.deserialize(data + extra)
    assert rest == extra
    assert new.ScanChannels == t.Channels.ALL_CHANNELS
    assert new.ScanDuration == 0xFE
    assert new.ScanCount is None
    assert new.nwkUpdateId == 0xAA
    assert new.nwkManagerAddr is None

    upd = types.NwkUpdate(t.Channels.ALL_CHANNELS, 0xFF, 0x04, 0xAA, 0x1234)
    data = upd.serialize()
    assert data == b"\x00\xf8\xff\x07\xff\xaa\x34\x12"

    new, rest = types.NwkUpdate.deserialize(data + extra)
    assert rest == extra
    assert new.ScanChannels == t.Channels.ALL_CHANNELS
    assert new.ScanDuration == 0xFF
    assert new.ScanCount is None
    assert new.nwkUpdateId == 0xAA
    assert new.nwkManagerAddr == 0x1234


def test_status_enum():
    """Test Status enums chaining."""
    status_names = [e.name for e in types.Status]
    aps_names = [e.name for e in t.APSStatus]
    nwk_names = [e.name for e in t.NWKStatus]
    mac_names = [e.name for e in t.MACStatus]

    status = types.Status(0x80)
    assert status.name in status_names
    assert status.name not in aps_names
    assert status.name not in nwk_names
    assert status.name not in mac_names

    status = types.Status(0xAE)
    assert status.name not in status_names
    assert status.name in aps_names
    assert status.name not in nwk_names
    assert status.name not in mac_names

    status = types.Status(0xD0)
    assert status.name not in status_names
    assert status.name not in aps_names
    assert status.name in nwk_names
    assert status.name not in mac_names

    status = types.Status(0xE9)
    assert status.name not in status_names
    assert status.name not in aps_names
    assert status.name not in nwk_names
    assert status.name in mac_names

    status = types.Status(0xFF)
    assert status.name not in status_names
    assert status.name not in aps_names
    assert status.name not in nwk_names
    assert status.name not in mac_names
    assert status.name == "undefined_0xff"


def test_neighbors():
    """Test ZDO neignbors struct."""

    data = (
        b"\x05\x00\x03\xa2\xaf\x8cY\xf5\x03\x96\xb4:p\x07\xfe\xffW\xb4\x14\xd1"
        b"\xb7\x12\x01\x01H\xa2\xaf\x8cY\xf5\x03\x96\xb4X\xb76\x02\x00\x8d\x15\x00=X"
        b"\x12\x01\x01:\xa2\xaf\x8cY\xf5\x03\x96\xb4\x9b-0\xfe\xff\xbd\x1b\xec$\xcb\x12"
        b"\x01\x01F"
    )
    extra = b"\x55\xaaextra\x00"
    neighbours, rest = types.Neighbors().deserialize(data + extra)
    assert rest == extra


def test_neighbor():
    """Test neighbor struct."""
    data = b"\xa2\xaf\x8cY\xf5\x03\x96\xb4:p\x07\xfe\xffW\xb4\x14\xd1\xb7\x12\x01\x01H"
    extra = b"\x55\xaaextra\x00"

    neighbor, rest = types.Neighbor.deserialize(data + extra)
    assert rest == extra

    assert str(neighbor.extended_pan_id) == "b4:96:03:f5:59:8c:af:a2"
    assert str(neighbor.ieee) == "14:b4:57:ff:fe:07:70:3a"
    assert neighbor.nwk == 0xB7D1
    assert neighbor.device_type == types.Neighbor.DeviceType.EndDevice
    assert neighbor.rx_on_when_idle == types.Neighbor.RxOnWhenIdle.Off
    assert neighbor.relationship == types.Neighbor.RelationShip.Child
    assert neighbor.reserved1 == 0
    assert neighbor.permit_joining == types.Neighbor.PermitJoins.Accepting
    assert neighbor.reserved2 == 0
    assert neighbor.device_type == types.Neighbor.DeviceType.EndDevice
    assert neighbor.relationship == types.Neighbor.RelationShip.Child
    assert neighbor.rx_on_when_idle == types.Neighbor.RxOnWhenIdle.Off
    assert neighbor.permit_joining == 1
    assert neighbor.depth == 1
    assert neighbor.lqi == 72


def test_neighbor_struct_device_type():
    """Test neighbor packed struct device_type."""

    for dev_type in range(3):
        struct = types.Neighbor()
        assert struct.device_type is None
        struct.device_type = dev_type
        assert struct.device_type == dev_type

    for i in range(127):
        struct = types.Neighbor(**types.Neighbor._parse_packed(i))
        orig_rx = struct.rx_on_when_idle
        orig_rel = struct.relationship
        for dev_type in range(3):
            struct.device_type = dev_type
            assert struct.rx_on_when_idle == orig_rx
            assert struct.relationship == orig_rel
            assert struct.device_type == dev_type


def test_neighbor_struct_rx_on_when_idle():
    """Test neighbor packed struct rx_on_when_idle."""

    for rx_on_when_idle in range(3):
        struct = types.Neighbor()
        assert struct.rx_on_when_idle is None
        struct.rx_on_when_idle = rx_on_when_idle
        assert struct.rx_on_when_idle == rx_on_when_idle

    for i in range(127):
        struct = types.Neighbor(**types.Neighbor._parse_packed(i))
        orig_dev_type = struct.device_type
        orig_rel = struct.relationship
        for rx_on_when_idle in range(3):
            struct.rx_on_when_idle = rx_on_when_idle
            assert struct.device_type == orig_dev_type
            assert struct.relationship == orig_rel
            assert struct.rx_on_when_idle == rx_on_when_idle


def test_neighbor_struct_relationship():
    """Test neighbor packed struct relationship."""

    for relationship in range(7):
        struct = types.Neighbor()
        assert struct.relationship is None
        struct.relationship = relationship
        assert struct.relationship == relationship

    for i in range(127):
        struct = types.Neighbor(**types.Neighbor._parse_packed(i))
        orig_dev_type = struct.device_type
        orig_rx = struct.rx_on_when_idle
        for relationship in range(7):
            struct.relationship = relationship
            assert struct.device_type == orig_dev_type
            assert struct.rx_on_when_idle == orig_rx
            assert struct.relationship == relationship