Allow disabling of topology scanner (#509)

Allow to skip coordinator neighbor scanning.

tests/test_topology.py

@@ -45,32 +45,46 @@ def device_3():
 
 
 @pytest.fixture
-def topology_f(device_1, device_2, device_3):
+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.CONF_TOPO_SCAN_PERIOD: 0}
+    app.config = zigpy.config.ZIGPY_SCHEMA({})
 
-    with patch("zigpy.topology.DELAY_INTER_DEVICE", 0):
+    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.CONF_TOPO_SCAN_PERIOD: 0}
-    with patch("zigpy.topology.Topology.scan", new=AsyncMock()) as scan:
+    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(device_1, device_2, topology_f, caplog):
+async def test_scan(coordinator, device_1, device_2, topology_f, caplog):
     """Test scanning."""
 
     assert topology_f.timestamp < time.time()
@@ -92,3 +106,40 @@ async def test_scan_preempts(device_1, device_2, topology_f, caplog):
     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

zigpy/config/__init__.py

@@ -15,7 +15,9 @@
     CONF_OTA_IKEA_DEFAULT,
     CONF_OTA_LEDVANCE_DEFAULT,
     CONF_OTA_OTAU_DIR_DEFAULT,
+    CONF_TOPO_SCAN_ENABLED_DEFAULT,
     CONF_TOPO_SCAN_PERIOD_DEFAULT,
+    CONF_TOPO_SKIP_COORDINATOR_DEFAULT,
 )
 from zigpy.config.validators import cv_boolean, cv_hex, cv_key
 import zigpy.types as t
@@ -39,6 +41,9 @@
 CONF_OTA_IKEA_URL = "ikea_update_url"
 CONF_OTA_LEDVANCE = "ledvance_provider"
 CONF_TOPO_SCAN_PERIOD = "topology_scan_period"
+CONF_TOPO_SCAN_ENABLED = "topology_scan_enabled"
+CONF_TOPO_SKIP_COORDINATOR = "topology_scan_skip_coordinator"
+
 
 SCHEMA_DEVICE = vol.Schema({vol.Required(CONF_DEVICE_PATH): str})
 SCHEMA_NETWORK = vol.Schema(
@@ -85,6 +90,12 @@
         vol.Optional(
             CONF_TOPO_SCAN_PERIOD, default=CONF_TOPO_SCAN_PERIOD_DEFAULT
         ): vol.All(int, vol.Range(min=20)),
+        vol.Optional(
+            CONF_TOPO_SCAN_ENABLED, default=CONF_TOPO_SCAN_ENABLED_DEFAULT
+        ): cv_boolean,
+        vol.Optional(
+            CONF_TOPO_SKIP_COORDINATOR, default=CONF_TOPO_SKIP_COORDINATOR_DEFAULT
+        ): cv_boolean,
     },
     extra=vol.ALLOW_EXTRA,
 )

zigpy/config/defaults.py

@@ -32,3 +32,5 @@
 CONF_OTA_LEDVANCE_DEFAULT = False
 CONF_OTA_OTAU_DIR_DEFAULT = None
 CONF_TOPO_SCAN_PERIOD_DEFAULT = 240  # 4 hours
+CONF_TOPO_SCAN_ENABLED_DEFAULT = True
+CONF_TOPO_SKIP_COORDINATOR_DEFAULT = False

zigpy/topology.py

@@ -38,7 +38,8 @@ def new(cls, app: zigpy.typing.ControllerApplicationType) -> "Topology":
         """Create Topology instance."""
 
         topo = cls(app)
-        asyncio.create_task(topo.scan_loop())
+        if app.config[zigpy.config.CONF_TOPO_SCAN_ENABLED]:
+            asyncio.create_task(topo.scan_loop())
         return topo
 
     async def scan_loop(self) -> None:
@@ -70,6 +71,11 @@ async def _scan(self) -> None:
             dev for dev in self._app.devices.values() if not dev.node_desc.is_end_device
         ]
         for device in devices_to_scan:
+            if (
+                self._app.config[zigpy.config.CONF_TOPO_SKIP_COORDINATOR]
+                and device.nwk == 0x0000
+            ):
+                continue
             if not device.neighbors.supported:
                 continue
             LOGGER.debug(