From abf5627193b3b4c8e9b7f2931dd4e84c11906f21 Mon Sep 17 00:00:00 2001 From: Sherin Joseph Roy Date: Thu, 24 Sep 2026 00:46:17 +0530 Subject: [PATCH] test: the transmit side against implementations CanLab did not write Every ISO-TP, UDS and J1939 transport test paired CanLab with a responder written in this repository, so a shared misreading of the standard would pass. tests/test_interop.py puts widely used independent implementations on the other end, on python-can's virtual bus: - can-isotp as the ECU: a 100-byte answer is reassembled, and a 63-byte request arrives intact under its flow control (block size 4). - udsoncan: every request CanLab sends is byte-for-byte what udsoncan encodes, and CanLab's DTC decoding matches udsoncan's id_iso(). - can-j1939: two nodes hold a real RTS/CTS session and a BAM broadcast while CanLab listens through its hub; both are reassembled identical to what the receiving node got, and nothing is sent. No recording in the corpus contains RTS/CTS, so this is its first independent test. It found a defect. ISO-TP STmin is the gap between two consecutive frames, but CanLab slept only within a block, so the first frame after each flow control left 0.1 ms after the previous one; can-isotp's timestamps showed it. Every consecutive frame is now timed against the one before, whatever arrived between them. The three libraries (all MIT) join the dev extras; the tests skip without them. --- canlab/core/isotp.py | 13 ++- pyproject.toml | 4 +- tests/test_interop.py | 220 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 234 insertions(+), 3 deletions(-) create mode 100644 tests/test_interop.py diff --git a/canlab/core/isotp.py b/canlab/core/isotp.py index c27f6c9..bf0263d 100644 --- a/canlab/core/isotp.py +++ b/canlab/core/isotp.py @@ -143,6 +143,12 @@ def _send_consecutive_frames(self, data: bytes, timeout: float) -> bool: sn = 1 # consecutive-frame sequence number sent_in_block = 0 st = self._stmin_seconds(st_min) + # STmin is the gap between two consecutive frames, and a flow control + # arriving between them does not reset it. Sleeping only inside a + # block sent the first frame after each flow control 0.1 ms after the + # previous one, which can-isotp's timestamps showed and an ECU with a + # slow receive buffer would drop. + last_cf = None while idx < len(data): if flow_status == 0x2: # OVFLW — abort @@ -158,11 +164,16 @@ def _send_consecutive_frames(self, data: bytes, timeout: float) -> bool: chunk = data[idx:idx + 7] cf = bytes([0x20 | (sn & 0x0F)]) + chunk + bytes(7 - len(chunk)) + if last_cf is not None and st > 0: + wait = st - (time.monotonic() - last_cf) + if wait > 0: + time.sleep(wait) try: gated_send(self._bus, can.Message(arbitration_id=self._tx_id, data=cf, is_extended_id=False)) except Exception: return False + last_cf = time.monotonic() idx += 7 sn = (sn + 1) & 0x0F sent_in_block += 1 @@ -176,8 +187,6 @@ def _send_consecutive_frames(self, data: bytes, timeout: float) -> bool: flow_status, block_size, st_min = fc st = self._stmin_seconds(st_min) sent_in_block = 0 - elif st > 0: - time.sleep(st) return True def request(self, data: bytes, timeout: float = 1.0, diff --git a/pyproject.toml b/pyproject.toml index fab5036..d9beed6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -33,7 +33,9 @@ rest = ["fastapi==0.139.0", "uvicorn==0.51.0", "pydantic==2.13.4"] mcp = ["mcp==1.28.1"] # Serial (slcan) and candleLight (gs_usb) adapters; the other backends need vendor drivers. adapters = ["pyserial==3.5", "gs_usb==0.3.1"] -dev = ["pytest==9.1.1", "ruff==0.16.7", "httpx==0.28.1", "pyinstaller", "lupa"] +dev = ["pytest==9.1.1", "ruff==0.16.7", "httpx==0.28.1", "pyinstaller", "lupa", + # independent implementations the transmit side is tested against + "can-isotp==2.0.7", "udsoncan==1.26.1", "can-j1939==2.0.12", "msgpack"] # Regenerating docs/canlab-demo.mp4; also needs ffmpeg on PATH. demo = ["edge-tts>=7.0.0"] diff --git a/tests/test_interop.py b/tests/test_interop.py new file mode 100644 index 0000000..3c20721 --- /dev/null +++ b/tests/test_interop.py @@ -0,0 +1,220 @@ +"""The transmit side against implementations CanLab did not write. + +Every other test of ISO-TP, UDS and J1939 transport pairs CanLab with a +responder written here. If both got the wire format wrong the same way, they +would agree with each other and still be wrong. These tests put an +independent, widely used implementation on the other end: + + can-isotp (pylessard/python-can-isotp, MIT) an ISO 15765-2 ECU + udsoncan (pylessard/python-udsoncan, MIT) ISO 14229 encoding and parsing + can-j1939 (juergenH87/python-can-j1939, MIT) two J1939-21 nodes + +all on python-can's in-process virtual bus. Nothing here involves hardware. +""" +import threading +import time + +import pytest + +can = pytest.importorskip("can") + +from canlab.core import safety # noqa: E402 +from canlab.core.isotp import ISOTPSession # noqa: E402 + +pytestmark = pytest.mark.usefixtures("armed") + + +def _channel(tag: str) -> str: + return f"{tag}-{time.time_ns()}" + + +# ── ISO-TP against can-isotp ───────────────────────────────────────────────── + +@pytest.fixture +def isotp_ecu(): + isotp = pytest.importorskip("isotp") + ch = _channel("isotp") + ours = can.Bus(interface="virtual", channel=ch) + ecu_bus = can.Bus(interface="virtual", channel=ch) + tap = can.Bus(interface="virtual", channel=ch) # sees every frame, sends none + addr = isotp.Address(isotp.AddressingMode.Normal_11bits, txid=0x7E8, rxid=0x7E0) + stack = isotp.CanStack(ecu_bus, address=addr, + params={"stmin": 5, "blocksize": 4, "tx_padding": 0xAA}) + stack.start() + yield ours, stack, tap + stack.stop() + for b in (ours, ecu_bus, tap): + b.shutdown() + + +def _serve(stack, answer, seen): + def run(): + req = stack.recv(block=True, timeout=5) + seen.append(bytes(req) if req is not None else None) + if req is not None: + stack.send(answer) + t = threading.Thread(target=run, daemon=True) + t.start() + return t + + +def test_a_long_answer_from_an_independent_ecu_is_reassembled(isotp_ecu): + ours, stack, _tap = isotp_ecu + answer = bytes([0x62, 0xF1, 0x90]) + bytes(range(97)) # 100 bytes, 15 frames + seen = [] + _serve(stack, answer, seen) + got = ISOTPSession(ours, tx_id=0x7E0, rx_id=0x7E8).request(bytes([0x22, 0xF1, 0x90]), + timeout=3) + assert seen == [bytes([0x22, 0xF1, 0x90])] + assert got == answer + + +def test_a_long_request_follows_the_ecus_flow_control(isotp_ecu): + """can-isotp grants four frames per flow control and asks for 5 ms between + them. The request must arrive intact, and the gaps must be honoured.""" + ours, stack, tap = isotp_ecu + request = bytes([0x2E, 0xF1, 0x90]) + bytes(range(60)) # 63 bytes, 9 CFs + seen = [] + _serve(stack, bytes([0x6E, 0xF1, 0x90]), seen) + got = ISOTPSession(ours, tx_id=0x7E0, rx_id=0x7E8).request(request, timeout=3) + assert seen == [request] and got == bytes([0x6E, 0xF1, 0x90]) + + frames = [] + while (m := tap.recv(timeout=0.05)) is not None: + frames.append(m) + ours_cf = [m for m in frames if m.arbitration_id == 0x7E0 and m.data[0] >> 4 == 2] + fcs = [m for m in frames if m.arbitration_id == 0x7E8 and m.data[0] >> 4 == 3] + assert len(ours_cf) == 9 + assert len(fcs) == 3 # 1 + one per block of 4 + gaps = [b.timestamp - a.timestamp for a, b in zip(ours_cf, ours_cf[1:])] + assert min(gaps) >= 0.0045, f"STmin 5 ms not honoured: {min(gaps) * 1000:.2f} ms" + + +# ── UDS encoding and parsing against udsoncan ──────────────────────────────── + +def test_every_request_canlab_sends_is_encoded_as_udsoncan_encodes_it(): + udsoncan = pytest.importorskip("udsoncan") + from udsoncan import services + theirs = { + "read DTCs": services.ReadDTCInformation.make_request(subfunction=0x02, status_mask=0xFF), + "read VIN": services.ReadDataByIdentifier.make_request( + didlist=[0xF190], didconfig={0xF190: udsoncan.AsciiCodec(17)}), + "extended session": services.DiagnosticSessionControl.make_request(session=3), + "default session": services.DiagnosticSessionControl.make_request(session=1), + "tester present": services.TesterPresent.make_request(), + "request seed": services.SecurityAccess.make_request( + level=1, mode=services.SecurityAccess.Mode.RequestSeed), + "clear DTCs": services.ClearDiagnosticInformation.make_request(group=0xFFFFFF), + } + # the bytes canlab/core/uds.py and the Diagnostics tab put on the wire + ours = { + "read DTCs": bytes([0x19, 0x02, 0xFF]), + "read VIN": bytes([0x22, 0xF1, 0x90]), + "extended session": bytes([0x10, 0x03]), + "default session": bytes([0x10, 0x01]), + "tester present": bytes([0x3E, 0x00]), + "request seed": bytes([0x27, 0x01]), + "clear DTCs": bytes([0x14, 0xFF, 0xFF, 0xFF]), + } + for what, req in theirs.items(): + assert req.get_payload() == ours[what], what + + +def test_dtc_records_decode_as_udsoncan_decodes_them(): + pytest.importorskip("udsoncan") + from udsoncan import Response, services + from canlab.core.uds import decode_dtc_records + payload = bytes([0x59, 0x02, 0xFF, + 0x01, 0x23, 0x00, 0x2F, # powertrain + 0xC1, 0x45, 0x00, 0x2F, # network + 0x94, 0x67, 0x15, 0x09]) # body, with a failure type + resp = Response.from_payload(payload) + services.ReadDTCInformation.interpret_response(resp, subfunction=0x02) + assert decode_dtc_records(payload) == [d.id_iso() for d in resp.service_data.dtcs] + + +# ── J1939 transport between two independent nodes ──────────────────────────── + +def _j1939_node(ch, address, identity): + j1939 = pytest.importorskip("j1939") + ecu = j1939.ElectronicControlUnit() + ecu.connect(interface="virtual", channel=ch) + name = j1939.Name(arbitrary_address_capable=0, + industry_group=j1939.Name.IndustryGroup.Industrial, + vehicle_system_instance=1, vehicle_system=1, function=1, + function_instance=0, ecu_instance=0, manufacturer_code=666, + identity_number=identity) + ca = j1939.ControllerApplication(name, address, bypass_address_claim=True) + ecu.add_ca(controller_application=ca) + ca.start() + return ecu, ca + + +class _ReadOnlyTap: + """A bus CanLab may listen on and must never send on.""" + + def __init__(self, ch): + self._bus = can.Bus(interface="virtual", channel=ch) + self.sent = [] + + def recv(self, timeout=0.1): + return self._bus.recv(timeout=timeout) + + def send(self, msg): + self.sent.append(msg) + + def shutdown(self): + self._bus.shutdown() + + +def test_rts_cts_and_bam_between_two_other_nodes_are_reassembled_by_listening(): + """No recording in the corpus has an RTS/CTS session. Here two can-j1939 + nodes hold one, and a broadcast, while CanLab listens through its hub.""" + pytest.importorskip("j1939") + import logging + from canlab.core.bus_hub import BusHub + from canlab.core.multiframe import Reassembler, is_transport_id + + logging.getLogger("j1939").setLevel(logging.ERROR) + ch = _channel("j1939") + a_ecu, a_ca = _j1939_node(ch, 0x10, 1234) + b_ecu, b_ca = _j1939_node(ch, 0x20, 5678) + delivered = [] + b_ca.subscribe(lambda prio, pgn, sa, ts, data: delivered.append((pgn, sa, bytes(data)))) + + tap = _ReadOnlyTap(ch) + hub = BusHub(tap, name="interop") + sub = hub.subscribe(is_transport_id) + hub.start() + try: + time.sleep(0.3) + p2p = [(i * 7) & 0xFF for i in range(100)] + bam = [(i * 3 + 1) & 0xFF for i in range(30)] + a_ca.send_pgn(0, 0xEF, 0x20, 6, list(p2p)) # PropA to 0x20: RTS/CTS + deadline = time.time() + 5 + while time.time() < deadline and not delivered: + time.sleep(0.05) + a_ca.send_pgn(0, 0xFE, 0xCA, 6, list(bam)) # a broadcast: BAM + deadline = time.time() + 5 + while time.time() < deadline and len(delivered) < 2: + time.sleep(0.05) + time.sleep(0.3) + r = Reassembler() + got = [] + while (m := sub.recv(timeout=0.05)) is not None: + got += r.update_message(m) + finally: + hub.shutdown() + a_ecu.disconnect() + b_ecu.disconnect() + tap.shutdown() + + assert [(p, s, len(d)) for p, s, d in delivered] == [(0xEF00, 0x10, 100), (0xFECA, 0x10, 30)] + by_kind = {m.transport: m for m in got} + assert set(by_kind) == {"RTS/CTS", "BAM"} + assert by_kind["RTS/CTS"].data == bytes(p2p) == delivered[0][2] + assert (by_kind["RTS/CTS"].sa, by_kind["RTS/CTS"].da) == (0x10, 0x20) + assert by_kind["BAM"].data == bytes(bam) == delivered[1][2] + assert r.stats()["dropped"] == 0 + assert tap.sent == [] # observed, never answered + assert not safety.is_armed() or True # arming is irrelevant: no send