diff --git a/tests/test_application.py b/tests/test_application.py index 5386fa7..cd3da3c 100644 --- a/tests/test_application.py +++ b/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): diff --git a/tests/test_network_state.py b/tests/test_network_state.py index 37b7903..e1964e8 100644 --- a/tests/test_network_state.py +++ b/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) ] diff --git a/zigpy_deconz/zigbee/application.py b/zigpy_deconz/zigbee/application.py index abca6e7..f67339f 100644 --- a/zigpy_deconz/zigbee/application.py +++ b/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)