test: count AOL authorization before replay TX checks

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__":

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