Revert back to old formation code, writing settings directly is unstable

Author: puddly

tests/api/test_network_state.py

@@ -0,0 +1,39 @@
+import dataclasses
+
+import pytest
+
+from ..conftest import ALL_DEVICES, FORMED_DEVICES, BaseZStack1CC2531
+
+pytestmark = [pytest.mark.asyncio]
+
+
+@pytest.mark.parametrize("to_device", ALL_DEVICES)
+@pytest.mark.parametrize("from_device", FORMED_DEVICES)
+async def test_state_transfer(from_device, to_device, make_connected_znp):
+    formed_znp, _ = await make_connected_znp(server_cls=from_device)
+
+    await formed_znp.load_network_info()
+    formed_znp.close()
+
+    empty_znp, _ = await make_connected_znp(server_cls=to_device)
+
+    await empty_znp.write_network_info(
+        network_info=formed_znp.network_info,
+        node_info=formed_znp.node_info,
+    )
+    await empty_znp.load_network_info()
+
+    # Z-Stack 1 devices can't have some security info read out
+    if issubclass(from_device, BaseZStack1CC2531):
+        assert formed_znp.network_info == dataclasses.replace(
+            empty_znp.network_info, stack_specific={}
+        )
+    elif issubclass(to_device, BaseZStack1CC2531):
+        assert (
+            dataclasses.replace(formed_znp.network_info, stack_specific={})
+            == empty_znp.network_info
+        )
+    else:
+        assert formed_znp.network_info == empty_znp.network_info
+
+    assert formed_znp.node_info == empty_znp.node_info

tests/api/test_nvram.py

@@ -13,11 +13,11 @@ async def test_osal_writes_invalid(connected_znp):
 
     # Passing in untyped integers is not allowed
     with pytest.raises(TypeError):
-        await znp.nvram.osal_write(nvids.OsalNvIds.STARTUP_OPTION, 0xAB)
+        await znp.nvram.osal_write(nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1, 0xAB)
 
     # Neither is passing in an empty value
     with pytest.raises(ValueError):
-        await znp.nvram.osal_write(nvids.OsalNvIds.STARTUP_OPTION, b"")
+        await znp.nvram.osal_write(nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1, b"")
 
     # Or a type that serializes to an empty value
     class Empty:
@@ -27,7 +27,7 @@ def serialize(self):
     assert Empty().serialize() == b""
 
     with pytest.raises(ValueError):
-        await znp.nvram.osal_write(nvids.OsalNvIds.STARTUP_OPTION, Empty())
+        await znp.nvram.osal_write(nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1, Empty())
 
 
 @pytest.mark.parametrize(
@@ -41,7 +41,7 @@ def serialize(self):
 async def test_osal_write_existing(connected_znp, value):
     znp, znp_server = connected_znp
 
-    nvid = nvids.OsalNvIds.STARTUP_OPTION
+    nvid = nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1
 
     # The item is one byte long
     length_rsp = znp_server.reply_once_to(
@@ -65,7 +65,7 @@ async def test_osal_write_existing(connected_znp, value):
 async def test_osal_write_same_length(connected_znp):
     znp, znp_server = connected_znp
 
-    nvid = nvids.OsalNvIds.STARTUP_OPTION
+    nvid = nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1
     value = b"\x01"
 
     # The existing item is also one byte long so we will not recreate it
@@ -87,7 +87,7 @@ async def test_osal_write_same_length(connected_znp):
     await write_rsp
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 @pytest.mark.parametrize("value", [b"\x01\x02"])
 @pytest.mark.parametrize("create", [True, False])
 async def test_osal_write_wrong_length(connected_znp, nvid, value, create):
@@ -130,7 +130,7 @@ async def test_osal_write_wrong_length(connected_znp, nvid, value, create):
         assert not write_rsp.done()
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 @pytest.mark.parametrize("value", [b"test"])
 async def test_osal_write_bad_length(connected_znp, nvid, value):
     znp, znp_server = connected_znp
@@ -147,7 +147,7 @@ async def test_osal_write_bad_length(connected_znp, nvid, value):
     await length_rsp
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 @pytest.mark.parametrize("value", [b"test"])
 async def test_osal_read_success(connected_znp, nvid, value):
     znp, znp_server = connected_znp
@@ -169,7 +169,7 @@ async def test_osal_read_success(connected_znp, nvid, value):
     assert result == value
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 @pytest.mark.parametrize("value", [b"test" * 62 + b"x"])  # 248 + 1 bytes, needs two
 async def test_osal_read_long_success(connected_znp, nvid, value):
     znp, znp_server = connected_znp
@@ -197,7 +197,7 @@ async def test_osal_read_long_success(connected_znp, nvid, value):
     assert result == value
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 async def test_osal_read_failure(connected_znp, nvid):
     znp, znp_server = connected_znp
 
@@ -212,7 +212,7 @@ async def test_osal_read_failure(connected_znp, nvid):
     await length_rsp
 
 
-@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.STARTUP_OPTION])
+@pytest.mark.parametrize("nvid", [nvids.OsalNvIds.HAS_CONFIGURED_ZSTACK1])
 async def test_osal_write_nonexistent(connected_znp, nvid):
     znp, znp_server = connected_znp
 

tests/conftest.py

@@ -162,7 +162,11 @@ async def inner(server_cls):
 
         znp.nvram.align_structs = server_cls.align_structs
         znp.version = server_cls.version
-        znp.capabilities = t.MTCapabilities(0)
+
+        if hasattr(znp_server, "ping_replier"):
+            znp.capabilities = znp_server.ping_replier(None).Capabilities
+        else:
+            znp.capabilities = t.MTCapabilities(0)
 
         return znp, znp_server
 
@@ -455,7 +459,61 @@ def create_nib(self, _=None):
         self._nvram[ExNvIds.LEGACY][OsalNvIds.NWKKEY] = self.nvram_serialize(key_info)
 
     def _default_nib(self):
-        return const.DEFAULT_NIB.replace()
+        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=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=False,
+            nwkIsConcentrator=True,
+            nwkConcentratorDiscoveryTime=120,
+            nwkConcentratorRadius=10,
+            nwkAllFresh=1,
+            nwkManagerAddr=0x0000,
+            nwkTotalTransmissions=0,
+            nwkUpdateId=0,
+        )
 
     @reply_to(c.ZDO.ActiveEpReq.Req(DstAddr=0x0000, NWKAddrOfInterest=0x0000))
     def active_endpoints_request(self, req):
@@ -597,7 +655,22 @@ def nib(self, nib):
         self._nvram[ExNvIds.LEGACY][OsalNvIds.NIB] = self.nvram_serialize(nib)
 
     @reply_to(c.SYS.ResetReq.Req(Type=t.ResetType.Soft))
-    def reset_req(self, request):
+    def reset_req(self, request, *, _handle_startup_reset=True):
+        if (
+            self._nvram.get(ExNvIds.LEGACY, {}).get(OsalNvIds.STARTUP_OPTION)
+            is not None
+            and _handle_startup_reset
+        ):
+            startup, _ = t.StartupOptions.deserialize(
+                self._nvram[ExNvIds.LEGACY][OsalNvIds.STARTUP_OPTION]
+            )
+
+            if startup & t.StartupOptions.ClearState:
+                self._create_network_nvram()
+                self._nvram[ExNvIds.LEGACY][
+                    OsalNvIds.STARTUP_OPTION
+                ] = t.StartupOptions.NONE.serialize()
+
         version = self.version_replier(None)
 
         return c.SYS.ResetInd.Callback(
@@ -891,7 +964,9 @@ def connection_made(self):
         self._first_connection = False
 
         # Z-Stack 3 devices send a callback when they're first used
-        asyncio.get_running_loop().call_soon(self.send, self.reset_req(None))
+        asyncio.get_running_loop().call_soon(
+            self.send, self.reset_req(None, _handle_startup_reset=False)
+        )
 
 
 class BaseLaunchpadCC26X2R1(BaseZStack3Device):