feat: validate source address on ControllerApplication construction

j1939/controller_application.py

@@ -38,6 +38,22 @@ class FieldValue:
         MAX_16 = 0xFAFF
         MAX_16_ARR = [0xFA, 0xFF]
 
+    @staticmethod
+    def _check_source_address(address):
+        """Validate that ``address`` is a claimable J1939 source address.
+
+        :param int address:
+            A claimable source address in the range 0..253. NULL (254) and
+            GLOBAL (255) are not claimable addresses.
+        :raises ValueError:
+            If ``address`` is outside the 0..253 range.
+        """
+        if address < 0 or address > 253:
+            raise ValueError(
+                f"Invalid source address '{address}': must be in the range 0..253 "
+                "(NULL (254) and GLOBAL (255) are not claimable addresses)"
+            )
+
     def __init__(self, name, device_address_preferred=None, bypass_address_claim=False):
         """
         :param name:
@@ -46,7 +62,12 @@ def __init__(self, name, device_address_preferred=None, bypass_address_claim=Fal
             The device_address this CA should claim on the bus.
         :param bypass_address_claim:
             Flag to bypass address claim procedure
+        :raises ValueError:
+            If ``device_address_preferred`` is not ``None`` and is outside the
+            claimable range 0..253.
         """
+        if device_address_preferred is not None:
+            ControllerApplication._check_source_address(device_address_preferred)
         self._name = name
         self._device_address_preferred = device_address_preferred
         if bypass_address_claim and (device_address_preferred is not None):
@@ -358,7 +379,9 @@ def _begin_address_claim(self, new_address):
         with self._lifecycle_lock:
             # Only 0..253 are valid (claimable) source addresses. NULL (254)
             # and GLOBAL (255) must never be claimed.
-            if new_address < 0 or new_address > 253:
+            try:
+                ControllerApplication._check_source_address(new_address)
+            except ValueError:
                 logger.warning("Ignoring address claim for invalid source address '%d'", new_address)
                 return False
             if self._ecu is None:

test/test_ca.py

@@ -254,6 +254,56 @@ def test_change_address_rejects_null_and_global_without_mutation(feeder):
     assert feeder.can_messages == []
 
 
+def _address_test_name():
+    return j1939.Name(
+        arbitrary_address_capable=0,
+        industry_group=j1939.Name.IndustryGroup.Global,
+        vehicle_system_instance=0,
+        vehicle_system=0,
+        function=0,
+        function_instance=0,
+        ecu_instance=0,
+        manufacturer_code=0,
+        identity_number=0,
+    )
+
+
+def test_constructor_accepts_min_valid_address():
+    """Address 0 is the lowest claimable source address."""
+    ca = j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=0)
+    assert ca._device_address_preferred == 0
+
+
+def test_constructor_accepts_max_valid_address():
+    """Address 253 is the highest claimable source address."""
+    ca = j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=253)
+    assert ca._device_address_preferred == 253
+
+
+def test_constructor_accepts_none_address():
+    """device_address_preferred=None (no preferred address yet) is allowed."""
+    ca = j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=None)
+    assert ca._device_address_preferred is None
+
+
+def test_constructor_rejects_null_address():
+    """NULL (254) is not a claimable source address and must raise."""
+    with pytest.raises(ValueError):
+        j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=254)
+
+
+def test_constructor_rejects_global_address():
+    """GLOBAL (255) is not a claimable source address and must raise."""
+    with pytest.raises(ValueError):
+        j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=255)
+
+
+def test_constructor_rejects_negative_address():
+    """Negative addresses are below the valid range and must raise."""
+    with pytest.raises(ValueError):
+        j1939.ControllerApplication(name=_address_test_name(), device_address_preferred=-1)
+
+
 def test_change_address_rearms_veto_for_started_ca(feeder):
     """A started CA changing into the veto range waits for the veto timeout."""
     name = _commanded_address_name()