diff --git a/selfdrive/controls/tests/test_startup.py b/selfdrive/controls/tests/test_startup.py new file mode 100755 index 0000000000..3da16e855d --- /dev/null +++ b/selfdrive/controls/tests/test_startup.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +import time +import unittest +from parameterized import parameterized + +from cereal import log, car +import cereal.messaging as messaging +from common.params import Params +from selfdrive.boardd.boardd_api_impl import can_list_to_can_capnp # pylint: disable=no-name-in-module,import-error +from selfdrive.car.fingerprints import _FINGERPRINTS +from selfdrive.car.hyundai.values import CAR as HYUNDAI +from selfdrive.car.mazda.values import CAR as MAZDA +from selfdrive.controls.lib.events import EVENT_NAME +from selfdrive.test.helpers import with_processes + +EventName = car.CarEvent.EventName + +class TestStartup(unittest.TestCase): + + @parameterized.expand([ + # TODO: test EventName.startup for release branches + + # officially supported car + (EventName.startupMaster, HYUNDAI.SONATA, False), + (EventName.startupMaster, HYUNDAI.SONATA, True), + + # community supported car + (EventName.startupMaster, HYUNDAI.KIA_STINGER, True), + (EventName.startupMaster, HYUNDAI.KIA_STINGER, False), + + # dashcamOnly car + (EventName.startupMaster, MAZDA.CX5, True), + (EventName.startupMaster, MAZDA.CX5, False), + + # unrecognized car + (EventName.startupNoCar, None, True), + (EventName.startupNoCar, None, False), + ]) + @with_processes(['controlsd']) + def test_startup_alert(self, expected_event, car, toggle_enabled): + + # TODO: this should be done without any real sockets + sm = messaging.SubMaster(['controlsState']) + pm = messaging.PubMaster(['can', 'health']) + Params().put("CommunityFeaturesToggle", b"1" if toggle_enabled else b"0") + + time.sleep(2) # wait for controlsd to be ready + + health = messaging.new_message('health') + health.health.hwType = log.HealthData.HwType.uno + pm.send('health', health) + + # fingerprint + if car is None: + finger = {addr: 1 for addr in range(1, 100)} + else: + finger = _FINGERPRINTS[car][0] + + for _ in range(500): + msgs = [[addr, 0, b'\x00'*length, 0] for addr, length in finger.items()] + pm.send('can', can_list_to_can_capnp(msgs)) + + time.sleep(0.05) + sm.update(0) + if sm.updated["controlsState"]: + event_name = sm["controlsState"].alertType.split("/")[0] + self.assertEqual(EVENT_NAME[expected_event], event_name, + f"expected {EVENT_NAME[expected_event]} for '{car}', got {event_name}") + break + else: + self.fail(f"failed to fingerprint {car}") + +if __name__ == "__main__": + unittest.main()