GM Control Tests

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())
 

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)))

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",