fix test_startup (#31103)

* fix test_startup

* real fix

selfdrive/controls/tests/test_startup.py

@@ -1,6 +1,4 @@
-#!/usr/bin/env python3
 import os
-import unittest
 from parameterized import parameterized
 
 from cereal import log, car
@@ -36,94 +34,87 @@ CX5_FW_VERSIONS = [
   (Ecu.transmission, 0x7e1, None, b'PYNC-21PS1-B\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'),
 ]
 
-class TestStartup(unittest.TestCase):
+
-
+@parameterized.expand([
-  def tearDown(self):
+  # TODO: test EventName.startup for release branches
-    managed_processes['controlsd'].stop()
+
-
+  # officially supported car
-  @parameterized.expand([
+  (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS, "toyota"),
-    # TODO: test EventName.startup for release branches
+  (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS, "toyota"),
-
+
-    # officially supported car
+  # dashcamOnly car
-    (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS, "toyota"),
+  (EventName.startupNoControl, MAZDA.CX5, CX5_FW_VERSIONS, "mazda"),
-    (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS, "toyota"),
+  (EventName.startupNoControl, MAZDA.CX5, CX5_FW_VERSIONS, "mazda"),
-
+
-    # dashcamOnly car
+  # unrecognized car with no fw
-    (EventName.startupNoControl, MAZDA.CX5, CX5_FW_VERSIONS, "mazda"),
+  (EventName.startupNoFw, None, None, ""),
-    (EventName.startupNoControl, MAZDA.CX5, CX5_FW_VERSIONS, "mazda"),
+  (EventName.startupNoFw, None, None, ""),
-
+
-    # unrecognized car with no fw
+  # unrecognized car
-    (EventName.startupNoFw, None, None, ""),
+  (EventName.startupNoCar, None, COROLLA_FW_VERSIONS[:1], "toyota"),
-    (EventName.startupNoFw, None, None, ""),
+  (EventName.startupNoCar, None, COROLLA_FW_VERSIONS[:1], "toyota"),
-
+
-    # unrecognized car
+  # fuzzy match
-    (EventName.startupNoCar, None, COROLLA_FW_VERSIONS[:1], "toyota"),
+  (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS_FUZZY, "toyota"),
-    (EventName.startupNoCar, None, COROLLA_FW_VERSIONS[:1], "toyota"),
+  (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS_FUZZY, "toyota"),
-
+])
-    # fuzzy match
+def test_startup_alert(expected_event, car_model, fw_versions, brand):
-    (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS_FUZZY, "toyota"),
+  controls_sock = messaging.sub_sock("controlsState")
-    (EventName.startupMaster, TOYOTA.COROLLA, COROLLA_FW_VERSIONS_FUZZY, "toyota"),
+  pm = messaging.PubMaster(['can', 'pandaStates'])
-  ])
+
-  def test_startup_alert(self, expected_event, car_model, fw_versions, brand):
+  params = Params()
-    controls_sock = messaging.sub_sock("controlsState")
+  params.put_bool("OpenpilotEnabledToggle", True)
-    pm = messaging.PubMaster(['can', 'pandaStates'])
+
-
+  # Build capnn version of FW array
-    params = Params()
+  if fw_versions is not None:
-    params.put_bool("OpenpilotEnabledToggle", True)
+    car_fw = []
-
+    cp = car.CarParams.new_message()
-    # Build capnn version of FW array
+    for ecu, addr, subaddress, version in fw_versions:
-    if fw_versions is not None:
+      f = car.CarParams.CarFw.new_message()
-      car_fw = []
+      f.ecu = ecu
-      cp = car.CarParams.new_message()
+      f.address = addr
-      for ecu, addr, subaddress, version in fw_versions:
+      f.fwVersion = version
-        f = car.CarParams.CarFw.new_message()
+      f.brand = brand
-        f.ecu = ecu
+
-        f.address = addr
+      if subaddress is not None:
-        f.fwVersion = version
+        f.subAddress = subaddress
-        f.brand = brand
+
-
+      car_fw.append(f)
-        if subaddress is not None:
+    cp.carVin = "1" * 17
-          f.subAddress = subaddress
+    cp.carFw = car_fw
-
+    params.put("CarParamsCache", cp.to_bytes())
-        car_fw.append(f)
+  else:
-      cp.carVin = "1" * 17
+    os.environ['SKIP_FW_QUERY'] = '1'
-      cp.carFw = car_fw
+
-      params.put("CarParamsCache", cp.to_bytes())
+  managed_processes['controlsd'].start()
-    else:
+
-      os.environ['SKIP_FW_QUERY'] = '1'
+  assert pm.wait_for_readers_to_update('can', 5)
-
+  pm.send('can', can_list_to_can_capnp([[0, 0, b"", 0]]))
-    managed_processes['controlsd'].start()
+
-
+  assert pm.wait_for_readers_to_update('pandaStates', 5)
-    assert pm.wait_for_readers_to_update('can', 5)
+  msg = messaging.new_message('pandaStates', 1)
-    pm.send('can', can_list_to_can_capnp([[0, 0, b"", 0]]))
+  msg.pandaStates[0].pandaType = log.PandaState.PandaType.uno
-
+  pm.send('pandaStates', msg)
-    msg = messaging.new_message('pandaStates', 1)
+
-    msg.pandaStates[0].pandaType = log.PandaState.PandaType.uno
+  # fingerprint
-    pm.send('pandaStates', msg)
+  if (car_model is None) or (fw_versions is not None):
-
+    finger = {addr: 1 for addr in range(1, 100)}
-    # fingerprint
+  else:
-    if (car_model is None) or (fw_versions is not None):
+    finger = _FINGERPRINTS[car_model][0]
-      finger = {addr: 1 for addr in range(1, 100)}
+
-    else:
+  msgs = [[addr, 0, b'\x00'*length, 0] for addr, length in finger.items()]
-      finger = _FINGERPRINTS[car_model][0]
+  for _ in range(1000):
-
+    # controlsd waits for boardd to echo back that it has changed the multiplexing mode
-    msgs = [[addr, 0, b'\x00'*length, 0] for addr, length in finger.items()]
+    if not params.get_bool("ObdMultiplexingChanged"):
-    for _ in range(1000):
+      params.put_bool("ObdMultiplexingChanged", True)
-      # controlsd waits for boardd to echo back that it has changed the multiplexing mode
+
-      if not params.get_bool("ObdMultiplexingChanged"):
+    pm.send('can', can_list_to_can_capnp(msgs))
-        params.put_bool("ObdMultiplexingChanged", True)
+    assert pm.wait_for_readers_to_update('can', 5, dt=0.001), f"step: {_}"
-
+
-      pm.send('can', can_list_to_can_capnp(msgs))
+    ctrls = messaging.drain_sock(controls_sock)
-      assert pm.wait_for_readers_to_update('can', 2, dt=0.001)
+    if len(ctrls):
-
+      event_name = ctrls[0].controlsState.alertType.split("/")[0]
-      ctrls = messaging.drain_sock(controls_sock)
+      assert EVENT_NAME[expected_event] == event_name, f"expected {EVENT_NAME[expected_event]} for '{car_model}', got {event_name}"
-      if len(ctrls):
+      break
-        event_name = ctrls[0].controlsState.alertType.split("/")[0]
+  else:
-        self.assertEqual(EVENT_NAME[expected_event], event_name,
+    raise Exception(f"failed to fingerprint {car_model}")
-                         f"expected {EVENT_NAME[expected_event]} for '{car_model}', got {event_name}")
-        break
-    else:
-      self.fail(f"failed to fingerprint {car_model}")
-
-if __name__ == "__main__":
-  unittest.main()