From 8d8f325a1b26d3bc65680c63dd325bd9eb15f50f Mon Sep 17 00:00:00 2001 From: firestar5683 <168790843+firestar5683@users.noreply.github.com> Date: Sun, 4 Oct 2026 17:46:51 -0500 Subject: [PATCH] GM Control Tests --- .../safety/tests/test_gm_cc_gateway_stock.py | 24 ++++-- .../starpilot/tests/test_gm_stock_caller.py | 80 +++++++++++++++++++ tools/ci/run_vehicle_tests.py | 1 + 3 files changed, 99 insertions(+), 6 deletions(-) create mode 100644 openpilot/starpilot/tests/test_gm_stock_caller.py diff --git a/opendbc_repo/opendbc/safety/tests/test_gm_cc_gateway_stock.py b/opendbc_repo/opendbc/safety/tests/test_gm_cc_gateway_stock.py index 4cce7b33a5..a41369f49f 100644 --- a/opendbc_repo/opendbc/safety/tests/test_gm_cc_gateway_stock.py +++ b/opendbc_repo/opendbc/safety/tests/test_gm_cc_gateway_stock.py @@ -16,6 +16,10 @@ class TestGmCcGatewayStock(unittest.TestCase): def setUp(self): self.safety = libsafety_py.libsafety + def tearDown(self): + self.safety.set_alternative_experience(0) + self.safety.set_safety_hooks(CarParams.SafetyModel.noOutput, 0) + @staticmethod def packet(msg): return libsafety_py.make_CANPacket(msg[0], msg[2], msg[1]) @@ -51,6 +55,11 @@ class TestGmCcGatewayStock(unittest.TestCase): self.assertTrue(state_out.canValid) state = interface.CS self.mode(cp.safetyConfigs[0].safetyParam) + # Establish required-source health before an advancing physical neutral + # observation earns cancel credit; cold incomplete RX must not grant it. + initial_frames = [gmcan.create_buttons(packer, 0, 1, 1) if frame[0] == 0x1E1 else frame for frame in frames] + self.feed(initial_frames) + self.safety.set_timer(1_010_000) self.feed(frames) controller = interface.CC controller.frame = 20 @@ -63,10 +72,11 @@ class TestGmCcGatewayStock(unittest.TestCase): control.actuators.torque = 0.03 control.actuators.accel = 1.5 _, commands = controller.update(control.as_reader(), state, now_nanos) + self.joined_cp, self.joined_interface, self.joined_state = cp, interface, state_out return packer, frames, commands def test_real_packed_controller_into_native_all_nine(self): - for car in CC_GATEWAY_STOCK_CAR: + for car in sorted(CC_GATEWAY_STOCK_CAR): with self.subTest(car=car): packer, frames, commands = self.joined(car) self.assertTrue(self.safety.get_controls_allowed()) @@ -76,7 +86,7 @@ class TestGmCcGatewayStock(unittest.TestCase): self.assertNotEqual(((steer[1][0] & 7) << 8) | steer[1][1], 0) self.assertEqual(cancel[2], 0) self.assertTrue(self.safety.safety_tx_hook(self.packet(steer))) - self.assertTrue(self.safety.safety_tx_hook(self.packet(cancel))) + self.assertTrue(self.safety.safety_tx_hook(self.packet(cancel)), (car, self.joined_cp.safetyConfigs[0].safetyParam, cancel)) self.assertFalse(self.safety.safety_tx_hook(self.packet(cancel))) # replay self.assertFalse(self.safety.safety_tx_hook(self.packet((cancel[0], cancel[1], 2)))) self.assertFalse(self.safety.safety_tx_hook(self.packet((cancel[0], cancel[1][:-1], 0)))) @@ -106,12 +116,14 @@ class TestGmCcGatewayStock(unittest.TestCase): _, _, commands = self.joined(car, cruise=cruise, main=main, brake=brake, gas=gas, active=active) steer = next(m for m in commands if m[0] == 0x180) - if gas and active and is_ordinary_cc_profile(params(car)): - # Stock conventional cruise retains lateral authority on gas. - self.assertGreater(((steer[1][0] & 7) << 8) | steer[1][1], 0) + if active and main and cruise and is_ordinary_cc_profile(params(car)): + # C160 consumes the caller axis; native independently rejects brake. + # Arbitrary active requests are retained here as adversarial probes. + self.assertGreater(((steer[1][0] & 7) << 8) | steer[1][1], 0, (car, cruise, main, brake, gas, active)) else: self.assertEqual(((steer[1][0] & 7) << 8) | steer[1][1], 0, (car, cruise, main, brake, gas, active)) - self.assertTrue(self.safety.safety_tx_hook(self.packet(steer))) + self.assertEqual(self.safety.safety_tx_hook(self.packet(steer)), + not (brake and active and main and cruise and is_ordinary_cc_profile(params(car)))) if not cruise or not main or brake: self.assertFalse(self.safety.get_controls_allowed()) diff --git a/openpilot/starpilot/tests/test_gm_stock_caller.py b/openpilot/starpilot/tests/test_gm_stock_caller.py new file mode 100644 index 0000000000..6b5509ee4b --- /dev/null +++ b/openpilot/starpilot/tests/test_gm_stock_caller.py @@ -0,0 +1,80 @@ +"""GM pedal events through the stock control state and native TX checks.""" +import ast +from pathlib import Path +from types import SimpleNamespace as NS +import unittest + +from openpilot.cereal import log +from opendbc.car.structs import car +from openpilot.selfdrive.selfdrived.events import Events +from openpilot.selfdrive.selfdrived.state import StateMachine +from openpilot.starpilot.aol.runtime import ordinary_lateral_requested +from opendbc.car.gm import gmcan +from opendbc.car.gm.values import CC_GATEWAY_STOCK_CAR, is_ordinary_cc_profile +from opendbc.safety.tests import test_gm_cc_gateway_stock as stock + + +def pedal_event_expression(): + # Compile the reached producer predicate rather than inventing pedal semantics. + import openpilot.selfdrive.selfdrived.selfdrived as module + tree = ast.parse(Path(module.__file__).read_text()) + matches = [node.test for node in ast.walk(tree) if isinstance(node, ast.If) + and any(isinstance(call, ast.Call) and isinstance(call.func, ast.Attribute) + and call.func.attr == "add" and call.args + and isinstance(call.args[0], ast.Attribute) + and call.args[0].attr == "pedalPressed" + for statement in node.body if isinstance(statement, ast.Expr) + for call in (statement.value,))] + if len(matches) != 1: + raise AssertionError("SelfdriveD pedal predicate changed; review its caller") + return compile(ast.Expression(matches[0]), "SelfdriveD.pedalPressed", "eval") + + +class TestGMStockCaller(unittest.TestCase): + def test_decoded_pedals_normal_axis_and_adversarial_native_boundary(self): + predicate = pedal_event_expression() + fixture = stock.TestGmCcGatewayStock() + fixture.setUp() + self.addCleanup(fixture.safety.set_safety_hooks, car.CarParams.SafetyModel.noOutput, 0) + self.addCleanup(fixture.safety.set_alternative_experience, 0) + for identity in sorted(CC_GATEWAY_STOCK_CAR): + for brake, gas, accelerator_disengage in ((False, False, False), (True, False, False), + (False, True, False), (False, True, True)): + with self.subTest(identity=identity, brake=brake, gas=gas, accelerator_disengage=accelerator_disengage): + # Capture the real decoded CS and CP via the actual CI/controller fixture. + packer, frames, _ = fixture.joined(identity, brake=brake, gas=gas, active=False, cancel=False) + cp, ci, cs = fixture.joined_cp, fixture.joined_interface, fixture.joined_state + self.assertEqual(cs.brakePressed, brake) + self.assertEqual(cs.gasPressed, gas) + self.assertFalse(cs.standstill) + previous = car.CarState.new_message() + events = Events() + producer = NS(CS_prev=previous, disengage_on_accelerator=accelerator_disengage) + if eval(predicate, {"CS": cs, "self": producer}): + events.add(log.OnroadEvent.EventName.pedalPressed) + machine = StateMachine() + machine.state = log.SelfdriveState.OpenpilotState.enabled + enabled, active = machine.update(events) + self.assertEqual(active, not (brake or (gas and accelerator_disengage))) + requested = ordinary_lateral_requested(active, cs, cp) + control = car.CarControl.new_message() + control.enabled = enabled + control.latActive = requested + control.actuators.torque = .03 + # Disabled steering is 10 Hz. Advance ten real 100-Hz input frames, + # keeping physical sources fresh before the next eligible TX slot. + for tick in range(1, 11): + stamp = 1_050_000_000 + tick * 10_000_000 + current_frames = [frame for frame in frames if frame[0] != 0x1E1] + if tick % 3 == 0: + current_frames.append(gmcan.create_buttons(packer, 0, (2 + tick // 3) % 4, 1)) + cs = ci.update([(stamp, current_frames)]) + self.assertTrue(cs.canValid) + fixture.safety.set_timer(stamp // 1000) + fixture.feed(current_frames) + ci.CC.frame = 30 + _, commands = ci.CC.update(control.as_reader(), ci.CS, stamp) + steering = next(msg for msg in commands if msg[0] == 0x180) + torque = ((steering[1][0] & 7) << 8) | steering[1][1] + self.assertEqual(torque > 0, requested and (not gas or is_ordinary_cc_profile(cp))) + self.assertTrue(fixture.safety.safety_tx_hook(fixture.packet(steering))) diff --git a/tools/ci/run_vehicle_tests.py b/tools/ci/run_vehicle_tests.py index 0cad12a541..b62ff8c0e8 100644 --- a/tools/ci/run_vehicle_tests.py +++ b/tools/ci/run_vehicle_tests.py @@ -87,6 +87,7 @@ RELEASE_TARGETS = ( "opendbc_repo/opendbc/safety/tests/test_gm_sdgm_stock.py", "opendbc_repo/opendbc/safety/tests/test_gm_cc_gateway_stock.py", "opendbc_repo/opendbc/safety/tests/test_gm_stock_cancel_delay.py", + "openpilot/starpilot/tests/test_gm_stock_caller.py", "opendbc_repo/opendbc/safety/tests/test_gm_camera_stock_four.py", "opendbc_repo/opendbc/safety/tests/test_gm_volt_gateway_mapping.py", "opendbc_repo/opendbc/safety/tests/test_gm_volt_alternate_brake.py",