Improve unit tests and test coverage

Author: puddly

tests/tools/test_network_backup_restore.py

@@ -20,6 +20,7 @@
     BaseZStack3CC2531,
     FormedZStack1CC2531,
     BaseLaunchpadCC26X2R1,
+    ResetLaunchpadCC26X2R1,
 )
 from ..application.test_startup import DEV_NETWORK_SETTINGS
 
@@ -91,13 +92,23 @@ def backup_json():
             {
                 "ieee_address": "aabbccddeeff0011",
                 "link_key": {
-                    "key": "01234567801234567801234567801234",  # Not derived from seed
+                    "key": "4dcc18441d843fe86d8ae1396648a92b",  # Derived from seed
                     "rx_counter": 112233,
                     "tx_counter": 445566,
                 },
                 "nwk_address": "abcd",
                 "is_child": False,
             },
+            {
+                "ieee_address": "abcdabcabcabcaaa",
+                "link_key": {
+                    "key": "76567876567867876718736817112312",  # Not derived from seed
+                    "rx_counter": 511223,
+                    "tx_counter": 844556,
+                },
+                "nwk_address": None,  # No known NWK address
+                "is_child": True,  # but a child
+            },
         ],
     }
 
@@ -204,6 +215,7 @@ async def test_network_restore_and_backup(
     backup_json2["metadata"]["source"] = backup_json["metadata"]["source"]
     backup_json2["network_key"]["frame_counter"] -= 2500
 
+    # Remove things Z-Stack 1 does not preserve
     if issubclass(device, BaseZStack1CC2531):
         del backup_json["stack_specific"]["zstack"]["tclk_seed"]
 
@@ -217,6 +229,31 @@ async def test_network_restore_and_backup(
     assert backup_json == backup_json2
 
 
+@pytest.mark.parametrize("device", [ResetLaunchpadCC26X2R1])
+@pytest.mark.asyncio
+async def test_network_restore_pick_optimal_tclk(
+    device, make_znp_server, backup_json, tmp_path
+):
+    # Pick a sub-optimal TCLK
+    old_tclk_seed = backup_json["stack_specific"]["zstack"]["tclk_seed"]
+    backup_json["stack_specific"]["zstack"]["tclk_seed"] = "ab" * 16
+
+    backup_file = tmp_path / "backup.json"
+    backup_file.write_text(json.dumps(backup_json))
+
+    znp_server = make_znp_server(server_cls=device)
+
+    await network_restore([znp_server._port_path, "-i", str(backup_file)])
+
+    backup_file2 = tmp_path / "backup2.json"
+    await network_backup([znp_server._port_path, "-o", str(backup_file2)])
+
+    backup_json2 = json.loads(backup_file2.read_text())
+
+    # Optimal TCLK is re-derived
+    assert backup_json2["stack_specific"]["zstack"]["tclk_seed"] == old_tclk_seed
+
+
 @pytest.mark.asyncio
 async def test_tc_frame_counter_zstack1(make_connected_znp):
     znp, znp_server = await make_connected_znp(BaseZStack1CC2531)

zigpy_znp/api.py

@@ -26,11 +26,7 @@
     CallbackResponseListener,
 )
 from zigpy_znp.frames import GeneralFrame
-from zigpy_znp.exceptions import (
-    SecurityError,
-    CommandNotRecognized,
-    InvalidCommandResponse,
-)
+from zigpy_znp.exceptions import CommandNotRecognized, InvalidCommandResponse
 from zigpy_znp.types.nvids import ExNvIds, OsalNvIds
 
 LOGGER = logging.getLogger(__name__)
@@ -113,24 +109,9 @@ async def load_network_info(self, *, load_devices=False):
         if not is_on_network:
             raise ValueError("Device is not a part of a network")
 
-        try:
-            key_desc = await self.nvram.osal_read(
-                OsalNvIds.NWK_ACTIVE_KEY_INFO, item_type=t.NwkKeyDesc
-            )
-        except SecurityError:
-            # XXX: Z-Stack 1 has no way to read some key info
-            key = await self.nvram.osal_read(OsalNvIds.PRECFGKEY, item_type=t.KeyData)
-            key_desc = t.NwkKeyDesc(
-                KeySeqNum=0,
-                Key=key,
-            )
-
-        tclk_seed = None
-
-        if self.version > 1.2:
-            tclk_seed = await self.nvram.osal_read(
-                OsalNvIds.TCLK_SEED, item_type=t.KeyData
-            )
+        key_desc = await self.nvram.osal_read(
+            OsalNvIds.NWK_ACTIVE_KEY_INFO, item_type=t.NwkKeyDesc
+        )
 
         tc_frame_counter = await security.read_tc_frame_counter(
             self, ext_pan_id=nib.extendedPANID
@@ -157,7 +138,11 @@ async def load_network_info(self, *, load_devices=False):
             stack_specific=None,
         )
 
-        if tclk_seed is not None:
+        if self.version > 1.2:
+            tclk_seed = await self.nvram.osal_read(
+                OsalNvIds.TCLK_SEED, item_type=t.KeyData
+            )
+
             network_info.stack_specific = {
                 "zstack": {
                     "tclk_seed": tclk_seed.serialize().hex(),
@@ -167,9 +152,14 @@ async def load_network_info(self, *, load_devices=False):
         # This takes a few seconds
         if load_devices:
             for dev in await security.read_devices(self):
+                if dev.node_info.nwk == 0xFFFE:
+                    dev = dev.replace(
+                        node_info=dataclasses.replace(dev.node_info, nwk=None)
+                    )
+
                 if dev.is_child:
                     network_info.children.append(dev.node_info)
-                else:
+                elif dev.node_info.nwk is not None:
                     network_info.nwk_addresses[dev.node_info.ieee] = dev.node_info.nwk
 
                 if dev.key is not None:
@@ -306,14 +296,14 @@ async def write_network_info(
         while True:
             try:
                 nib = await self.nvram.osal_read(OsalNvIds.NIB, item_type=t.NIB)
-
+            except KeyError:
+                pass
+            else:
                 LOGGER.debug("Current NIB is %s", nib)
 
                 # Usually this works after the first attempt
                 if nib.nwkLogicalChannel != 0 and nib.nwkPanId != 0xFFFE:
                     break
-            except KeyError:
-                pass
 
             await asyncio.sleep(1)
 

zigpy_znp/tools/network_restore.py

@@ -53,14 +53,19 @@ def json_backup_to_zigpy_state(
 
     for obj in backup["devices"]:
         node = zigpy.state.NodeInfo()
-        node.nwk, _ = t.NWK.deserialize(bytes.fromhex(obj["nwk_address"])[::-1])
+
+        if obj["nwk_address"] is not None:
+            node.nwk, _ = t.NWK.deserialize(bytes.fromhex(obj["nwk_address"])[::-1])
+        else:
+            node.nwk = None
+
         node.ieee, _ = t.EUI64.deserialize(bytes.fromhex(obj["ieee_address"])[::-1])
         node.logical_type = None
 
         # The `is_child` key is currently optional
         if obj.get("is_child", True):
             network_info.children.append(node)
-        else:
+        elif node.nwk is not None:
             network_info.nwk_addresses[node.ieee] = node.nwk
 
         if "link_key" in obj:
@@ -73,6 +78,9 @@ def json_backup_to_zigpy_state(
 
             network_info.key_table.append(key)
 
+        # XXX: Devices that are not children, have no NWK address, and have no link key
+        #      are effectively ignored, since there is no place to write them
+
     return network_info, node_info
 
 

zigpy_znp/znp/security.py

@@ -366,7 +366,7 @@ def find_optimal_tclk_seed(
     if not keys:
         return tclk_seed
 
-    best_count, best_seed = max(iter_seed_candidates(keys))
+    best_count, best_seed = max(sorted(iter_seed_candidates(keys)))
     tclk_count = count_seed_matches(keys, tclk_seed)
     assert tclk_count <= best_count