diff --git a/j1939/controller_application.py b/j1939/controller_application.py index f31a67c..798a04d 100644 --- a/j1939/controller_application.py +++ b/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: diff --git a/test/test_ca.py b/test/test_ca.py index 3f1cbc5..c45e223 100644 --- a/test/test_ca.py +++ b/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()