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
|
import asyncio
from unittest.mock import call
import pytest
import zigpy_znp.types as t
import zigpy_znp.commands as c
from zigpy_znp.api import ZNP
from ..conftest import BaseServerZNP, CoroutineMock, config_for_port_path
async def test_connect_no_test(make_znp_server):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
await znp.connect(test_port=False)
# Nothing will be sent
assert znp_server._uart.data_received.call_count == 0
await znp.disconnect()
@pytest.mark.parametrize("work_after_attempt", [1, 2, 3])
async def test_connect_skip_bootloader(make_znp_server, mocker, work_after_attempt):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
mocker.patch.object(znp.nvram, "determine_alignment", new=CoroutineMock())
mocker.patch.object(znp, "detect_zstack_version", new=CoroutineMock())
num_pings = 0
def ping_rsp(req):
nonlocal num_pings
num_pings += 1
# Ignore the first few pings
if num_pings >= work_after_attempt:
return c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
znp_server.reply_to(c.SYS.Ping.Req(), responses=[ping_rsp])
await znp.connect(test_port=True)
await znp.disconnect()
async def test_connect_skip_bootloader_batched_rsp(make_znp_server, mocker):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
mocker.patch.object(znp.nvram, "determine_alignment", new=CoroutineMock())
mocker.patch.object(znp, "detect_zstack_version", new=CoroutineMock())
num_pings = 0
def ping_rsp(req):
nonlocal num_pings
num_pings += 1
if num_pings == 3:
# CC253x radios sometimes buffer requests until they send a `ResetInd`
return (
[
c.SYS.ResetInd.Callback(
Reason=t.ResetReason.PowerUp,
TransportRev=0x00,
ProductId=0x12,
MajorRel=0x01,
MinorRel=0x02,
MaintRel=0x03,
)
]
+ [c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)] * num_pings,
)
elif num_pings >= 3:
return c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
znp_server.reply_to(c.SYS.Ping.Req(), responses=[ping_rsp])
await znp.connect(test_port=True)
await znp.disconnect()
async def test_connect_skip_bootloader_failure(make_znp_server):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
with pytest.raises(asyncio.TimeoutError):
await znp.connect(test_port=True)
await znp.disconnect()
async def test_connect_skip_bootloader_rts_dtr_pins(make_znp_server, mocker):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
mocker.patch.object(znp.nvram, "determine_alignment", new=CoroutineMock())
mocker.patch.object(znp, "detect_zstack_version", new=CoroutineMock())
znp_server.reply_to(
c.SYS.Ping.Req(), responses=[c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)]
)
await znp.connect(test_port=True)
serial = znp._uart._transport
assert serial._mock_dtr_prop.mock_calls == [call(False), call(False), call(False)]
assert serial._mock_rts_prop.mock_calls == [call(False), call(True), call(False)]
await znp.disconnect()
async def test_connect_skip_bootloader_config(make_znp_server, mocker):
znp_server = make_znp_server(server_cls=BaseServerZNP)
znp = ZNP(config_for_port_path(znp_server.port_path))
znp._znp_config["skip_bootloader"] = False
mocker.patch.object(znp.nvram, "determine_alignment", new=CoroutineMock())
mocker.patch.object(znp, "detect_zstack_version", new=CoroutineMock())
znp_server.reply_to(
c.SYS.Ping.Req(), responses=[c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)]
)
await znp.connect(test_port=True)
serial = znp._uart._transport
assert serial._mock_dtr_prop.called is False
assert serial._mock_rts_prop.called is False
await znp.disconnect()
async def test_api_close(connected_znp, mocker):
znp, znp_server = connected_znp
uart = znp._uart
mocker.spy(uart, "close")
await znp.disconnect()
# Make sure our UART was actually closed
assert znp._uart is None
assert znp._app is None
assert uart.close.call_count == 1
# ZNP.disconnect should not throw any errors if called multiple times
await znp.disconnect()
await znp.disconnect()
def dict_minus(d, minus):
return {k: v for k, v in d.items() if k not in minus}
ignored_keys = ["_sync_request_lock", "nvram"]
# Closing ZNP should reset it completely to that of a fresh object
# We have to ignore our mocked method and the lock
znp2 = ZNP(znp._config)
assert znp2._sync_request_lock.locked() == znp._sync_request_lock.locked()
assert dict_minus(znp.__dict__, ignored_keys) == dict_minus(
znp2.__dict__, ignored_keys
)
await znp2.disconnect()
await znp2.disconnect()
assert dict_minus(znp.__dict__, ignored_keys) == dict_minus(
znp2.__dict__, ignored_keys
)
|
