Keep native test fixtures aligned with current lifecycle contracts

opendbc_repo/opendbc/car/gm/tests/test_bolt_pedal.py

@@ -790,6 +790,14 @@ class TestBoltPedalStartupParser(unittest.TestCase):
               self.assertEqual(decoder.vl['ASCMLKASteeringCmd']['LKASteeringCmd'], 0)
               self.assertTrue(native.safety_tx_hook(libsafety_py.make_CANPacket(address, bus, raw)))
 
+  def test_gen2_drive_does_not_inherit_previous_alternative_experience(self):
+    from opendbc.safety.tests.libsafety import libsafety_py
+    for run in (self.test_gen2_pedal_drive_actual_parser_controller_and_native,
+                self.test_gen2_pedal_drive_withdraws_on_source_or_driver_change):
+      with self.subTest(contract=run.__name__):
+        libsafety_py.libsafety.set_alternative_experience(32)
+        run()
+
   def test_gen2_pedal_drive_actual_parser_controller_and_native(self):
     from opendbc.car.gm.bolt_cc import button_bytes
     from opendbc.safety.tests.libsafety import libsafety_py
@@ -806,6 +814,7 @@ class TestBoltPedalStartupParser(unittest.TestCase):
       with self.subTest(identity=identity, gear=gear, mode=mode):
         cp = params(identity, pedal=True, camera=True)
         native = libsafety_py.libsafety
+        native.set_alternative_experience(cp.alternativeExperience)
         self.assertEqual(native.set_safety_hooks(cp.safetyConfigs[0].safetyModel.raw, cp.safetyConfigs[0].safetyParam), 0)
         native.init_tests()
         ci, _ = self.stream(cp, native=native)
@@ -851,6 +860,7 @@ class TestBoltPedalStartupParser(unittest.TestCase):
       with self.subTest(fault=fault):
         cp = params(CAR.CHEVROLET_BOLT_CC_2022_2023, pedal=True, camera=True)
         native = libsafety_py.libsafety
+        native.set_alternative_experience(cp.alternativeExperience)
         self.assertEqual(native.set_safety_hooks(cp.safetyConfigs[0].safetyModel.raw, cp.safetyConfigs[0].safetyParam), 0)
         native.init_tests()
         ci, _ = self.stream(cp, native=native)

openpilot/selfdrive/car/tests/test_distance_personality_claim.py

@@ -205,6 +205,8 @@ class TestDistancePersonalityClaim(unittest.TestCase):
     drive.big_model_ready_t = 0.0
     drive.jetlink_swaps = None
     drive.enabled = False
+    drive.active = False
+    drive.aol_replay = False
     drive.nostalgia_enabled = False
     drive.CS_prev = car.CarState.new_message()
     drive.dm_lockout_set = False

openpilot/selfdrive/car/tests/test_init_lifecycle.py

@@ -13,6 +13,34 @@ from openpilot.starpilot.vehicle_startup import VehicleStartupOwner
 
 
 class CardInitLifecycleTest(unittest.TestCase):
+  def assert_inactive_gm_stream(self, sent, *, alpha, brake_bus, inactive_regen):
+    expected = {(0x180, 0)}
+    if alpha:
+      expected |= {(0x184, 2), (0x2CB, 0), (0x315, brake_bus), (0x370, 0)}
+    self.assertEqual({(address, bus) for address, _, bus in sent}, expected)
+    for address, data, _bus in sent:
+      if address == 0x180:
+        self.assertEqual(data[0] & 15, 0)
+        self.assertEqual(data[1], 0)
+      elif address == 0x184:
+        self.assertEqual(data, bytes.fromhex('0000200000200000'))
+      elif address == 0x2CB:
+        self.assertFalse(data[0] & 1)  # Gas/regen command inactive.
+        self.assertFalse(data[1] & 0x20)  # No full-stop request.
+        # DBC: 19-bit Motorola command, scale .125 and offset -22534.
+        raw = (inactive_regen + 22534) * 8
+        self.assertEqual(int.from_bytes(data[1:4], 'big') & 0x7FFFF, raw)
+        counter = data[0] >> 6
+        command = bytes((counter << 6, 0x40 | (raw >> 16), (raw >> 8) & 0xFF, raw & 0xFF))
+        # Inactive checksum bit, complemented bytes 1/2, then byte 3 + counter complement.
+        checksum = bytes((1, 0xFF - command[1], 0xFF - command[2], (-command[3] - counter) & 0xFF))
+        self.assertEqual(data, command + checksum)
+      elif address == 0x315:
+        self.assertEqual(data[0] >> 4, 1)  # Released friction-brake mode.
+        self.assertEqual(int.from_bytes(data[:2], 'big') & 0xFFF, 0)
+      elif address == 0x370:
+        self.assertEqual(data, bytes.fromhex('010200000100'))  # Inactive ACC, no speed or warning.
+
   def defer_radar_fixture(self, card):
     eager = card.state_update
     batch = []
@@ -213,10 +241,7 @@ class CardInitLifecycleTest(unittest.TestCase):
         self.assertFalse(cc.enabled or cc.latActive or cc.longActive)
         self.assertEqual((cc.actuators.torque, cc.actuators.accel), (0., 0.))
       self.assertTrue(sent)
-      for address, data, bus in sent:
-        self.assertEqual((address, bus), (0x180, 0))
-        self.assertEqual(data[0] & 15, 0)
-        self.assertEqual(data[1], 0)
+      self.assert_inactive_gm_stream(sent, alpha=True, brake_bus=2, inactive_regen=-500)
       panda.safetyParam = 0x5087
       card.step()
       self.assertEqual(apply.call_count, 30)
@@ -255,7 +280,9 @@ class CardInitLifecycleTest(unittest.TestCase):
       cp.safetyConfigs[0].safetyParam = word
     self.assertEqual(cp.safetyConfigs[0].safetyParam, word)
     cadence = ((0x180, 110_000_000),)
-    ready_cadence = cadence + ((0x2CB, 50_000_000), (0x315, 50_000_000)) if alpha else cadence
+    if alpha:
+      cadence += ((0x2CB, 50_000_000), (0x315, 50_000_000), (0x370, 50_000_000))
+    ready_cadence = cadence
     cp.alternativeExperience = 32
     card = Car.__new__(Car)
     card.CP = cp
@@ -367,10 +394,7 @@ class CardInitLifecycleTest(unittest.TestCase):
         self.assertFalse(any(address in (0x2CB, 0x315) for address, _, _ in sent))
         self.assertTrue(card.CP.pcmCruise)
         self.assertFalse(card.CP.openpilotLongitudinalControl)
-      for address, data, bus in sent:
-        self.assertEqual((address, bus), (0x180, 0))
-        self.assertEqual(data[0] & 15, 0)
-        self.assertEqual(data[1], 0)
+      self.assert_inactive_gm_stream(sent, alpha=alpha, brake_bus=0, inactive_regen=-650)
       panda.safetyParam = word ^ 1
       card.step()
       self.assertEqual(apply.call_count, 601)

openpilot/starpilot/aol/tests/test_late_lateral_inputs.py

@@ -40,6 +40,10 @@ BASE = 10_000_000_000
     'disabled_long_candidate',
     'clock_crosses_deadline',
     'old_long_axis',
+    'old_long_axis_empty',
+    'old_long_axis_stale',
+    'old_long_active_only',
+    'old_long_desired_only',
     'stock_cruise_off',
     'stock_cruise_off_denied',
     'new_host_long',
@@ -114,10 +118,10 @@ def test_queued_fresh_sdgm_axis_is_consumed_before_steering_gate(c9, case):
       return [r for r in captured if r.which() != 'carState'] + [state.as_reader(), axis.as_reader(), wire.as_reader()]
 
     old = batch(BASE)
-    if case == 'old_long_axis':
+    if case.startswith('old_long_'):
       changed = old[-2].as_builder()
-      changed.aolAxisState.longitudinalActive = True
-      changed.aolAxisState.desiredLongitudinal = True
+      changed.aolAxisState.longitudinalActive = case != 'old_long_desired_only'
+      changed.aolAxisState.desiredLongitudinal = case != 'old_long_active_only'
       old[-2] = changed.as_reader()
     controls.sm.update_msgs(BASE / 1e9, old)
     with patch('openpilot.selfdrive.controls.controlsd.time.monotonic_ns', return_value=BASE):
@@ -169,7 +173,7 @@ def test_queued_fresh_sdgm_axis_is_consumed_before_steering_gate(c9, case):
       values['aolSafetyWire'].valid = False
     elif case == 'future_axis':
       values['aolAxisState'].aolAxisState.observedMonoTime = now + 1
-    elif case == 'stale_axis':
+    elif case in ('stale_axis', 'old_long_axis_stale'):
       values['aolAxisState'].aolAxisState.validUntilMonoTime = now - 1
     elif case == 'native_denied':
       values['aolSafetyWire'].aolSafetyWire = encode_safety(
@@ -206,7 +210,7 @@ def test_queued_fresh_sdgm_axis_is_consumed_before_steering_gate(c9, case):
           1, True, BASE + 25_000_000, BASE + 225_000_000, 4, cp.safetyConfigs[0].safetyParam, True, True, True, True, 'native-test-panda', 'late-sdgm'
         )
       )
-    pending = [] if case == 'empty' else [v.as_reader() for v in values.values()]
+    pending = [] if case in ('empty', 'old_long_axis_empty') else [v.as_reader() for v in values.values()]
 
     def receive(timeout):
       nonlocal now
@@ -221,12 +225,18 @@ def test_queued_fresh_sdgm_axis_is_consumed_before_steering_gate(c9, case):
       patch('openpilot.selfdrive.controls.controlsd.time.monotonic_ns', return_value=now) as decision_clock,
     ):
       command, _ = controls.state_control()
-    expected = case in ('healthy', 'live_snapshot', 'disabled_long_candidate', 'stock_sibling', 'stock_cruise_off', 'new_host_long')
+    # GM may refresh an expired combined proof, but cannot start LONG on a LAT-only receipt.
+    expected = case in ('healthy', 'live_snapshot', 'disabled_long_candidate', 'stock_sibling', 'stock_cruise_off', 'new_host_long', 'old_long_axis')
     assert command.latActive == expected and not command.longActive, (case, command.to_dict())
-    if case in ('live_snapshot', 'old_long_axis'):
+    if case in ('live_snapshot', 'old_long_active_only', 'old_long_desired_only'):
       receive_latest.assert_not_called()
     else:
       receive_latest.assert_called_once_with(0)
+    if case == 'old_long_axis':
+      refreshed = controls.sm['aolAxisState']
+      assert refreshed.sourceCarStateMonoTime == BASE + 25_000_000
+      assert refreshed.validUntilMonoTime == BASE + 55_000_000
+      assert not refreshed.longitudinalActive and not refreshed.desiredLongitudinal
     safety.set_timer(now // 1000)
     safety.set_aol_test_heartbeat(True)
     if case != 'new_host_long':

openpilot/starpilot/galaxy/tests/test_models.py

@@ -10,6 +10,7 @@ import unittest
 from openpilot.starpilot.galaxy.access import GalaxyAccessOwner
 from openpilot.starpilot.galaxy.server import make_server
 from openpilot.starpilot.models.manager import ModelManager, preferences
+from openpilot.starpilot.models.jetlink_settings import JetlinkSettings
 
 
 PAYLOAD = {"schemaVersion": 1, "catalog": [{"id": "bundled-current", "name": "Bundled driving model",
@@ -91,11 +92,14 @@ class LocalModelManagerHttpTest(unittest.TestCase):
     with tempfile.TemporaryDirectory() as tmp:
       root = Path(tmp)
       parked = [True]
-      class LinkSettings:
+      class LinkSettings(JetlinkSettings):
         mode = "off"
 
         def snapshot(self):
-          return {"mode": self.mode}
+          return {"mode": self.mode, "supported": True, "chestnut": False}
+
+        def pick(self):
+          return None
 
         def configure(self, payload):
           self.mode = payload["mode"]
@@ -132,6 +136,10 @@ class LocalModelManagerHttpTest(unittest.TestCase):
         self.assertFalse(snapshot['randomizer'])
         self.assertEqual(request('/api/models/jetlink', {'mode': 'usb'})[0], 200)
         self.assertEqual(link.mode, 'usb')
+        status, snapshot = request('/api/models/manager')
+        self.assertEqual(status, 200)
+        self.assertTrue(snapshot['jetlinkBig'])
+        self.assertEqual(snapshot['activeBigModel'], '')
         status, laboratory = request('/api/models/laboratory')
         self.assertEqual(status, 200)
         self.assertFalse(laboratory['runtimeSupported'])

openpilot/starpilot/lateral/tests/test_lane_host.py

@@ -11,6 +11,7 @@ import struct
 import sys
 import tempfile
 import unittest
+from unittest.mock import patch
 
 from openpilot.common.basedir import BASEDIR
 
@@ -21,10 +22,14 @@ def probe(output, reference_method=None):
   ):
     raise RuntimeError("Native lane-host probe requires isolated IPC without simulation flags")
   from openpilot.cereal import messaging
+  from openpilot.common.constants import ACCELERATION_DUE_TO_GRAVITY
   from openpilot.common.realtime import DT_CTRL
   from openpilot.common.params import Params
   from openpilot.selfdrive.controls.controlsd import Controls
-  from openpilot.selfdrive.controls.lib.drive_helpers import clip_curvature
+  from openpilot.selfdrive.controls.lib.drive_helpers import (MAX_CURVATURE, MAX_LATERAL_ACCEL_NO_ROLL,
+                                                            MAX_LATERAL_JERK, MIN_SPEED, clip_curvature)
+  from openpilot.starpilot.lateral.lane_clearance import current_delay, limit_tracking
+  from openpilot.starpilot.lateral.lane_tracking import SingleLineTracker
   from openpilot.selfdrive.controls.lib.latcontrol_torque import LatControlTorque
   from openpilot.selfdrive.controls.lib.longcontrol import LongControl
   from openpilot.starpilot.longitudinal.ioniq6_start import eligible as ioniq6_start_eligible
@@ -83,6 +88,7 @@ def probe(output, reference_method=None):
 
   for change in changes:
     candidate, reference = host(), host()
+    tracking_reference = SingleLineTracker()
     previous_correction = 0.0
     # Each gate starts with positive acquired state; absence/disabled equivalence
     # starts disabled from the beginning to compare all controller history exactly.
@@ -109,7 +115,8 @@ def probe(output, reference_method=None):
       lp = message("vehicleParameters")
       lp.stiffnessFactor, lp.steerRatio = 1.0, candidate.CP.steerRatio
       message("longitudinalPlan").aTarget = 0.5
-      message("lateralDelay").lateralDelay = 0.1
+      if tick % 25 == 0:  # lateralDelay is a 4 Hz service, not a control-rate packet.
+        message("lateralDelay").lateralDelay = 0.1
       message("onroadEvents", 0)
       if tick % 5 == 0 and phase != "stale_model":
         event = messaging.new_message("modelV2", valid=phase != "invalid_model", logMonoTime=timestamp)
@@ -138,23 +145,62 @@ def probe(output, reference_method=None):
       elif phase == "invalid_request":
         configured = "invalid"
       prior_curvature = candidate.desired_curvature
-      command, lateral = candidate.state_control(lane_centering=configured)
-      base_command, _ = reference.state_control() if reference_method is None else reference_method(reference)
+      # Supplied packet times and the native freshness clock share one epoch.
+      with patch("openpilot.selfdrive.controls.controlsd.time.monotonic_ns", return_value=timestamp):
+        command, lateral = candidate.state_control(lane_centering=configured)
+        base_command, _ = reference.state_control() if reference_method is None else reference_method(reference)
       after = input_bytes(candidate)
       if before != after:
         raise AssertionError("Host changed its supplied messages")
       result = candidate.last_lane_centering_result
+      delay = current_delay(candidate.sm, timestamp)
+      if tick >= 25 and delay != candidate.sm["lateralDelay"].lateralDelay:
+        raise AssertionError("Fresh native delay failed qualification")
+      # Track the supplied geometry independently, including source-gap expiry
+      # and signal release, without reading or mutating the host's tracker.
+      if (isinstance(configured, LaneCenteringRequest) and configured.settings.enabled and configured.mode.lateral_authority and
+          command.latActive and candidate.sm.all_checks(["modelV2"]) and not cs.steeringPressed and
+          not configured.time_discontinuity and not cs.leftBlinker):
+        tracking_reference.target(candidate.sm["modelV2"], candidate.sm.logMonoTime["modelV2"], cs.vEgo, DT_CTRL, 0.0)
+      else:
+        tracking_reference.reset()
+      # These fixtures never request lane-change shaping: the native clip is
+      # followed only by the two-line tracking guard, which may shrink its delta.
+      if candidate.lane_change_smoother.previous_factor != 1.0:
+        raise AssertionError("Unexpected lane-change shaping in the fixture")
+      speed_squared = max(cs.vEgo, MIN_SPEED) ** 2
+      curvature = candidate.desired_curvature
+      if (not math.isfinite(curvature) or abs(curvature) > MAX_CURVATURE + 1e-12 or
+          abs(curvature - prior_curvature) > MAX_LATERAL_JERK / speed_squared * DT_CTRL + 1e-12 or
+          abs(curvature * speed_squared - lp.roll * ACCELERATION_DUE_TO_GRAVITY) > MAX_LATERAL_ACCEL_NO_ROLL + 1e-12):
+        raise AssertionError("Final curvature exceeded independent native bounds")
       if result is not None and result.candidate_curvature is not None:
-        expected, _ = clip_curvature(cs.vEgo, prior_curvature, result.candidate_curvature, lp.roll)
+        clipped, _ = clip_curvature(cs.vEgo, prior_curvature, result.candidate_curvature, lp.roll)
+        expected = clipped
+        if result.correction != 0.0:
+          current_model = candidate.sm["modelV2"]
+          baseline_input = (candidate.sm["lateralManeuverPlan"].desiredCurvature
+                            if candidate.sm.valid["lateralManeuverPlan"] else current_model.action.desiredCurvature)
+          if not command.latActive:
+            baseline_input = 0.0
+          if not math.isclose(result.candidate_curvature - result.correction, baseline_input, abs_tol=1e-12):
+            raise AssertionError("Lane correction changed its native baseline")
+          without_lane, _ = clip_curvature(cs.vEgo, prior_curvature, baseline_input, lp.roll)
+          # Apply the pure guard to independently tracked input geometry;
+          # invoking guard_applied again would mutate the acquired correction.
+          allowed = clipped - without_lane
+          if tracking_reference.path is not None:
+            allowed = (0.0 if delay is None else
+                       limit_tracking(tracking_reference.path, without_lane, allowed, 0.0, cs.vEgo, delay))
+          expected = without_lane + allowed
+          if not min(without_lane, clipped) - 1e-12 <= expected <= max(without_lane, clipped) + 1e-12:
+            raise AssertionError("Tracking guard increased or reversed the native contribution")
+          if not math.isclose(candidate.lane_centering_applied, curvature - without_lane, abs_tol=1e-12):
+            raise AssertionError("Lane feedback did not reflect the final command limiter")
         published_precision = struct.unpack("f", struct.pack("f", expected))[0]
-        if candidate.desired_curvature != expected or command.actuators.curvature != published_precision:
-          raise AssertionError("Lane candidate bypassed native curvature limiting")
-      if result is not None and result.correction != 0.0:
-        baseline_input = result.candidate_curvature - result.correction
-        without_lane, _ = clip_curvature(cs.vEgo, prior_curvature, baseline_input, lp.roll)
-        if not math.isclose(candidate.lane_centering_applied, candidate.desired_curvature - without_lane, abs_tol=1e-12):
-          raise AssertionError("Lane feedback did not reflect the final command limiter")
-      elif candidate.lane_centering_applied != 0.0:
+        if curvature != expected or command.actuators.curvature != published_precision:
+          raise AssertionError(f"Lane candidate bypassed native clipping or tracking guard: {change=} {tick=} {curvature=} {expected=}")
+      if (result is None or result.correction == 0.0) and candidate.lane_centering_applied != 0.0:
         raise AssertionError("Inactive lane contribution remained visible")
       records.append({"scenario": change, "phase": phase, "tick": tick,
                       "model_sample": candidate.sm.logMonoTime["modelV2"], "model_checks": candidate.sm.all_checks(["modelV2"]),
@@ -187,9 +233,10 @@ class TestLaneHost(unittest.TestCase):
         self.assertTrue(math.isfinite(command["actuators"]["torque"]))
         if row["scenario"] == "disabled_setting":
           self.assertEqual(command, reference)
-        elif phase == "warm" and row["tick"] >= 20:
+        elif phase == "warm" and row["tick"] >= 30:
+          # Two fresh delay packets qualify the guard; its correction may settle.
           self.assertGreater(result["correction"], 0.0)
-          self.assertGreater(result["correction"], row["previous_correction"])
+          self.assertGreater(command["actuators"]["curvature"], reference["actuators"]["curvature"])
         elif phase in ("off", "long_only", "fault", "disabled_host", "override", "invalid_model", "clock_reset", "invalid_request"):
           self.assertEqual(result["correction"], 0.0)
         elif phase == "absent_request":
@@ -206,7 +253,7 @@ class TestLaneHost(unittest.TestCase):
     stale = [row for row in records if row["phase"] == "stale_model"]
     self.assertTrue(any(row["model_checks"] for row in stale))
     self.assertTrue(any(not row["model_checks"] for row in stale))
-    held = [row for row in records if row["scenario"] == "lateral_only" and row["tick"] in (21, 22)]
+    held = [row for row in records if row["scenario"] == "lateral_only" and row["tick"] in (31, 32)]
     self.assertEqual(held[0]["model_sample"], held[1]["model_sample"])
     self.assertGreater(held[1]["result"]["correction"], held[0]["result"]["correction"])
 

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "7b00d3f8766a5fdcffd5e98ac639dc711e5cab6601c20817a88209a9248d80ce", "schema_version": 1}
+{"protected_inventory_sha256": "567658c0f7e93cec0b15bec55dcabded810f6cd2c1d3321009976fa147360f4a", "schema_version": 1}