Add validation for module addresses and pgm counts; update test function names

Author: yozik04

paradox/paradox.py

@@ -22,6 +22,7 @@
     async_loop_unhandled_exception_handler,
 )
 from paradox.hardware import Panel, create_panel
+from paradox.hardware.evo.parsers import MODULE_PGM_PACKET_SLOTS
 from paradox.lib import ps
 from paradox.lib.async_message_manager import ErrorMessageHandler, EventMessageHandler
 from paradox.lib.handlers import PersistentHandler
@@ -492,6 +493,22 @@ async def control_partition(self, partition: str, command: str) -> bool:
 
     def _init_module_pgms(self):
         for addr, pgm_count in cfg.MODULE_PGM_ADDRESSES.items():
+            if not isinstance(addr, int) or not (1 <= addr <= 254):
+                logger.warning(
+                    "MODULE_PGM_ADDRESSES: invalid module address %r (expected int 1-254), skipping",
+                    addr,
+                )
+                continue
+            if not isinstance(pgm_count, int) or not (
+                1 <= pgm_count <= MODULE_PGM_PACKET_SLOTS
+            ):
+                logger.warning(
+                    "MODULE_PGM_ADDRESSES: invalid pgm_count %r for address %d (expected int 1-%d), skipping",
+                    pgm_count,
+                    addr,
+                    MODULE_PGM_PACKET_SLOTS,
+                )
+                continue
             for pgm_index in range(1, pgm_count + 1):
                 key = f"module{addr}_pgm{pgm_index}"
                 self.storage.get_container("module_pgm")[key] = {
@@ -533,15 +550,19 @@ async def control_output(self, output, command) -> bool:
                         out["module_address"], out["pgm_index"], command
                     )
                 except NotImplementedError:
-                    logger.error("control_module_pgm_outputs is not implemented for this alarm type")
+                    logger.error(
+                        "control_module_pgm_outputs is not implemented for this alarm type"
+                    )
                 except asyncio.CancelledError:
-                    logger.error("control_output canceled")
+                    logger.error("control_module_pgm_output canceled")
                     raise
                 except asyncio.TimeoutError:
                     logger.error("control_output timeout")
                 if accepted:
                     is_on = command in ("on", "on_override")
-                    self.storage.update_container_object("module_pgm", out["key"], {"on": is_on})
+                    self.storage.update_container_object(
+                        "module_pgm", out["key"], {"on": is_on}
+                    )
             return accepted
 
         logger.error("No outputs selected")

tests/hardware/evo/test_pgm.py

@@ -1,9 +1,7 @@
-from binascii import hexlify, unhexlify
+from binascii import unhexlify
 
 from paradox.hardware.evo.adapters import PGMFlags
-from paradox.hardware.evo.parsers import (BroadcastRequest, BroadcastResponse,
-                                          PerformPGMAction,
-                                          PGMBroadcastCommand)
+from paradox.hardware.evo.parsers import PerformActionResponse, PerformPGMAction
 
 
 def test_pgm3_activate_and_monitor():
@@ -50,7 +48,7 @@ def test_pgm3_deactivate_and_monitor():
     assert a == expected_out
 
 
-def test_module_pgm_activate_and_monitor():
+def test_pgm4_activate_and_monitor():
     expected_out = unhexlify("40130600000008000000000000000300000064")
 
     pgms = [4]
@@ -60,7 +58,7 @@ def test_module_pgm_activate_and_monitor():
     assert a == expected_out
 
 
-def test_module_pgm_deactivate_and_monitor():
+def test_pgm4_deactivate_and_monitor():
     expected_out = unhexlify("40130600000008000000000000000100000062")
 
     pgms = [4]
@@ -119,4 +117,10 @@ def test_pgm_flags_3():
 
 
 def test_pgm_confirmation():
-    payload = unhexlify("42070000000049")
+    raw = unhexlify("42070000000049")
+    parsed = PerformActionResponse.parse(raw)
+
+    assert parsed.fields.value.po.command == 0x4
+    assert parsed.fields.value.po.status.Winload_connected is True
+    assert parsed.fields.value.po.status.alarm_reporting_pending is False
+    assert parsed.fields.value.packet_length == 7