From d34a7569292dd7c2495ce8b88d7aa313a3d9f340 Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Mon, 21 Sep 2026 15:16:17 -0500 Subject: [PATCH] test: count AOL authorization before replay TX checks --- .../tests/safety_replay/replay_drive.py | 15 ++++- .../tests/safety_replay/test_replay_drive.py | 63 +++++++++++++++++++ 2 files changed, 75 insertions(+), 3 deletions(-) create mode 100644 opendbc_repo/opendbc/safety/tests/safety_replay/test_replay_drive.py diff --git a/opendbc_repo/opendbc/safety/tests/safety_replay/replay_drive.py b/opendbc_repo/opendbc/safety/tests/safety_replay/replay_drive.py index c829cbc216..d829be897d 100755 --- a/opendbc_repo/opendbc/safety/tests/safety_replay/replay_drive.py +++ b/opendbc_repo/opendbc/safety/tests/safety_replay/replay_drive.py @@ -20,6 +20,7 @@ def replay_drive(msgs, safety_mode, param, alternative_experience): init_segment(safety, msgs, safety_mode, param) rx_tot, rx_invalid, tx_tot, tx_blocked, tx_controls, tx_controls_blocked = 0, 0, 0, 0, 0, 0 + tx_lateral, tx_lateral_blocked = 0, 0 safety_tick_rx_invalid = False blocked_addrs = Counter() invalid_addrs = set() @@ -38,14 +39,20 @@ def replay_drive(msgs, safety_mode, param, alternative_experience): if msg.which() == 'sendcan': for canmsg in msg.sendcan: _msg = package_can_msg(canmsg) + # TX hooks can revoke permission on a violation. Count the permission + # before checking the message, including lateral-only AOL operation. + controls_allowed = safety.get_controls_allowed() + lateral_allowed = controls_allowed or safety.get_aol_allowed() sent = safety.safety_tx_hook(_msg) if not sent: tx_blocked += 1 - tx_controls_blocked += safety.get_controls_allowed() + tx_controls_blocked += controls_allowed + tx_lateral_blocked += lateral_allowed blocked_addrs[canmsg.address] += 1 carlog.debug("blocked bus %d msg %d at %f" % (canmsg.src, canmsg.address, (msg.logMonoTime - start_t) / 1e9)) - tx_controls += safety.get_controls_allowed() + tx_controls += controls_allowed + tx_lateral += lateral_allowed tx_tot += 1 elif msg.which() == 'can': # ignore msgs we sent @@ -68,9 +75,11 @@ def replay_drive(msgs, safety_mode, param, alternative_experience): print("total msgs with controls allowed:", tx_controls) print("blocked msgs:", tx_blocked) print("blocked with controls allowed:", tx_controls_blocked) + print("total msgs with lateral allowed:", tx_lateral) + print("blocked with lateral allowed:", tx_lateral_blocked) print("blocked addrs:", blocked_addrs) - return tx_controls_blocked == 0 and rx_invalid == 0 and not safety_tick_rx_invalid + return tx_lateral_blocked == 0 and rx_invalid == 0 and not safety_tick_rx_invalid if __name__ == "__main__": diff --git a/opendbc_repo/opendbc/safety/tests/safety_replay/test_replay_drive.py b/opendbc_repo/opendbc/safety/tests/safety_replay/test_replay_drive.py new file mode 100644 index 0000000000..22ef050179 --- /dev/null +++ b/opendbc_repo/opendbc/safety/tests/safety_replay/test_replay_drive.py @@ -0,0 +1,63 @@ +import importlib.util +import sys +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + + +@pytest.fixture +def replay_module(monkeypatch): + # Load the sibling source explicitly, without native safety libraries or a + # host-runtime snapshot. These tests isolate replay accounting, not CAN rules. + monkeypatch.setitem(sys.modules, "opendbc.car.carlog", SimpleNamespace(carlog=Mock())) + monkeypatch.setitem(sys.modules, "opendbc.safety.tests.libsafety", SimpleNamespace(libsafety_py=SimpleNamespace())) + monkeypatch.setitem(sys.modules, "opendbc.safety.tests.safety_replay.helpers", + SimpleNamespace(package_can_msg=lambda msg: msg, init_segment=Mock())) + spec = importlib.util.spec_from_file_location("replay_drive_accounting", Path(__file__).with_name("replay_drive.py")) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + module.tqdm = lambda msgs: msgs + return module + + +@pytest.mark.parametrize("controls,aol,accepted,post_controls,post_aol", [ + (False, True, False, False, True), # AOL-only denial must fail replay. + (False, True, False, False, False), # A TX hook can revoke AOL permission. + (True, False, False, False, False), # A TX hook can revoke controls permission. + (False, False, False, False, False), # Expected inactive blocks remain allowed. + (False, False, False, True, True), # Post-hook permission must not misclassify a block. + (False, True, True, False, True), + (True, False, True, True, False), + (True, True, True, True, True), # Count overlapping permissions only once. +]) +def test_tx_authorization_accounted_before_hook(replay_module, capsys, controls, aol, accepted, post_controls, post_aol): + state = SimpleNamespace(controls=controls, aol=aol) + + def tx_hook(msg): + state.controls = post_controls + state.aol = post_aol + return accepted + + safety = Mock() + safety.set_safety_hooks.return_value = 0 + safety.get_controls_allowed.side_effect = lambda: state.controls + safety.get_aol_allowed.side_effect = lambda: state.aol + safety.safety_tx_hook.side_effect = tx_hook + replay_module.libsafety_py.libsafety = safety + packet = SimpleNamespace(address=0x488, src=0, dat=b"\x00" * 4) + msg = SimpleNamespace(logMonoTime=0, sendcan=[packet], which=lambda: "sendcan") + + result = replay_module.replay_drive([msg], 10, 0, 0) + + lateral_allowed = controls or aol + assert result == (accepted or not lateral_allowed) + safety.safety_tx_hook.assert_called_once_with(packet) + output = capsys.readouterr().out + assert "total openpilot msgs: 1\n" in output + assert f"total msgs with controls allowed: {int(controls)}\n" in output + assert f"blocked msgs: {int(not accepted)}\n" in output + assert f"blocked with controls allowed: {int(controls and not accepted)}\n" in output + assert f"total msgs with lateral allowed: {int(lateral_allowed)}\n" in output + assert f"blocked with lateral allowed: {int(lateral_allowed and not accepted)}\n" in output