1
- from unittest .mock import AsyncMock , MagicMock , patch , sentinel , call
1
+ import logging
2
+ from unittest .mock import AsyncMock , MagicMock , call , patch
2
3
3
4
import pytest
4
- import logging
5
- import zigpy .types as zigpy_t
6
5
import zigpy .exceptions
6
+ import zigpy .types as zigpy_t
7
7
8
8
import zigpy_zigate .api
9
- import zigpy_zigate .types as t
10
9
import zigpy_zigate .config as config
10
+ import zigpy_zigate .types as t
11
11
import zigpy_zigate .zigbee .application
12
12
13
13
APP_CONFIG = zigpy_zigate .zigbee .application .ControllerApplication .SCHEMA (
16
16
config .CONF_DATABASE : None ,
17
17
}
18
18
)
19
- FAKE_FIRMWARE_VERSION = ' 3.1z'
19
+ FAKE_FIRMWARE_VERSION = " 3.1z"
20
20
21
21
22
22
@pytest .fixture
@@ -41,7 +41,7 @@ def test_zigpy_ieee(app):
41
41
42
42
def test_model_detection (app ):
43
43
device = zigpy_zigate .zigbee .application .ZiGateDevice (app , 0 , 0 )
44
- assert device .model == ' ZiGate USB-TTL {}' .format (FAKE_FIRMWARE_VERSION )
44
+ assert device .model == " ZiGate USB-TTL {}" .format (FAKE_FIRMWARE_VERSION )
45
45
46
46
47
47
@pytest .mark .asyncio
@@ -52,16 +52,17 @@ async def test_form_network_success(app):
52
52
app ._api .reset = AsyncMock ()
53
53
54
54
async def mock_start_network ():
55
- return [[0x00 , 0x1234 , 0x0123456789abcdef ], 0 ]
55
+ return [[0x00 , 0x1234 , 0x0123456789ABCDEF ], 0 ]
56
+
56
57
app ._api .start_network = mock_start_network
57
58
58
59
async def mock_get_network_state ():
59
60
return [
60
61
[
61
62
0x0000 ,
62
- t .EUI64 ([0xef , 0xcd , 0xab , 0x89 , 0x67 , 0x45 , 0x23 , 0x01 ]),
63
+ t .EUI64 ([0xEF , 0xCD , 0xAB , 0x89 , 0x67 , 0x45 , 0x23 , 0x01 ]),
63
64
0x1234 ,
64
- 0x1234abcdef012345 ,
65
+ 0x1234ABCDEF012345 ,
65
66
0x11 ,
66
67
],
67
68
0 ,
@@ -88,11 +89,15 @@ async def test_form_network_failed(app):
88
89
app ._api .set_channel = AsyncMock ()
89
90
app ._api .set_extended_panid = AsyncMock ()
90
91
app ._api .reset = AsyncMock ()
92
+
91
93
async def mock_start_network ():
92
94
return [[0x06 ], 0 ]
95
+
93
96
app ._api .start_network = mock_start_network
97
+
94
98
async def mock_get_network_state ():
95
- return [[0xffff , 0x0123456789abcdef , 0x1234 , 0 , 0x11 ], 0 ]
99
+ return [[0xFFFF , 0x0123456789ABCDEF , 0x1234 , 0 , 0x11 ], 0 ]
100
+
96
101
app ._api .get_network_state = mock_get_network_state
97
102
98
103
with pytest .raises (zigpy .exceptions .FormationFailure ):
@@ -139,10 +144,10 @@ async def test_disconnect_multiple(app):
139
144
140
145
@pytest .mark .asyncio
141
146
@patch ("zigpy_zigate.zigbee.application.ZiGate.new" )
142
- @pytest .mark .parametrize ("version_rsp, expected_version" , [
143
- [(( 261 , 798 ), 0 ), "3.1e" ] ,
144
- [(( 5 , 801 ), 0 ), "3.21" ]
145
- ] )
147
+ @pytest .mark .parametrize (
148
+ "version_rsp, expected_version" ,
149
+ [[(( 261 , 798 ), 0 ), "3.1e" ], [(( 5 , 801 ), 0 ), "3.21" ]],
150
+ )
146
151
async def test_startup_connect (zigate_new , app , version_rsp , expected_version ):
147
152
api = zigate_new .return_value
148
153
api .version .return_value = version_rsp
@@ -153,15 +158,37 @@ async def test_startup_connect(zigate_new, app, version_rsp, expected_version):
153
158
154
159
155
160
@pytest .mark .asyncio
156
- @pytest .mark .parametrize ("version, addr_mode" , [
157
- ["3.1z" , t .AddressMode .NWK_NO_ACK ],
158
- ["3.1d" , t .AddressMode .NWK ],
159
- ])
161
+ @pytest .mark .parametrize (
162
+ "version, addr_mode" ,
163
+ [
164
+ ["3.1z" , t .AddressMode .NWK_NO_ACK ],
165
+ ["3.1d" , t .AddressMode .NWK ],
166
+ ],
167
+ )
160
168
async def test_send_unicast_request (app , version , addr_mode ):
161
- packet = zigpy_t .ZigbeePacket (src = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .NWK , address = 0x0000 ), src_ep = 1 , dst = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .NWK , address = 0xFA5D ), dst_ep = 1 , source_route = None , extended_timeout = False , tsn = 20 , profile_id = 260 , cluster_id = 6 , data = zigpy_t .SerializableBytes (b'\x01 \x14 \x00 ' ), tx_options = zigpy_t .TransmitOptions .NONE , radius = 0 , non_member_radius = 0 , lqi = None , rssi = None )
169
+ packet = zigpy_t .ZigbeePacket (
170
+ src = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .NWK , address = 0x0000 ),
171
+ src_ep = 1 ,
172
+ dst = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .NWK , address = 0xFA5D ),
173
+ dst_ep = 1 ,
174
+ source_route = None ,
175
+ extended_timeout = False ,
176
+ tsn = 20 ,
177
+ profile_id = 260 ,
178
+ cluster_id = 6 ,
179
+ data = zigpy_t .SerializableBytes (b"\x01 \x14 \x00 " ),
180
+ tx_options = zigpy_t .TransmitOptions .NONE ,
181
+ radius = 0 ,
182
+ non_member_radius = 0 ,
183
+ lqi = None ,
184
+ rssi = None ,
185
+ )
162
186
163
187
app .version = version
164
- app ._api .raw_aps_data_request .return_value = ([t .Status .Success , 163 , 1328 , b'\x00 \x00 ' ], 0 )
188
+ app ._api .raw_aps_data_request .return_value = (
189
+ [t .Status .Success , 163 , 1328 , b"\x00 \x00 " ],
190
+ 0 ,
191
+ )
165
192
await app .send_packet (packet )
166
193
167
194
# The packet was sent with ACKs, even though zigpy didn't ask for it
@@ -172,9 +199,28 @@ async def test_send_unicast_request(app, version, addr_mode):
172
199
173
200
@pytest .mark .asyncio
174
201
async def test_send_group_request (app ):
175
- packet = zigpy_t .ZigbeePacket (src = None , src_ep = 1 , dst = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .Group , address = 0x0002 ), dst_ep = None , source_route = None , extended_timeout = False , tsn = 21 , profile_id = 260 , cluster_id = 6 , data = zigpy_t .SerializableBytes (b'\x01 \x15 \x00 ' ), tx_options = zigpy_t .TransmitOptions .NONE , radius = 0 , non_member_radius = 3 , lqi = None , rssi = None )
202
+ packet = zigpy_t .ZigbeePacket (
203
+ src = None ,
204
+ src_ep = 1 ,
205
+ dst = zigpy_t .AddrModeAddress (addr_mode = zigpy_t .AddrMode .Group , address = 0x0002 ),
206
+ dst_ep = None ,
207
+ source_route = None ,
208
+ extended_timeout = False ,
209
+ tsn = 21 ,
210
+ profile_id = 260 ,
211
+ cluster_id = 6 ,
212
+ data = zigpy_t .SerializableBytes (b"\x01 \x15 \x00 " ),
213
+ tx_options = zigpy_t .TransmitOptions .NONE ,
214
+ radius = 0 ,
215
+ non_member_radius = 3 ,
216
+ lqi = None ,
217
+ rssi = None ,
218
+ )
176
219
177
- app ._api .raw_aps_data_request .return_value = ([t .Status .Success , 0 , 1328 , b'\x01 \xea \x00 \x00 ' ], 0 )
220
+ app ._api .raw_aps_data_request .return_value = (
221
+ [t .Status .Success , 0 , 1328 , b"\x01 \xea \x00 \x00 " ],
222
+ 0 ,
223
+ )
178
224
await app .send_packet (packet )
179
225
180
226
app ._api .raw_aps_data_request .assert_called_once ()
@@ -183,7 +229,9 @@ async def test_send_group_request(app):
183
229
@pytest .mark .asyncio
184
230
async def test_energy_scanning (app , caplog ):
185
231
with caplog .at_level (logging .WARNING ):
186
- scan_results = await app .energy_scan (channels = zigpy_t .Channels .ALL_CHANNELS , duration_exp = 2 , count = 5 )
232
+ scan_results = await app .energy_scan (
233
+ channels = zigpy_t .Channels .ALL_CHANNELS , duration_exp = 2 , count = 5
234
+ )
187
235
188
236
assert scan_results == {c : 0 for c in zigpy_t .Channels .ALL_CHANNELS }
189
237
0 commit comments