additional coverage

Author: dmulcahey

tests/conftest.py

@@ -342,9 +342,11 @@ def __init__(
         self.zha_data = zha_data
         self.ws_gateway: WebSocketServerGateway = ws_gateway
         self.client_gateway: WebSocketClientGateway = client_gateway
-        self.application_controller: ControllerApplication = (
-            self.ws_gateway.application_controller
-        )
+
+    @property
+    def application_controller(self) -> ControllerApplication:
+        """Return the Zigpy application controller."""
+        return self.ws_gateway.application_controller
 
     @property
     def config(self) -> ZHAData:

tests/websocket/test_client_controller.py

@@ -183,6 +183,15 @@ async def test_ws_client_gateway_devices(
     await zha_gateway.async_block_till_done()
     assert len(ws_client_gateway.devices) == 1
 
+    # lets kill the network and then start it back up to make sure everything is still in working order
+    await ws_client_gateway.network.stop_network()
+
+    assert zha_gateway.application_controller is None
+
+    await ws_client_gateway.network.start_network()
+
+    assert zha_gateway.application_controller is not None
+
     # let's add it back
     zha_device = await join_zigpy_device(zha_gateway, zigpy_device)
     await zha_gateway.async_block_till_done()
@@ -313,6 +322,17 @@ async def test_ws_client_gateway_devices(
         )
     )
 
+    # test topology scan
+    zha_gateway.application_controller.topology.scan = AsyncMock()
+    await ws_client_gateway.network.update_topology()
+    assert zha_gateway.application_controller.topology.scan.await_count == 1
+
+    # test permit join
+    zha_gateway.application_controller.permit = AsyncMock()
+    await ws_client_gateway.network.permit_joining(60)
+    assert zha_gateway.application_controller.permit.await_count == 1
+    assert zha_gateway.application_controller.permit.await_args == call(60, None)
+
 
 @pytest.mark.parametrize(
     "zha_gateway",

zha/application/gateway.py

@@ -757,7 +757,7 @@ async def async_remove_zigpy_group(self, group_id: int) -> None:
                 await asyncio.gather(*tasks)
         self.application_controller.groups.pop(group_id)
 
-    async def shutdown(self) -> None:
+    async def shutdown(self, call_super=True) -> None:
         """Stop ZHA Controller Application."""
         if self.shutting_down:
             _LOGGER.debug("Ignoring duplicate shutdown event")
@@ -780,7 +780,8 @@ async def shutdown(self) -> None:
             self.application_controller = None
             await asyncio.sleep(0.1)  # give bellows thread callback a chance to run
 
-        await super().shutdown()
+        if call_super:
+            await super().shutdown()
 
         self._devices.clear()
         self._groups.clear()
@@ -859,6 +860,15 @@ async def stop_server(self) -> None:
 
         self._stopped_event.set()
 
+    async def start_network(self) -> None:
+        """Start the Zigbee network."""
+        await super().async_initialize()  # we do this to avoid 2x event registration
+        await self.async_initialize_devices_and_entities()
+
+    async def stop_network(self) -> None:
+        """Stop the Zigbee network."""
+        await self.shutdown(call_super=False)
+
     async def wait_closed(self) -> None:
         """Wait until the server is not running."""
         await self._stopped_event.wait()