Improved network reset, attempt 2

tests/test_application.py

@@ -553,10 +553,20 @@ async def read_param(param_id, index):
 
 
 async def test_reset_network_info(app):
-    app.form_network = AsyncMock()
+    app.write_network_info = AsyncMock()
     await app.reset_network_info()
 
-    app.form_network.assert_called_once_with(fast=True)
+    assert len(app.write_network_info.mock_calls) == 1
+    network_info = app.write_network_info.mock_calls[0].kwargs["network_info"]
+    node_info = app.write_network_info.mock_calls[0].kwargs["node_info"]
+
+    # Verify invalid network settings are written
+    assert network_info.pan_id == 0xFFFF
+    assert network_info.extended_pan_id == zigpy.types.EUI64.convert(
+        "FF:FF:FF:FF:FF:FF:FF:FF"
+    )
+    assert network_info.channel is None
+    assert node_info.nwk == 0xFFFF
 
 
 async def test_energy_scan_conbee_2(app):

tests/test_network_state.py

@@ -12,6 +12,7 @@
 import zigpy_deconz
 import zigpy_deconz.api
 import zigpy_deconz.exception
+import zigpy_deconz.types
 import zigpy_deconz.zigbee.application as application
 
 from tests.async_mock import AsyncMock, patch
@@ -415,13 +416,20 @@ async def write_parameter(param, *args):
 
     app._change_network_state = AsyncMock()
     app._api.write_parameter = AsyncMock(side_effect=write_parameter)
-    app.backups = AsyncMock()
-    app.backups.restore_backup = AsyncMock()
+    app._api.read_parameter = AsyncMock(
+        return_value=t.EUI64.convert("00:11:22:33:44:55:66:77")
+    )
 
-    # Should not raise an error because reset_network_info calls form_network(fast=True)
+    # Should not raise an error despite frame counter not being supported
     await app.reset_network_info()
 
-    # Verify that restore_backup was called once (via form_network)
-    assert app.backups.restore_backup.mock_calls == [
-        call(backup=ANY, counter_increment=0, allow_incomplete=True, create_new=False)
+    # Verify that write_parameter was called (including the frame counter attempt)
+    assert any(
+        mock_call.args[0] == zigpy_deconz.api.NetworkParameter.nwk_frame_counter
+        for mock_call in app._api.write_parameter.mock_calls
+    )
+
+    # Verify network state changes were awaited
+    assert app._change_network_state.mock_calls == [
+        call(zigpy_deconz.api.NetworkState.OFFLINE)
     ]

zigpy_deconz/zigbee/application.py

@@ -180,7 +180,29 @@ async def change_loop():
 
     async def reset_network_info(self):
         # TODO: There does not appear to be a way to factory reset a Conbee
-        await self.form_network(fast=True)
+        await self.write_network_info(
+            network_info=zigpy.state.NetworkInfo(
+                pan_id=0xFFFF,
+                extended_pan_id=zigpy.types.EUI64.convert("FF:FF:FF:FF:FF:FF:FF:FF"),
+                channel=None,
+                channel_mask=zigpy.types.Channels(0),
+                nwk_update_id=0,
+                network_key=zigpy.state.Key(
+                    key=zigpy.types.KeyData.convert(
+                        "FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF:FF"
+                    )
+                ),
+                tc_link_key=zigpy.state.Key(
+                    key=zigpy.types.KeyData.convert(b"ZigBeeAlliance09".hex())
+                ),
+                security_level=0x05,
+            ),
+            node_info=zigpy.state.NodeInfo(
+                logical_type=zdo_t.LogicalType.Coordinator,
+                ieee=zigpy.types.EUI64.UNKNOWN,
+                nwk=0xFFFF,
+            ),
+        )
 
     async def write_network_info(self, *, network_info, node_info):
         try:
@@ -291,6 +313,14 @@ async def write_network_info(self, *, network_info, node_info):
         # Note: Changed network configuration parameters become only affective after
         # sending a Leave Network Request followed by a Create or Join Network Request
         await self._change_network_state(NetworkState.OFFLINE)
+
+        if (
+            network_info.pan_id == 0xFFFF
+            or network_info.channel_mask == zigpy.types.Channels(0)
+        ):
+            # Network is being reset, it will never enter the CONNECTED state
+            return
+
         await asyncio.sleep(CHANGE_NETWORK_STATE_DELAY)
         await self._change_network_state(NetworkState.CONNECTED)
 
@@ -341,7 +371,7 @@ async def load_network_info(self, *, load_devices=False):
             NetworkParameter.aps_extended_panid
         )
 
-        if network_info.extended_pan_id == zigpy.types.EUI64.convert(
+        if network_info.extended_pan_id == zigpy.types.ExtendedPanId.convert(
             "00:00:00:00:00:00:00:00"
         ):
             network_info.extended_pan_id = await self._api.read_parameter(
@@ -358,8 +388,16 @@ async def load_network_info(self, *, load_devices=False):
             NetworkParameter.nwk_update_id
         )
 
-        if network_info.channel == 0:
-            raise NetworkNotFormed("Network channel is zero")
+        if (
+            node_info.nwk == 0xFFFF
+            or network_info.pan_id == 0xFFFF
+            or (
+                network_info.extended_pan_id
+                == zigpy.types.ExtendedPanId.convert("FF:FF:FF:FF:FF:FF:FF:FF")
+            )
+            or network_info.channel == 0
+        ):
+            raise NetworkNotFormed("Network is not formed")
 
         indexed_key = await self._api.read_parameter(NetworkParameter.network_key, 0)