Regression test fixtures

This commit is contained in:
firestar5683
2026-10-08 00:44:18 -05:00
parent f7c0090274
commit 65a2b29b34
33 changed files with 176 additions and 98 deletions
@@ -1,6 +1,4 @@
import unittest
import tempfile
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch
@@ -9,7 +7,7 @@ from opendbc.car.hyundai.blended_longitudinal import candidate_from_stock
from opendbc.car.hyundai.tests.test_palisade_2023 import params
from opendbc.car.hyundai.values import CarControllerParams, CAR
from opendbc.car.hyundai.interface import CarInterface
from openpilot.starpilot.longitudinal.output_max import final_output, OutputMaximum, KEY
from openpilot.starpilot.longitudinal.output_max import final_output
from openpilot.selfdrive.controls.lib.longcontrol import LongControl
from openpilot.starpilot.longitudinal.extension import LongitudinalContext
from openpilot.starpilot.longitudinal.inputs import LongitudinalInputs
@@ -103,14 +101,9 @@ class TestBlendedOutput(unittest.TestCase):
output = control.update(True, self.cs, 3.5, False, CarInterface.get_pid_accel_limits(candidate, 0., 0.),
context=LongitudinalContext(experimental_mode=False))
self.assertAlmostEqual(output, 3.5)
self.assertAlmostEqual(final_output(output, None, 0), 3.5)
with tempfile.TemporaryDirectory() as directory:
saved_params = SimpleNamespace(get_param_path=lambda key: str(Path(directory) / key))
maximum = OutputMaximum(saved_params, candidate)
self.assertAlmostEqual(final_output(output, maximum, 0), 3.5)
(Path(directory) / KEY).write_bytes(b'1.2')
self.assertAlmostEqual(final_output(output, maximum, 999_999_999), 3.5)
self.assertAlmostEqual(final_output(output, maximum, 1_000_000_000), 1.2)
self.assertAlmostEqual(final_output(output), 3.5)
self.assertAlmostEqual(final_output(5.), 4.)
self.assertAlmostEqual(final_output(-3.5), -3.5)
self.assertEqual(CarControllerParams(stock).ACCEL_MAX, 2.)
self.assertEqual(CarInterface.get_pid_accel_limits(stock, 0., 0.), (-3.5, 2.))
sibling = CarInterface.get_params(CAR.HYUNDAI_IONIQ_6, {0: {}, 1: {}, 2: {}}, [], False, False, False)
@@ -172,7 +172,8 @@ class HondaHostTests(unittest.TestCase):
def update(self, can_packets):
return None
with OpenpilotPrefix(), mock.patch.dict('os.environ', {'AOL_REPLAY_RUNTIME': '1', 'SIMULATION': '1'}):
with (OpenpilotPrefix(), mock.patch.dict('os.environ', {'AOL_REPLAY_RUNTIME': '1', 'SIMULATION': '1'}),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.SIMULATION', False)):
messaging.reset_context()
params = Params()
params.put_bool('OpenpilotEnabledToggle', True, block=True)
@@ -193,6 +194,7 @@ class HondaHostTests(unittest.TestCase):
self.assertTrue(intent.seen['aolIntentWire'])
self.assertTrue(decode_intent(intent['aolIntentWire']).allowedLatch)
selfdrive.step()
self.assertFalse(selfdrive.initialized) # Native acknowledgment must precede bootstrap.
controls.sm.update(100)
cc, _ = controls.state_control()
self.assertFalse(cc.latActive or cc.longActive) # no native capability/ack
@@ -138,8 +138,9 @@ class CardInitLifecycleTest(unittest.TestCase):
card.aol_replay = True
card.ioniq6_long_prearmed = False
card.ioniq6_long_selected = False
card.can_callbacks = (lambda **_: [], lambda _: None)
card.params = SimpleNamespace(put_bool=Mock())
card.can_callbacks = (lambda wait_for_one=False: [], lambda _: None)
params = SimpleNamespace(put_bool=Mock())
self.enterContext(patch.object(card, 'params', params, create=True))
card.state_publish = Mock()
packer = CANPacker(DBC[cp.carFingerprint][Bus.pt])
cs, _ = feed_camera(card.CI, packer, 22_000_000_000, active=False)
@@ -164,9 +165,10 @@ class CardInitLifecycleTest(unittest.TestCase):
def all_alive(self, services):
return all(self.alive[name] for name in services)
card.sm = SubMaster()
sm = SubMaster()
self.enterContext(patch.object(card, 'sm', sm, create=True))
sent = []
card.publish_sendcan = lambda frames, valid=True: sent.extend(frames)
self.enterContext(patch.object(card, 'publish_sendcan', lambda frames, valid=True: sent.extend(frames)))
# These saved requests cannot acquire startup brake ownership.
card.CI.CC.gm_auto_hold = True
card.CI.CC.volt_one_pedal = True
@@ -180,7 +182,7 @@ class CardInitLifecycleTest(unittest.TestCase):
card.step()
self.assertEqual(apply.call_count, 30)
init.assert_called_once()
card.params.put_bool.assert_called_once_with('ControlsReady', True)
params.put_bool.assert_called_once_with('ControlsReady', True)
self.assertEqual(card.CI.CC.gm_auto_hold_state.drive_ns, 0)
self.assertEqual(card.CI.CC.apply_brake, 0)
for call in apply.call_args_list:
@@ -202,16 +204,16 @@ class CardInitLifecycleTest(unittest.TestCase):
card.step()
self.assertEqual(apply.call_count, 30)
panda.safetyParam = 0x5007
card.sm.logMonoTime['pandaStates'] = boot - 300_000_001
sm.logMonoTime['pandaStates'] = boot - 300_000_001
card.step()
self.assertEqual(apply.call_count, 30)
card.sm.logMonoTime['pandaStates'] = boot
sm.logMonoTime['pandaStates'] = boot
active = structs.CarControl.new_message(latActive=True)
control = active.as_reader()
card.step()
self.assertEqual(apply.call_count, 30)
control = structs.CarControl.new_message().as_reader()
card.sm.events = []
sm.events = []
card.step()
self.assertEqual(apply.call_count, 31)
self.assertEqual(apply.call_args.args[1], boot)
@@ -29,10 +29,16 @@ class Sources(messaging.SubMaster):
class Capture:
def __init__(self):
self.message = None
self.companion = None
def send(self, service, message):
assert service == 'longitudinalPlan'
self.message = messaging.log_from_bytes(message.to_bytes())
assert service in ('longitudinalPlan', 'starpilotLongitudinalPlan')
decoded = messaging.log_from_bytes(message.to_bytes())
assert decoded.which() == service
if service == 'longitudinalPlan':
self.message = decoded
else:
self.companion = decoded
class LongitudinalPlanValidityTest(unittest.TestCase):
@@ -45,10 +51,15 @@ class LongitudinalPlanValidityTest(unittest.TestCase):
with self.subTest(delay_ns=delay_ns):
plan = messaging.new_message('longitudinalPlan')
plan.logMonoTime = sources.logMonoTime['modelV2'] + delay_ns
with patch('openpilot.selfdrive.controls.lib.longitudinal_planner.messaging.new_message', return_value=plan):
new_message = messaging.new_message
with patch('openpilot.selfdrive.controls.lib.longitudinal_planner.messaging.new_message',
side_effect=lambda service, plan=plan, new_message=new_message: (
plan if service == 'longitudinalPlan' else new_message(service))):
planner.publish(sources, published)
self.assertEqual(published.message.longitudinalPlan.modelMonoTime, sources.logMonoTime['modelV2'])
self.assertAlmostEqual(published.message.longitudinalPlan.processingDelay, delay_ns / 1e9, places=8)
self.assertEqual(published.companion.logMonoTime, plan.logMonoTime)
self.assertEqual(published.companion.starpilotLongitudinalPlan.sourcePlanMonoTime, plan.logMonoTime)
def test_optional_vision_frequency_and_dashboard_absence_do_not_invalidate_native_plan(self):
cp = CarInterface.get_non_essential_params(CAR.HONDA_CIVIC)
@@ -31,12 +31,13 @@ class FakeMaster:
class FakeAlerts:
def __init__(self):
self.alerts = []
self.current_alert = None
def add_many(self, _frame, alerts):
self.alerts = alerts
def process_alerts(self, _frame, _clear):
pass
self.current_alert = max(self.alerts, key=lambda alert: alert.priority, default=None)
class TestLaneChangeAlert(unittest.TestCase):
@@ -161,7 +161,7 @@ def test_transform_guard_preserves_render_projection_updates(renderer, monkeypat
@pytest.mark.parametrize("renderer_class", [ModelRenderer, LargeModelRenderer])
def test_radar_only_reuses_lane_geometry_without_changing_path_filters_or_clip(renderer, renderer_class):
def test_radar_only_reuses_lane_geometry_without_changing_path_filters_or_clip(renderer, renderer_class, monkeypatch):
import copy
if renderer_class is LargeModelRenderer:
@@ -176,8 +176,8 @@ def test_radar_only_reuses_lane_geometry_without_changing_path_filters_or_clip(r
renderer._road_edges = [ModelPoints(raw_points=points + np.array([0, y, 0], dtype=np.float32)) for y in (-6, 6)]
renderer._lane_line_probs = np.array([.6, .9, .8, .5], dtype=np.float32)
renderer._acceleration_x = np.array([.2] * 33, dtype=np.float32)
renderer._acceleration_x_filter = Mock()
renderer._acceleration_x_filter2 = Mock()
monkeypatch.setattr(renderer, "_acceleration_x_filter", Mock(), raising=False)
monkeypatch.setattr(renderer, "_acceleration_x_filter2", Mock(), raising=False)
renderer._experimental_mode = False
renderer._car_space_transform[1, 2] = 480
renderer._path_offset_z = 1.22
@@ -231,7 +231,7 @@ def test_large_render_preserves_lead_cadence_and_lane_invalidation(renderer, mon
radarState=SimpleNamespace(leadOne=None))
monkeypatch.setattr(model_renderer, "ui_state", SimpleNamespace(sm=sm, started_frame=0))
view = object.__new__(LargeModelRenderer)
view._path = SimpleNamespace(raw_points=np.array([[10, 0, 0]], dtype=np.float32))
view._path = ModelPoints(raw_points=np.array([[10, 0, 0]], dtype=np.float32))
view._longitudinal_control = True
view._update_model = Mock()
view._update_raw_points = Mock()
@@ -772,7 +772,7 @@ class IpcAxisContractTests(unittest.TestCase):
logMonoTime = {'aolSafetyWire': now}
freq_ok = {'aolSafetyWire': True}
def update(self, _timeout):
def update(self, *args, **kwargs):
pass
def all_checks(self, _services=None):
@@ -1004,9 +1004,9 @@ class IpcAxisContractTests(unittest.TestCase):
self.assertEqual(configured.alternativeExperience, cp.alternativeExperience)
physical = selected.CS_prev
now = now[0]
receipt_now = now[0]
# Synthetic zero-axis Panda configuration receipt is withheld until actual Card wrote ControlsReady.
native = SafetyState(1, True, now, now + 200_000_000, int(cp.safetyConfigs[0].safetyModel.raw),
native = SafetyState(1, True, receipt_now, receipt_now + 200_000_000, int(cp.safetyConfigs[0].safetyModel.raw),
int(cp.safetyConfigs[0].safetyParam), False, False, False, False, 'panda', 'drive-session')
class SM(dict):
@@ -1014,10 +1014,10 @@ class IpcAxisContractTests(unittest.TestCase):
valid = {'aolSafetyWire': True, 'aolIntentWire': False}
alive = {'aolSafetyWire': True, 'aolIntentWire': False}
seen = {'aolSafetyWire': False, 'aolIntentWire': False}
logMonoTime = {'aolSafetyWire': now}
logMonoTime = {'aolSafetyWire': receipt_now}
freq_ok = {'aolSafetyWire': True}
def update(self, _timeout):
def update(self, *args, **kwargs):
pass
def all_checks(self, _services=None):
@@ -1051,15 +1051,15 @@ class IpcAxisContractTests(unittest.TestCase):
sd.publish_selfdriveState = mock.Mock()
self.enterContext(mock.patch.object(sd, 'update_events', lambda _cs, sd=sd: sd.events.clear()))
message = messaging.new_message('carState')
message.valid, message.logMonoTime, message.carState = True, now, sd.CS_prev
message.valid, message.logMonoTime, message.carState = True, receipt_now, sd.CS_prev
with (mock.patch('openpilot.selfdrive.selfdrived.selfdrived.messaging.recv_one', return_value=message),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', return_value=now),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', return_value=receipt_now),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.REPLAY', False),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.SIMULATION', False),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.VisionIpcClient',
SimpleNamespace(available_streams=lambda *_args, **_kwargs: [])),
mock.patch('openpilot.selfdrive.selfdrived.selfdrived.cloudlog.event')):
for receipt in (None, replace(native, observedMonoTime=now - 250_000_000, validUntilMonoTime=now - 1),
for receipt in (None, replace(native, observedMonoTime=receipt_now - 250_000_000, validUntilMonoTime=receipt_now - 1),
replace(native, axisSessionId='old-session'), replace(native, safetyParam=native.safetyParam + 1)):
sm.seen['aolSafetyWire'] = receipt is not None
sm['aolSafetyWire'] = encode_safety(receipt) if receipt else b''
@@ -1277,7 +1277,7 @@ class UiProcessIntentTests(unittest.TestCase):
if key == 'onroadEvents':
return self.events.to_msg()
raise KeyError(key)
card.sm = SM()
self.enterContext(mock.patch.object(card, 'sm', SM(), create=True))
return card
def observe(self, card, stamp, failed):
@@ -238,29 +238,46 @@ def test_disengaged_notification_priority_duration_and_no_replayed_chime():
def test_actual_selfdrived_alert_path_works_disengaged_and_resets_offroad(monkeypatch):
from openpilot.cereal import messaging
from openpilot.selfdrive.selfdrived.selfdrived import SelfdriveD
from openpilot.selfdrive.selfdrived.events import ET, Events
from openpilot.selfdrive.selfdrived.events import Events
from openpilot.selfdrive.selfdrived.state import StateMachine
from openpilot.starpilot.controllers.mode_actions import SwitchbackCooldown
from openpilot.starpilot.longitudinal.force_stop_alert import HoldAlertState
daemon = SelfdriveD.__new__(SelfdriveD)
daemon.CP, daemon.sm, daemon.AM = vehicle(), Messages(), AlertManager()
daemon.CP, daemon.AM = vehicle(), AlertManager()
daemon.sm = messaging.SubMaster(['deviceState', 'radarTracks'])
daemon.events, daemon.enabled, daemon.personality, daemon.is_metric = Events(), False, 1, True
daemon.state_machine = SimpleNamespace(current_alert_types=[ET.PERMANENT], soft_disable_timer=0)
daemon.state_machine = StateMachine()
daemon.switchback_capable = False
daemon.switchback_cooldown = SimpleNamespace(allow=lambda *args, **kwargs: True)
daemon.force_stop_hold_alert = SimpleNamespace(active=lambda *args, **kwargs: False)
daemon.switchback_cooldown = SwitchbackCooldown()
daemon.force_stop_hold_alert = HoldAlertState()
daemon.radar_recovery_enabled = True
now = [10_000_000_000]
daemon.switchback_setting_ns = now[0] + 100_000_000_000
daemon.switchback_cooldown_ns = 0
monkeypatch.setattr('openpilot.selfdrive.selfdrived.selfdrived.time.monotonic_ns', lambda: now[0])
for tick in range(24):
now[0] += 100_000_000
daemon.sm.frame = tick * 10
daemon.sm.logMonoTime['radarTracks'] = now[0]
daemon.sm.errors['radarFault'] = tick == 1
daemon.sm.valid = {'radarTracks': tick != 1}
def publish(*, fault=False, started=True, update_device=False):
radar = messaging.new_message('radarTracks', valid=not fault, logMonoTime=now[0])
radar.radarTracks.errors.radarFault = fault
messages = [radar.as_reader()]
if update_device:
device = messaging.new_message('deviceState', valid=True, logMonoTime=now[0])
device.deviceState.started, device.deviceState.startedMonoTime = started, 1
messages.append(device.as_reader())
daemon.sm.update_msgs(now[0] / 1e9, messages)
device = messaging.new_message('deviceState', valid=True, logMonoTime=now[0] - 450_000_000)
device.deviceState.started, device.deviceState.startedMonoTime = True, 1
daemon.sm.update_msgs((now[0] - 450_000_000) / 1e9, [device.as_reader()])
for tick in range(48):
now[0] += 50_000_000
publish(fault=tick == 1, update_device=tick % 10 == 0)
daemon.sm.frame = tick * 5
daemon.update_alerts(SimpleNamespace())
assert not daemon.enabled and daemon.AM.current_alert.alert_type == ALERT_TYPE
daemon.sm.device.started = False
publish(started=False, update_device=True)
daemon.sm.frame += 1
daemon.update_alerts(SimpleNamespace())
assert not daemon.enabled and daemon.AM.current_alert.alert_type != ALERT_TYPE
@@ -107,7 +107,7 @@ class SlcFallbackTests(unittest.TestCase):
'SLCPriority1': 'Dashboard', 'SLCPriority2': 'Dashboard'})
self.assertEqual(evaluate(self.step(settings), settings).reason, Reason.OTHER_FALLBACK)
self.assertEqual(evaluate(absent, replace(self.settings, fallback_choice=True)).reason, Reason.INVALID_FALLBACK)
disabled = parse({'SpeedLimitController': False})
disabled = parse({'SpeedLimitController': False, 'ShowSpeedLimits': False})
self.assertEqual(evaluate(self.step(disabled), disabled).reason, Reason.DISABLED)
display = parse({'SpeedLimitController': False, 'ShowSpeedLimits': True})
self.assertEqual(evaluate(self.step(display), display).reason, Reason.DISPLAY_ONLY)
@@ -108,7 +108,7 @@ class TestControllerCruise(unittest.TestCase):
instance.CS_prev = self.cs
vars(instance)['CI'] = SimpleNamespace(CS=SimpleNamespace(), update=lambda packets:self.cs)
instance.car_gps_publisher = card.CarGpsPublisher()
instance.pm = SimpleNamespace(send=Mock())
self.enterContext(patch.object(instance, 'pm', SimpleNamespace(send=Mock()), create=True))
vars(instance)['RI'] = SimpleNamespace(update=lambda packets:None)
directory = tempfile.TemporaryDirectory()
self.addCleanup(directory.cleanup)
@@ -164,7 +164,7 @@ class TestGmSurface(unittest.TestCase):
knobs['ff_gain_left'] = .6
self.assertEqual(surface.knobs['ff_gain_left'], .3)
with self.assertRaises(TypeError):
surface.knobs['ff_gain_left'] = .5
mock.Mock(wraps=setitem)(surface.knobs, 'ff_gain_left', .5)
self.assertEqual(GmSurface.from_document(surface.document()).document(), surface.document())
self.assertGreater(surface.deadband(5.), 0.)
self.assertAlmostEqual(surface.base_threshold(10.), .20)
@@ -27,9 +27,12 @@ class PreviewContractTest(unittest.TestCase):
scripts = "\n".join(path.read_text() for path in (WEB / "js").glob("*.js"))
self.assertEqual(sorted(re.findall(r"\bfetch\s*\(\s*[\"']([^\"']+)", scripts)),
['./api/auth/session', './api/drive-state/action', './api/drive-state/status',
'./api/ui/layout', '/_gateway/devices'])
'/_gateway/devices'])
config = json.loads((WEB / 'data/runtime.json').read_text())
self.assertEqual(config, {'schemaVersion': 1, 'monitor': 'sample'})
self.assertIn('./api/ui/layout', scripts)
self.assertIn('./api/ui/layout/preview', scripts)
self.assertIn('this.fetcher', (WEB / 'js/onroad-layout.js').read_text())
self.assertIn('./api/system/monitor', scripts)
self.assertIn('./data/system-monitor.sample.json', scripts)
self.assertIn('./api/software/status', scripts)
@@ -35,10 +35,12 @@ class TestSavedTorquePage(unittest.TestCase):
rows = {row['label']: row for row in page['rows']}
self.assertEqual(set(rows), {'Steering Controller', 'Turn Assist', 'Prep My Vehicle for Tuning',
'Automatic Steering Learning', 'Lat Accel', 'Lateral acceleration — Reset to Default',
'Friction', 'Friction — Reset to Default'})
'Friction', 'Friction — Reset to Default', 'Pause steering below',
'Steering resume delay', 'Pause only while signaling'})
self.assertIn('Saved vehicle settings', page['subtitle'])
self.assertFalse(any(row['page'] for row in page['rows']))
for label in ('Steering Controller', 'Turn Assist', 'Prep My Vehicle for Tuning', 'Lat Accel', 'Friction'):
for label in ('Steering Controller', 'Turn Assist', 'Prep My Vehicle for Tuning', 'Lat Accel', 'Friction',
'Pause steering below', 'Steering resume delay', 'Pause only while signaling'):
self.assertTrue(rows[label]['available'], label)
self.assertFalse(rows['Automatic Steering Learning']['available'])
index = next(i for i, row in enumerate(page['rows']) if row['label'] == 'Friction')
@@ -73,7 +73,9 @@ def test_no_native_settings_or_favorite_registration(setup):
assert all(row.key != KEY for row in state.rows)
assert all(KEY not in keys for keys in FEATURE_KEYS.values())
galaxy_row = next(row for row in gateway._state('vehicle', gateway.context.sample()).rows if row.key == KEY)
assert not owner.apply(row_change(galaxy_row))
request = row_change(galaxy_row)
assert request is not None
assert not owner.apply(request)
@pytest.mark.parametrize('change', ['vehicle', 'vin', 'saved', 'session', 'preference'])
@@ -167,8 +167,8 @@ class TestGmGeometryRuntime(unittest.TestCase):
expected_delay = delay.to_dict()
expected_delay['lateralDelay']['lateralDelay'] = delayed.lateralDelay.lateralDelay
self.assertEqual(delayed.to_dict(),expected_delay)
self.assertEqual(original.to_bytes(), raw)
self.assertEqual(delay.to_bytes(), lag_raw)
self.assertEqual(original.as_reader().as_builder().to_bytes(), raw)
self.assertEqual(delay.as_reader().as_builder().to_bytes(), lag_raw)
self.assertTrue((learner.kf.x==state).all() and (learner.kf.P==p).all())
self.assertAlmostEqual(lag.initial_lag, cp.steerActuatorDelay+.2)
stored(saved, edit(cp,profiles,'learning','force_auto'))
@@ -186,8 +186,8 @@ class TestGmGeometryRuntime(unittest.TestCase):
owner.last_read_ns = None
self.assertIs(owner.parameters(original),original)
self.assertAlmostEqual(owner.delay(delay).lateralDelay.lateralDelay,cp.steerActuatorDelay+.2)
self.assertEqual(original.to_bytes(),raw)
self.assertEqual(delay.to_bytes(),lag_raw)
self.assertEqual(original.as_reader().as_builder().to_bytes(),raw)
self.assertEqual(delay.as_reader().as_builder().to_bytes(),lag_raw)
def test_ui_guarded_geometry_choices_and_source_staleness(self):
cp = volt_params(CAR.CHEVROLET_VOLT_CAMERA)
@@ -263,6 +263,7 @@ class TestGmGeometryRuntime(unittest.TestCase):
self.assertEqual(row.default_key, f'torque:{field}:mode')
request = row_default(row)
self.assertIsNotNone(request, field)
assert request is not None
self.assertEqual(request.key, row.default_key)
self.assertTrue(owner.apply(request), field)
publisher.last_read_ns = None
@@ -112,7 +112,7 @@ class TestGmManualTorque(unittest.TestCase):
for document in (None,b'{',serialize_document(other_profile)):
with OpenpilotPrefix(),patch.dict(os.environ,{'SIMULATION':'1','REPLAY':'1','TORQUE_REPLAY_RUNTIME':'0'}):
saved=Params()
saved.put('CarParams',cp.to_bytes(),block=True)
saved.put('CarParams',cp.as_reader().as_builder().to_bytes(),block=True)
saved.put('LateralControllerSelection',json.loads(replace_mode(None,cp,mode)),block=True)
baseline=Controls()
saved.put('SteerLatAccel',cp.lateralTuning.torque.latAccelFactor*1.1,block=True)
@@ -147,7 +147,7 @@ class TestGmManualTorque(unittest.TestCase):
gain=gain_basis(cp,lateral)
profiles=replace_gain(profiles,str(cp.carFingerprint),gain,'custom',.7)
stored(saved,profiles)
native_cp=cp.to_bytes()
native_cp=cp.as_reader().as_builder().to_bytes()
owner=None
for legacy in (None,b'0',b'1',b'malformed'):
lateral=LatControlTorque(cp.as_reader(),CarInterface(cp),.01)
@@ -163,7 +163,7 @@ class TestGmManualTorque(unittest.TestCase):
owner=create_gain_owner(saved,cp,lateral)
self.assertIsNotNone(owner)
self.assertEqual(lateral.pid._k_p,[[0.],[.7]])
self.assertEqual(cp.to_bytes(),native_cp)
self.assertEqual(cp.as_reader().as_builder().to_bytes(),native_cp)
raw=serialize_document(profiles)
for invalid in (raw.replace(b'CHEVROLET_MALIBU_ASCM',b'UNKNOWN_GM'),b'{',raw.replace(b'0.8',b'NaN')):
with self.assertRaises((ValueError,UnicodeError)):
@@ -178,7 +178,7 @@ class TestGmManualTorque(unittest.TestCase):
for mode in ControllerMode:
with OpenpilotPrefix(), patch.dict(os.environ, {'SIMULATION':'1', 'REPLAY':'1', 'TORQUE_REPLAY_RUNTIME':'0'}):
saved = Params()
saved.put('CarParams', cp.to_bytes(), block=True)
saved.put('CarParams', cp.as_reader().as_builder().to_bytes(), block=True)
saved.put('LateralControllerSelection', json.loads(replace_mode(None, cp, mode)), block=True)
baseline = Controls()
self.assertIsNone(baseline.torque_host)
@@ -221,7 +221,7 @@ class TestGmManualTorque(unittest.TestCase):
for mode in ControllerMode:
with OpenpilotPrefix(), patch.dict(os.environ, {'SIMULATION':'1','REPLAY':'1','TORQUE_REPLAY_RUNTIME':'0'}):
saved = Params()
saved.put('CarParams', cp.to_bytes(), block=True)
saved.put('CarParams', cp.as_reader().as_builder().to_bytes(), block=True)
choice = json.loads(replace_mode(None, cp, mode))
saved.put('LateralControllerSelection', choice, block=True)
baseline = Controls()
@@ -290,7 +290,7 @@ class TestGmManualTorque(unittest.TestCase):
cp = volt_params(CAR.CHEVROLET_VOLT_CAMERA)
with OpenpilotPrefix():
saved=Params()
saved.put('CarParams',cp.to_bytes(),block=True)
saved.put('CarParams',cp.as_reader().as_builder().to_bytes(),block=True)
saved.put_bool('AdvancedLateralTune',True,block=True)
owner=FeatureSettingsOwner(saved,lambda _:True,vehicle_fingerprint=lambda:cp.carFingerprint,vehicle_params=lambda:cp)
def rows():
@@ -4,6 +4,7 @@ from pathlib import Path
import tempfile
import unittest
from opendbc.car.car_helpers import interfaces
from openpilot.starpilot.lateral.torque_runtime import read_settings
from openpilot.starpilot.lateral.torque_tuning import TorqueSource, TorqueTuning
from openpilot.starpilot.lateral.torque_settings import DOCUMENT_KEY, parse_document
@@ -31,7 +32,14 @@ class SavedParams:
class Owner:
def __init__(self, params):
self.params = params
self.capability = ('HYUNDAI_IONIQ_6', '', '', '', '', 3.0, 0.0, 0.1)
self.cp = interfaces['HYUNDAI_IONIQ_6'].get_non_essential_params('HYUNDAI_IONIQ_6')
tune = self.cp.lateralTuning.torque
tune.latAccelFactor, tune.latAccelOffset, tune.friction = 3.0, 0.0, 0.1
self.capability = (str(self.cp.carFingerprint), str(self.cp.brand), str(self.cp.steerControlType),
str(self.cp.lateralTuning.which()), bool(self.cp.dashcamOnly),
float(tune.latAccelFactor), float(tune.latAccelOffset), float(tune.friction))
def vehicle_params(self):
return self.cp
def authority(self, group):
return True
def vehicle_fingerprint(self):
@@ -51,8 +59,8 @@ class TestInferredTorque(unittest.TestCase):
self.directory = tempfile.TemporaryDirectory()
self.addCleanup(self.directory.cleanup)
self.params = SavedParams(self.directory.name)
self.vehicle = TorqueTuning(TorqueSource.VEHICLE, 'HYUNDAI_IONIQ_6', 3.0, 0.0, 0.1)
self.owner = Owner(self.params)
self.vehicle = TorqueTuning(TorqueSource.VEHICLE, self.owner.capability[0], *self.owner.capability[5:8])
self.editor = TorqueFeature(self.owner)
def rows(self):
valid, rows = self.editor.rows(self.owner.capability, True)
@@ -33,7 +33,9 @@ class TestNativeInferredTorque(unittest.TestCase):
owner.params = params
owner.authority = lambda group: True
owner.vehicle_fingerprint = lambda: stock.vehicle
cap = (stock.vehicle, '', '', '', '', stock.lat_accel_factor, stock.lat_accel_offset, stock.friction)
owner.vehicle_params = lambda: cp
cap = (stock.vehicle, str(cp.brand), str(cp.steerControlType), str(cp.lateralTuning.which()),
bool(cp.dashcamOnly), stock.lat_accel_factor, stock.lat_accel_offset, stock.friction)
owner._capability = lambda group: cap
def raw(key):
path = Path(params.get_param_path(key))
@@ -313,7 +313,7 @@ class GlobalPowertrainPresetTests(unittest.TestCase):
from opendbc.car.gm.values import CAR as GmCAR
from openpilot.starpilot.longitudinal.profile_runtime import read_global_powertrain_preset, GlobalPowertrainPreset
truck = params(GmCAR.CHEVROLET_SILVERADO, alpha=True)
before = truck.to_bytes()
before = truck.as_reader().as_builder().to_bytes()
self.assertIsNone(read_global_powertrain_preset(self.params, truck))
self.params.put_bool('EVTuning', True, block=True)
self.assertEqual(read_global_powertrain_preset(self.params, truck), GlobalPowertrainPreset(True, False))
@@ -327,7 +327,7 @@ class GlobalPowertrainPresetTests(unittest.TestCase):
stock = params(GmCAR.CHEVROLET_SILVERADO, alpha=True, release=True)
self.assertIsNone(read_global_powertrain_preset(self.params, stock))
self.assertIsNone(read_global_powertrain_preset(self.params, self.cp))
self.assertEqual(truck.to_bytes(), before)
self.assertEqual(truck.as_reader().as_builder().to_bytes(), before)
path = Path(self.params.get_param_path('EVTuning'))
path.write_bytes(b'1\n')
self.assertIsNone(read_global_powertrain_preset(self.params, truck))
@@ -235,8 +235,10 @@ def test_native_ui_keeps_loading_during_enumeration_then_reports_failure(previou
chestnut_enum = next(node for node in body if isinstance(node, ast.ClassDef) and node.name == "ChestnutState")
ui_class = next(node for node in body if isinstance(node, ast.ClassDef) and node.name == "UIState")
update = next(node for node in ui_class.body if isinstance(node, ast.FunctionDef) and node.name == "_update_chestnut_state")
display_time = next(node for node in ui_class.body if isinstance(node, ast.FunctionDef) and node.name == "display_time_ns")
env: dict = {"Enum": Enum, "time": SimpleNamespace(monotonic_ns=lambda: 1_000_000_000)}
execute([chestnut_enum, update], env)
execute([chestnut_enum, update, display_time], env)
env["UIState"] = SimpleNamespace(display_time_ns=env["display_time_ns"])
states = env["ChestnutState"]
class Messages(dict):
@@ -338,9 +340,11 @@ def test_fallback_releases_only_chestnut_dependencies(opened):
amd, unrelated = devices["AMD"], object()
for device in devices.values():
device.pending = {amd: "gpu fence", unrelated: "other fence"}
env = {"Device": devices}
env: dict[str, object] = {"Device": devices}
execute([node], env)
env["drop_chestnut"]()
drop = env["drop_chestnut"]
assert callable(drop)
drop()
for name in opened:
assert unrelated in devices[name].pending
assert (amd in devices[name].pending) is ("AMD" not in opened)
@@ -757,6 +757,7 @@ class RuntimeReplayTests(unittest.TestCase):
replay.sm.logMonoTime['slcState'] = recorded_ns + 20_000_000
replay.sm.logMonoTime['carControl'] = recorded_ns + 30_000_000
packet.logMonoTime = recorded_ns + 30_000_000
packet.clear_write_flag()
with (mock.patch.object(card, 'REPLAY', True),
mock.patch.object(card.time, 'monotonic_ns', return_value=host_ns),
mock.patch.object(card.messaging, 'drain_sock_raw', return_value=[packet.to_bytes()]),
@@ -1063,7 +1064,7 @@ class RuntimeReplayTests(unittest.TestCase):
self.sm.advance(9_100_000_000)
stale = runtime.step(self.sm, self.cp, now_ns=9_100_000_000)
self.assertEqual(runtime._observations(self.sm, self.cp, 9_100_000_000)[Source.DASHBOARD].kind, ObservationKind.STALE)
self.assertEqual(stale.message.slcState.observationKind, 'unknown') # unimplemented map source remains unknown
self.assertEqual(stale.message.slcState.observationKind, 'stale') # both configured slots use the expired dashboard
self.assertFalse(stale.message.slcState.hasCeiling)
planner.update(self.sm, cruise_ceiling=stale.result.ceiling)
self.assertEqual(planner.last_cruise_ceiling_status, 'absent')
@@ -41,7 +41,7 @@ class TestBoltCcControlsLow(unittest.TestCase):
self.assertEqual(cp.alternativeExperience, 0)
cp.fingerprintSource = structs.CarParams.FingerprintSource.can
cp.carFw = []
params.put('CarParams', cp.to_bytes(), block=True)
params.put('CarParams', cp.as_reader().as_builder().to_bytes(), block=True)
controls = Controls()
self.assertEqual(controls.CP.to_dict(), cp.to_dict())
self.assertEqual(controls.CP.safetyConfigs[0].safetyParam, 0x1CD if stock_acc else 0x9D if pedal else 0xC121 if removed else 0xC120)
@@ -54,9 +54,15 @@ class TestBoltCcControlsLow(unittest.TestCase):
drive_pedal_frames = low_pedal_frames = stock_withdrawals = 0
last_feeds = {}
internal_frames = 0
internal_status_frames = 0
from opendbc.safety.tests.test_gm_bolt_pedal import TestGmBoltPedalSafety
recorder = TestGmBoltPedalSafety()
recorder.safety = libsafety_py.libsafety
if pedal and not stock_acc:
# An arbitrary host status cannot acquire the physical RX-owned feed.
libsafety_py.libsafety.reset_recorded_can()
self.assertFalse(native('tx', (0x3D1, bytes(8), 0), 900_000))
self.assertEqual(recorder.recorded(), [])
for tick in range(360):
now = 1_000_000_000 + tick * 10_000_000
gas = 120 <= tick < 150
@@ -89,6 +95,7 @@ class TestBoltCcControlsLow(unittest.TestCase):
self.assertEqual(bus, 0)
self.assertEqual(payload, last_feeds[address])
internal_frames += 1
internal_status_frames += int(address == 0x3D1)
libsafety_py.libsafety.safety_tick()
cs = ci.update([(now, frames)])
self.assertTrue(cs.canValid)
@@ -161,7 +168,9 @@ class TestBoltCcControlsLow(unittest.TestCase):
_, commands = ci.apply(cc.as_reader(), now + 1_000_000)
for frame in commands:
allowed = native('tx', frame, (now + 1_000_000) // 1000)
if pedal and frame[0] in (0xBD, 0x1F5):
if pedal and frame[0] in (0xBD, 0x1F5, 0x3D1):
# The host schedules these exact feeds; matching physical RX emits
# them internally. They never grant ordinary host TX permission.
self.assertFalse(allowed)
last_feeds[frame[0]] = frame[1]
else:
@@ -213,6 +222,7 @@ class TestBoltCcControlsLow(unittest.TestCase):
if pedal:
if not stock_acc:
self.assertGreater(internal_frames, 0)
self.assertGreater(internal_status_frames, 0)
for index, button in enumerate((3, 1)):
rearm = [frame for frame in frames if frame[0] not in (0x1E1, 0x201, 0xC9)]
rearm.append(packer.make_can_msg("ECMEngineStatus", 0, {"CruiseMainOn": 1}))
@@ -10,7 +10,7 @@ from opendbc.car.gm.values import CAR
from opendbc.car.gm.tests.test_camera_acc_pedal import CAMERA_IDS, fingerprint as camera_pedal_fingerprint
from openpilot.common.params import Params
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.starpilot.feature_runtime import enabled, slc_runtime_settings
from openpilot.starpilot.feature_runtime import enabled, slc_runtime_settings, vision_control_enabled
from openpilot.starpilot.longitudinal.profile_document import default_personality_profiles, profile_document
from openpilot.starpilot.vehicle_preferences import VehicleStartupPreferences
@@ -86,15 +86,17 @@ class TestGmFeatureRuntime(unittest.TestCase):
self.assertTrue(longitudinal_supported(cp))
for feature in ('conditional', 'curve', 'slc', 'profile'):
self.assertTrue(enabled(params, cp, feature, {}), feature)
self.assertFalse(enabled(params, cp, 'vision', {}))
self.assertTrue(enabled(params, cp, 'vision', {}))
for identity, kwargs in STOCK_CASES:
with self.subTest(identity=identity, kwargs=kwargs):
cp = configured(params, identity, **kwargs)
self.assertFalse(cp.openpilotLongitudinalControl, (identity, kwargs, cp.safetyConfigs, cp.flags))
self.assertFalse(longitudinal_supported(cp))
self.assertTrue(display_supported(cp), (identity, kwargs, cp.safetyConfigs, cp.flags, cp.dashcamOnly))
for feature in ('conditional', 'curve', 'profile', 'vision'):
for feature in ('conditional', 'curve', 'profile'):
self.assertFalse(enabled(params, cp, feature, {}), feature)
self.assertTrue(enabled(params, cp, 'vision', {}))
self.assertFalse(vision_control_enabled(params, cp))
self.assertTrue(enabled(params, cp, 'slc', {}))
settings = slc_runtime_settings(params, cp, {})
self.assertTrue(settings.display) # Saved controller-on also requests display.
@@ -109,8 +111,14 @@ class TestGmFeatureRuntime(unittest.TestCase):
cp = configured(params, CAR.CHEVROLET_VOLT_CC)
self.assertTrue(enabled(params, cp, 'conditional', {})) # Readable factory CEM.
self.assertTrue(enabled(params, cp, 'profile', {})) # Fresh Comfort is the readable missing-document default.
for feature in ('curve', 'slc', 'vision'):
self.assertFalse(enabled(params, cp, feature, {}))
self.assertFalse(enabled(params, cp, 'curve', {}))
self.assertTrue(enabled(params, cp, 'slc', {}))
self.assertTrue(enabled(params, cp, 'vision', {}))
settings = slc_runtime_settings(params, cp, {})
self.assertTrue(settings.display)
self.assertFalse(settings.enabled)
self.assertTrue(settings.acceptance.display_only)
self.assertFalse(vision_control_enabled(params, cp))
params.put('ConditionalModeConfig', json.loads(encode_preferences(SavedPreferences(mode=ModeChoice.STOCK))), block=True)
self.assertTrue(enabled(params, cp, 'conditional', {})) # Owner retains Stock session lifecycle.
params.put_bool('SafeMode', True, block=True)
@@ -339,7 +339,9 @@ class TestGMPitchStartupAndSettings(unittest.TestCase):
with patch.object(card, 'Params', return_value=self.saved), \
patch.object(card, 'feature_requested', return_value=False), \
patch.object(card.messaging, 'sub_sock', return_value=Mock()), \
patch.object(card.messaging, 'SubMaster', return_value=Mock()), \
patch.object(card.messaging, 'SubMaster', side_effect=lambda *_args, **kwargs: SimpleNamespace(
ignore_alive=list(kwargs.get('ignore_alive', [])), ignore_valid=list(kwargs.get('ignore_valid', [])),
ignore_average_freq=list(kwargs.get('ignore_average_freq', [])))), \
patch.object(card.messaging, 'PubMaster', return_value=SimpleNamespace(sock={'sendcan': Mock()})), \
patch.object(card, 'Ratekeeper', return_value=Mock()), \
patch.object(card, 'get_cache', return_value=None), patch.object(card, 'put_cache'), \
@@ -156,6 +156,7 @@ class TestGmVisionSource(unittest.TestCase):
def test_stock_ui_can_select_display_source_with_stale_request_guard(self):
with OpenpilotPrefix():
params = Params()
params.put('SLCPriority1', 'Dashboard', block=True)
cp = configured(params, CAR.CHEVROLET_VOLT_CC, disable=True)
owner = FeatureSettingsOwner(params, authority=lambda group: group == 'preferences',
vehicle_fingerprint=lambda: cp.carFingerprint, vehicle_params=lambda: cp,
@@ -45,7 +45,9 @@ class TestVehicleSelection(unittest.TestCase):
self.assertEqual(len(selected), 1)
self.assertTrue(self.owner.choose(None, selected[0].platform).committed)
self.assertEqual(startup_candidate(self.params), CAR.KIA_K4_2025)
cp = CarInterface.get_params(startup_candidate(self.params), gen_empty_fingerprint(), [], False, False, False)
candidate = startup_candidate(self.params)
assert candidate is not None
cp = CarInterface.get_params(candidate, gen_empty_fingerprint(), [], False, False, False)
self.assertEqual(cp.carFingerprint, CAR.KIA_K4_2025)
self.assertFalse(cp.openpilotLongitudinalControl)
@@ -33,6 +33,7 @@ class CompactVisualsSlcTests(unittest.TestCase):
def test_sign_choice_stays_editable_with_system_long_off(self):
with tempfile.TemporaryDirectory() as path:
params = Params(path)
params.put_bool("ShowSpeedLimits", False, block=True)
appearance = AppearanceOwner(params, lambda: True)
feature = FeatureSettingsOwner(params, lambda group: group == "preferences", vehicle_fingerprint=lambda: None)
@@ -70,6 +71,7 @@ class CompactVisualsSlcTests(unittest.TestCase):
def test_original_speed_limit_rows_refresh_with_parent_and_leave_control_off(self):
with tempfile.TemporaryDirectory() as path:
params = Params(path)
params.put_bool("ShowSpeedLimits", False, block=True)
parked = True
appearance = AppearanceOwner(params, lambda: parked)
feature = FeatureSettingsOwner(params, lambda group: parked, vehicle_fingerprint=lambda: "TEST CAR")
@@ -26,16 +26,16 @@ class TestCalibrationReset(unittest.TestCase):
self.ui = NS(params=self.params, engaged=False, ignition=False, is_offroad=lambda: True,
prime_state=NS(is_paired=lambda: False))
self.enterContext(patch.object(device_layout, "ui_state", self.ui))
self.enterContext(patch.object(device_layout, "Params", return_value=self.params))
self.enterContext(patch.object(device_layout, "calibration_reset", self.reset))
self.enterContext(patch.object(device, "ui_state", self.ui))
self.enterContext(patch.object(device, "calibration_reset", self.reset))
self.enterContext(patch.object(gui_app, "font", return_value=rl.Font()))
self.enterContext(patch.object(gui_app, "texture", return_value=NS(width=64, height=64)))
self.pushed = self.enterContext(patch.object(gui_app, "push_widget"))
self.panel = device_layout.DeviceLayoutMici()
with patch.object(device_layout, "Params", return_value=self.params):
self.panel = device_layout.DeviceLayoutMici()
self.button = next(item for item in self.panel._scroller._items
if hasattr(item, "get_text") and item.get_text() == "reset calibration")
if isinstance(item, device_layout.EngagedConfirmationButton) and item.get_text() == "reset calibration")
def wait_for_reset(self):
deadline = time.monotonic() + 3
@@ -44,6 +44,7 @@ class TestCalibrationReset(unittest.TestCase):
self.assertFalse(self.reset.pending, "Calibration reset did not finish")
def confirm(self):
assert self.button._click_callback is not None
self.button._click_callback()
self.pushed.call_args.args[0]._confirm_callback()
@@ -75,6 +76,7 @@ class TestCalibrationReset(unittest.TestCase):
def test_confirmation_rechecks_engagement(self):
self.params.put("CalibrationParams", b"saved", block=True)
assert self.button._click_callback is not None
self.button._click_callback()
confirm = self.pushed.call_args.args[0]._confirm_callback
self.ui.engaged = True
@@ -50,7 +50,8 @@ class DisplaySettingsTests(unittest.TestCase):
session.display_owner = self.owner
session.power_owner = PowerOwner(self.params, lambda: self.parked)
session.map_snapshot = Mock(return_value=FeatureSettingsState())
session.drive_state = NS(snapshot=lambda: {"mode": "auto", "revision": None, "available": False, "effective": None})
self.enterContext(patch.object(session, 'drive_state',
NS(snapshot=lambda: {"mode": "auto", "revision": None, "available": False, "effective": None}), create=True))
session.display_scroll, session.display_edit_key = 0, None
session.sidebar_expanded = expanded
session._snapshot_cache = None
@@ -119,7 +119,7 @@ class FeatureNavigationTests(unittest.TestCase):
with self.subTest(profile=profile):
session = runtime_app.StarShellSession.__new__(runtime_app.StarShellSession)
session._mode, session.profile = runtime_app.ShellMode.SETTINGS, profile
session._unavailable = lambda _reason: None
self.enterContext(patch.object(session, '_unavailable', lambda _reason: None))
session.feature_owner = FeatureSettingsOwner(params, session._feature_authority,
vehicle_fingerprint=lambda: vehicle.carFingerprint, vehicle_params=lambda: vehicle)
with patch.object(runtime_app, 'ui_state', NS(CP=vehicle)):
@@ -55,12 +55,12 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.fingerprint = cp.carFingerprint
self.owner.vehicle_params = lambda: cp
self.owner.configuration_vehicle = lambda: True
original = cp.to_bytes()
original = cp.as_reader().as_builder().to_bytes()
row = self._row('aol', 'AlwaysOnLateral')
self.assertTrue(row.available)
self.assertTrue(self.owner.apply(required_change(row)))
self.assertFalse(qualified_gm(cp))
self.assertEqual(cp.to_bytes(), original)
self.assertEqual(cp.as_reader().as_builder().to_bytes(), original)
self.owner.configuration_vehicle = lambda: False
cp = malibu_hybrid_params(pedal=True, removed=True)
row = self._row('vehicle', 'GMPedalLongitudinal')
@@ -78,7 +78,7 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
from openpilot.starpilot.aol.vehicle import policy_for
self.params.put_bool('AlwaysOnLateral', False, block=True)
cp = CarInterface.get_params(CAR.TOYOTA_PRIUS_RETROFIT, gen_empty_fingerprint(), [], False, False, False)
original = cp.to_bytes()
original = cp.as_reader().as_builder().to_bytes()
self.fingerprint = cp.carFingerprint
self.owner.vehicle_params = lambda: cp
self.owner.configuration_vehicle = lambda: True
@@ -87,7 +87,7 @@ class FeatureSettingsOwnerTests(unittest.TestCase):
self.assertIn('next startup', row.reason)
self.assertTrue(self.owner.apply(required_change(row)))
self.assertTrue(self.params.get_bool('AlwaysOnLateral'))
self.assertEqual(cp.to_bytes(), original)
self.assertEqual(cp.as_reader().as_builder().to_bytes(), original)
self.assertFalse(policy_for(cp).runtime_supported)
self.assertFalse(policy_for(cp).intent_supported)
self.owner.configuration_vehicle = lambda: False
@@ -750,7 +750,7 @@ class TruckTuningFeatureTests(unittest.TestCase):
path.write_bytes(b'1\n')
row = self._row('vehicle', 'TruckTuning')
request = row_default(row)
self.assertIsNotNone(request)
assert request is not None
self.assertFalse(self.owner.apply(replace(request, value='On')))
self.allowed = False
self.assertFalse(self.owner.apply(request))
@@ -792,7 +792,7 @@ class EvPresetFeatureTests(unittest.TestCase):
path = Path(self.params.get_param_path('EVTuning'))
path.write_bytes(b'1\n')
request = row_default(self._row('vehicle','EVTuning'))
self.assertIsNotNone(request)
assert request is not None
self.assertFalse(self.owner.apply(replace(request,value='On')))
self.allowed = False
self.assertFalse(self.owner.apply(request))
@@ -177,7 +177,7 @@ class TestOnroadAxes(unittest.TestCase):
state = self.state(True, True)
for profile in (Profile.LARGE, Profile.COMPACT):
view = onroad.OnroadView.__new__(onroad.OnroadView)
view.fonts = NS(profile=profile, draw=Mock())
self.enterContext(patch.object(view, 'fonts', NS(profile=profile, draw=Mock()), create=True))
view._driving_mode(state)
view.fonts.draw.assert_not_called()
state.customization['layouts'][profile][MODE_WIDGET]['enabled'] = True
@@ -23,7 +23,8 @@ from openpilot.starpilot.ui import clip
class DrawerNativeTest(unittest.TestCase):
def test_qualified_source_rows_roundtrip_and_do_not_change_selection(self):
sm, cp = replay_inputs()
runtime = Runtime(parse({'SpeedLimitController': True}), session_id='drawer-native')
runtime = Runtime(parse({'SpeedLimitController': True, 'SLCPriority1': 'Dashboard', 'SLCPriority2': 'Map Data'}),
session_id='drawer-native')
out = runtime.step(sm, cp, now_ns=START).message
with log.Event.from_bytes(out.to_bytes()) as decoded:
obs = speed_limit_from_message(decoded.slcState)
@@ -112,7 +113,7 @@ class DrawerNativeTest(unittest.TestCase):
def test_disabled_and_invalid_rows_have_no_value_or_ui_authority(self):
sm, cp = replay_inputs()
runtime = Runtime(parse({}), session_id='drawer-disabled')
runtime = Runtime(parse({'ShowSpeedLimits': False}), session_id='drawer-disabled')
message = runtime.step(sm, cp, now_ns=START).message.slcState
self.assertFalse(message.enabled)
self.assertTrue(all(not row.enabled for row in message.sourceReadings))