controlsd: add state machine tests (#24107)

* Handle NO_ENTRY in PRE_ENABLED * add test * add preEnabled NO_ENTRY test * stash * test soft disable * tuples * remove overriding until it's merged in * use Events class * fix tests and split out * don't rely on controlsd's counter
2026-10-01 03:43:51 +08:00 · 2022-04-04 17:13:29 -07:00
parent 7dffd0ac0b
commit d4f330447a
2 changed files with 128 additions and 16 deletions
@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+import unittest
+
+from cereal import car, log
+from common.realtime import DT_CTRL
+from selfdrive.car.car_helpers import interfaces
+from selfdrive.controls.controlsd import Controls, SOFT_DISABLE_TIME
+from selfdrive.controls.lib.events import Events, ET, Alert, Priority, AlertSize, AlertStatus, VisualAlert, \
+                                          AudibleAlert, EVENTS
+
+State = log.ControlsState.OpenpilotState
+
+# The event types that maintain the current state
+MAINTAIN_STATES = {State.enabled: None, State.disabled: None, State.softDisabling: ET.SOFT_DISABLE,
+                   State.preEnabled: ET.PRE_ENABLE}
+ALL_STATES = tuple((state for state in State.schema.enumerants.values() if
+                    state != State.overriding))  # TODO: remove overriding exception
+# The event types checked in DISABLED section of state machine
+ENABLE_EVENT_TYPES = (ET.ENABLE, ET.PRE_ENABLE)
+
+
+def make_event(event_types):
+  event = {}
+  for ev in event_types:
+    event[ev] = Alert("", "", AlertStatus.normal, AlertSize.small, Priority.LOW,
+                      VisualAlert.none, AudibleAlert.none, 1.)
+  EVENTS[0] = event
+  return 0
+
+
+class TestStateMachine(unittest.TestCase):
+
+  def setUp(self):
+    CarInterface, CarController, CarState = interfaces["mock"]
+    CP = CarInterface.get_params("mock")
+    CI = CarInterface(CP, CarController, CarState)
+
+    self.controlsd = Controls(CI=CI)
+    self.controlsd.events = Events()
+    self.controlsd.soft_disable_timer = int(SOFT_DISABLE_TIME / DT_CTRL)
+    self.CS = car.CarState()
+
+  def test_immediate_disable(self):
+    for state in ALL_STATES:
+      self.controlsd.events.add(make_event([MAINTAIN_STATES[state], ET.IMMEDIATE_DISABLE]))
+      self.controlsd.state = state
+      self.controlsd.state_transition(self.CS)
+      self.assertEqual(State.disabled, self.controlsd.state)
+      self.controlsd.events.clear()
+
+  def test_user_disable(self):
+    for state in ALL_STATES:
+      self.controlsd.events.add(make_event([MAINTAIN_STATES[state], ET.USER_DISABLE]))
+      self.controlsd.state = state
+      self.controlsd.state_transition(self.CS)
+      self.assertEqual(State.disabled, self.controlsd.state)
+      self.controlsd.events.clear()
+
+  def test_soft_disable(self):
+    for state in ALL_STATES:
+      if state == State.preEnabled:  # preEnabled considers NO_ENTRY instead
+        continue
+      self.controlsd.events.add(make_event([MAINTAIN_STATES[state], ET.SOFT_DISABLE]))
+      self.controlsd.state = state
+      self.controlsd.state_transition(self.CS)
+      self.assertEqual(self.controlsd.state, State.disabled if state == State.disabled else State.softDisabling)
+      self.controlsd.events.clear()
+
+  def test_soft_disable_timer(self):
+    self.controlsd.state = State.enabled
+    self.controlsd.events.add(make_event([ET.SOFT_DISABLE]))
+    self.controlsd.state_transition(self.CS)
+    for _ in range(int(SOFT_DISABLE_TIME / DT_CTRL)):
+      self.assertEqual(self.controlsd.state, State.softDisabling)
+      self.controlsd.state_transition(self.CS)
+
+    self.assertEqual(self.controlsd.state, State.disabled)
+
+  def test_no_entry(self):
+    # disabled with enable events
+    for et in ENABLE_EVENT_TYPES:
+      self.controlsd.events.add(make_event([ET.NO_ENTRY, et]))
+      self.controlsd.state_transition(self.CS)
+      self.assertEqual(self.controlsd.state, State.disabled)
+      self.controlsd.events.clear()
+
+  def test_no_entry_pre_enable(self):
+    # preEnabled with preEnabled event
+    self.controlsd.state = State.preEnabled
+    self.controlsd.events.add(make_event([ET.NO_ENTRY, ET.PRE_ENABLE]))
+    self.controlsd.state_transition(self.CS)
+    self.assertEqual(self.controlsd.state, State.disabled)
+
+  def test_maintain_states(self):
+    # Given current state's event type, we should maintain state
+    for state in ALL_STATES:
+      self.controlsd.state = state
+      self.controlsd.events.add(make_event([MAINTAIN_STATES[state]]))
+      self.controlsd.state_transition(self.CS)
+      self.assertEqual(self.controlsd.state, state)
+      self.controlsd.events.clear()
+
+
+if __name__ == "__main__":
+  unittest.main()