-
Notifications
You must be signed in to change notification settings - Fork 156
/
test_topology.py
145 lines (115 loc) · 4.05 KB
/
test_topology.py
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
"""Test Topology class."""
import asyncio
import logging
import time
import pytest
import zigpy.config
import zigpy.neighbor
import zigpy.topology
from .async_mock import AsyncMock, MagicMock, patch, sentinel
@pytest.fixture
def device_1():
dev = MagicMock()
dev.ieee = sentinel.ieee_1
dev.node_desc.is_end_device = False
dev.nwk = 0x1111
dev.neighbors.scan = AsyncMock()
return dev
@pytest.fixture
def device_2():
dev = MagicMock()
dev.ieee = sentinel.ieee_2
dev.node_desc.is_end_device = False
dev.nwk = 0x2222
dev.neighbors.scan = AsyncMock()
return dev
@pytest.fixture
def device_3():
dev = MagicMock()
dev.ieee = sentinel.ieee_3
dev.node_desc.is_end_device = False
dev.nwk = 0x3333
dev.neighbors.scan = AsyncMock()
dev.neighbors.supported = False
return dev
@pytest.fixture
def coordinator():
dev = MagicMock()
dev.ieee = sentinel.ieee_4
dev.node_desc.is_end_device = False
dev.nwk = 0x0000
dev.neighbors.scan = AsyncMock()
dev.neighbors.supported = True
return dev
@pytest.fixture
def topology_f(coordinator, device_1, device_2, device_3):
app = MagicMock()
app.devices = {
device_1.ieee: device_1,
device_2.ieee: device_2,
device_3.ieee: device_3,
coordinator.ieee: coordinator,
}
app.config = zigpy.config.ZIGPY_SCHEMA({})
p2 = patch.dict(app.config, {zigpy.config.CONF_TOPO_SCAN_PERIOD: 0})
with patch("zigpy.topology.DELAY_INTER_DEVICE", 0), p2:
yield zigpy.topology.Topology(app)
async def test_new():
"""Test creating new instance."""
app = MagicMock()
app.config = zigpy.config.ZIGPY_SCHEMA({})
p2 = patch.dict(app.config, {zigpy.config.CONF_TOPO_SCAN_PERIOD: 0})
with patch("zigpy.topology.Topology.scan", new=AsyncMock()) as scan, p2:
zigpy.topology.Topology.new(app)
await asyncio.sleep(0)
await asyncio.sleep(0)
assert scan.await_count >= 1
async def test_scan(coordinator, device_1, device_2, topology_f, caplog):
"""Test scanning."""
assert topology_f.timestamp < time.time()
ts = topology_f.timestamp
with caplog.at_level(logging.DEBUG):
await topology_f.scan()
assert device_1.neighbors.scan.await_count == 1
assert device_2.neighbors.scan.await_count == 1
assert topology_f.timestamp != ts
assert "Scanning" in caplog.text
async def test_scan_preempts(device_1, device_2, topology_f, caplog):
"""Test scanning."""
with caplog.at_level(logging.DEBUG):
await asyncio.gather(topology_f.scan(), topology_f.scan())
assert "Cancelling old" in caplog.text
assert "Cancelled topology" in caplog.text
assert device_1.neighbors.scan.await_count == 1
assert device_2.neighbors.scan.await_count == 1
async def test_scan_coordinator_skip(
coordinator, device_1, device_2, device_3, topology_f, caplog
):
"""Test scanning skips coordinator."""
assert topology_f.timestamp < time.time()
ts = topology_f.timestamp
with caplog.at_level(logging.DEBUG):
await topology_f.scan()
assert device_1.neighbors.scan.await_count == 1
assert device_2.neighbors.scan.await_count == 1
assert device_3.neighbors.scan.call_count == 0
assert device_3.neighbors.scan.await_count == 0
assert coordinator.neighbors.scan.call_count == 1
assert coordinator.neighbors.scan.await_count == 1
assert topology_f.timestamp != ts
assert "Scanning" in caplog.text
device_1.neighbors.scan.reset_mock()
device_2.neighbors.scan.reset_mock()
coordinator.neighbors.scan.reset_mock()
p2 = patch.dict(
topology_f._app.config, {zigpy.config.CONF_TOPO_SKIP_COORDINATOR: True}
)
with p2:
await topology_f.scan()
assert device_1.neighbors.scan.await_count == 1
assert device_2.neighbors.scan.await_count == 1
assert device_3.neighbors.scan.call_count == 0
assert device_3.neighbors.scan.await_count == 0
assert coordinator.neighbors.scan.call_count == 0
assert coordinator.neighbors.scan.await_count == 0
assert topology_f.timestamp != ts