test_models: add fuzzy tx test

selfdrive/car/tests/test_models.py

@@ -17,6 +17,7 @@
 from opendbc.car import DT_CTRL, gen_empty_fingerprint, structs
 from opendbc.car.fingerprints import all_known_cars, MIGRATION
 from opendbc.car.car_helpers import FRAME_FINGERPRINT, interfaces
+from opendbc.car.toyota.values import ANGLE_CONTROL_CAR as TOYOTA_ANGLE_CONTROL_CAR
 from opendbc.car.honda.values import CAR as HONDA, HondaFlags
 from opendbc.car.values import Platform
 from opendbc.car.tests.routes import non_tested_cars, routes, CarTestRoute
@@ -26,8 +27,10 @@
 from openpilot.system.hardware.hw import DEFAULT_DOWNLOAD_CACHE_ROOT
 from openpilot.tools.lib.logreader import LogReader, auto_source, internal_source, openpilotci_source, openpilotci_source_zst
 from openpilot.tools.lib.route import SegmentName
+from openpilot.selfdrive.test.fuzzy_generation import FuzzyGenerator
 
 from panda.tests.libpanda import libpanda_py
+from panda.tests.safety.common import VEHICLE_SPEED_FACTOR
 
 EventName = car.OnroadEvent.EventName
 PandaType = log.PandaState.PandaType
@@ -326,10 +329,13 @@ def test_car_controller(car_control):
   @settings(max_examples=MAX_EXAMPLES, deadline=None,
             phases=(Phase.reuse, Phase.generate, Phase.shrink))
   @given(data=st.data())
-  def test_panda_safety_carstate_fuzzy(self, data):
+  def test_panda_safety_carstate_tx_fuzzy(self, data):
     """
       For each example, pick a random CAN message on the bus and fuzz its data,
       checking for panda state mismatches.
+
+      We also randomize a CarControl message, and set it up to be consistent with the
+      current carstate and check for logic mismatches between OP carcontrollers and panda
     """
 
     if self.CP.dashcamOnly:
@@ -386,6 +392,63 @@ def test_panda_safety_carstate_fuzzy(self, data):
         if self.safety.get_acc_main_on() != prev_panda_acc_main_on:
           self.assertEqual(CS.cruiseState.available, self.safety.get_acc_main_on())
 
+
+    if self.CP.notCar:
+      self.skipTest("no need to check safety tx hooks for notCar")
+
+    cc_msg = FuzzyGenerator.get_random_msg(data.draw, car.CarControl, real_floats=True)
+
+    def set_controls_allowed(status):
+      self.safety.set_controls_allowed(status)
+      self.safety.set_cruise_engaged_prev(status)
+
+    # check for controls_allowed and set True if not already in these cases:
+    # - cancel button messages before controls_allowed
+    # - actuator commands before controls_allowed
+    # - resume button message before controls_allowed
+    controls_state = any([cc_msg["enabled"], cc_msg["latActive"], cc_msg["longActive"]])
+    panda_controls_allowed = any([controls_state, cc_msg["cruiseControl"]["resume"], cc_msg["cruiseControl"]["cancel"]])
+    set_controls_allowed(panda_controls_allowed)
+
+    # relay_malfunction might be set during randomizing carState
+    if self.safety.get_relay_malfunction():
+      self.safety.set_relay_malfunction(False)
+
+    # Nissan calculate vEgoRaw with avg speed of 4 wheels
+    # while panda get vehicle_speed from rear wheels only,
+    # so there might be a big enough difference between the two
+    if self.CP.carName == "nissan":
+      panda_vehicle_speed_last = self.safety.get_vehicle_speed_last() / VEHICLE_SPEED_FACTOR
+      if abs(self.CI.CS.out.vEgoRaw - panda_vehicle_speed_last) > 0.2:
+        cs_msg = self.CI.CS.out.to_dict()
+        cs_msg["vEgoRaw"] = panda_vehicle_speed_last
+        self.CI.CS.out = car.CarState.new_message(**cs_msg).as_reader()
+
+    # no long controls if gas_pressed_prev
+    if self.safety.get_gas_pressed_prev():
+      cc_msg["longActive"] = False
+      cc_msg["enabled"] = False
+
+    if cc_msg["longActive"]:
+      cc_msg["enabled"] = True
+    else:
+      # follow logic of LoC in controlsd to set accel to 0 if longActive is False
+      cc_msg["actuators"]["accel"] = 0
+
+    if self.CP.carName == "honda" and self.safety.get_honda_fwd_brake():
+      self.safety.set_honda_fwd_brake(False)
+
+    CC = car.CarControl.new_message(**cc_msg).as_reader()
+
+    self.safety.set_timer(int(self.tx_fuzzy_ts_nanos / 1e3))
+    new_actuators, sendcans = self.CI.apply(CC, self.tx_fuzzy_ts_nanos)
+    new_actuators = new_actuators.as_reader()
+    self.tx_fuzzy_ts_nanos += DT_CTRL * 1e9
+
+    for addr, _, dat, bus in sendcans:
+      to_send = libpanda_py.make_CANPacket(addr, bus % 4, dat)
+      self.assertTrue(self.safety.safety_tx_hook(to_send), (addr, dat, bus))
+
   def test_panda_safety_carstate(self):
     """
       Assert that panda safety matches openpilot's carState