Test and fix from-scratch network formation and startup

Author: puddly

tests/test_api.py

@@ -294,11 +294,11 @@ def test_command_deduplication():
                 partial=True,
                 Status=c.app_config.BDBCommissioningStatus.InProgress,
                 Mode=c.app_config.BDBCommissioningMode.NwkFormation,
-                RemainingModes=c.app_config.BDBRemainingCommissioningModes.InitiatorTl,
+                RemainingModes=c.app_config.BDBCommissioningMode.InitiatorTouchLink,
             ),
             c.AppConfig.BDBCommissioningNotification.Callback(
                 partial=True,
-                RemainingModes=c.app_config.BDBRemainingCommissioningModes.InitiatorTl,
+                RemainingModes=c.app_config.BDBCommissioningMode.InitiatorTouchLink,
             ),
         ]
     )
@@ -312,7 +312,7 @@ def test_command_deduplication():
         ),
         c.AppConfig.BDBCommissioningNotification.Callback(
             partial=True,
-            RemainingModes=c.app_config.BDBRemainingCommissioningModes.InitiatorTl,
+            RemainingModes=c.app_config.BDBCommissioningMode.InitiatorTouchLink,
         ),
     }
 

tests/test_application.py

@@ -66,7 +66,7 @@ async def callback(request):
             callback.called = True
 
             for response in responses:
-                await asyncio.sleep(0.1)
+                await asyncio.sleep(0.01)
                 LOGGER.debug("Replying to %s with %s", request, response)
 
                 try:
@@ -86,7 +86,7 @@ async def callback(request):
             callback.call_count += 1
 
             for response in responses:
-                await asyncio.sleep(0.1)
+                await asyncio.sleep(0.01)
                 LOGGER.debug("Replying to %s with %s", request, response)
 
                 try:
@@ -159,19 +159,55 @@ def application(znp_server):
         ],
     )
 
+    active_eps = [100, 12, 11, 8, 1]
+
     znp_server.reply_to(
         request=c.ZDO.ActiveEpReq.Req(DstAddr=0x0000, NWKAddrOfInterest=0x0000),
         responses=[
             c.ZDO.ActiveEpReq.Rsp(Status=t.Status.Success),
             c.ZDO.ActiveEpRsp.Callback(
-                Src=0x0000, Status=t.ZDOStatus.SUCCESS, NWK=0x0000, ActiveEndpoints=[]
+                Src=0x0000,
+                Status=t.ZDOStatus.SUCCESS,
+                NWK=0x0000,
+                ActiveEndpoints=active_eps,
             ),
         ],
     )
 
     znp_server.reply_to(
-        request=c.AF.Register.Req(partial=True),
-        responses=[c.AF.Register.Rsp(Status=t.Status.Success)],
+        request=c.ZDO.ActiveEpReq.Req(DstAddr=0x0000, NWKAddrOfInterest=0x0000),
+        responses=[
+            c.ZDO.ActiveEpReq.Rsp(Status=t.Status.Success),
+            c.ZDO.ActiveEpRsp.Callback(
+                Src=0x0000,
+                Status=t.ZDOStatus.SUCCESS,
+                NWK=0x0000,
+                ActiveEndpoints=active_eps,
+            ),
+        ],
+    )
+
+    def on_endpoint_registration(req):
+        assert req.Endpoint not in active_eps
+
+        active_eps.append(req.Endpoint)
+        active_eps.sort(reverse=True)
+
+        return c.AF.Register.Rsp(Status=t.Status.Success)
+
+    znp_server.reply_to(
+        request=c.AF.Register.Req(partial=True), responses=[on_endpoint_registration],
+    )
+
+    def on_endpoint_deletion(req):
+        assert req.Endpoint in active_eps
+
+        active_eps.remove(req.Endpoint)
+
+        return c.AF.Delete.Rsp(Status=t.Status.Success)
+
+    znp_server.reply_to(
+        request=c.AF.Delete.Req(partial=True), responses=[on_endpoint_deletion],
     )
 
     znp_server.reply_to(
@@ -183,12 +219,7 @@ def application(znp_server):
             c.AppConfig.BDBCommissioningNotification.Callback(
                 Status=c.app_config.BDBCommissioningStatus.Success,
                 Mode=c.app_config.BDBCommissioningMode.NwkSteering,
-                RemainingModes=c.app_config.BDBRemainingCommissioningModes.NONE,
-            ),
-            c.AppConfig.BDBCommissioningNotification.Callback(
-                Status=c.app_config.BDBCommissioningStatus.NoNetwork,
-                Mode=c.app_config.BDBCommissioningMode.NwkSteering,
-                RemainingModes=c.app_config.BDBRemainingCommissioningModes.NONE,
+                RemainingModes=c.app_config.BDBCommissioningMode.NONE,
             ),
         ],
     )
@@ -206,30 +237,71 @@ def application(znp_server):
             responses=[c.Sys.OSALNVWrite.Rsp(Status=t.Status.Success)],
         )
 
+    znp_server.reply_to(
+        request=c.Sys.OSALNVRead.Req(Id=NwkNvIds.HAS_CONFIGURED_ZSTACK3, Offset=0),
+        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.Success, Value=b"\x55")],
+    )
+
+    znp_server.reply_to(
+        request=c.Util.GetDeviceInfo.Req(),
+        responses=[
+            c.Util.GetDeviceInfo.Rsp(
+                Status=t.Status.Success,
+                IEEE=t.EUI64([0x00, 0x12, 0x4B, 0x00, 0x1C, 0xAA, 0xAC, 0x5C]),
+                NWK=t.NWK(0xFFFE),
+                DeviceType=t.DeviceLogicalType(7),
+                DeviceState=t.DeviceState.InitializedNotStarted,
+                AssociatedDevices=[],
+            )
+        ],
+    )
+
+    znp_server.reply_to(
+        request=c.ZDO.StartupFromApp.Req(partial=True),
+        responses=[
+            c.ZDO.StartupFromApp.Rsp(State=c.zdo.StartupState.RestoredNetworkState),
+            c.ZDO.StateChangeInd.Callback(State=t.DeviceState.StartedAsCoordinator),
+        ],
+    )
+
     return app, znp_server
 
 
 @pytest_mark_asyncio_timeout(seconds=5)
-async def test_application_startup(application):
+async def test_application_startup_endpoints(application):
     app, znp_server = application
 
-    num_endpoints = 5
     endpoints = []
+    znp_server.callback_for_response(c.AF.Register.Req(partial=True), endpoints.append)
 
-    def register_endpoint(request):
-        nonlocal num_endpoints
-        num_endpoints -= 1
+    await app.startup(auto_form=False)
 
-        endpoints.append(request)
+    assert len(endpoints) == 5
 
-        if num_endpoints < 0:
-            raise RuntimeError("Too many endpoints registered")
 
-    znp_server.callback_for_response(c.AF.Register.Req(partial=True), register_endpoint)
+@pytest_mark_asyncio_timeout(seconds=5)
+async def test_application_startup_failure(application):
+    app, znp_server = application
 
-    await app.startup(auto_form=False)
+    # Prevent the fixture's default response
+    znp_server._response_listeners[c.Sys.OSALNVRead.Req.header].clear()
 
-    assert len(endpoints) == 5
+    znp_server.reply_once_to(
+        request=c.Sys.OSALNVRead.Req(Id=NwkNvIds.HAS_CONFIGURED_ZSTACK3, Offset=0),
+        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.InvalidParameter, Value=b"")],
+    )
+
+    # We cannot start the application if Z-Stack is not configured and without auto_form
+    with pytest.raises(RuntimeError):
+        await app.startup(auto_form=False)
+
+    znp_server.reply_once_to(
+        request=c.Sys.OSALNVRead.Req(Id=NwkNvIds.HAS_CONFIGURED_ZSTACK3, Offset=0),
+        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.Success, Value=b"\x00")],
+    )
+
+    with pytest.raises(RuntimeError):
+        await app.startup(auto_form=False)
 
 
 @pytest_mark_asyncio_timeout(seconds=3)
@@ -858,17 +930,9 @@ async def test_force_remove(application, mocker):
 async def test_auto_form_unnecessary(application, mocker):
     app, znp_server = application
 
-    # b"\x55" means that Z-Stack is already configured?
-    read_zstack_configured = znp_server.reply_once_to(
-        request=c.Sys.OSALNVRead.Req(Id=NwkNvIds.HAS_CONFIGURED_ZSTACK3, Offset=0),
-        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.Success, Value=b"\x55")],
-    )
-
     mocker.patch.object(app, "form_network")
 
     await app.startup(auto_form=True)
-
-    await read_zstack_configured
     assert app.form_network.call_count == 0
 
 
@@ -881,22 +945,31 @@ async def test_auto_form_necessary(application, mocker):
     mocker.patch.object(app, "_reset", new=CoroutineMock())
 
     def nvram_writer(req):
-        if req.Id in nvram:
-            raise ValueError("Unexpected overwrite")
-
         nvram[req.Id] = req.Value
 
         return c.Sys.OSALNVWrite.Rsp(Status=t.Status.Success)
 
+    def nvram_init(req):
+        nvram[req.Id] = req.Value
+
+        return c.Sys.OSALNVItemInit.Rsp(Status=t.Status.Success)
+
+    # Prevent the fixture's default response
+    znp_server._response_listeners[c.Sys.OSALNVRead.Req.header].clear()
+
     read_zstack_configured = znp_server.reply_once_to(
         request=c.Sys.OSALNVRead.Req(Id=NwkNvIds.HAS_CONFIGURED_ZSTACK3, Offset=0),
-        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.Success, Value=b"\x00")],
+        responses=[c.Sys.OSALNVRead.Rsp(Status=t.Status.InvalidParameter, Value=b"")],
     )
 
     znp_server.reply_to(
         request=c.Sys.OSALNVWrite.Req(Offset=0, partial=True), responses=[nvram_writer]
     )
 
+    znp_server.reply_to(
+        request=c.Sys.OSALNVItemInit.Req(partial=True), responses=[nvram_init]
+    )
+
     znp_server.reply_to(
         request=c.AppConfig.BDBStartCommissioning.Req(
             Mode=c.app_config.BDBCommissioningMode.NwkFormation
@@ -919,8 +992,9 @@ def nvram_writer(req):
     await read_zstack_configured
 
     assert app.update_network.call_count == 1
-    assert app._reset.call_count == 2
+    assert app._reset.call_count == 1
 
+    assert nvram[NwkNvIds.HAS_CONFIGURED_ZSTACK3] == b"\x55"
     assert nvram[NwkNvIds.STARTUP_OPTION] == t.StartupOptions.ClearState.serialize()
     assert nvram[NwkNvIds.LOGICAL_TYPE] == t.DeviceLogicalType.Coordinator.serialize()
     assert nvram[NwkNvIds.ZDO_DIRECT_CB] == t.Bool(True).serialize()

zigpy_znp/commands/app_config.py

@@ -49,23 +49,14 @@ class BDBCommissioningStatus(t.enum_uint8):
     Failure = 0x0E
 
 
-class BDBCommissioningMode(t.enum_uint8):
-    Initialization = 0x00
-    NwkSteering = 0x01
-    NwkFormation = 0x02
-    FindingBinding = 0x03
-    Touchlink = 0x04
-    ParentLost = 0x05
-
-
-class BDBRemainingCommissioningModes(t.enum_flag_uint8):
+class BDBCommissioningMode(t.enum_flag_uint8):
     NONE = 0
 
-    InitiatorTl = 1 << 0
+    InitiatorTouchLink = 1 << 0
     NwkSteering = 1 << 1
     NwkFormation = 1 << 2
     FindingBinding = 1 << 3
-    Initialization = 1 << 4
+    Touchlink = 1 << 4
     ParentLost = 1 << 5
 
 
@@ -245,7 +236,7 @@ class AppConfig(t.CommandsBase, subsystem=t.Subsystem.APPConfig):
                 ),
                 t.Param(
                     "RemainingModes",
-                    BDBRemainingCommissioningModes,
+                    BDBCommissioningMode,
                     (
                         "Bitmask of the remaining commissioning modes after "
                         "this notification"

zigpy_znp/commands/sys.py

@@ -16,7 +16,7 @@ class Sys(t.CommandsBase, subsystem=t.Subsystem.SYS):
             (
                 t.Param(
                     "Type",
-                    t.uint8_t,
+                    t.ResetType,
                     (
                         "This command will reset the device by using a hardware reset "
                         "(i.e. watchdog reset) if ‘Type’ is zero. Otherwise a soft "

zigpy_znp/types/named.py

@@ -205,9 +205,9 @@ class KeySource(basic.FixedList, item_type=basic.uint8_t, length=8):
 
 
 class StartupOptions(basic.enum_flag_uint8):
-    ClearConfig = 1 << 1
-    ClearState = 1 << 2
-    AutoStart = 1 << 3
+    ClearConfig = 1 << 0
+    ClearState = 1 << 1
+    AutoStart = 1 << 2
 
 
 class DeviceLogicalType(basic.enum_flag_uint8):