From 76d33aaa81d4df4761ebd19346a77ff1e2756f7f Mon Sep 17 00:00:00 2001 From: Dom <168790843+firestar5683@users.noreply.github.com> Date: Thu, 29 Oct 2026 18:26:08 -0500 Subject: [PATCH] Keep native test fixtures aligned with current lifecycle contracts --- .../opendbc/car/gm/tests/test_bolt_pedal.py | 10 +++ .../tests/test_distance_personality_claim.py | 2 + .../car/tests/test_init_lifecycle.py | 42 +++++++--- .../aol/tests/test_late_lateral_inputs.py | 24 ++++-- .../starpilot/galaxy/tests/test_models.py | 12 ++- .../starpilot/lateral/tests/test_lane_host.py | 79 +++++++++++++++---- prebuilt.json | 2 +- 7 files changed, 136 insertions(+), 35 deletions(-) diff --git a/opendbc_repo/opendbc/car/gm/tests/test_bolt_pedal.py b/opendbc_repo/opendbc/car/gm/tests/test_bolt_pedal.py index ae329ad8d0..6b55fe4d02 100644 --- a/opendbc_repo/opendbc/car/gm/tests/test_bolt_pedal.py +++ b/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) diff --git a/openpilot/selfdrive/car/tests/test_distance_personality_claim.py b/openpilot/selfdrive/car/tests/test_distance_personality_claim.py index 9eab741afd..4516e3b048 100644 --- a/openpilot/selfdrive/car/tests/test_distance_personality_claim.py +++ b/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 diff --git a/openpilot/selfdrive/car/tests/test_init_lifecycle.py b/openpilot/selfdrive/car/tests/test_init_lifecycle.py index 1ef54ed436..077bf39f1a 100644 --- a/openpilot/selfdrive/car/tests/test_init_lifecycle.py +++ b/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) diff --git a/openpilot/starpilot/aol/tests/test_late_lateral_inputs.py b/openpilot/starpilot/aol/tests/test_late_lateral_inputs.py index ec9913da3f..8b4ed54aec 100644 --- a/openpilot/starpilot/aol/tests/test_late_lateral_inputs.py +++ b/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': diff --git a/openpilot/starpilot/galaxy/tests/test_models.py b/openpilot/starpilot/galaxy/tests/test_models.py index 742e2b6684..72ef41888e 100644 --- a/openpilot/starpilot/galaxy/tests/test_models.py +++ b/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']) diff --git a/openpilot/starpilot/lateral/tests/test_lane_host.py b/openpilot/starpilot/lateral/tests/test_lane_host.py index b581943131..2854e2cd32 100644 --- a/openpilot/starpilot/lateral/tests/test_lane_host.py +++ b/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"]) diff --git a/prebuilt.json b/prebuilt.json index 5bc0a53aed..e9bf5ebd9f 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "7b00d3f8766a5fdcffd5e98ac639dc711e5cab6601c20817a88209a9248d80ce", "schema_version": 1} +{"protected_inventory_sha256": "567658c0f7e93cec0b15bec55dcabded810f6cd2c1d3321009976fa147360f4a", "schema_version": 1}