Handle new formation corner cases and use zigpy.state more

Author: puddly

tests/application/test_update_network.py

@@ -16,6 +16,7 @@
 import zigpy.endpoint
 import zigpy.zdo.types as zdo_t
 
+import zigpy_znp.const as const
 import zigpy_znp.types as t
 import zigpy_znp.config as conf
 import zigpy_znp.commands as c
@@ -404,16 +405,11 @@ def _create_network_nvram(self):
             FrameCounter=0,
         )
 
-        self._nvram[ExNvIds.LEGACY][OsalNvIds.STARTUP_OPTION] = self.nvram_serialize(
-            t.StartupOptions.NONE
-        )
-        self._nvram[ExNvIds.LEGACY][OsalNvIds.NWKKEY] = self.nvram_serialize(empty_key)
-        self._nvram[ExNvIds.LEGACY][
-            OsalNvIds.NWK_ACTIVE_KEY_INFO
-        ] = self.nvram_serialize(empty_key.Active)
-        self._nvram[ExNvIds.LEGACY][
-            OsalNvIds.NWK_ALTERN_KEY_INFO
-        ] = self.nvram_serialize(empty_key.Active)
+        legacy = self._nvram[ExNvIds.LEGACY]
+        legacy[OsalNvIds.STARTUP_OPTION] = self.nvram_serialize(t.StartupOptions.NONE)
+        legacy[OsalNvIds.NWKKEY] = self.nvram_serialize(empty_key)
+        legacy[OsalNvIds.NWK_ACTIVE_KEY_INFO] = self.nvram_serialize(empty_key.Active)
+        legacy[OsalNvIds.NWK_ALTERN_KEY_INFO] = self.nvram_serialize(empty_key.Active)
 
     def update_device_state(self, state):
         self.device_state = state
@@ -426,7 +422,7 @@ def create_nib(self, _=None):
         nib.channelList, _ = t.Channels.deserialize(
             self._nvram[ExNvIds.LEGACY][OsalNvIds.CHANLIST]
         )
-        nib.nwkLogicalChannel = list(nib.channelList)[0]
+        nib.nwkLogicalChannel = (list(nib.channelList) + [11])[0]
 
         if OsalNvIds.APS_USE_EXT_PANID in self._nvram[ExNvIds.LEGACY]:
             epid = self._nvram[ExNvIds.LEGACY][OsalNvIds.APS_USE_EXT_PANID]
@@ -459,61 +455,7 @@ def create_nib(self, _=None):
         self._nvram[ExNvIds.LEGACY][OsalNvIds.NWKKEY] = self.nvram_serialize(key_info)
 
     def _default_nib(self):
-        return t.NIB(
-            SequenceNum=0,
-            PassiveAckTimeout=5,
-            MaxBroadcastRetries=2,
-            MaxChildren=0,
-            MaxDepth=20,
-            MaxRouters=0,
-            dummyNeighborTable=0,
-            BroadcastDeliveryTime=30,
-            ReportConstantCost=0,
-            RouteDiscRetries=0,
-            dummyRoutingTable=0,
-            SecureAllFrames=1,
-            SecurityLevel=5,
-            SymLink=1,
-            CapabilityFlags=143,
-            TransactionPersistenceTime=7,
-            nwkProtocolVersion=2,
-            RouteDiscoveryTime=5,
-            RouteExpiryTime=30,
-            nwkDevAddress=0xFFFE,
-            nwkLogicalChannel=0,
-            nwkCoordAddress=0xFFFE,
-            nwkCoordExtAddress=t.EUI64.convert("00:00:00:00:00:00:00:00"),
-            nwkPanId=0xFFFF,
-            nwkState=t.NwkState.NWK_INIT,
-            channelList=t.Channels.NO_CHANNELS,
-            beaconOrder=15,
-            superFrameOrder=15,
-            scanDuration=0,
-            battLifeExt=0,
-            allocatedRouterAddresses=0,
-            allocatedEndDeviceAddresses=0,
-            nodeDepth=0,
-            extendedPANID=t.EUI64.convert("00:00:00:00:00:00:00:00"),
-            nwkKeyLoaded=t.Bool.false,
-            spare1=t.NwkKeyDesc(
-                KeySeqNum=0, Key=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
-            ),
-            spare2=t.NwkKeyDesc(
-                KeySeqNum=0, Key=[0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0]
-            ),
-            spare3=0,
-            spare4=0,
-            nwkLinkStatusPeriod=60,
-            nwkRouterAgeLimit=3,
-            nwkUseMultiCast=t.Bool.false,
-            nwkIsConcentrator=t.Bool.true,
-            nwkConcentratorDiscoveryTime=120,
-            nwkConcentratorRadius=10,
-            nwkAllFresh=1,
-            nwkManagerAddr=0x0000,
-            nwkTotalTransmissions=0,
-            nwkUpdateId=0,
-        )
+        return const.DEFAULT_NIB.replace()
 
     @reply_to(c.ZDO.ActiveEpReq.Req(DstAddr=0x0000, NWKAddrOfInterest=0x0000))
     def active_endpoints_request(self, req):
@@ -898,7 +840,7 @@ def handle_bdb_start_commissioning(self, request):
                     RemainingModes=c.app_config.BDBCommissioningMode.NwkFormation,
                 ),
                 c.AppConfig.BDBCommissioningNotification.Callback(
-                    Status=c.app_config.BDBCommissioningStatus.FormationFailure,
+                    Status=c.app_config.BDBCommissioningStatus.Success,
                     Mode=c.app_config.BDBCommissioningMode.NwkFormation,
                     RemainingModes=c.app_config.BDBCommissioningMode.NONE,
                 ),
@@ -957,6 +899,14 @@ class BaseLaunchpadCC26X2R1(BaseZStack3Device):
     align_structs = True
     code_revision = 20200805
 
+    def _create_network_nvram(self):
+        super()._create_network_nvram()
+        self._nvram[ExNvIds.LEGACY][OsalNvIds.APS_LINK_KEY_TABLE] = b"\xFF" * 20
+        self._nvram[ExNvIds.ADDRMGR] = {
+            addr: self.nvram_serialize(const.EMPTY_ADDR_MGR_ENTRY)
+            for addr in range(0x0000, 0x0100 + 1)
+        }
+
     def create_nib(self, _=None):
         super().create_nib()
 
@@ -1065,6 +1015,13 @@ class BaseZStack3CC2531(BaseZStack3Device):
     align_structs = False
     code_revision = 20190425
 
+    def _create_network_nvram(self):
+        super()._create_network_nvram()
+        self._nvram[ExNvIds.LEGACY][OsalNvIds.APS_LINK_KEY_TABLE] = b"\xFF" * 17
+        self._nvram[ExNvIds.LEGACY][OsalNvIds.ADDRMGR] = 124 * self.nvram_serialize(
+            const.EMPTY_ADDR_MGR_ENTRY
+        )
+
     def create_nib(self, _=None):
         super().create_nib()
 
@@ -1105,21 +1062,6 @@ def version_replier(self, request):
     def led_responder(self, req):
         return req.Rsp(Status=t.Status.SUCCESS)
 
-    @reply_to(
-        c.AppConfig.BDBStartCommissioning.Req(
-            Mode=c.app_config.BDBCommissioningMode.NwkFormation
-        )
-    )
-    def handle_bdb_start_commissioning(self, request):
-        result = super().handle_bdb_start_commissioning(request)
-
-        # This item is only created after a network is formed
-        self._nvram[ExNvIds.LEGACY][OsalNvIds.ADDRMGR] = 124 * self.nvram_serialize(
-            t.EMPTY_ADDR_MGR_ENTRY
-        )
-
-        return result
-
 
 class FormedLaunchpadCC26X2R1(BaseLaunchpadCC26X2R1):
     def __init__(self, *args, **kwargs):

tests/conftest.py

@@ -6,20 +6,16 @@
 from jsonschema import ValidationError
 
 import zigpy_znp.types as t
-import zigpy_znp.config as conf
-from zigpy_znp.api import ZNP
 from zigpy_znp.znp import security
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 from zigpy_znp.tools.common import validate_backup_json
-from zigpy_znp.zigbee.application import ControllerApplication
 from zigpy_znp.tools.network_backup import main as network_backup
 from zigpy_znp.tools.network_restore import main as network_restore
 
 from ..conftest import (
     ALL_DEVICES,
     EMPTY_DEVICES,
     FORMED_DEVICES,
-    CoroutineMock,
     BaseZStack1CC2531,
     BaseZStack3CC2531,
     FormedZStack1CC2531,
@@ -177,135 +173,41 @@ async def test_network_backup_formed(device, make_znp_server, tmp_path):
     assert len(backup["devices"]) > 1
 
 
-@pytest.mark.parametrize("device", FORMED_DEVICES)
-@pytest.mark.asyncio
-async def test_network_restore_full(
-    device, make_znp_server, backup_json, tmp_path, mocker
-):
-    backup_file = tmp_path / "backup.json"
-    backup_file.write_text(json.dumps(backup_json))
-
-    znp_server = make_znp_server(server_cls=device)
-
-    # Perform the "restore"
-    await network_restore([znp_server._port_path, "-i", str(backup_file), "-c", "2500"])
-
-
 @pytest.mark.parametrize("device", ALL_DEVICES)
 @pytest.mark.asyncio
-async def test_network_restore(device, make_znp_server, backup_json, tmp_path, mocker):
+async def test_network_restore_and_backup(
+    device, make_znp_server, backup_json, tmp_path
+):
     backup_file = tmp_path / "backup.json"
     backup_file.write_text(json.dumps(backup_json))
 
     znp_server = make_znp_server(server_cls=device)
 
-    async def mock_startup(self, *, force_form):
-        assert force_form
-
-        config = self.config[conf.CONF_NWK]
-
-        assert config[conf.CONF_NWK_KEY] == t.KeyData(
-            bytes.fromhex("37668fd64e35e03342e5ef9f35ccf4ab")
-        )
-        assert config[conf.CONF_NWK_PAN_ID] == 0xFEED
-        assert config[conf.CONF_NWK_CHANNEL] == 25
-        assert config[conf.CONF_NWK_EXTENDED_PAN_ID] == t.EUI64.convert(
-            "ab:de:fa:bc:de:fa:bc:de"
-        )
-
-        znp = ZNP(self.config)
-        await znp.connect()
-
-        if OsalNvIds.APS_LINK_KEY_TABLE not in znp_server._nvram[ExNvIds.LEGACY]:
-            znp_server._nvram[ExNvIds.LEGACY][OsalNvIds.APS_LINK_KEY_TABLE] = (
-                b"\x00" * 1000
-            )
-
-        if OsalNvIds.NIB not in znp_server._nvram[ExNvIds.LEGACY]:
-            znp_server._nvram[ExNvIds.LEGACY][
-                OsalNvIds.NIB
-            ] = znp_server.nvram_serialize(znp_server._default_nib())
-
-        self._znp = znp
-        self._znp.set_application(self)
-
-        self._bind_callbacks()
-
-    startup_mock = mocker.patch.object(
-        ControllerApplication, "startup", side_effect=mock_startup, autospec=True
-    )
-
-    async def mock_load_network_info(self, *args, **kwargs):
-        self.network_info = BARE_NETWORK_INFO
-        self.node_info = BARE_NODE_INFO
-
-    load_nwk_info_mock = mocker.patch.object(
-        ZNP, "load_network_info", side_effect=mock_load_network_info, autospec=True
-    )
-
-    write_tc_counter_mock = mocker.patch(
-        "zigpy_znp.tools.network_restore.write_tc_frame_counter", new=CoroutineMock()
-    )
-    write_devices_mock = mocker.patch(
-        "zigpy_znp.tools.network_restore.write_devices", new=CoroutineMock()
-    )
-
-    # Perform the "restore"
+    # Restore our backup on top of an existing network (or onto an empty device)
     await network_restore([znp_server._port_path, "-i", str(backup_file), "-c", "2500"])
 
-    # The NIB should contain correct values
-    nib = znp_server.nvram_deserialize(
-        znp_server._nvram[ExNvIds.LEGACY][OsalNvIds.NIB], t.NIB
-    )
-    assert nib.channelList == t.Channels.from_channel_list([15, 20, 25])
-    assert nib.nwkUpdateId == 2
-    assert nib.SecurityLevel == 5
+    # And then back that restored device up
+    backup_file2 = tmp_path / "backup2.json"
+    await network_backup([znp_server._port_path, "-o", str(backup_file2)])
 
-    # And validate that the low-level functions were called appropriately
-    assert startup_mock.call_count == 1
-    assert startup_mock.mock_calls[0][2]["force_form"] is True
+    backup_json2 = json.loads(backup_file2.read_text())
 
-    assert load_nwk_info_mock.call_count == 1
+    # Fix up some inconsequential metadata
+    backup_json2["metadata"]["internal"] = backup_json["metadata"]["internal"]
+    backup_json2["metadata"]["source"] = backup_json["metadata"]["source"]
+    backup_json2["network_key"]["frame_counter"] -= 2500
 
-    assert write_tc_counter_mock.call_count == 1
-    assert write_tc_counter_mock.mock_calls[0][1][1] == 66781 + 2500
+    if issubclass(device, BaseZStack1CC2531):
+        del backup_json["stack_specific"]["zstack"]["tclk_seed"]
 
-    assert write_devices_mock.call_count == 1
-    write_devices_call = write_devices_mock.mock_calls[0]
+        if not backup_json["stack_specific"]["zstack"]:
+            del backup_json["stack_specific"]
 
-    assert write_devices_call[2]["counter_increment"] == 2500
+        for device in backup_json["devices"]:
+            if "link_key" in device:
+                del device["link_key"]
 
-    if issubclass(device, BaseZStack1CC2531):
-        assert write_devices_call[2]["tclk_seed"] is None
-    else:
-        assert write_devices_call[2]["tclk_seed"] == bytes.fromhex(
-            "c04884427c8a1ed7bb8412815ccce7aa"
-        )
-
-    assert sorted(write_devices_call[1][1], key=lambda d: d.nwk) == [
-        security.StoredDevice(
-            ieee=t.EUI64.convert("00:0b:57:ff:fe:38:b2:12"),
-            nwk=0x9672,
-            aps_link_key=t.KeyData.deserialize(
-                bytes.fromhex("d2fabcbc83dd15d7a9362a7fa39becaa")
-            )[0],
-            rx_counter=123,
-            tx_counter=456,
-        ),
-        security.StoredDevice(
-            ieee=t.EUI64.convert("aa:bb:cc:dd:ee:ff:00:11"),
-            nwk=0xABCD,
-            aps_link_key=t.KeyData.deserialize(
-                bytes.fromhex("01234567801234567801234567801234")
-            )[0],
-            rx_counter=112233,
-            tx_counter=445566,
-        ),
-        security.StoredDevice(
-            ieee=t.EUI64.convert("00:0b:57:ff:fe:36:b9:a0"),
-            nwk=0xF319,
-        ),
-    ]
+    assert backup_json == backup_json2
 
 
 @pytest.mark.asyncio