tests: migrate sunnypilot tests to unittest and remove pytest

This commit is contained in:
Jason Wen
2026-08-13 17:47:37 -04:00
parent 9fb9b73620
commit a3012de889
38 changed files with 621 additions and 839 deletions
-101
View File
@@ -1,101 +0,0 @@
# TODO-SP: upstream migrated from pytest to unittest with a custom test runner (tools/test_runner.py).
# Once sunnypilot test files are converted to unittest, this conftest and the pytest deps can be removed.
import contextlib
import gc
import os
import pytest
from openpilot.common.prefix import OpenpilotPrefix
from openpilot.system.manager import manager
from openpilot.common.hardware import COMMA_HARDWARE, HARDWARE
# these are heavy CI-only tests, invoked explicitly in .github/workflows/tests.yaml
collect_ignore = [
"openpilot/selfdrive/test/process_replay/test_processes.py",
"openpilot/selfdrive/test/process_replay/test_regen.py",
"openpilot/tools/sim/",
# tinygrad JIT has process-global state. Other test files import modeld → tinygrad,
# which corrupts JIT captures for test_warp.py in the same process. Run separately in CI.
"openpilot/sunnypilot/modeld_v2/tests/test_warp.py",
]
def pytest_sessionstart(session):
# TODO: fix tests and enable test order randomization
if session.config.pluginmanager.hasplugin('randomly'):
session.config.option.randomly_reorganize = False
@pytest.hookimpl(hookwrapper=True, trylast=True)
def pytest_runtest_call(item):
# ensure we run as a hook after capturemanager's
if item.get_closest_marker("nocapture") is not None:
capmanager = item.config.pluginmanager.getplugin("capturemanager")
with capmanager.global_and_fixture_disabled():
yield
else:
yield
@contextlib.contextmanager
def clean_env():
starting_env = dict(os.environ)
yield
os.environ.clear()
os.environ.update(starting_env)
@pytest.fixture(scope="function", autouse=True)
def openpilot_function_fixture(request):
with clean_env():
# setup a clean environment for each test
with OpenpilotPrefix(shared_download_cache=request.node.get_closest_marker("shared_download_cache") is not None) as prefix:
prefix = os.environ["OPENPILOT_PREFIX"]
yield
# ensure the test doesn't change the prefix
assert "OPENPILOT_PREFIX" in os.environ and prefix == os.environ["OPENPILOT_PREFIX"]
# cleanup any started processes
manager.manager_cleanup()
# some processes disable gc for performance, re-enable here
if not gc.isenabled():
gc.enable()
gc.collect()
# If you use setUpClass, the environment variables won't be cleared properly,
# so we need to hook both the function and class pytest fixtures
@pytest.fixture(scope="class", autouse=True)
def openpilot_class_fixture():
with clean_env():
yield
@pytest.fixture(scope="function")
def tici_setup_fixture(request, openpilot_function_fixture):
"""Ensure a consistent state for tests on-device. Needs the openpilot function fixture to run first."""
if 'skip_tici_setup' in request.keywords:
return
HARDWARE.initialize_hardware()
HARDWARE.set_power_save(False)
os.system("pkill -9 -f athena")
@pytest.hookimpl(tryfirst=True)
def pytest_collection_modifyitems(config, items):
skipper = pytest.mark.skip(reason="Skipping tici test on PC")
for item in items:
if "tici" in item.keywords:
if not COMMA_HARDWARE:
item.add_marker(skipper)
else:
item.fixturenames.append('tici_setup_fixture')
if "xdist_group_class_property" in item.keywords:
class_property_name = item.get_closest_marker('xdist_group_class_property').args[0]
class_property_value = getattr(item.cls, class_property_name)
item.add_marker(pytest.mark.xdist_group(class_property_value))
+1
View File
@@ -83,6 +83,7 @@ def parameterized_class(attrs, input_list=None):
new_cls = type(name, (cls,), dict(params))
new_cls.__module__ = cls.__module__
new_cls.__unittest_skip__ = False
new_cls.__unittest_skip_why__ = "" # else inherited from the base and the collector drops it
globs[name] = new_cls
# Don't collect the un-parametrised base.
cls.__unittest_skip__ = True
@@ -1,4 +1,4 @@
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.common.test import OpenpilotTestCase
from openpilot.cereal import log
@@ -251,28 +251,30 @@ def _build_sm(selfdrive_enabled, lat_active, steering_pressed, gas_pressed):
}
@pytest.mark.parametrize("selfdrive_enabled, lat_active, steering, gas, expected_op_engaged, expected_driver_engaged", [
(False, False, False, False, False, False), # disabled
(True, False, False, False, True, False), # OP enabled
(False, True, False, False, True, False), # MADS lat-only
(True, True, False, False, True, False), # both active
(False, True, False, True, True, False), # MADS lat-only + gas
(True, True, False, True, True, True), # full op + gas: override
(False, True, True, False, True, True), # MADS lat-only + wheel touch: override
])
def test_run_step_engagement(selfdrive_enabled, lat_active, steering, gas,
expected_op_engaged, expected_driver_engaged):
sm = _build_sm(selfdrive_enabled, lat_active, steering, gas)
dm = DriverMonitoring()
captured = {}
orig = dm._update_events
class TestRunStepEngagement(OpenpilotTestCase):
@parameterized.expand([
(False, False, False, False, False, False), # disabled
(True, False, False, False, True, False), # OP enabled
(False, True, False, False, True, False), # MADS lat-only
(True, True, False, False, True, False), # both active
(False, True, False, True, True, False), # MADS lat-only + gas
(True, True, False, True, True, True), # full op + gas: override
(False, True, True, False, True, True), # MADS lat-only + wheel touch: override
], names=["selfdrive_enabled", "lat_active", "steering", "gas",
"expected_op_engaged", "expected_driver_engaged"])
def test_run_step_engagement(self, selfdrive_enabled, lat_active, steering, gas,
expected_op_engaged, expected_driver_engaged):
sm = _build_sm(selfdrive_enabled, lat_active, steering, gas)
dm = DriverMonitoring()
captured = {}
orig = dm._update_events
def spy(driver_engaged, op_engaged, lowspeed, wrong_gear):
captured['driver_engaged'] = driver_engaged
captured['op_engaged'] = op_engaged
return orig(driver_engaged, op_engaged, lowspeed, wrong_gear)
def spy(driver_engaged, op_engaged, lowspeed, wrong_gear):
captured['driver_engaged'] = driver_engaged
captured['op_engaged'] = op_engaged
return orig(driver_engaged, op_engaged, lowspeed, wrong_gear)
object.__setattr__(dm, '_update_events', spy)
dm.run_step(sm, demo=False)
assert captured['op_engaged'] == expected_op_engaged
assert captured['driver_engaged'] == expected_driver_engaged
object.__setattr__(dm, '_update_events', spy)
dm.run_step(sm, demo=False)
assert captured['op_engaged'] == expected_op_engaged
assert captured['driver_engaged'] == expected_driver_engaged
@@ -5,14 +5,13 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pytest
from pytest_mock import MockerFixture
from openpilot.cereal import custom
from openpilot.common.realtime import DT_CTRL
from openpilot.sunnypilot.mads.state import StateMachine, SOFT_DISABLE_TIME
from openpilot.selfdrive.selfdrived.events import ET, NormalPermanentAlert, Events
from openpilot.sunnypilot.selfdrive.selfdrived.events import EventsSP, EVENTS_SP
from openpilot.common.test import OpenpilotTestCase
State = custom.ModularAssistiveDrivingSystem.ModularAssistiveDrivingSystemState
EventNameSP = custom.OnroadEventSP.EventName
@@ -34,16 +33,16 @@ def make_event(event_types):
class MockMADS:
def __init__(self, mocker: MockerFixture):
def __init__(self, mocker):
self.selfdrive = mocker.MagicMock()
self.selfdrive.state_machine = mocker.MagicMock()
self.selfdrive.events = Events()
self.selfdrive.events_sp = EventsSP()
class TestMADSStateMachine:
@pytest.fixture(autouse=True)
def setup_method(self, mocker: MockerFixture):
class TestMADSStateMachine(OpenpilotTestCase):
def setup_method(self):
mocker = self._fixture("mocker")
self.mads = MockMADS(mocker)
self.state_machine = StateMachine(self.mads)
self.events = self.mads.selfdrive.events
@@ -5,7 +5,7 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.cereal import log, custom
from opendbc.car import structs
@@ -14,6 +14,7 @@ from openpilot.sunnypilot.selfdrive.selfdrived.events import EventsSP
from openpilot.sunnypilot.mads.helpers import MadsSteeringModeOnBrake, read_steering_mode_param
from openpilot.sunnypilot.mads.mads import ModularAssistiveDrivingSystem
from opendbc.sunnypilot.car.tesla.values import MadsScreenButtonType, TeslaFlagsSP
from openpilot.common.test import OpenpilotTestCase
State = custom.ModularAssistiveDrivingSystem.ModularAssistiveDrivingSystemState
EventName = log.OnroadEvent.EventName
@@ -80,8 +81,8 @@ def run_frames(mads, sd, cs, n=1):
# should_silent_lkas_enable across all modes
class TestShouldSilentLkasEnable:
@pytest.mark.parametrize("brake,regen", [(True, False), (False, True)])
class TestShouldSilentLkasEnable(OpenpilotTestCase):
@parameterized.expand([(True, False), (False, True)], names=["brake", "regen"])
def test_pause_blocks_reenable_on_braking_at_standstill(self, mocker, brake, regen):
mads, _ = make_mads(mocker, MadsSteeringModeOnBrake.PAUSE)
cs = make_car_state(brake_pressed=brake, regen_braking=regen, standstill=True)
@@ -105,7 +106,7 @@ class TestShouldSilentLkasEnable:
# pause
class TestPauseMode:
class TestPauseMode(OpenpilotTestCase):
def test_stays_paused_at_standstill_brake_held(self, mocker):
mads, sd = make_mads(mocker, MadsSteeringModeOnBrake.PAUSE)
mads.state_machine.state = State.enabled
@@ -150,7 +151,7 @@ class TestPauseMode:
# disengage
class TestDisengageMode:
class TestDisengageMode(OpenpilotTestCase):
def test_brake_while_enabled_disables(self, mocker):
mads, sd = make_mads(mocker, MadsSteeringModeOnBrake.DISENGAGE)
mads.state_machine.state = State.enabled
@@ -174,7 +175,7 @@ class TestDisengageMode:
# remain active
class TestRemainActiveMode:
class TestRemainActiveMode(OpenpilotTestCase):
def test_brake_does_not_pause_or_disable(self, mocker):
mads, sd = make_mads(mocker, MadsSteeringModeOnBrake.REMAIN_ACTIVE)
mads.state_machine.state = State.enabled
@@ -188,7 +189,7 @@ class TestRemainActiveMode:
# lateral mismatch counter
class TestLateralMismatchCounter:
class TestLateralMismatchCounter(OpenpilotTestCase):
def test_no_accumulation_while_paused(self, mocker):
mads, sd = make_mads(mocker, MadsSteeringModeOnBrake.PAUSE)
mads.state_machine.state = State.paused
@@ -212,7 +213,7 @@ class TestLateralMismatchCounter:
# brand restrictions
class TestBrandSteeringModeRestrictions:
class TestBrandSteeringModeRestrictions(OpenpilotTestCase):
def test_rivian_forced_to_disengage(self, mocker):
CP = structs.CarParams()
CP.brand = "rivian"
@@ -229,9 +230,9 @@ class TestBrandSteeringModeRestrictions:
params = mocker.MagicMock()
assert read_steering_mode_param(CP, CP_SP, params) == MadsSteeringModeOnBrake.DISENGAGE
@pytest.mark.parametrize("screen_button", [MadsScreenButtonType.THREE_FINGER,
@parameterized.expand([MadsScreenButtonType.THREE_FINGER,
MadsScreenButtonType.FOUR_FINGER,
MadsScreenButtonType.FIVE_FINGER])
MadsScreenButtonType.FIVE_FINGER], names=["screen_button"])
def test_tesla_with_vehicle_bus_uses_param(self, mocker, screen_button):
CP = structs.CarParams()
CP.brand = "tesla"
@@ -250,7 +251,7 @@ class TestBrandSteeringModeRestrictions:
"MadsSteeringMode": MadsSteeringModeOnBrake.REMAIN_ACTIVE})
assert read_steering_mode_param(CP, CP_SP, params) == MadsSteeringModeOnBrake.DISENGAGE
@pytest.mark.parametrize("brand", ["hyundai", "toyota", "honda", "gm"])
@parameterized.expand(["hyundai", "toyota", "honda", "gm"], names=["brand"])
def test_other_brands_use_param(self, mocker, brand):
CP = structs.CarParams()
CP.brand = brand
@@ -7,9 +7,10 @@ See the LICENSE.md file in the root directory for more details.
from openpilot.sunnypilot import get_file_hash
from openpilot.sunnypilot.mapd import MAPD_PATH
from openpilot.sunnypilot.mapd.update_version import MAPD_HASH_PATH
from openpilot.common.test import OpenpilotTestCase
class TestMapdVersion:
class TestMapdVersion(OpenpilotTestCase):
def test_compare_versions(self):
mapd_hash = get_file_hash(MAPD_PATH)
@@ -5,8 +5,9 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pathlib
import pickle
import pytest
import tempfile
import openpilot.sunnypilot.models.helpers as helpers
import openpilot.sunnypilot.modeld_v2.modeld as modeld_module
@@ -181,16 +182,19 @@ def make_bundle(archetype):
)
@pytest.fixture
def tmp_path():
with tempfile.TemporaryDirectory() as d:
yield pathlib.Path(d)
def patch_modeld(monkeypatch):
def _patch(bundle):
monkeypatch.setattr(helpers, 'get_active_bundle', lambda params=None: bundle, raising=False)
monkeypatch.setattr(modeld_module, 'get_active_bundle', lambda params=None: bundle, raising=False)
monkeypatch.setattr(helpers, 'get_active_bundle', lambda params=None: bundle)
monkeypatch.setattr(modeld_module, 'get_active_bundle', lambda params=None: bundle)
return _patch
@pytest.fixture
def model_state_factory(tmp_path, monkeypatch, patch_modeld):
from openpilot.common.hardware import hw
@@ -9,6 +9,7 @@ import numpy as np
from openpilot.common.transformations.camera import DEVICE_CAMERAS
from openpilot.common.transformations.model import get_warp_matrix
from openpilot.sunnypilot.modeld_v2.camera_offset_helper import CameraOffsetHelper
from openpilot.common.test import OpenpilotTestCase
class MockStruct:
@@ -20,7 +21,7 @@ class MockStruct:
return getattr(self, item)
class TestCameraOffset:
class TestCameraOffset(OpenpilotTestCase):
def setup_method(self):
self.camera_offset = CameraOffsetHelper()
self.dc = DEVICE_CAMERAS[('mici', 'os04c10')]
@@ -5,20 +5,27 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pytest
from openpilot.common.parameterized import parameterized
import openpilot.sunnypilot.models.helpers as helpers
import openpilot.sunnypilot.modeld_v2.modeld as modeld_module
from openpilot.sunnypilot.modeld_v2.modeld import _find_driving_pkl
from openpilot.sunnypilot.modeld_v2.tests.conftest import DummyModel, DummyBundle, ARCHETYPES, CAM_W, CAM_H, \
from openpilot.sunnypilot.modeld_v2.tests import helpers as tests_helpers
from openpilot.sunnypilot.modeld_v2.tests.helpers import DummyModel, DummyBundle, ARCHETYPES, CAM_W, CAM_H, \
SPLIT_VISION_INPUT_SHAPES, SPLIT_POLICY_INPUT_SHAPES
from openpilot.common.test import OpenpilotTestCase
# resolved by name from this module when a test asks for them
tmp_path = tests_helpers.tmp_path
patch_modeld = tests_helpers.patch_modeld
model_state_factory = tests_helpers.model_state_factory
ModelState = modeld_module.ModelState
# Pkl discovery
class TestFindDrivingPkl:
class TestFindDrivingPkl(OpenpilotTestCase):
def test_returns_none_when_no_bundle(self):
assert _find_driving_pkl(None) is None
@@ -49,16 +56,16 @@ class TestFindDrivingPkl:
# Init — assertion guard
class TestModelStateCombinedInit:
class TestModelStateCombinedInit(OpenpilotTestCase):
def test_asserts_when_no_pkl(self, monkeypatch):
bundle = DummyBundle(models=[], is_20hz=True)
monkeypatch.setattr(helpers, 'get_active_bundle', lambda params=None: bundle, raising=False)
monkeypatch.setattr(modeld_module, 'get_active_bundle', lambda params=None: bundle, raising=False)
with pytest.raises(AssertionError, match="No driving pkl found"):
monkeypatch.setattr(helpers, 'get_active_bundle', lambda params=None: bundle)
monkeypatch.setattr(modeld_module, 'get_active_bundle', lambda params=None: bundle)
with self.assertRaisesRegex(AssertionError, "No driving pkl found"):
ModelState(cam_w=CAM_W, cam_h=CAM_H)
class TestStockEquivalence:
class TestStockEquivalence(OpenpilotTestCase):
def test_split_queue_keys_match_stock(self, model_state_factory):
from openpilot.selfdrive.modeld.compile_modeld import make_input_queues
@@ -100,8 +107,8 @@ class TestStockEquivalence:
ARCHETYPE_NAMES = list(ARCHETYPES.keys())
class TestModelTypeDetection:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestModelTypeDetection(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_combined_model_type(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -109,8 +116,8 @@ class TestModelTypeDetection:
f"{arch.name}: got {state._combined_model_type}, expected {arch.expected_model_type}"
class TestConstantsSelection:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestConstantsSelection(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_constants_class(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -118,8 +125,8 @@ class TestConstantsSelection:
f"{arch.name}: got {type(state.constants).__name__}, expected {arch.expected_constants_class.__name__}"
class TestParserSelection:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestParserSelection(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_parser_module(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -128,8 +135,8 @@ class TestParserSelection:
f"{arch.name}: parser from {parser_module}, expected module ending with {arch.expected_parser_module}"
class TestDesireKeyDetection:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestDesireKeyDetection(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_desire_key(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -137,8 +144,8 @@ class TestDesireKeyDetection:
f"{arch.name}: got {state.desire_key}, expected {arch.expected_desire_key}"
class TestVisionInputNames:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestVisionInputNames(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_vision_names_contain_img(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -147,14 +154,14 @@ class TestVisionInputNames:
assert 'img' in name, f"{arch.name}: vision input name '{name}' missing 'img'"
class TestOutputSlices:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestOutputSlices(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_vision_slices_populated(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
assert len(state.vision_output_slices) > 0, f"{arch.name}: vision_output_slices empty"
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_policy_slices_match_type(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -164,14 +171,14 @@ class TestOutputSlices:
assert len(state.policy_output_slices) > 0, f"{arch.name}: split/multi should have policy slices"
class TestInputQueueCreation:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestInputQueueCreation(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_queues_not_empty(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
assert len(state.input_queues) > 0, f"{arch.name}: input_queues empty"
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_npy_contains_transforms(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -180,7 +187,7 @@ class TestInputQueueCreation:
assert state.numpy_inputs['tfm'].shape == (3, 3)
assert state.numpy_inputs['big_tfm'].shape == (3, 3)
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_npy_contains_desire(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -188,8 +195,8 @@ class TestInputQueueCreation:
f"{arch.name}: '{arch.expected_desire_key}' missing from npy"
class TestFrameBufferParams:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestFrameBufferParams(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_frame_buf_params_per_vision_input(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
@@ -199,29 +206,29 @@ class TestFrameBufferParams:
assert len(nv12_info) >= 4, f"{arch.name}: nv12_info for '{name}' too short"
class TestBundleOverrides:
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
class TestBundleOverrides(OpenpilotTestCase):
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_smoothing_params_from_overrides(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
assert state.LAT_SMOOTH_SECONDS == 0.1
assert state.LONG_SMOOTH_SECONDS == 0.3
@pytest.mark.parametrize("archetype_name", ARCHETYPE_NAMES)
@parameterized.expand(ARCHETYPE_NAMES, names=["archetype_name"])
def test_generation_from_bundle(self, archetype_name, model_state_factory):
arch = ARCHETYPES[archetype_name]
state = model_state_factory(arch)
assert state.generation == 10
class TestMlsimProperty:
class TestMlsimProperty(OpenpilotTestCase):
def test_mlsim_false_for_gen10(self, model_state_factory):
state = model_state_factory(ARCHETYPES['supercombo_non20hz'])
assert state.mlsim is False
def test_mlsim_true_for_gen11(self, tmp_path, monkeypatch, patch_modeld):
from openpilot.common.hardware import hw
from openpilot.sunnypilot.modeld_v2.tests.conftest import write_pkl, ARCHETYPES as A
from openpilot.sunnypilot.modeld_v2.tests.helpers import write_pkl, ARCHETYPES as A
arch = A['supercombo_non20hz']
write_pkl(tmp_path, arch)
@@ -233,10 +240,10 @@ class TestMlsimProperty:
assert state.mlsim is True
class TestCrossArchetypeMismatch:
class TestCrossArchetypeMismatch(OpenpilotTestCase):
def test_wrong_is_20hz_changes_constants(self, tmp_path, monkeypatch, patch_modeld):
from openpilot.common.hardware import hw
from openpilot.sunnypilot.modeld_v2.tests.conftest import write_pkl
from openpilot.sunnypilot.modeld_v2.tests.helpers import write_pkl
from openpilot.sunnypilot.modeld_v2.constants import ModelConstants
arch = ARCHETYPES['vision_policy_split']
@@ -6,12 +6,13 @@ See the LICENSE.md file in the root directory for more details.
"""
import numpy as np
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.sunnypilot.modeld_v2.compile_modeld import derive_frame_skip, _detect_desire_key
from openpilot.common.test import OpenpilotTestCase
class TestDeriveFrameSkip:
class TestDeriveFrameSkip(OpenpilotTestCase):
def test_non20hz_supercombo(self):
vision = {}
policy = {'features_buffer': (1, 99, 512), 'desire': (1, 100, 8)}
@@ -31,21 +32,21 @@ class TestDeriveFrameSkip:
assert derive_frame_skip({}, {}) == 1
class TestFrameSkipBufferLengthEquivalence:
@pytest.mark.parametrize("frame_skip,expected_buffer_length", [
class TestFrameSkipBufferLengthEquivalence(OpenpilotTestCase):
@parameterized.expand([
(1, 2),
(4, 5),
])
], names=["frame_skip", "expected_buffer_length"])
def test_img_buffer_size_matches_warp_buffer_length(self, frame_skip, expected_buffer_length):
n_frames = 2
img_buf_dim0 = frame_skip * (n_frames - 1) + 1
assert img_buf_dim0 == expected_buffer_length, \
f"frame_skip={frame_skip}: img_buf[0]={img_buf_dim0}, expected {expected_buffer_length}"
@pytest.mark.parametrize("is_20hz,expected_frame_skip,expected_buffer_length", [
@parameterized.expand([
(False, 1, 2),
(True, 4, 5),
])
], names=["is_20hz", "expected_frame_skip", "expected_buffer_length"])
def test_is_20hz_to_frame_skip_to_buffer_length(self, is_20hz, expected_frame_skip, expected_buffer_length):
if is_20hz:
policy_shapes = {'features_buffer': (1, 24, 512)}
@@ -59,7 +60,7 @@ class TestFrameSkipBufferLengthEquivalence:
assert img_buf_dim0 == expected_buffer_length
class TestTemporalSamplingEquivalence:
class TestTemporalSamplingEquivalence(OpenpilotTestCase):
def test_non20hz_desire_sampling_identity(self):
buf = np.random.default_rng(0).standard_normal((100, 1, 8)).astype(np.float32)
frame_skip = 1
@@ -94,12 +95,12 @@ class TestTemporalSamplingEquivalence:
np.testing.assert_array_equal(sampled, buf[:, 0, :])
class TestTemporalIdxEquivalence:
@pytest.mark.parametrize("mode,desire_shape,fb_shape,frame_skip", [
class TestTemporalIdxEquivalence(OpenpilotTestCase):
@parameterized.expand([
('non20hz', (1, 100, 8), (1, 99, 512), 1),
('20hz', (1, 25, 8), (1, 24, 512), 4),
('split', (1, 25, 8), (1, 25, 512), 4),
])
], names=["mode", "desire_shape", "fb_shape", "frame_skip"])
def test_features_buffer_idx_equivalence(self, mode, desire_shape, fb_shape, frame_skip):
history = fb_shape[1]
@@ -118,11 +119,11 @@ class TestTemporalIdxEquivalence:
assert len(modelstate_idxs) == fb_shape[1], \
f"{mode}: ModelState idx count {len(modelstate_idxs)} != input shape {fb_shape[1]}"
@pytest.mark.parametrize("mode,desire_shape,fb_shape,frame_skip", [
@parameterized.expand([
('non20hz', (1, 100, 8), (1, 99, 512), 1),
('20hz', (1, 25, 8), (1, 24, 512), 4),
('split', (1, 25, 8), (1, 25, 512), 4),
])
], names=["mode", "desire_shape", "fb_shape", "frame_skip"])
def test_desire_idx_equivalence(self, mode, desire_shape, fb_shape, frame_skip):
history = desire_shape[1]
@@ -132,7 +133,7 @@ class TestTemporalIdxEquivalence:
f"{mode}: compile desire samples {compile_sampled_count} != model input {history}"
class TestDetectDesireKey:
class TestDetectDesireKey(OpenpilotTestCase):
def test_finds_desire(self):
shapes = {'features_buffer': (1, 99, 512), 'desire': (1, 100, 8), 'traffic_convention': (1, 2)}
assert _detect_desire_key(shapes) == 'desire'
@@ -146,7 +147,7 @@ class TestDetectDesireKey:
assert _detect_desire_key(shapes) is None
class TestOutputSlicePreservation:
class TestOutputSlicePreservation(OpenpilotTestCase):
def test_vision_hidden_state_slice_used_for_features(self):
mock_slices = {'hidden_state': slice(0, 512), 'plan': slice(512, 1024)}
features_slice = mock_slices['hidden_state']
@@ -7,6 +7,7 @@ from openpilot.cereal import log
from openpilot.sunnypilot.modeld_v2.constants import Plan
from openpilot.sunnypilot.modeld_v2.modeld import ModelState
import openpilot.sunnypilot.modeld_v2.modeld as modeld
from openpilot.common.test import OpenpilotTestCase
class MockStruct:
@@ -15,58 +16,59 @@ class MockStruct:
setattr(self, k, v)
def test_recovery_power_scaling():
state: Any = MockStruct(
PLANPLUS_CONTROL=0.75,
LONG_SMOOTH_SECONDS=0.3,
LAT_SMOOTH_SECONDS=0.1,
MIN_LAT_CONTROL_SPEED=0.3,
mlsim=True,
generation=12,
constants=MockStruct(T_IDXS=np.arange(100), DESIRE_LEN=8)
)
prev_action = log.ModelDataV2.Action()
recorded_vel: list = []
recorded_curv_plans: list = []
class TestRecoveryPower(OpenpilotTestCase):
def test_recovery_power_scaling(self):
state: Any = MockStruct(
PLANPLUS_CONTROL=0.75,
LONG_SMOOTH_SECONDS=0.3,
LAT_SMOOTH_SECONDS=0.1,
MIN_LAT_CONTROL_SPEED=0.3,
mlsim=True,
generation=12,
constants=MockStruct(T_IDXS=np.arange(100), DESIRE_LEN=8)
)
prev_action = log.ModelDataV2.Action()
recorded_vel: list = []
recorded_curv_plans: list = []
def mock_accel(plan_vel, plan_accel, t_idxs, action_t=0.0):
recorded_vel.append(plan_vel.copy())
return 0.0, False
def mock_accel(plan_vel, plan_accel, t_idxs, action_t=0.0):
recorded_vel.append(plan_vel.copy())
return 0.0, False
def mock_curvature(output, plan, vego, lat_action_t, mlsim):
recorded_curv_plans.append(plan.copy())
return 0.0
def mock_curvature(output, plan, vego, lat_action_t, mlsim):
recorded_curv_plans.append(plan.copy())
return 0.0
modeld.get_accel_from_plan = mock_accel # ty: ignore[invalid-assignment]
modeld.get_curvature_from_output = mock_curvature # ty: ignore[invalid-assignment]
plan = np.random.default_rng(0).random((1, 100, 15)).astype(np.float32)
planplus = np.random.default_rng(1).random((1, 100, 15)).astype(np.float32)
merged_plan = plan + planplus
modeld.get_accel_from_plan = mock_accel # ty: ignore[invalid-assignment]
modeld.get_curvature_from_output = mock_curvature # ty: ignore[invalid-assignment]
plan = np.random.default_rng(0).random((1, 100, 15)).astype(np.float32)
planplus = np.random.default_rng(1).random((1, 100, 15)).astype(np.float32)
merged_plan = plan + planplus
model_output: dict = {
'plan': merged_plan.copy(),
'planplus': planplus.copy()
}
model_output: dict = {
'plan': merged_plan.copy(),
'planplus': planplus.copy()
}
test_cases: list = [
# (control, v_ego)
(0.55, 20.0),
(1.0, 25.0),
(1.5, 25.1),
(2.0, 20.0),
(0.75, 19.0),
(0.8, 25.1),
]
test_cases: list = [
# (control, v_ego)
(0.55, 20.0),
(1.0, 25.0),
(1.5, 25.1),
(2.0, 20.0),
(0.75, 19.0),
(0.8, 25.1),
]
for control, v_ego in test_cases:
state.PLANPLUS_CONTROL = control
recorded_vel.clear()
recorded_curv_plans.clear()
ModelState.get_action_from_model(state, model_output, prev_action, 0.0, 0.0, v_ego) # type: ignore[arg-type]
for control, v_ego in test_cases:
state.PLANPLUS_CONTROL = control
recorded_vel.clear()
recorded_curv_plans.clear()
ModelState.get_action_from_model(state, model_output, prev_action, 0.0, 0.0, v_ego) # type: ignore[arg-type]
expected_accel_plan_vel = plan[0, :, Plan.VELOCITY][:, 0] + planplus[0, :, Plan.VELOCITY][:, 0]
np.testing.assert_allclose(recorded_vel[0], expected_accel_plan_vel, rtol=1e-5, atol=1e-6)
expected_accel_plan_vel = plan[0, :, Plan.VELOCITY][:, 0] + planplus[0, :, Plan.VELOCITY][:, 0]
np.testing.assert_allclose(recorded_vel[0], expected_accel_plan_vel, rtol=1e-5, atol=1e-6)
# For the below, yes, I know this isn't the same slicing as fillmodlmsg. This is to show that the values are only scaled on curv
expected_curv_plan_vel = plan[0, :, Plan.VELOCITY][:, 0] + control * planplus[0, :, Plan.VELOCITY][:, 0]
np.testing.assert_allclose(recorded_curv_plans[0][:, Plan.VELOCITY][:, 0], expected_curv_plan_vel, rtol=1e-5, atol=1e-6)
# For the below, yes, I know this isn't the same slicing as fillmodlmsg. This is to show that the values are only scaled on curv
expected_curv_plan_vel = plan[0, :, Plan.VELOCITY][:, 0] + control * planplus[0, :, Plan.VELOCITY][:, 0]
np.testing.assert_allclose(recorded_curv_plans[0][:, Plan.VELOCITY][:, 0], expected_curv_plan_vel, rtol=1e-5, atol=1e-6)
@@ -8,9 +8,10 @@ See the LICENSE.md file in the root directory for more details.
from openpilot.sunnypilot import get_file_hash
from openpilot.sunnypilot.models.default_model import MODEL_HASH_PATH, SUPERCOMBO_ONNX_PATH
import hashlib
from openpilot.common.test import OpenpilotTestCase
class TestDefaultModel:
class TestDefaultModel(OpenpilotTestCase):
def test_compare_onnx_hashes(self):
supercombo_hash = get_file_hash(SUPERCOMBO_ONNX_PATH)
@@ -2,6 +2,7 @@ import requests
from openpilot.sunnypilot.models.tinygrad_ref import get_tinygrad_ref
from openpilot.sunnypilot.models.fetcher import ModelFetcher
from openpilot.common.test import OpenpilotTestCase
def fetch_tinygrad_ref():
@@ -11,13 +12,14 @@ def fetch_tinygrad_ref():
return json_data.get("tinygrad_ref")
def test_tinygrad_ref():
current_ref = get_tinygrad_ref()
remote_ref = fetch_tinygrad_ref()
assert remote_ref == current_ref, (
f"""tinygrad_repo ref does not match remote tinygrad_ref of current compiled driving models json.
Current: {current_ref}
Remote: {remote_ref}
Please run build-all workflow to update models."""
)
print("tinygrad_repo ref matches current compiled driving models json ref.")
class TestTinygradRef(OpenpilotTestCase):
def test_tinygrad_ref(self):
current_ref = get_tinygrad_ref()
remote_ref = fetch_tinygrad_ref()
assert remote_ref == current_ref, (
f"""tinygrad_repo ref does not match remote tinygrad_ref of current compiled driving models json.
Current: {current_ref}
Remote: {remote_ref}
Please run build-all workflow to update models."""
)
print("tinygrad_repo ref matches current compiled driving models json ref.")
@@ -1,94 +0,0 @@
import pytest
from openpilot.sunnypilot.selfdrive.controls.lib.dec.dec import DynamicExperimentalController
class MockLeadOne:
def __init__(self, status=0.0):
self.status = status
class MockRadarState:
def __init__(self, status=0.0):
self.leadOne = MockLeadOne(status=status)
class MockCarState:
def __init__(self, vEgo=0.0, vCruise=0.0, standstill=False):
self.vEgo = vEgo
self.vCruise = vCruise
self.standstill = standstill
class MockModelData:
def __init__(self, valid=True):
size = 33 if valid else 10 # incomplete if invalid
self.position = type("Pos", (), {"x": [0.0] * size})()
self.orientation = type("Ori", (), {"x": [0.0] * size})()
class MockSelfDriveState:
def __init__(self, experimentalMode=False):
self.experimentalMode = experimentalMode
class MockParams:
def get_bool(self, name):
return True
@pytest.fixture
def default_sm():
sm = {
'carState': MockCarState(vEgo=10.0, vCruise=20.0),
'radarState': MockRadarState(status=1.0),
'modelV2': MockModelData(valid=True),
'selfdriveState': MockSelfDriveState(experimentalMode=True),
}
return sm
@pytest.fixture
def mock_cp():
class CP:
radarUnavailable = False
return CP()
@pytest.fixture
def mock_mpc():
class MPC:
crash_cnt = 0
return MPC()
# Fake Kalman Filter that always returns a given value
class FakeKalman:
def __init__(self, value=1.0):
self.value = value
def add_data(self, v): pass
def get_value(self): return self.value
def get_confidence(self): return 1.0
def reset_data(self): pass
def test_initial_mode_is_acc(mock_cp, mock_mpc):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
assert controller.mode() == "acc"
def test_standstill_triggers_blended(mock_cp, mock_mpc, default_sm):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
default_sm['carState'].standstill = True
for _ in range(10):
controller.update(default_sm)
assert controller.mode() == "blended"
def test_emergency_blended_on_fcw(mock_cp, mock_mpc, default_sm):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
mock_mpc.crash_cnt = 1 # simulate FCW
for _ in range(2):
controller.update(default_sm)
assert controller.mode() == "blended"
def test_radarless_slowdown_triggers_blended(mock_cp, mock_mpc, default_sm):
mock_cp.radarUnavailable = True
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
# Force conditions to simulate slowdown
controller._slow_down_filter = FakeKalman(value=1.0) # ty: ignore[invalid-assignment]
controller._v_ego_kph = 35.0
default_sm['modelV2'] = MockModelData(valid=False) # Incomplete trajectory
for _ in range(3):
controller.update(default_sm)
assert controller.mode() == "blended"
@@ -0,0 +1,91 @@
from openpilot.common.test import OpenpilotTestCase
from openpilot.sunnypilot.selfdrive.controls.lib.dec.dec import DynamicExperimentalController
class MockLeadOne:
def __init__(self, present=0.0):
self.present = present
class MockRadarState:
def __init__(self, present=0.0):
self.leadOne = MockLeadOne(present=present)
class MockCarState:
def __init__(self, vEgo=0.0, vCruise=0.0, standstill=False):
self.vEgo = vEgo
self.vCruise = vCruise
self.standstill = standstill
class MockModelData:
def __init__(self, valid=True):
size = 33 if valid else 10 # incomplete if invalid
self.position = type("Pos", (), {"x": [0.0] * size})()
self.orientation = type("Ori", (), {"x": [0.0] * size})()
class MockSelfDriveState:
def __init__(self, experimentalMode=False):
self.experimentalMode = experimentalMode
class MockParams:
def get_bool(self, name):
return True
def default_sm():
sm = {
'carState': MockCarState(vEgo=10.0, vCruise=20.0),
'radarState': MockRadarState(present=1.0),
'modelV2': MockModelData(valid=True),
'selfdriveState': MockSelfDriveState(experimentalMode=True),
}
return sm
def mock_cp():
class CP:
radarUnavailable = False
return CP()
def mock_mpc():
class MPC:
crash_cnt = 0
return MPC()
# Fake Kalman Filter that always returns a given value
class FakeKalman:
def __init__(self, value=1.0):
self.value = value
def add_data(self, v): pass
def get_value(self): return self.value
def get_confidence(self): return 1.0
def reset_data(self): pass
class TestDynamicExperimentalController(OpenpilotTestCase):
def test_initial_mode_is_acc(self, mock_cp, mock_mpc):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
assert controller.mode() == "acc"
def test_standstill_triggers_blended(self, mock_cp, mock_mpc, default_sm):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
default_sm['carState'].standstill = True
for _ in range(10):
controller.update(default_sm)
assert controller.mode() == "blended"
def test_emergency_blended_on_fcw(self, mock_cp, mock_mpc, default_sm):
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
mock_mpc.crash_cnt = 1 # simulate FCW
for _ in range(2):
controller.update(default_sm)
assert controller.mode() == "blended"
def test_radarless_slowdown_triggers_blended(self, mock_cp, mock_mpc, default_sm):
mock_cp.radarUnavailable = True
controller = DynamicExperimentalController(mock_cp, mock_mpc, params=MockParams())
# Force conditions to simulate slowdown
controller._slow_down_filter = FakeKalman(value=1.0) # ty: ignore[invalid-assignment]
controller._v_ego_kph = 35.0
default_sm['modelV2'] = MockModelData(valid=False) # Incomplete trajectory
for _ in range(3):
controller.update(default_sm)
assert controller.mode() == "blended"
@@ -7,6 +7,7 @@ from opendbc.car.tesla.values import CAR as TESLA
from openpilot.common.parameterized import parameterized
from openpilot.common.params import Params
from openpilot.sunnypilot.selfdrive.car import interfaces as sunnypilot_interfaces
from openpilot.common.test import OpenpilotTestCase
FINGERPRINT_EXACT_MATCH = [HONDA.HONDA_CIVIC_BOSCH, TOYOTA.TOYOTA_RAV4_TSS2_2022, HYUNDAI.HYUNDAI_IONIQ_5]
@@ -14,7 +15,7 @@ FINGERPRINT_FUZZY_MATCH = [HONDA.HONDA_CIVIC_BOSCH_DIESEL, HYUNDAI.GENESIS_G70_2
FINGERPRINT_ANGLE_NO_MATCH = [TOYOTA.TOYOTA_RAV4_TSS2_2023, NISSAN.NISSAN_LEAF, TESLA.TESLA_MODEL_3]
class TestNNLCFingerprintBase:
class TestNNLCFingerprintBase(OpenpilotTestCase):
@staticmethod
def _setup_platform(car_name):
@@ -8,9 +8,10 @@ from openpilot.common.realtime import DT_CTRL
from openpilot.selfdrive.car.helpers import convert_to_capnp
from openpilot.selfdrive.controls.lib.latcontrol_torque import LatControlTorque
from openpilot.sunnypilot.selfdrive.car import interfaces as sunnypilot_interfaces
from openpilot.common.test import OpenpilotTestCase
class TestNNTorqueModel:
class TestNNTorqueModel(OpenpilotTestCase):
@parameterized.expand([HONDA.HONDA_CIVIC, TOYOTA.TOYOTA_RAV4, HYUNDAI.HYUNDAI_SANTA_CRUZ_1ST_GEN])
def test_load_model(self, car_name):
@@ -17,6 +17,7 @@ from openpilot.selfdrive.locationd.helpers import Pose
from openpilot.common.mock.generators import generate_deviceMotion
from openpilot.sunnypilot.selfdrive.car import interfaces as sunnypilot_interfaces
from openpilot.selfdrive.modeld.constants import ModelConstants
from openpilot.common.test import OpenpilotTestCase
def generate_modelV2():
@@ -42,7 +43,7 @@ def generate_modelV2():
return model
class TestNeuralNetworkLateralControl:
class TestNeuralNetworkLateralControl(OpenpilotTestCase):
@parameterized.expand([HONDA.HONDA_CIVIC, TOYOTA.TOYOTA_RAV4, HYUNDAI.HYUNDAI_SANTA_CRUZ_1ST_GEN, GM.CHEVROLET_BOLT_EUV])
def test_saturation(self, car_name):
@@ -81,7 +82,7 @@ class TestNeuralNetworkLateralControl:
for _ in range(1000):
controller.extension.update_model_v2(model_v2)
controller.extension.update_lateral_lag(test_lag)
controller.update_live_torque_params(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.update_torque_parameters(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.extension.update_limits()
_, _, lac_log = controller.update(True, CS, VM, params, False, 0, pose, True, 0.2)
assert lac_log.saturated
@@ -89,7 +90,7 @@ class TestNeuralNetworkLateralControl:
for _ in range(1000):
controller.extension.update_model_v2(model_v2)
controller.extension.update_lateral_lag(test_lag)
controller.update_live_torque_params(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.update_torque_parameters(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.extension.update_limits()
_, _, lac_log = controller.update(True, CS, VM, params, False, 0, pose, False, 0.2)
assert not lac_log.saturated
@@ -97,7 +98,7 @@ class TestNeuralNetworkLateralControl:
for _ in range(1000):
controller.extension.update_model_v2(model_v2)
controller.extension.update_lateral_lag(test_lag)
controller.update_live_torque_params(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.update_torque_parameters(torque_params.latAccelFactor, torque_params.latAccelOffset, torque_params.friction)
controller.extension.update_limits()
_, _, lac_log = controller.update(True, CS, VM, params, False, 1, pose, False, 0.2)
assert lac_log.saturated
@@ -8,18 +8,18 @@ import json
import math
import platform
import pytest
from openpilot.cereal import custom
from openpilot.common.params import Params
from openpilot.common.realtime import DT_MDL
from openpilot.selfdrive.car.cruise import V_CRUISE_UNSET
from openpilot.sunnypilot.selfdrive.controls.lib.smart_cruise_control.map_controller import R, SmartCruiseControlMap
from openpilot.common.test import OpenpilotTestCase
MapState = VisionState = custom.LongitudinalPlanSP.SmartCruiseControl.MapState
class TestSmartCruiseControlMap:
class TestSmartCruiseControlMap(OpenpilotTestCase):
def setup_method(self):
self.params = Params()
@@ -70,6 +70,6 @@ class TestSmartCruiseControlMap:
self.scc_m.update(True, False, 25.0, 0.0, 30.0)
assert self.scc_m.v_target == pytest.approx(24.0)
self.assertAlmostEqual(self.scc_m.v_target, 24.0, delta=24.0 * 1e-6)
# TODO-SP: mock data from modelV2 to test other states
@@ -7,7 +7,7 @@ See the LICENSE.md file in the root directory for more details.
from typing import Any
import numpy as np
import pytest
from openpilot.common.parameterized import parameterized
import openpilot.cereal.messaging as messaging
from openpilot.cereal import custom, log
@@ -17,6 +17,7 @@ from openpilot.selfdrive.car.cruise import V_CRUISE_UNSET
from openpilot.selfdrive.modeld.constants import ModelConstants
from openpilot.sunnypilot.selfdrive.controls.lib.smart_cruise_control import MIN_V
from openpilot.sunnypilot.selfdrive.controls.lib.smart_cruise_control.vision_controller import SmartCruiseControlVision, _ENTERING_PRED_LAT_ACC_TH
from openpilot.common.test import OpenpilotTestCase
VisionState = custom.LongitudinalPlanSP.SmartCruiseControl.VisionState
@@ -105,7 +106,7 @@ def generate_controlsState():
return controls_state
class TestSmartCruiseControlVision:
class TestSmartCruiseControlVision(OpenpilotTestCase):
def setup_method(self):
self.params = Params()
@@ -145,19 +146,11 @@ class TestSmartCruiseControlVision:
self.scc_v.update(self.sm, True, False, 0., 0., 0.)
assert self.scc_v.state == VisionState.enabled
@pytest.mark.parametrize(
"case, should_enter",
[
@parameterized.expand([
("p97_just_above_threshold", True),
("single_spike_filtered", False),
("persistent_high_values", True),
],
ids=[
"p97>threshold_enters",
"single_spike_max_large_but_p97_below_threshold",
"high_values_persist_trigger_entering",
],
)
], names=["case", "should_enter"])
def test_max_pred_lat_acc_uses_p97_and_threshold(self, case, should_enter):
n = len(ModelConstants.T_IDXS)
th = float(_ENTERING_PRED_LAT_ACC_TH)
@@ -7,7 +7,7 @@ See the LICENSE.md file in the root directory for more details.
import time
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.cereal import custom
from opendbc.car.car_helpers import interfaces
@@ -27,6 +27,7 @@ from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit.speed_limit_assist
PRE_ACTIVE_GUARD_PERIOD, ACTIVE_STATES, CRUISE_BUTTON_CONFIRM_HOLD
from openpilot.sunnypilot.selfdrive.selfdrived.button_state_tracker import ButtonStateTracker
from openpilot.sunnypilot.selfdrive.selfdrived.events import EventsSP
from openpilot.common.test import OpenpilotTestCase
ButtonEvent = car.CarState.ButtonEvent
ButtonType = car.CarState.ButtonEvent.Type
@@ -45,21 +46,10 @@ SPEED_LIMITS = {
DEFAULT_CAR = TOYOTA.TOYOTA_RAV4_TSS2
@pytest.fixture
def car_name(request):
return getattr(request, "param", DEFAULT_CAR)
class TestSpeedLimitAssist(OpenpilotTestCase):
car_name = DEFAULT_CAR
@pytest.fixture(autouse=True)
def set_car_name_on_instance(request, car_name):
instance = getattr(request, "instance", None)
if instance:
instance.car_name = car_name
class TestSpeedLimitAssist:
def setup_method(self, method):
def setup_method(self):
self.params = Params()
self.reset_custom_params()
self.events_sp = EventsSP()
@@ -69,7 +59,7 @@ class TestSpeedLimitAssist:
self.pcm_long_max_set_speed = PCM_LONG_REQUIRED_MAX_SET_SPEED[self.sla.is_metric][1] # use 80 MPH for now
self.speed_conv = CV.MS_TO_KPH if self.sla.is_metric else CV.MS_TO_MPH
def teardown_method(self, method):
def teardown_method(self):
self.reset_state()
def _setup_platform(self, car_name):
@@ -112,13 +102,16 @@ class TestSpeedLimitAssist:
assert not self.sla.is_active
assert V_CRUISE_UNSET == self.sla.get_v_target_from_control()
@pytest.mark.parametrize("car_name", [RIVIAN.RIVIAN_R1, TESLA.TESLA_MODEL_Y], indirect=True)
@parameterized.expand([RIVIAN.RIVIAN_R1, TESLA.TESLA_MODEL_Y], names=["car_name"])
def test_disallowed_brands(self, car_name):
"""
Speed Limit Assist is disabled for the following brands and conditions:
- All Tesla and is a release branch;
- All Rivian
"""
self.car_name = car_name
self.openpilot_setup_method() # rebuild the platform for this brand
assert not self.sla.enabled
# stay disallowed even when the param may have changed from somewhere else
@@ -285,9 +278,10 @@ class TestSpeedLimitAssist:
assert self.sla.state in ACTIVE_STATES
class TestButtonStateTrackerSLAIntegration:
class TestButtonStateTrackerSLAIntegration(OpenpilotTestCase):
def setup_method(self):
def setup_method(self, method):
self.tracker = ButtonStateTracker()
self.params = Params()
self.params.put("IsReleaseSpBranch", True, block=True)
@@ -7,26 +7,26 @@ See the LICENSE.md file in the root directory for more details.
import random
import time
import pytest
from pytest_mock import MockerFixture
from openpilot.common.parameterized import parameterized
from openpilot.cereal import custom
from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit import LIMIT_MAX_MAP_DATA_AGE
from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit.speed_limit_resolver import SpeedLimitResolver, ALL_SOURCES
from openpilot.sunnypilot.selfdrive.controls.lib.speed_limit.common import Policy
from openpilot.common.test import OpenpilotTestCase
SpeedLimitSource = custom.LongitudinalPlanSP.SpeedLimit.Source
def create_mock(properties, mocker: MockerFixture):
def create_mock(properties, mocker):
mock = mocker.MagicMock()
for _property, value in properties.items():
setattr(mock, _property, value)
return mock
def setup_sm_mock(mocker: MockerFixture):
def setup_sm_mock(mocker):
cruise_speed_limit = random.uniform(0, 120)
live_map_data_limit = random.uniform(0, 120)
@@ -58,21 +58,24 @@ def setup_sm_mock(mocker: MockerFixture):
return sm_mock
parametrized_policies = pytest.mark.parametrize(
"policy, sm_key, function_key", [
parametrized_policies = parameterized.expand(
[
(Policy.car_state_only, 'carStateSP', SpeedLimitSource.car),
(Policy.car_state_priority, 'carStateSP', SpeedLimitSource.car),
(Policy.map_data_only, 'liveMapDataSP', SpeedLimitSource.map),
(Policy.map_data_priority, 'liveMapDataSP', SpeedLimitSource.map),
],
ids=lambda val: val.name if hasattr(val, 'name') else str(val)
names=["policy", "sm_key", "function_key"]
)
@pytest.mark.parametrize("resolver_class", [SpeedLimitResolver])
class TestSpeedLimitResolverValidation:
def resolver_class():
return SpeedLimitResolver
@pytest.mark.parametrize("policy", list(Policy), ids=lambda policy: policy.name)
class TestSpeedLimitResolverValidation(OpenpilotTestCase):
@parameterized.expand(list(Policy), names=["policy"])
def test_initial_state(self, resolver_class, policy):
resolver = resolver_class()
resolver.policy = policy
@@ -82,7 +85,7 @@ class TestSpeedLimitResolverValidation:
assert resolver.distance_solutions[source] == 0.
@parametrized_policies
def test_resolver(self, resolver_class, policy, sm_key, function_key, mocker: MockerFixture):
def test_resolver(self, resolver_class, policy, sm_key, function_key, mocker):
resolver = resolver_class()
resolver.policy = policy
sm_mock = setup_sm_mock(mocker)
@@ -93,7 +96,7 @@ class TestSpeedLimitResolverValidation:
assert resolver.speed_limit == source_speed_limit
assert resolver.source == ALL_SOURCES[function_key]
def test_resolver_combined(self, resolver_class, mocker: MockerFixture):
def test_resolver_combined(self, resolver_class, mocker):
resolver = resolver_class()
resolver.policy = Policy.combined
sm_mock = setup_sm_mock(mocker)
@@ -108,7 +111,7 @@ class TestSpeedLimitResolverValidation:
assert resolver.source == socket_to_source[minimum_key]
@parametrized_policies
def test_parser(self, resolver_class, policy, sm_key, function_key, mocker: MockerFixture):
def test_parser(self, resolver_class, policy, sm_key, function_key, mocker):
resolver = resolver_class()
resolver.policy = policy
sm_mock = setup_sm_mock(mocker)
@@ -119,8 +122,8 @@ class TestSpeedLimitResolverValidation:
assert resolver.limit_solutions[ALL_SOURCES[function_key]] == source_speed_limit
assert resolver.distance_solutions[ALL_SOURCES[function_key]] == 0.
@pytest.mark.parametrize("policy", list(Policy), ids=lambda policy: policy.name)
def test_resolve_interaction_in_update(self, resolver_class, policy, mocker: MockerFixture):
@parameterized.expand(list(Policy), names=["policy"])
def test_resolve_interaction_in_update(self, resolver_class, policy, mocker):
v_ego = 50
resolver = resolver_class()
resolver.policy = policy
@@ -133,8 +136,8 @@ class TestSpeedLimitResolverValidation:
assert resolver.distance is not None
assert resolver.source is not None
@pytest.mark.parametrize("policy", list(Policy), ids=lambda policy: policy.name)
def test_old_map_data_ignored(self, resolver_class, policy, mocker: MockerFixture):
@parameterized.expand(list(Policy), names=["policy"])
def test_old_map_data_ignored(self, resolver_class, policy, mocker):
resolver = resolver_class()
resolver.policy = policy
sm_mock = mocker.MagicMock()
@@ -9,6 +9,7 @@ from openpilot.common.realtime import DT_MDL
from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper, LaneChangeState, LaneChangeDirection
from openpilot.sunnypilot.selfdrive.controls.lib.auto_lane_change import AutoLaneChangeController, AutoLaneChangeMode, \
AUTO_LANE_CHANGE_TIMER, ONE_SECOND_DELAY
from openpilot.common.test import OpenpilotTestCase
AUTO_LANE_CHANGE_TIMER_COMBOS = [
(AutoLaneChangeMode.NUDGELESS, AUTO_LANE_CHANGE_TIMER[AutoLaneChangeMode.NUDGELESS]),
@@ -19,7 +20,7 @@ AUTO_LANE_CHANGE_TIMER_COMBOS = [
]
class TestAutoLaneChangeController:
class TestAutoLaneChangeController(OpenpilotTestCase):
def setup_method(self):
self.DH = DesireHelper()
self.alc = AutoLaneChangeController(self.DH)
@@ -8,9 +8,10 @@ from opendbc.car.structs import car
from openpilot.common.constants import CV
from openpilot.sunnypilot.selfdrive.controls.lib.blinker_pause_lateral import BlinkerPauseLateral
from openpilot.common.test import OpenpilotTestCase
class TestBlinkerPauseLateral:
class TestBlinkerPauseLateral(OpenpilotTestCase):
def setup_method(self):
self.blinker_pause_lateral = BlinkerPauseLateral()
@@ -1,6 +1,7 @@
import pytest
from openpilot.cereal import log, custom
from openpilot.common.params import Params
from openpilot.common.parameterized import parameterized
from openpilot.common.test import OpenpilotTestCase
from openpilot.selfdrive.controls.lib.desire_helper import DesireHelper, LaneChangeState, LaneChangeDirection
from openpilot.sunnypilot.selfdrive.controls.lib.lane_turn_desire import LaneTurnController, LANE_CHANGE_SPEED_MIN
@@ -10,65 +11,63 @@ from openpilot.sunnypilot.selfdrive.controls.lib.auto_lane_change import AutoLan
TurnDirection = custom.ModelDataV2SP.TurnDirection
@pytest.mark.parametrize("left_blinker,right_blinker,v_ego,blindspot_left,blindspot_right,expected", [
(True, False, 5, False, False, TurnDirection.turnLeft),
(False, True, 6, False, False, TurnDirection.turnRight),
(True, False, 9, False, False, TurnDirection.none),
(True, False, 7, True, False, TurnDirection.none),
(False, True, 6, False, True, TurnDirection.none),
(False, False, 5, False, False, TurnDirection.none),
(True, True, 5, False, False, TurnDirection.none),
])
def test_lane_turn_desire_conditions(left_blinker, right_blinker, v_ego, blindspot_left, blindspot_right, expected):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(blindspot_left, blindspot_right, left_blinker, right_blinker, v_ego)
assert controller.get_turn_direction() == expected
class TestLaneTurnDesire(OpenpilotTestCase):
@parameterized.expand([
(True, False, 5, False, False, TurnDirection.turnLeft),
(False, True, 6, False, False, TurnDirection.turnRight),
(True, False, 9, False, False, TurnDirection.none),
(True, False, 7, True, False, TurnDirection.none),
(False, True, 6, False, True, TurnDirection.none),
(False, False, 5, False, False, TurnDirection.none),
(True, True, 5, False, False, TurnDirection.none),
])
def test_lane_turn_desire_conditions(self, left_blinker, right_blinker, v_ego, blindspot_left, blindspot_right, expected):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(blindspot_left, blindspot_right, left_blinker, right_blinker, v_ego)
assert controller.get_turn_direction() == expected
def test_lane_turn_desire_disabled(self):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = False
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(False, False, True, False, 7)
assert controller.get_turn_direction() == TurnDirection.none
def test_lane_turn_desire_disabled():
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = False
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(False, False, True, False, 7)
assert controller.get_turn_direction() == TurnDirection.none
def test_lane_turn_overrides_lane_change(self):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
# left turn desire
controller.update_lane_turn(False, False, True, False, 5)
assert controller.get_turn_direction() == TurnDirection.turnLeft
# right turn desire
controller.update_lane_turn(False, False, False, True, 6)
assert controller.get_turn_direction() == TurnDirection.turnRight
# no turn
controller.update_lane_turn(False, False, False, False, 7)
assert controller.get_turn_direction() == TurnDirection.none
def test_lane_turn_overrides_lane_change():
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
# left turn desire
controller.update_lane_turn(False, False, True, False, 5)
assert controller.get_turn_direction() == TurnDirection.turnLeft
# right turn desire
controller.update_lane_turn(False, False, False, True, 6)
assert controller.get_turn_direction() == TurnDirection.turnRight
# no turn
controller.update_lane_turn(False, False, False, False, 7)
assert controller.get_turn_direction() == TurnDirection.none
@pytest.mark.parametrize("v_ego,expected", [
(8.93, TurnDirection.turnLeft), # just below threshold
(8.96, TurnDirection.none), # above threshold
(8.95, TurnDirection.none), # just above threshold
])
def test_lane_turn_desire_speed_boundary(v_ego, expected):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(False, True, True, False, v_ego)
assert controller.get_turn_direction() == expected
@parameterized.expand([
(8.93, TurnDirection.turnLeft), # just below threshold
(8.96, TurnDirection.none), # above threshold
(8.95, TurnDirection.none), # just above threshold
])
def test_lane_turn_desire_speed_boundary(self, v_ego, expected):
dh = DesireHelper()
controller = LaneTurnController(dh)
controller.enabled = True
controller.lane_turn_value = LANE_CHANGE_SPEED_MIN
controller.turn_direction = TurnDirection.none
controller.update_lane_turn(False, True, True, False, v_ego)
assert controller.get_turn_direction() == expected
class DummyCarState:
@@ -84,43 +83,42 @@ class DummyCarState:
self.brakePressed = brakePressed
@pytest.fixture
def set_lane_turn_params():
params = Params()
params.put("LaneTurnDesire", True)
params.put("LaneTurnValue", 20.0)
@pytest.mark.parametrize("carstate, lateral_active, lane_change_prob, expected_desire", [
# Lane turn desire overrides lane change desire
(DummyCarState(vEgo=5, leftBlinker=True, rightBlinker=False, leftBlindspot=False, rightBlindspot=False), True, 1.0,
log.Desire.turnLeft),
(DummyCarState(vEgo=7, leftBlinker=False, rightBlinker=True, leftBlindspot=False, rightBlindspot=False), True, 1.0,
log.Desire.turnRight),
# Lane change desire only (no turn desires)
(DummyCarState(vEgo=9, leftBlinker=True, rightBlinker=False, leftBlindspot=False, rightBlindspot=False,
steeringPressed=True, steeringTorque=1), True, 1.0, log.Desire.laneChangeLeft),
(DummyCarState(vEgo=9, leftBlinker=False, rightBlinker=True, leftBlindspot=False, rightBlindspot=False,
steeringPressed=True, steeringTorque=-1), True, 1.0, log.Desire.laneChangeRight),
# No desire (inactive)
(DummyCarState(vEgo=9, leftBlinker=False, rightBlinker=False), False, 1.0, log.Desire.none),
(DummyCarState(vEgo=4, leftBlinker=False, rightBlinker=False), True, 1.0, log.Desire.none), # No blinkers? no desire!
])
def test_desire_helper_integration(carstate, lateral_active, lane_change_prob, expected_desire, set_lane_turn_params):
dh = DesireHelper()
dh.alc.lane_change_set_timer = AutoLaneChangeMode.NUDGE
for _ in range(10):
dh.update(carstate, lateral_active, lane_change_prob,
left_edge_detected=False, right_edge_detected=False)
assert dh.desire == expected_desire
class TestDesireHelperIntegration(OpenpilotTestCase):
@parameterized.expand([
# Lane turn desire overrides lane change desire
(DummyCarState(vEgo=5, leftBlinker=True, rightBlinker=False, leftBlindspot=False, rightBlindspot=False), True, 1.0,
log.Desire.turnLeft),
(DummyCarState(vEgo=7, leftBlinker=False, rightBlinker=True, leftBlindspot=False, rightBlindspot=False), True, 1.0,
log.Desire.turnRight),
# Lane change desire only (no turn desires)
(DummyCarState(vEgo=9, leftBlinker=True, rightBlinker=False, leftBlindspot=False, rightBlindspot=False,
steeringPressed=True, steeringTorque=1), True, 1.0, log.Desire.laneChangeLeft),
(DummyCarState(vEgo=9, leftBlinker=False, rightBlinker=True, leftBlindspot=False, rightBlindspot=False,
steeringPressed=True, steeringTorque=-1), True, 1.0, log.Desire.laneChangeRight),
# No desire (inactive)
(DummyCarState(vEgo=9, leftBlinker=False, rightBlinker=False), False, 1.0, log.Desire.none),
(DummyCarState(vEgo=4, leftBlinker=False, rightBlinker=False), True, 1.0, log.Desire.none), # No blinkers? no desire!
], names=["carstate", "lateral_active", "lane_change_prob", "expected_desire"])
def test_desire_helper_integration(self, carstate, lateral_active, lane_change_prob, expected_desire, set_lane_turn_params):
dh = DesireHelper()
dh.alc.lane_change_set_timer = AutoLaneChangeMode.NUDGE
for _ in range(10):
dh.update(carstate, lateral_active, lane_change_prob,
left_edge_detected=False, right_edge_detected=False)
assert dh.desire == expected_desire
def test_edge_blocks_lane_change(set_lane_turn_params):
dh = DesireHelper()
dh.alc.lane_change_set_timer = AutoLaneChangeMode.NUDGE
carstate = DummyCarState(vEgo=15, leftBlinker=True, steeringPressed=True, steeringTorque=1)
for _ in range(10):
dh.update(carstate, True, 1.0, left_edge_detected=True, right_edge_detected=False)
assert dh.lane_change_state == LaneChangeState.preLaneChange
assert dh.lane_change_direction == LaneChangeDirection.left
assert dh.desire == log.Desire.none
def test_edge_blocks_lane_change(self, set_lane_turn_params):
dh = DesireHelper()
dh.alc.lane_change_set_timer = AutoLaneChangeMode.NUDGE
carstate = DummyCarState(vEgo=15, leftBlinker=True, steeringPressed=True, steeringTorque=1)
for _ in range(10):
dh.update(carstate, True, 1.0, left_edge_detected=True, right_edge_detected=False)
assert dh.lane_change_state == LaneChangeState.preLaneChange
assert dh.lane_change_direction == LaneChangeDirection.left
assert dh.desire == log.Desire.none
@@ -19,6 +19,7 @@ from openpilot.selfdrive.locationd.helpers import Pose
from openpilot.common.mock.generators import generate_deviceMotion
from openpilot.sunnypilot.selfdrive.car import interfaces as sunnypilot_interfaces
from openpilot.selfdrive.modeld.constants import ModelConstants
from openpilot.common.test import OpenpilotTestCase
def _make_controller(enhanced=False, nnlc=False):
@@ -71,7 +72,7 @@ def _run_update(controller, VM):
return controller.update(True, CS, VM, params, False, 0.5, pose, False, 0.2)
class TestLatControlTorqueExt:
class TestLatControlTorqueExt(OpenpilotTestCase):
def test_init_enhanced_only(self):
controller, VM, _ = _make_controller(enhanced=True, nnlc=False)
assert controller.extension._jerk_aware_enabled
@@ -4,7 +4,8 @@ Copyright (c) 2021-, rav4kumar, sunnypilot, and a number of other contributors.
This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.common.test import OpenpilotTestCase
from openpilot.common.realtime import DT_MDL
from openpilot.sunnypilot.selfdrive.controls.lib.relc import (
@@ -29,7 +30,6 @@ CLOSE_EDGES = edges(-2.0, 1.5)
FAR_EDGES = edges(-10.0, 10.0)
@pytest.fixture
def relc(mocker):
mocker.patch("openpilot.sunnypilot.selfdrive.controls.lib.relc.Params")
controller = RoadEdgeLaneChangeController()
@@ -42,128 +42,129 @@ def drive(controller, road_edge_stds, lane_line_probs, seconds, v_ego=V_HIGH, ro
controller.update(road_edge_stds, lane_line_probs, v_ego, road_edges)
@pytest.mark.parametrize("road_edge_stds,lane_line_probs,attr", [
([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], "left_edge_detected"),
([0.9, 0.0], [0.8, 0.8, 0.8, 0.0], "right_edge_detected"),
])
def test_edge_detection(relc, road_edge_stds, lane_line_probs, attr):
drive(relc, road_edge_stds, lane_line_probs, EDGE_REACTION_TIME + 0.1)
assert getattr(relc, attr)
class TestRELC(OpenpilotTestCase):
@parameterized.expand([
([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], "left_edge_detected"),
([0.9, 0.0], [0.8, 0.8, 0.8, 0.0], "right_edge_detected"),
], names=["road_edge_stds", "lane_line_probs", "attr"])
def test_edge_detection(self, relc, road_edge_stds, lane_line_probs, attr):
drive(relc, road_edge_stds, lane_line_probs, EDGE_REACTION_TIME + 0.1)
assert getattr(relc, attr)
def test_edge_detection_requires_time(relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME - 0.05)
assert not relc.left_edge_detected
def test_edge_detection_requires_time(self, relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME - 0.05)
assert not relc.left_edge_detected
def test_both_edges_detected(relc):
drive(relc, [0.0, 0.0], [0.0, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
assert relc.right_edge_detected
def test_both_edges_detected(self, relc):
drive(relc, [0.0, 0.0], [0.0, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
assert relc.right_edge_detected
def test_noise_doesnt_clear(relc):
edge = ([0.0, 0.9], [0.0, 0.8, 0.8, 0.8])
clear = ([0.9, 0.9], [0.8, 0.8, 0.8, 0.8])
def test_noise_doesnt_clear(self, relc):
edge = ([0.0, 0.9], [0.0, 0.8, 0.8, 0.8])
clear = ([0.9, 0.9], [0.8, 0.8, 0.8, 0.8])
drive(relc, *edge, EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
drive(relc, *edge, EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
relc.update(*clear, V_HIGH, CLOSE_EDGES)
relc.update(*edge, V_HIGH, CLOSE_EDGES)
assert relc.left_edge_detected
relc.update(*clear, V_HIGH, CLOSE_EDGES)
relc.update(*edge, V_HIGH, CLOSE_EDGES)
assert relc.left_edge_detected
def test_clears_after_window(relc):
edge = ([0.0, 0.9], [0.0, 0.8, 0.8, 0.8])
clear = ([0.9, 0.9], [0.8, 0.8, 0.8, 0.8])
def test_clears_after_window(self, relc):
edge = ([0.0, 0.9], [0.0, 0.8, 0.8, 0.8])
clear = ([0.9, 0.9], [0.8, 0.8, 0.8, 0.8])
drive(relc, *edge, EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
drive(relc, *edge, EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
drive(relc, *clear, EDGE_CLEAR_TIME + 0.05)
assert not relc.left_edge_detected
assert relc.left_edge_timer == 0.0
drive(relc, *clear, EDGE_CLEAR_TIME + 0.05)
assert not relc.left_edge_detected
assert relc.left_edge_timer == 0.0
def test_low_speed_skips(relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1, v_ego=V_LOW)
assert not relc.left_edge_detected
assert relc.left_edge_timer == 0.0
def test_low_speed_skips(self, relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1, v_ego=V_LOW)
assert not relc.left_edge_detected
assert relc.left_edge_timer == 0.0
def test_speed_drop_resets(relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
def test_speed_drop_resets(self, relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
relc.update([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], V_LOW, CLOSE_EDGES)
assert not relc.left_edge_detected
relc.update([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], V_LOW, CLOSE_EDGES)
assert not relc.left_edge_detected
def test_param_off_resets(relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
def test_param_off_resets(self, relc):
drive(relc, [0.0, 0.9], [0.0, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert relc.left_edge_detected
relc.params.get_bool.return_value = False
relc.read_params()
relc.update([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], V_HIGH, CLOSE_EDGES)
assert not relc.left_edge_detected
assert not relc.right_edge_detected
relc.params.get_bool.return_value = False
relc.read_params()
relc.update([0.0, 0.9], [0.0, 0.8, 0.8, 0.8], V_HIGH, CLOSE_EDGES)
assert not relc.left_edge_detected
assert not relc.right_edge_detected
def test_lane_line_prevents_detection(relc):
drive(relc, [0.0, 0.9], [0.8, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert not relc.left_edge_detected
def test_lane_line_prevents_detection(self, relc):
drive(relc, [0.0, 0.9], [0.8, 0.8, 0.8, 0.8], EDGE_REACTION_TIME + 0.1)
assert not relc.left_edge_detected
def test_one_side_blocks_other_allows(relc):
drive(relc, [0.9, 0.0], [0.8, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert relc.right_edge_detected
assert not relc.left_edge_detected
def test_one_side_blocks_other_allows(self, relc):
drive(relc, [0.9, 0.0], [0.8, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert relc.right_edge_detected
assert not relc.left_edge_detected
def test_disabled_no_detection(relc):
relc.enabled = False
relc.params.get_bool.return_value = False
drive(relc, [0.0, 0.0], [0.0, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert not relc.left_edge_detected
assert not relc.right_edge_detected
def test_disabled_no_detection(self, relc):
relc.enabled = False
relc.params.get_bool.return_value = False
drive(relc, [0.0, 0.0], [0.0, 0.8, 0.8, 0.0], EDGE_REACTION_TIME + 0.1)
assert not relc.left_edge_detected
assert not relc.right_edge_detected
def test_far_edge_no_block(relc):
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1, road_edges=FAR_EDGES)
assert not relc.left_edge_detected
def test_far_edge_no_block(self, relc):
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1, road_edges=FAR_EDGES)
assert not relc.left_edge_detected
def test_close_edge_blocks(relc):
drive(relc, [0.9, 0.0], [0.05, 0.8, 0.8, 0.05], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-8.0, 1.5))
assert relc.right_edge_detected
assert not relc.left_edge_detected
def test_close_edge_blocks(self, relc):
drive(relc, [0.9, 0.0], [0.05, 0.8, 0.8, 0.05], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-8.0, 1.5))
assert relc.right_edge_detected
assert not relc.left_edge_detected
def test_wide_road_no_lines_no_block(relc):
drive(relc, [0.0, 0.0], [0.05, 0.4, 0.4, 0.05], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-8.0, 8.0))
assert not relc.left_edge_detected
assert not relc.right_edge_detected
def test_wide_road_no_lines_no_block(self, relc):
drive(relc, [0.0, 0.0], [0.05, 0.4, 0.4, 0.05], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-8.0, 8.0))
assert not relc.left_edge_detected
assert not relc.right_edge_detected
def test_narrow_road_both_block(relc):
drive(relc, [0.0, 0.0], [0.02, 0.4, 0.4, 0.02], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-2.5, 2.5))
assert relc.left_edge_detected
assert relc.right_edge_detected
def test_narrow_road_both_block(self, relc):
drive(relc, [0.0, 0.0], [0.02, 0.4, 0.4, 0.02], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-2.5, 2.5))
assert relc.left_edge_detected
assert relc.right_edge_detected
def test_clearance_boundary(relc):
boundary = VEHICLE_EDGE_MARGIN + EDGE_CLEARANCE # 4.78m
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-(boundary - 0.1), 10.0))
assert relc.left_edge_detected
def test_clearance_boundary(self, relc):
boundary = VEHICLE_EDGE_MARGIN + EDGE_CLEARANCE # 4.78m
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-(boundary - 0.1), 10.0))
assert relc.left_edge_detected
relc.reset()
relc.reset()
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-(boundary + 0.1), 10.0))
assert not relc.left_edge_detected
drive(relc, [0.0, 0.9], [0.05, 0.5, 0.5, 0.08], EDGE_REACTION_TIME + 0.1,
road_edges=edges(-(boundary + 0.1), 10.0))
assert not relc.left_edge_detected
@@ -1,5 +1,5 @@
import pytest
import platform
import unittest
import json
import random
import time
@@ -11,13 +11,11 @@ from openpilot.common.params import Params
from openpilot.common.transformations.coordinates import ecef2geodetic
from openpilot.system.manager.process_config import managed_processes
from openpilot.common.test import OpenpilotTestCase
if platform.system() == 'Darwin':
pytest.skip("Skipping locationd test on macOS due to unsupported msgq.", allow_module_level=True)
class TestLocationdProc:
@unittest.skipIf(platform.system() == 'Darwin', "msgq unsupported on macOS")
class TestLocationdProc(OpenpilotTestCase):
LLD_MSGS = ['gpsLocationExternal', 'cameraOdometry', 'carState', 'extrinsicsCalibration',
'accelerometer', 'gyroscope']
@@ -88,6 +86,6 @@ class TestLocationdProc:
time.sleep(1) # wait for async params write
lastGPS = json.loads(self.params.get('LastGPSPositionLLK'))
assert lastGPS['latitude'] == pytest.approx(self.lat, abs=0.001)
assert lastGPS['longitude'] == pytest.approx(self.lon, abs=0.001)
assert lastGPS['altitude'] == pytest.approx(self.alt, abs=0.001)
self.assertAlmostEqual(lastGPS['latitude'], self.lat, delta=0.001)
self.assertAlmostEqual(lastGPS['longitude'], self.lon, delta=0.001)
self.assertAlmostEqual(lastGPS['altitude'], self.alt, delta=0.001)
@@ -6,12 +6,13 @@ See the LICENSE.md file in the root directory for more details.
"""
from opendbc.car.structs import car
from openpilot.sunnypilot.selfdrive.selfdrived.button_state_tracker import ButtonStateTracker
from openpilot.common.test import OpenpilotTestCase
ButtonEvent = car.CarState.ButtonEvent
ButtonType = car.CarState.ButtonEvent.Type
class TestButtonStateTracker:
class TestButtonStateTracker(OpenpilotTestCase):
def setup_method(self) -> None:
self.tracker = ButtonStateTracker()
@@ -5,9 +5,10 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
from openpilot.sunnypilot.sunnylink.athena import sunnylinkd
from openpilot.common.test import OpenpilotTestCase
class TestSunnylinkdMethods:
class TestSunnylinkdMethods(OpenpilotTestCase):
def setup_method(self):
self.saved_params = []
@@ -12,7 +12,6 @@ the same commit so the bump shows up in code review.
"""
from __future__ import annotations
import pytest
from openpilot.sunnypilot.sunnylink.capabilities import (
CAPABILITY_DEFAULTS,
@@ -21,18 +20,18 @@ from openpilot.sunnypilot.sunnylink.capabilities import (
PROTOCOL_VERSION,
generate_capabilities,
)
from openpilot.common.test import OpenpilotTestCase
KNOWN_PROTOCOL_VERSIONS = (1,)
LATEST_KNOWN = max(KNOWN_PROTOCOL_VERSIONS)
@pytest.fixture(scope="module")
def caps():
return generate_capabilities()
class TestProtocolVersion:
class TestProtocolVersion(OpenpilotTestCase):
def test_protocol_version_in_capability_fields(self):
assert "protocol_version" in CAPABILITY_FIELDS
@@ -63,7 +62,7 @@ class TestProtocolVersion:
)
class TestOpaquePerBrandFlags:
class TestOpaquePerBrandFlags(OpenpilotTestCase):
def test_subaru_has_sng_field_present(self):
assert "subaru_has_sng" in CAPABILITY_FIELDS
@@ -77,7 +76,7 @@ class TestOpaquePerBrandFlags:
assert caps["hyundai_alpha_long_available"] is False
class TestCapabilitiesShape:
class TestCapabilitiesShape(OpenpilotTestCase):
def test_all_fields_present(self, caps):
for field in CAPABILITY_FIELDS:
assert field in caps, f"capabilities missing {field}"
@@ -19,7 +19,6 @@ import difflib
import json
import os
import pytest
import yaml
from openpilot.sunnypilot.sunnylink.tools.compile_settings_ui import (
@@ -29,20 +28,19 @@ from openpilot.sunnypilot.sunnylink.tools.compile_settings_ui import (
_resolve_refs,
compile_schema,
)
from openpilot.common.test import OpenpilotTestCase
@pytest.fixture(scope="module")
def compiled() -> dict:
return compile_schema(DEFAULT_SRC)
@pytest.fixture(scope="module")
def committed() -> dict:
with open(DEFAULT_OUT) as f:
return json.load(f)
class TestRoundtrip:
class TestRoundtrip(OpenpilotTestCase):
def test_compiled_matches_committed(self, compiled, committed):
"""Compiled output must match the checked-in JSON."""
if compiled == committed:
@@ -54,7 +52,7 @@ class TestRoundtrip:
tofile="settings_ui.json (freshly compiled)",
lineterm="",
))
pytest.fail(f"settings_ui.json schema mismatch — run compile_settings_ui.py\n\n{diff}")
self.fail(f"settings_ui.json schema mismatch — run compile_settings_ui.py\n\n{diff}")
def test_committed_file_is_canonical(self):
"""Compiled output must byte-match the checked-in file (including trailing newline).
@@ -72,10 +70,10 @@ class TestRoundtrip:
tofile="settings_ui.json (freshly compiled)",
lineterm="",
))
pytest.fail(f"settings_ui.json out of sync — run compile_settings_ui.py\n\n{diff}")
self.fail(f"settings_ui.json out of sync — run compile_settings_ui.py\n\n{diff}")
class TestRefResolution:
class TestRefResolution(OpenpilotTestCase):
def test_list_context_splices(self):
macros = {"a": [{"type": "offroad_only"}], "b": [{"type": "not_engaged"}]}
out = _resolve_refs([{"$ref": "#/macros/a"}, {"$ref": "#/macros/b"}], macros)
@@ -95,12 +93,12 @@ class TestRefResolution:
assert out == [{"type": "offroad_only"}]
def test_unknown_macro_raises(self):
with pytest.raises(CompileError, match="unknown macro"):
with self.assertRaisesRegex(CompileError, "unknown macro"):
_resolve_refs([{"$ref": "#/macros/missing"}], {})
def test_cycle_raises(self):
macros = {"a": [{"$ref": "#/macros/b"}], "b": [{"$ref": "#/macros/a"}]}
with pytest.raises(CompileError, match="cycle"):
with self.assertRaisesRegex(CompileError, "cycle"):
_resolve_refs([{"$ref": "#/macros/a"}], macros)
def test_depth_limit(self):
@@ -111,20 +109,20 @@ class TestRefResolution:
"l3": [{"$ref": "#/macros/l4"}],
"l4": [{"type": "offroad_only"}],
}
with pytest.raises(CompileError, match="depth"):
with self.assertRaisesRegex(CompileError, "depth"):
_resolve_refs([{"$ref": "#/macros/l1"}], macros)
def test_invalid_ref_scheme(self):
with pytest.raises(CompileError, match="unsupported"):
with self.assertRaisesRegex(CompileError, "unsupported"):
_resolve_refs([{"$ref": "https://example.com/x"}], {})
def test_scalar_macro_in_list_context_raises(self):
macros = {"x": {"type": "offroad_only"}} # macro is a single rule (dict), not a list
with pytest.raises(CompileError, match="must resolve to a list"):
with self.assertRaisesRegex(CompileError, "must resolve to a list"):
_resolve_refs([{"$ref": "#/macros/x"}], macros)
class TestCompiledShape:
class TestCompiledShape(OpenpilotTestCase):
def test_panels_present(self, compiled):
assert isinstance(compiled["panels"], list)
assert len(compiled["panels"]) == 9
@@ -145,7 +143,7 @@ class TestCompiledShape:
def walk(node):
if isinstance(node, dict):
if "$ref" in node:
pytest.fail(f"unresolved $ref: {node}")
self.fail(f"unresolved $ref: {node}")
for v in node.values():
walk(v)
elif isinstance(node, list):
@@ -154,7 +152,7 @@ class TestCompiledShape:
walk(compiled)
class TestSourceTreeIntegrity:
class TestSourceTreeIntegrity(OpenpilotTestCase):
def test_macros_yaml_well_formed(self):
with open(os.path.join(DEFAULT_SRC, "_macros.yaml")) as f:
doc = yaml.safe_load(f)
@@ -15,7 +15,7 @@ import json
import os
from typing import Any
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import (
DEFINITION_PATH,
@@ -24,6 +24,7 @@ from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import (
_load_torque_versions,
generate_schema,
)
from openpilot.common.test import OpenpilotTestCase
SCHEMA_VALIDATOR_PATH = os.path.join(os.path.dirname(DEFINITION_PATH), "settings_ui.schema.json")
@@ -105,12 +106,11 @@ def _references_capability_field(rules: list[dict[str, Any]] | None, field: str)
return found
@pytest.fixture(scope="module")
def schema():
return generate_schema()
class TestMadsBrandGates:
class TestMadsBrandGates(OpenpilotTestCase):
def test_mads_main_cruise_has_brand_gate(self, schema):
"""MadsMainCruiseAllowed must gate on brand and tesla_has_vehicle_bus."""
item = _find_item(schema, "MadsMainCruiseAllowed")
@@ -126,7 +126,7 @@ class TestMadsBrandGates:
assert _references_capability_field(item.get("enablement"), "tesla_has_vehicle_bus")
class TestTestManeuversSection:
class TestTestManeuversSection(OpenpilotTestCase):
def test_lateral_maneuver_mode_in_test_maneuvers(self, schema):
section = _find_section(schema, "developer", "test_maneuvers")
assert section is not None, "developer.test_maneuvers section missing"
@@ -153,10 +153,13 @@ class TestTestManeuversSection:
"test_maneuvers must gate ShowAdvancedControls via enablement"
class TestValidator:
class TestValidator(OpenpilotTestCase):
def test_validator_accepts_real_json(self):
"""settings_ui.json validates against settings_ui.schema.json."""
jsonschema = pytest.importorskip("jsonschema")
try:
import jsonschema
except ImportError:
self.skipTest("jsonschema not installed")
with open(DEFINITION_PATH) as f:
data = json.load(f)
with open(SCHEMA_VALIDATOR_PATH) as f:
@@ -164,7 +167,7 @@ class TestValidator:
jsonschema.validate(instance=data, schema=validator)
class TestTorqueOptionGeneration:
class TestTorqueOptionGeneration(OpenpilotTestCase):
def test_torque_versions_match_generated_options(self, schema):
versions = _load_torque_versions()
assert versions, "latcontrol_torque_versions.json must have at least one version"
@@ -179,11 +182,11 @@ class TestTorqueOptionGeneration:
)
class TestReleaseBranchGates:
@pytest.mark.parametrize("key", [
class TestReleaseBranchGates(OpenpilotTestCase):
@parameterized.expand([
"EnableGithubRunner",
"QuickBootToggle",
])
], names=["key"])
def test_sp_dev_items_gate_on_is_sp_release(self, schema, key):
"""sunnypilot dev items must hide on sunnypilot release branches (is_sp_release gate)."""
item = _find_item(schema, key)
@@ -192,7 +195,7 @@ class TestReleaseBranchGates:
assert _references_capability_field(rules, "is_sp_release"), f"{key} missing is_sp_release gate"
class TestSpuriousOffroadGatesDropped:
class TestSpuriousOffroadGatesDropped(OpenpilotTestCase):
def test_disengage_on_accelerator_has_no_offroad_only(self, schema):
item = _find_item(schema, "DisengageOnAccelerator")
assert item is not None
@@ -204,12 +207,12 @@ class TestSpuriousOffroadGatesDropped:
assert "offroad_only" not in _flatten_rule_types(item.get("enablement"))
class TestNotEngagedReplacement:
@pytest.mark.parametrize("key", [
class TestNotEngagedReplacement(OpenpilotTestCase):
@parameterized.expand([
"AlphaLongitudinalEnabled",
"ToyotaEnforceStockLongitudinal",
"ToyotaStopAndGoHack",
])
], names=["key"])
def test_offroad_only_replaced_with_not_engaged(self, schema, key):
"""These items should use not_engaged, not offroad_only."""
item = _find_item(schema, key)
@@ -5,7 +5,6 @@ This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import json
import pytest
from openpilot.common.params import Params
from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import (
@@ -16,6 +15,7 @@ from openpilot.sunnypilot.sunnylink.tools.generate_settings_schema import (
collect_capability_refs,
)
from openpilot.sunnypilot.sunnylink.capabilities import CAPABILITY_FIELDS
from openpilot.common.test import OpenpilotTestCase
VALID_WIDGET_TYPES = {"toggle", "option", "multiple_button", "button", "info"}
@@ -55,18 +55,16 @@ def _brand_items(brand_data) -> list[dict]:
return []
@pytest.fixture(scope="module")
def schema():
return generate_schema()
@pytest.fixture(scope="module")
def all_param_keys():
"""All keys registered in the device param store."""
return {k.decode("utf-8") for k in Params().all_keys()}
class TestSchemaStructure:
class TestSchemaStructure(OpenpilotTestCase):
def test_schema_is_valid_json(self):
"""Schema serializes to valid JSON."""
raw = generate_schema_json()
@@ -141,16 +139,16 @@ class TestSchemaStructure:
for item in _iter_panel_items(panel):
key = item["key"]
if key in seen:
pytest.fail(f"Key '{key}' appears in both panel '{seen[key]}' and '{panel['id']}'")
self.fail(f"Key '{key}' appears in both panel '{seen[key]}' and '{panel['id']}'")
seen[key] = panel["id"]
for sub in item.get("sub_items", []):
sub_key = sub["key"]
if sub_key in seen:
pytest.fail(f"Sub-item key '{sub_key}' appears in both '{seen[sub_key]}' and '{panel['id']}'")
self.fail(f"Sub-item key '{sub_key}' appears in both '{seen[sub_key]}' and '{panel['id']}'")
seen[sub_key] = panel["id"]
class TestSchemaCoverage:
class TestSchemaCoverage(OpenpilotTestCase):
def test_all_schema_keys_exist_in_params(self, schema, all_param_keys):
"""Schema keys must exist in Params().all_keys()."""
schema_keys = collect_all_keys(schema)
@@ -169,7 +167,7 @@ class TestSchemaCoverage:
assert set(schema["capability_fields"]) == set(CAPABILITY_FIELDS)
class TestRuleWellFormedness:
class TestRuleWellFormedness(OpenpilotTestCase):
def _validate_rule(self, rule: dict, context: str = ""):
"""Recursively validate a single rule dict."""
assert "type" in rule, f"Rule missing 'type' in {context}"
@@ -232,7 +230,7 @@ class TestRuleWellFormedness:
key = item.get("key")
for rule in item.get(rules_field, []):
if rule.get("type") == "param" and rule.get("key") == key:
pytest.fail(f"Item {key} has self-referencing {rules_field} rule")
self.fail(f"Item {key} has self-referencing {rules_field} rule")
for panel in schema["panels"]:
for item in _iter_panel_items(panel):
@@ -245,7 +243,7 @@ class TestRuleWellFormedness:
_check_self_ref(item, "enablement")
class TestKnownPanels:
class TestKnownPanels(OpenpilotTestCase):
def test_expected_panels_exist(self, schema):
panel_ids = {p["id"] for p in schema["panels"]}
expected = {"steering", "cruise", "display", "visuals", "device", "software", "developer"}
@@ -279,7 +277,7 @@ class TestKnownPanels:
assert "NeuralNetworkLateralControl" in enhanced_enable_keys
class TestKnownVehicleSettings:
class TestKnownVehicleSettings(OpenpilotTestCase):
def test_hyundai_has_longitudinal_tuning(self, schema):
keys = {i["key"] for i in _brand_items(schema["vehicle_settings"].get("hyundai"))}
assert "HyundaiLongitudinalTuning" in keys
@@ -299,7 +297,7 @@ class TestKnownVehicleSettings:
assert "SubaruStopAndGoManualParkingBrake" in keys
class TestItemCompleteness:
class TestItemCompleteness(OpenpilotTestCase):
def _collect_all_items(self, schema):
"""Collect all items and sub_items from panels and vehicle_settings."""
items = []
@@ -319,7 +317,7 @@ class TestItemCompleteness:
"""All items must have titles."""
missing = [i["key"] for i in self._collect_all_items(schema) if "title" not in i]
if len(missing) > MAX_ALLOWED_MISSING_TITLES:
pytest.fail(f"Items without titles ({len(missing)}): {missing[:10]}")
self.fail(f"Items without titles ({len(missing)}): {missing[:10]}")
def test_no_default_titles(self, schema):
"""Item titles must differ from keys."""
@@ -1,6 +1,5 @@
import os
import subprocess
import pytest
import time
import numpy as np
from collections import namedtuple, defaultdict
@@ -12,6 +11,7 @@ from openpilot.common.gpio import get_irqs_for_action
from openpilot.common.timeout import Timeout
from openpilot.common.hardware import HARDWARE
from openpilot.system.manager.process_config import managed_processes
from openpilot.common.test import OpenpilotTestCase
BMX = {
('bmx055', 'acceleration'),
@@ -103,8 +103,9 @@ def read_sensor_events(duration_sec):
return {k: v for k, v in events.items() if len(v) > 0}
@pytest.mark.tici
class TestSensord:
class TestSensord(OpenpilotTestCase):
COMMA_HARDWARE_TEST = True
@classmethod
def setup_class(cls):
# enable LSM self test
@@ -4,70 +4,72 @@ Copyright (c) 2021-, Haibin Wen, sunnypilot, and a number of other contributors.
This file is part of sunnypilot and is licensed under the MIT License.
See the LICENSE.md file in the root directory for more details.
"""
import pytest
from openpilot.common.parameterized import parameterized
from openpilot.common.test import OpenpilotTestCase
from openpilot.common.params import Params
from openpilot.system.updated.updated import Updater
@pytest.mark.parametrize(("device_type", "branch", "expected"), [
("tici", "staging-c3-new", "staging-tici"),
("tici", "dev-c3-new", "staging-tici"),
("tici", "master", "master-tici"),
("tici", "master-dev-c3-new", "master-tici"),
("tizi", "staging-c3-new", "staging"),
("tizi", "dev-c3-new", "dev"),
("tizi", "master-dev-c3-new", "master-dev"),
("tizi", "release3", "release-tizi"),
("tizi", "release3-staging", "release-tizi-staging"),
("mici", "release3", "release-mici"),
("mici", "release3-staging", "release-mici-staging"),
])
def test_sp_branch_migrations_from_current_branch(mocker, device_type, branch, expected):
params = Params()
params.remove("UpdaterTargetBranch")
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
mocker.patch.object(Updater, "get_branch", return_value=branch)
assert Updater().target_branch == expected
@pytest.mark.parametrize(("device_type", "branch", "expected"), [
("tici", "staging-c3-new", "staging-tici"),
("tici", "dev-c3-new", "staging-tici"),
("tici", "master", "master-tici"),
("tici", "master-dev-c3-new", "master-tici"),
("tizi", "staging-c3-new", "staging"),
("tizi", "dev-c3-new", "dev"),
("tizi", "master-dev-c3-new", "master-dev"),
("tizi", "release3", "release-tizi"),
("tizi", "release3-staging", "release-tizi-staging"),
("mici", "release3", "release-mici"),
("mici", "release3-staging", "release-mici-staging"),
])
def test_sp_branch_migrations_from_param(mocker, device_type, branch, expected):
params = Params()
params.put("UpdaterTargetBranch", branch, block=True)
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
try:
assert Updater().target_branch == expected
finally:
class TestBranchMigrations(OpenpilotTestCase):
@parameterized.expand([
("tici", "staging-c3-new", "staging-tici"),
("tici", "dev-c3-new", "staging-tici"),
("tici", "master", "master-tici"),
("tici", "master-dev-c3-new", "master-tici"),
("tizi", "staging-c3-new", "staging"),
("tizi", "dev-c3-new", "dev"),
("tizi", "master-dev-c3-new", "master-dev"),
("tizi", "release3", "release-tizi"),
("tizi", "release3-staging", "release-tizi-staging"),
("mici", "release3", "release-mici"),
("mici", "release3-staging", "release-mici-staging"),
], names=["device_type", "branch", "expected"])
def test_sp_branch_migrations_from_current_branch(self, mocker, device_type, branch, expected):
params = Params()
params.remove("UpdaterTargetBranch")
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
mocker.patch.object(Updater, "get_branch", return_value=branch)
@pytest.mark.parametrize(("device_type", "branch"), [
("tici", "unknown"),
("tizi", "unknown"),
("mici", "unknown"),
])
def test_sp_branch_migrations_passthrough(mocker, device_type, branch):
params = Params()
params.remove("UpdaterTargetBranch")
assert Updater().target_branch == expected
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
mocker.patch.object(Updater, "get_branch", return_value=branch)
assert Updater().target_branch == branch
@parameterized.expand([
("tici", "staging-c3-new", "staging-tici"),
("tici", "dev-c3-new", "staging-tici"),
("tici", "master", "master-tici"),
("tici", "master-dev-c3-new", "master-tici"),
("tizi", "staging-c3-new", "staging"),
("tizi", "dev-c3-new", "dev"),
("tizi", "master-dev-c3-new", "master-dev"),
("tizi", "release3", "release-tizi"),
("tizi", "release3-staging", "release-tizi-staging"),
("mici", "release3", "release-mici"),
("mici", "release3-staging", "release-mici-staging"),
], names=["device_type", "branch", "expected"])
def test_sp_branch_migrations_from_param(self, mocker, device_type, branch, expected):
params = Params()
params.put("UpdaterTargetBranch", branch, block=True)
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
try:
assert Updater().target_branch == expected
finally:
params.remove("UpdaterTargetBranch")
@parameterized.expand([
("tici", "unknown"),
("tizi", "unknown"),
("mici", "unknown"),
], names=["device_type", "branch"])
def test_sp_branch_migrations_passthrough(self, mocker, device_type, branch):
params = Params()
params.remove("UpdaterTargetBranch")
mocker.patch("openpilot.system.updated.updated.HARDWARE.get_device_type", return_value=device_type)
mocker.patch.object(Updater, "get_branch", return_value=branch)
assert Updater().target_branch == branch
-24
View File
@@ -65,11 +65,6 @@ dev = [
testing = [
"coverage",
"ty",
"pytest",
"pytest-cpp",
"pytest-subtests",
"pytest-xdist @ git+https://github.com/sshane/pytest-xdist@2b4372bd62699fb412c4fe2f95bf9f01bd2018da",
"pytest-mock",
"ruff",
"codespell",
]
@@ -115,24 +110,6 @@ packages = [
[tool.hatch.metadata]
allow-direct-references = true
# TODO-SP: upstream replaced pytest with a custom unittest runner (tools/test_runner.py).
# Remove this section and the pytest deps once sunnypilot tests are migrated to unittest.
[tool.pytest.ini_options]
minversion = "6.0"
addopts = "-Werror --strict-config --strict-markers --durations=10 -n auto --dist=loadgroup"
python_files = "test_*.py"
markers = [
"slow: tests that take awhile to run and can be skipped with -m 'not slow'",
"tici: tests that are only meant to run on the C3/C3X",
"skip_tici_setup: mark test to skip tici setup fixture",
"nocapture: don't capture test output",
"shared_download_cache: share download cache between tests",
"xdist_group_class_property: group tests by a property of the class that contains them",
]
testpaths = [
"openpilot",
]
[tool.codespell]
quiet-level = 3
# if you've got a short variable name that's getting flagged, add it here
@@ -171,7 +148,6 @@ exclude = [
lint.flake8-implicit-str-concat.allow-multiline = false
[tool.ruff.lint.flake8-tidy-imports.banned-api]
"pytest.main".msg = "pytest.main requires special handling that is easy to mess up!"
"time.time".msg = "Use time.monotonic. time.time can skip due to its reference clock, you probably want a monotonic clock"
"pyray.measure_text_ex".msg = "Use openpilot.system.ui.lib.text_measure"
"pyray.is_mouse_button_pressed".msg = "This can miss events. Use Widget._handle_mouse_press"
Generated
-108
View File
@@ -18,15 +18,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/da/35/f2287558c17e29fafc8ef3daf819bb9834061cfa43bff8014f7df7f63bdc/anyio-4.14.2-py3-none-any.whl", hash = "sha256:9f505dda5ac9f0c8309b5e8bd445a8c2bf7246f3ce950121e45ea15bc41d1494", size = 125813, upload-time = "2026-07-12T20:29:05.763Z" },
]
[[package]]
name = "attrs"
version = "26.1.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/9a/8e/82a0fe20a541c03148528be8cac2408564a6c9a0cc7e9171802bc1d26985/attrs-26.1.0.tar.gz", hash = "sha256:d03ceb89cb322a8fd706d4fb91940737b6642aa36998fe130a9bc96c985eff32", size = 952055, upload-time = "2026-03-19T14:22:25.026Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/64/b4/17d4b0b2a2dc85a6df63d1157e028ed19f90d4cd97c36717afef2bc2f395/attrs-26.1.0-py3-none-any.whl", hash = "sha256:c647aa4a12dfbad9333ca4e71fe62ddc36f4e63b2d260a37a8b83d2f043ac309", size = 67548, upload-time = "2026-03-19T14:22:23.645Z" },
]
[[package]]
name = "certifi"
version = "2026.7.22"
@@ -378,15 +369,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/51/25/2a75b47cb057b1e164c604fb81ab690a6cdb5e2260ce651194eae90f64a3/deepmerge-2.1.0-py3-none-any.whl", hash = "sha256:8f148339a91d680a75ecb74ade235d9e759a93df373a0b04e9d31c8666cfeb75", size = 14345, upload-time = "2026-06-22T05:46:06.742Z" },
]
[[package]]
name = "execnet"
version = "2.1.2"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/bf/89/780e11f9588d9e7128a3f87788354c7946a9cbb1401ad38a48c4db9a4f07/execnet-2.1.2.tar.gz", hash = "sha256:63d83bfdd9a23e35b9c6a3261412324f964c2ec8dcd8d3c6916ee9373e0befcd", size = 166622, upload-time = "2025-11-12T09:56:37.75Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/ab/84/02fc1827e8cdded4aa65baef11296a9bbe595c474f0d6d758af082d849fd/execnet-2.1.2-py3-none-any.whl", hash = "sha256:67fba928dd5a544b783f6056f449e5e3931a5c378b128bc18501f7ea79e296ec", size = 40708, upload-time = "2025-11-12T09:56:36.333Z" },
]
[[package]]
name = "filelock"
version = "3.32.2"
@@ -513,15 +495,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/8a/db/55a262f3606bebcae07cc14095338471ad7c0bbcaa37707e6f0ee49725b7/importlib_resources-7.1.0-py3-none-any.whl", hash = "sha256:1bd7b48b4088eddb2cd16382150bb515af0bd2c70128194392725f82ad2c96a1", size = 37232, upload-time = "2026-04-12T16:36:08.219Z" },
]
[[package]]
name = "iniconfig"
version = "2.3.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/72/34/14ca021ce8e5dfedc35312d08ba8bf51fdd999c576889fc2c24cb97f4f10/iniconfig-2.3.0.tar.gz", hash = "sha256:c76315c77db068650d49c5b56314774a7804df16fee4402c1f19d6d15d8c4730", size = 20503, upload-time = "2025-10-18T21:55:43.219Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
[[package]]
name = "inputs"
version = "0.5"
@@ -808,11 +781,6 @@ submodules = [
testing = [
{ name = "codespell" },
{ name = "coverage" },
{ name = "pytest" },
{ name = "pytest-cpp" },
{ name = "pytest-mock" },
{ name = "pytest-subtests" },
{ name = "pytest-xdist" },
{ name = "ruff" },
{ name = "ty" },
]
@@ -856,11 +824,6 @@ requires-dist = [
{ name = "pandacan", marker = "extra == 'submodules'", editable = "panda" },
{ name = "pycapnp", specifier = "==2.1.0" },
{ name = "pyjwt", extras = ["crypto"] },
{ name = "pytest", marker = "extra == 'testing'" },
{ name = "pytest-cpp", marker = "extra == 'testing'" },
{ name = "pytest-mock", marker = "extra == 'testing'" },
{ name = "pytest-subtests", marker = "extra == 'testing'" },
{ name = "pytest-xdist", marker = "extra == 'testing'", git = "https://github.com/sshane/pytest-xdist?rev=2b4372bd62699fb412c4fe2f95bf9f01bd2018da" },
{ name = "pyzmq" },
{ name = "rednose", marker = "extra == 'submodules'", editable = "rednose_repo" },
{ name = "requests" },
@@ -939,15 +902,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/de/47/4845a0a6c0dbf1db8456bd9fc791f13c5ced7ced20606d08a0aacfd25b49/pillow-12.3.0-cp312-cp312-win_arm64.whl", hash = "sha256:331b624368d4f1d069149002f25f44bc61c8919ce8ddb3c45bdad8f6e2d89510", size = 2568267, upload-time = "2026-07-01T11:54:24.051Z" },
]
[[package]]
name = "pluggy"
version = "1.6.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f9/e2/3e91f31a7d2b083fe6ef3fa267035b518369d9511ffab804f839851d2779/pluggy-1.6.0.tar.gz", hash = "sha256:7dcc130b76258d33b90f61b658791dede3486c3e6bfb003ee5c9bfb396dd22f3", size = 69412, upload-time = "2025-05-15T12:30:07.975Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/54/20/4d324d65cc6d9205fabedc306948156824eb9f0ee1633355a8f7ec5c66bf/pluggy-1.6.0-py3-none-any.whl", hash = "sha256:e920276dd6813095e9377c0bc5566d94c932c33b27a3e3945d8389c374dd4746", size = 20538, upload-time = "2025-05-15T12:30:06.134Z" },
]
[[package]]
name = "pycapnp"
version = "2.1.0"
@@ -1041,68 +995,6 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/10/bd/c038d7cc38edc1aa5bf91ab8068b63d4308c66c4c8bb3cbba7dfbc049f9c/pyparsing-3.3.2-py3-none-any.whl", hash = "sha256:850ba148bd908d7e2411587e247a1e4f0327839c40e2e5e6d05a007ecc69911d", size = 122781, upload-time = "2026-01-21T03:57:55.912Z" },
]
[[package]]
name = "pytest"
version = "9.1.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
{ name = "iniconfig" },
{ name = "packaging" },
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/e4/47/b9efed96c114afcfa3c9d3fe98a76a1d14c74a9e266d397cf6eb64be5e01/pytest-9.1.1.tar.gz", hash = "sha256:1088fbde8f2b49d95a549a195707afa7a76a3ce9bcadc26b6d71f0ffda5fe313", size = 1636369, upload-time = "2026-06-19T10:58:32.857Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" },
]
[[package]]
name = "pytest-cpp"
version = "2.6.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/cf/a1/c2679d7ff2da20a0f89c7820ae2739cde739eac9b43c192531117b31b5f4/pytest_cpp-2.6.0.tar.gz", hash = "sha256:c2f49d3c038539ac84786a94d852e4f4619c34c95979c2bc69c20b3bdf051d85", size = 465490, upload-time = "2024-09-18T00:08:08.251Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/2a/44/dc2f5d53165264ae5831f361fe7723c45da05718a97015b2eddc452cf503/pytest_cpp-2.6.0-py3-none-any.whl", hash = "sha256:b33de94609450feea2fba9efff3558b8ac8f1fdf40a99e263b395d4798b911bb", size = 15074, upload-time = "2024-09-18T00:08:06.415Z" },
]
[[package]]
name = "pytest-mock"
version = "3.15.1"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/68/14/eb014d26be205d38ad5ad20d9a80f7d201472e08167f0bb4361e251084a9/pytest_mock-3.15.1.tar.gz", hash = "sha256:1849a238f6f396da19762269de72cb1814ab44416fa73a8686deac10b0d87a0f", size = 34036, upload-time = "2025-09-16T16:37:27.081Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/5a/cc/06253936f4a7fa2e0f48dfe6d851d9c56df896a9ab09ac019d70b760619c/pytest_mock-3.15.1-py3-none-any.whl", hash = "sha256:0a25e2eb88fe5168d535041d09a4529a188176ae608a6d249ee65abc0949630d", size = 10095, upload-time = "2025-09-16T16:37:25.734Z" },
]
[[package]]
name = "pytest-subtests"
version = "0.15.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "attrs" },
{ name = "pytest" },
]
sdist = { url = "https://files.pythonhosted.org/packages/bb/d9/20097971a8d315e011e055d512fa120fd6be3bdb8f4b3aa3e3c6bf77bebc/pytest_subtests-0.15.0.tar.gz", hash = "sha256:cb495bde05551b784b8f0b8adfaa27edb4131469a27c339b80fd8d6ba33f887c", size = 18525, upload-time = "2025-10-20T16:26:18.358Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/23/64/bba465299b37448b4c1b84c7a04178399ac22d47b3dc5db1874fe55a2bd3/pytest_subtests-0.15.0-py3-none-any.whl", hash = "sha256:da2d0ce348e1f8d831d5a40d81e3aeac439fec50bd5251cbb7791402696a9493", size = 9185, upload-time = "2025-10-20T16:26:17.239Z" },
]
[[package]]
name = "pytest-xdist"
version = "3.7.1.dev24+g2b4372b"
source = { git = "https://github.com/sshane/pytest-xdist?rev=2b4372bd62699fb412c4fe2f95bf9f01bd2018da#2b4372bd62699fb412c4fe2f95bf9f01bd2018da" }
dependencies = [
{ name = "execnet" },
{ name = "pytest" },
]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"