Files
StarPilot/openpilot/starpilot/controllers/tests/test_runtime.py
T
firestar5683 a82935f326 User interfaces and Android Auto
Integrate onroad views, configurable widgets, local settings, steering-profile contracts, Bluetooth and Android Auto with their device and runtime services.

Co-authored-by: Adeeb Shihadeh <adeebshihadeh@gmail.com>
Co-authored-by: Daniel Koepping <elkoled@gmail.com>
Co-authored-by: Danny <dannyirwin31@yahoo.com>
Co-authored-by: DimitriChrysafis <Dimitri.chrysafis@gmail.com>
Co-authored-by: Dom <168790843+firestar5683@users.noreply.github.com>
Co-authored-by: Harald Schäfer <harald.the.engineer@gmail.com>
Co-authored-by: Prabhaav Pillai <prabhaav@msu.edu>
Co-authored-by: Shane Smiskol <shane@smiskol.com>
Co-authored-by: dirwin31 <83434411+dirwin31@users.noreply.github.com>
Co-authored-by: firestarsdog <229254897+firestarsdog@users.noreply.github.com>
Co-authored-by: stef <19478336+stefpi@users.noreply.github.com>
Co-authored-by: whoisdomi <dcantu83@gmail.com>

Co-authored-by: Intelli <contact@intelli.software>
Co-authored-by: pharmacomaniac <blittle65@gmail.com>

Co-authored-by: prabhaavp <prabhaav@msu.edu>
2026-10-14 20:02:39 -05:00

59 lines
2.6 KiB
Python

from unittest import TestCase
from unittest.mock import Mock, patch
from openpilot.starpilot.controllers.input import Press
from openpilot.starpilot.controllers.runtime import ControllerRuntime
class TestControllerRuntime(TestCase):
def runtime(self):
authority, reader, owner, service = Mock(), Mock(), Mock(), Mock()
reader.poll.return_value = []
reader.devices.return_value = []
with patch('openpilot.starpilot.controllers.runtime.ControllerOwner', return_value=owner), \
patch('openpilot.starpilot.controllers.runtime.ControllerService', return_value=service):
runtime = ControllerRuntime(None, actions=Mock(), favorites=Mock(), invoke_favorite=Mock(),
authority=authority, reader=reader)
self.addCleanup(runtime.close)
return runtime, authority, reader, owner, service
def test_input_inventory_and_existing_presses_precede_new_bindings(self):
runtime, _authority, reader, owner, service = self.runtime()
reader.poll.return_value = [Press('device', 304, 10)]
reader.devices.return_value = [{'id': 'device', 'name': 'Gamepad', 'bus': 5}]
events = []
owner.set_devices.side_effect = lambda devices: events.append(('devices', devices))
owner.tick.side_effect = lambda: events.append(('tick',))
owner.feed.side_effect = lambda press: events.append(('press', press))
service.poll.side_effect = lambda: events.append(('settings',))
runtime.poll()
self.assertEqual([item[0] for item in events], ['devices', 'tick', 'press', 'settings'])
self.assertEqual(events[0][1], reader.devices.return_value)
def test_input_failure_stops_actions_and_releases_all_resources(self):
runtime, authority, reader, owner, service = self.runtime()
reader.poll.side_effect = OSError('input disconnected')
with self.assertLogs('openpilot.starpilot.controllers.runtime', level='ERROR'):
runtime.poll()
runtime.poll()
owner.feed.assert_not_called()
service.poll.assert_not_called()
reader.close.assert_called_once()
authority.close.assert_called_once()
service.close.assert_called_once()
def test_idle_poll_expires_learning_even_without_input(self):
runtime, _authority, _reader, owner, service = self.runtime()
runtime.poll()
owner.tick.assert_called_once()
service.poll.assert_called_once()
owner.feed.assert_not_called()
def test_close_is_idempotent(self):
runtime, authority, reader, _owner, service = self.runtime()
runtime.close()
runtime.close()
reader.close.assert_called_once()
authority.close.assert_called_once()
service.close.assert_called_once()