Uppercase command subsystem names (Util => UTIL, App => APP)

Author: puddly

tests/api/test_request.py

@@ -55,7 +55,7 @@ async def test_cleanup_timeout_internal(connected_znp):
     assert not znp._listeners
 
     with pytest.raises(asyncio.TimeoutError):
-        await znp.request(c.Util.TimeAlive.Req())
+        await znp.request(c.UTIL.TimeAlive.Req())
 
     # We should be cleaned up
     assert not znp._listeners
@@ -69,7 +69,7 @@ async def test_cleanup_timeout_external(connected_znp):
     # This request will timeout because we didn't send anything back
     with pytest.raises(asyncio.TimeoutError):
         async with async_timeout.timeout(0.1):
-            await znp.request(c.Util.TimeAlive.Req())
+            await znp.request(c.UTIL.TimeAlive.Req())
 
     # We should be cleaned up
     assert not znp._listeners
@@ -84,7 +84,7 @@ async def test_callback_rsp_cleanup_timeout_external(connected_znp):
     with pytest.raises(asyncio.TimeoutError):
         async with async_timeout.timeout(0.1):
             await znp.request_callback_rsp(
-                request=c.Util.TimeAlive.Req(),
+                request=c.UTIL.TimeAlive.Req(),
                 callback=c.SYS.ResetInd.Callback(partial=True),
             )
 
@@ -103,7 +103,7 @@ async def test_callback_rsp_cleanup_timeout_internal(background, connected_znp):
     # This request will timeout because we didn't send anything back
     with pytest.raises(asyncio.TimeoutError):
         await znp.request_callback_rsp(
-            request=c.Util.TimeAlive.Req(),
+            request=c.UTIL.TimeAlive.Req(),
             callback=c.SYS.ResetInd.Callback(partial=True),
             background=background,
         )
@@ -122,7 +122,7 @@ async def test_callback_rsp_cleanup_background_error(connected_znp):
     # This request will timeout because we didn't send anything back
     with pytest.raises(asyncio.TimeoutError):
         await znp.request_callback_rsp(
-            request=c.Util.TimeAlive.Req(),
+            request=c.UTIL.TimeAlive.Req(),
             callback=c.SYS.ResetInd.Callback(partial=True),
             background=True,
         )
@@ -141,7 +141,7 @@ async def test_callback_rsp_cleanup_background_timeout(connected_znp):
     # This request will timeout because we didn't send anything back
     with pytest.raises(asyncio.TimeoutError):
         await znp.request_callback_rsp(
-            request=c.Util.TimeAlive.Req(),
+            request=c.UTIL.TimeAlive.Req(),
             callback=c.SYS.ResetInd.Callback(partial=True),
             background=True,
         )
@@ -158,15 +158,15 @@ async def test_callback_rsp_cleanup_concurrent(connected_znp, event_loop, mocker
     assert not znp._listeners
 
     def send_responses():
-        znp_server.send(c.Util.TimeAlive.Rsp(Seconds=123))
-        znp_server.send(c.Util.TimeAlive.Rsp(Seconds=456))
+        znp_server.send(c.UTIL.TimeAlive.Rsp(Seconds=123))
+        znp_server.send(c.UTIL.TimeAlive.Rsp(Seconds=456))
         znp_server.send(c.SYS.OSALTimerExpired.Callback(Id=0xAB))
         znp_server.send(c.SYS.OSALTimerExpired.Callback(Id=0xCD))
 
     event_loop.call_soon(send_responses)
 
     callback_rsp = await znp.request_callback_rsp(
-        request=c.Util.TimeAlive.Req(),
+        request=c.UTIL.TimeAlive.Req(),
         callback=c.SYS.OSALTimerExpired.Callback(partial=True),
     )
 
@@ -178,7 +178,7 @@ def send_responses():
     # Even though all four requests were sent in the same tick, they should be handled
     # correctly by request_callback_rsp and in the correct order
     assert znp._unhandled_command.mock_calls == [
-        mocker.call(c.Util.TimeAlive.Rsp(Seconds=456)),
+        mocker.call(c.UTIL.TimeAlive.Rsp(Seconds=456)),
         mocker.call(c.SYS.OSALTimerExpired.Callback(Id=0xCD)),
     ]
 

tests/api/test_response.py

@@ -213,7 +213,7 @@ async def test_response_callback_simple(connected_znp, event_loop, mocker):
 # being implicitly marked as "asyncio"
 async def test_command_deduplication_simple():
     c1 = c.SYS.Ping.Rsp(partial=True)
-    c2 = c.Util.TimeAlive.Rsp(Seconds=12)
+    c2 = c.UTIL.TimeAlive.Rsp(Seconds=12)
 
     assert _deduplicate_commands([]) == ()
     assert _deduplicate_commands([c1]) == (c1,)
@@ -230,8 +230,8 @@ async def test_command_deduplication_complex():
             c.SYS.Ping.Rsp(partial=True),
             c.SYS.Ping.Rsp(partial=True),
             # Matching against different command types should also work
-            c.Util.TimeAlive.Rsp(Seconds=12),
-            c.Util.TimeAlive.Rsp(Seconds=10),
+            c.UTIL.TimeAlive.Rsp(Seconds=12),
+            c.UTIL.TimeAlive.Rsp(Seconds=10),
             c.AppConfig.BDBCommissioningNotification.Callback(
                 partial=True, Status=c.app_config.BDBCommissioningStatus.InProgress
             ),
@@ -255,8 +255,8 @@ async def test_command_deduplication_complex():
 
     assert set(result) == {
         c.SYS.Ping.Rsp(partial=True),
-        c.Util.TimeAlive.Rsp(Seconds=12),
-        c.Util.TimeAlive.Rsp(Seconds=10),
+        c.UTIL.TimeAlive.Rsp(Seconds=12),
+        c.UTIL.TimeAlive.Rsp(Seconds=10),
         c.AppConfig.BDBCommissioningNotification.Callback(
             partial=True, Status=c.app_config.BDBCommissioningStatus.InProgress
         ),
@@ -285,7 +285,7 @@ async def async_callback(response):
 
     good_response1 = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
     good_response2 = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.APP)
-    good_response3 = c.Util.TimeAlive.Rsp(Seconds=12)
+    good_response3 = c.UTIL.TimeAlive.Rsp(Seconds=12)
     bad_response1 = c.SYS.SetExtAddr.Rsp(Status=t.Status.SUCCESS)
     bad_response2 = c.SYS.NVWrite.Req(
         SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
@@ -296,16 +296,16 @@ async def async_callback(response):
         c.SYS.Ping.Rsp(partial=True),
         c.SYS.Ping.Rsp(partial=True),
         # Matching against different response types should also work
-        c.Util.TimeAlive.Rsp(Seconds=12),
+        c.UTIL.TimeAlive.Rsp(Seconds=12),
         c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS),
         c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS),
-        c.Util.TimeAlive.Rsp(Seconds=10),
+        c.UTIL.TimeAlive.Rsp(Seconds=10),
     ]
 
     assert set(_deduplicate_commands(responses)) == {
         c.SYS.Ping.Rsp(partial=True),
-        c.Util.TimeAlive.Rsp(Seconds=12),
-        c.Util.TimeAlive.Rsp(Seconds=10),
+        c.UTIL.TimeAlive.Rsp(Seconds=12),
+        c.UTIL.TimeAlive.Rsp(Seconds=10),
     }
 
     # We shouldn't see any effects from receiving a frame early
@@ -335,7 +335,7 @@ async def test_wait_for_responses(connected_znp, event_loop):
 
     response1 = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS)
     response2 = c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.APP)
-    response3 = c.Util.TimeAlive.Rsp(Seconds=12)
+    response3 = c.UTIL.TimeAlive.Rsp(Seconds=12)
     response4 = c.SYS.SetExtAddr.Rsp(Status=t.Status.SUCCESS)
     response5 = c.SYS.NVWrite.Req(
         SysId=0x12, ItemId=0x3456, SubId=0x7890, Offset=0x00, Value=b"asdfoo"
@@ -352,22 +352,22 @@ async def test_wait_for_responses(connected_znp, event_loop):
     # Will match the first response3 and detach
     future2 = znp.wait_for_responses(
         [
-            c.Util.TimeAlive.Rsp(Seconds=12),
+            c.UTIL.TimeAlive.Rsp(Seconds=12),
             c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.UTIL),
         ]
     )
 
     # Will not match anything
-    future3 = znp.wait_for_responses([c.Util.TimeAlive.Rsp(Seconds=10)])
+    future3 = znp.wait_for_responses([c.UTIL.TimeAlive.Rsp(Seconds=10)])
 
     # Will match response1 the second time around
     future4 = znp.wait_for_responses(
         [
             # Matching against different response types should also work
-            c.Util.TimeAlive.Rsp(Seconds=12),
+            c.UTIL.TimeAlive.Rsp(Seconds=12),
             c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS),
             c.SYS.Ping.Rsp(Capabilities=t.MTCapabilities.SYS),
-            c.Util.TimeAlive.Rsp(Seconds=10),
+            c.UTIL.TimeAlive.Rsp(Seconds=10),
         ]
     )
 
@@ -401,6 +401,6 @@ async def test_wait_for_responses(connected_znp, event_loop):
 
     await asyncio.sleep(0)
 
-    znp.frame_received(c.Util.TimeAlive.Rsp(Seconds=10).to_frame())
+    znp.frame_received(c.UTIL.TimeAlive.Rsp(Seconds=10).to_frame())
     assert future3.done()
-    assert (await future3) == c.Util.TimeAlive.Rsp(Seconds=10)
+    assert (await future3) == c.UTIL.TimeAlive.Rsp(Seconds=10)

tests/application/test_requests.py

@@ -777,12 +777,12 @@ def assoc_get_with_addr(req):
 
         if assoc_device is None:
             dev, _ = c.util.Device.deserialize(b"\xFF" * 100)
-            return c.Util.AssocGetWithAddress.Rsp(Device=dev)
+            return c.UTIL.AssocGetWithAddress.Rsp(Device=dev)
 
-        return c.Util.AssocGetWithAddress.Rsp(Device=assoc_device)
+        return c.UTIL.AssocGetWithAddress.Rsp(Device=assoc_device)
 
     did_assoc_get = znp_server.reply_once_to(
-        c.Util.AssocGetWithAddress.Req(IEEE=device.ieee, partial=True),
+        c.UTIL.AssocGetWithAddress.Req(IEEE=device.ieee, partial=True),
         responses=[assoc_get_with_addr],
     )
 
@@ -796,23 +796,23 @@ def assoc_remove(req):
             nonlocal assoc_device
 
             if assoc_device is None:
-                return c.Util.AssocRemove.Rsp(Status=t.Status.FAILURE)
+                return c.UTIL.AssocRemove.Rsp(Status=t.Status.FAILURE)
 
             assoc_device = None
-            return c.Util.AssocRemove.Rsp(Status=t.Status.SUCCESS)
+            return c.UTIL.AssocRemove.Rsp(Status=t.Status.SUCCESS)
 
         did_assoc_remove = znp_server.reply_once_to(
-            c.Util.AssocRemove.Req(IEEE=device.ieee),
+            c.UTIL.AssocRemove.Req(IEEE=device.ieee),
             responses=[assoc_remove],
         )
 
         did_assoc_add = znp_server.reply_once_to(
-            c.Util.AssocAdd.Req(
+            c.UTIL.AssocAdd.Req(
                 NWK=device.nwk,
                 IEEE=device.ieee,
                 NodeRelation=c.util.NodeRelation.CHILD_FFD_RX_IDLE,
             ),
-            responses=[c.Util.AssocAdd.Rsp(Status=t.Status.SUCCESS)],
+            responses=[c.UTIL.AssocAdd.Rsp(Status=t.Status.SUCCESS)],
         )
     else:
         did_assoc_remove = None

tests/application/test_startup.py

@@ -175,10 +175,10 @@ async def test_led_mode(device, make_application):
     # Z-Stack just does not respond to this command if HAL_LED is not enabled
     # It does not send the usual "command not recognized" response
     set_led_mode = znp_server.reply_once_to(
-        request=c.Util.LEDControl.Req(partial=True),
+        request=c.UTIL.LEDControl.Req(partial=True),
         responses=[]
         if device is FormedLaunchpadCC26X2R1
-        else [c.Util.LEDControl.Rsp(Status=t.Status.SUCCESS)],
+        else [c.UTIL.LEDControl.Rsp(Status=t.Status.SUCCESS)],
     )
 
     await app.startup(auto_form=False)

tests/conftest.py

@@ -637,14 +637,14 @@ def reset_req(self, request, *, startup=True):
             MaintRel=version.MaintRel,
         )
 
-    @reply_to(c.Util.GetDeviceInfo.Req())
+    @reply_to(c.UTIL.GetDeviceInfo.Req())
     def util_device_info(self, request):
         nwk = 0xFFFE
 
         if self.device_state == t.DeviceState.StartedAsCoordinator:
             nwk = 0x0000
 
-        return c.Util.GetDeviceInfo.Rsp(
+        return c.UTIL.GetDeviceInfo.Rsp(
             Status=t.Status.SUCCESS,
             IEEE=t.EUI64.deserialize(self._nvram[ExNvIds.LEGACY][OsalNvIds.EXTADDR])[0],
             NWK=nwk,
@@ -801,7 +801,7 @@ def permit_join(self, request):
             c.ZDO.PermitJoinInd.Callback(Duration=0),
         ]
 
-    @reply_to(c.Util.LEDControl.Req(partial=True))
+    @reply_to(c.UTIL.LEDControl.Req(partial=True))
     def led_responder(self, req):
         return req.Rsp(Status=t.Status.SUCCESS)
 
@@ -1030,7 +1030,7 @@ def node_desc_responder(self, req):
             ),
         ]
 
-    @reply_to(c.Util.LEDControl.Req(partial=True))
+    @reply_to(c.UTIL.LEDControl.Req(partial=True))
     def led_responder(self, req):
         # XXX: Yes, there is *no response*
         return
@@ -1076,7 +1076,7 @@ def version_replier(self, request):
 
     node_desc_responder = BaseZStack1CC2531.node_desc_responder
 
-    @reply_to(c.Util.LEDControl.Req(partial=True))
+    @reply_to(c.UTIL.LEDControl.Req(partial=True))
     def led_responder(self, req):
         return req.Rsp(Status=t.Status.SUCCESS)
 

tests/test_commands.py

@@ -184,23 +184,23 @@ def test_command_param_binding():
 
     # Invalid type
     with pytest.raises(ValueError):
-        c.Util.TimeAlive.Rsp(Seconds=b"asd")
+        c.UTIL.TimeAlive.Rsp(Seconds=b"asd")
 
     # Valid type but invalid value
     with pytest.raises(ValueError):
-        c.Util.SetPreConfigKey.Req(PreConfigKey=t.KeyData([1, 2, 3]))
+        c.UTIL.SetPreConfigKey.Req(PreConfigKey=t.KeyData([1, 2, 3]))
 
     # Coerced numerical type
-    a = c.Util.TimeAlive.Rsp(Seconds=12)
-    b = c.Util.TimeAlive.Rsp(Seconds=t.uint32_t(12))
+    a = c.UTIL.TimeAlive.Rsp(Seconds=12)
+    b = c.UTIL.TimeAlive.Rsp(Seconds=t.uint32_t(12))
 
     assert a == b
     assert a.Seconds == b.Seconds
     assert type(a.Seconds) == type(b.Seconds) == t.uint32_t  # noqa: E721
 
     # Overflowing integer types
     with pytest.raises(ValueError):
-        c.Util.TimeAlive.Rsp(Seconds=10 ** 20)
+        c.UTIL.TimeAlive.Rsp(Seconds=10 ** 20)
 
     # Integers will not be coerced to enums
     assert t.MTCapabilities.SYS == 0x0001
@@ -223,15 +223,15 @@ def test_command_param_binding():
     assert isinstance(cmd.Value, t.ShortBytes)
 
     # Lists are converted to typed LVLists
-    c.Util.BindAddEntry.Req(
+    c.UTIL.BindAddEntry.Req(
         DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
         DstEndpoint=0x56,
         ClusterIdList=[0x12, 0x45],
     )
 
     # Type errors within containers are also caught
     with pytest.raises(ValueError):
-        c.Util.BindAddEntry.Req(
+        c.UTIL.BindAddEntry.Req(
             DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
             DstEndpoint=0x56,
             ClusterIdList=[0x12, 0x457890],  # 0x457890 doesn't fit into a uint8_t
@@ -372,7 +372,7 @@ def test_simple_descriptor():
 
 
 def test_command_str_repr():
-    command = c.Util.BindAddEntry.Req(
+    command = c.UTIL.BindAddEntry.Req(
         DstAddrModeAddr=t.AddrModeAddress(mode=t.AddrMode.NWK, address=0x1234),
         DstEndpoint=0x56,
         ClusterIdList=[0x12, 0x34],

zigpy_znp/commands/__init__.py

@@ -1,25 +1,25 @@
 from .af import AF
-from .app import App
+from .app import APP
 from .mac import MAC
 from .sys import SYS
 from .ubl import UBL
 from .zdo import ZDO
 from .zgp import ZGP
 from .znp import ZNP
 from .sapi import SAPI
-from .util import Util
+from .util import UTIL
 from .rpc_error import RPCError
 from .app_config import AppConfig
 
 ALL_COMMANDS = [
     RPCError,
     AF,
-    App,
+    APP,
     AppConfig,
     MAC,
     SAPI,
     SYS,
-    Util,
+    UTIL,
     ZDO,
     ZGP,
     ZNP,

zigpy_znp/commands/app.py

@@ -1,7 +1,7 @@
 import zigpy_znp.types as t
 
 
-class App(t.CommandsBase, subsystem=t.Subsystem.APP):
+class APP(t.CommandsBase, subsystem=t.Subsystem.APP):
     # This command is sent to the target in order to test the functions defined
     # for individual applications.
     # This command sends a raw data to an application

zigpy_znp/commands/util.py

@@ -69,7 +69,7 @@ class LEDMode(t.enum_uint8):
     TOGGLE = 4
 
 
-class Util(t.CommandsBase, subsystem=t.Subsystem.UTIL):
+class UTIL(t.CommandsBase, subsystem=t.Subsystem.UTIL):
     # MAC Reset command to reset MAC state machine
     GetDeviceInfo = t.CommandDef(
         t.CommandType.SREQ,