convert tests to unittest (#38387)

.github/workflows/tests.yaml

@@ -20,7 +20,7 @@ concurrency:
 env:
   CI: 1
   PYTHONPATH: ${{ github.workspace }}
-  PYTEST: pytest --continue-on-collection-errors --durations=0 -n logical
+  PYTEST: pytest --continue-on-collection-errors --durations=0 -n logical --dist worksteal
   GIT_CONFIG_COUNT: 1
   GIT_CONFIG_KEY_0: lfs.fetchexclude
   GIT_CONFIG_VALUE_0: openpilot/selfdrive/modeld/models/big_driving_supercombo.onnx
@@ -110,9 +110,9 @@ jobs:
       env:
         RAYLIB_BACKEND: headless
       run: |
-        # Pre-compile Python bytecode so each pytest worker doesn't need to
-        $PYTEST --collect-only -m 'not slow' -qq
-        MAX_EXAMPLES=1 $PYTEST -m 'not slow'
+        # Pre-compile Python bytecode so each worker doesn't need to.
+        python -m compileall -q -j 0 openpilot
+        MAX_EXAMPLES=1 SKIP_SLOW=1 $PYTEST
 
   process_replay:
     name: process replay

conftest.py

@@ -1,95 +1,15 @@
-import contextlib
-import gc
+"""Pytest runner configuration for the unittest suite."""
+
 import os
-import pytest
 
-from openpilot.common.prefix import OpenpilotPrefix
-from openpilot.system.manager import manager
-from openpilot.common.hardware import TICI, HARDWARE
-
-# these are heavy CI-only tests, invoked explicitly in .github/workflows/tests.yaml
+# Heavy CI-only tests are invoked explicitly by their dedicated jobs.
 collect_ignore = [
   "openpilot/selfdrive/test/process_replay/test_processes.py",
   "openpilot/selfdrive/test/process_replay/test_regen.py",
-
   "openpilot/tools/sim/",
 ]
 
 
-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 TICI:
-        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))
+def pytest_collection_modifyitems(items):
+  if os.environ.get("SKIP_SLOW"):
+    items[:] = [item for item in items if not getattr(getattr(item, "cls", None), "SLOW_TEST", False)]

openpilot/cereal/messaging/tests/test_messaging.py

@@ -4,6 +4,7 @@ import numbers
 import random
 import threading
 import time
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.cereal import log
@@ -46,7 +47,7 @@ def delayed_send(delay, sock, dat):
   threading.Timer(delay, send_func).start()
 
 
-class TestMessaging:
+class TestMessaging(OpenpilotTestCase):
   @parameterized.expand(events)
   def test_new_message(self, evt):
     try:

openpilot/cereal/messaging/tests/test_pub_sub_master.py

@@ -3,13 +3,14 @@ import time
 from typing import cast
 from collections.abc import Sized
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.messaging.tests.test_messaging import events, random_sock, random_socks, \
                                                   random_bytes, random_carstate, assert_carstate
 from openpilot.cereal.services import SERVICE_LIST
 
 
-class TestSubMaster:
+class TestSubMaster(OpenpilotTestCase):
 
   def test_init(self):
     sm = messaging.SubMaster(events)
@@ -107,7 +108,7 @@ class TestSubMaster:
     assert sm[sock].vEgo == n
 
 
-class TestPubMaster:
+class TestPubMaster(OpenpilotTestCase):
 
   def test_init(self):
     messaging.PubMaster(events)

openpilot/cereal/messaging/tests/test_services.py

@@ -1,13 +1,14 @@
 import subprocess
 import tempfile
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 import openpilot.cereal.services as services
 from openpilot.cereal.services import SERVICE_LIST
 
 
-class TestServices:
+class TestServices(OpenpilotTestCase):
 
   @parameterized.expand(SERVICE_LIST.keys())
   def test_services(self, s):

openpilot/common/hardware/tici/tests/test_agnos_updater.py

@@ -6,7 +6,8 @@ TEST_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)))
 MANIFEST = os.path.join(TEST_DIR, "../agnos.json")
 
 
-class TestAgnosUpdater:
+from openpilot.common.test import OpenpilotTestCase
+class TestAgnosUpdater(OpenpilotTestCase):
 
   def test_manifest(self):
     with open(MANIFEST) as f:

openpilot/common/hardware/tici/tests/test_amplifier.py

@@ -1,18 +1,14 @@
-import pytest
 import time
 import subprocess
 
 from panda import Panda
-from openpilot.common.hardware import TICI, HARDWARE
+from openpilot.common.test import OpenpilotTestCase
+from openpilot.common.hardware import HARDWARE
 from openpilot.common.hardware.tici.amplifier import Amplifier
 
 
-class TestAmplifier:
-
-  @classmethod
-  def setup_class(cls):
-    if not TICI:
-      pytest.skip()
+class TestAmplifier(OpenpilotTestCase):
+  TICI_TEST = True
 
   def setup_method(self):
     # clear dmesg
@@ -65,4 +61,4 @@ class TestAmplifier:
       if self._check_for_i2c_errors(True):
         break
     else:
-      pytest.fail("didn't hit any i2c errors")
+      self.fail("didn't hit any i2c errors")

openpilot/common/parameterized.py

@@ -1,7 +1,7 @@
 import re
 import sys
-import pytest
 import inspect
+import unittest
 
 
 def _to_safe_name(s):
@@ -10,22 +10,60 @@ def _to_safe_name(s):
 
 class parameterized:
   @staticmethod
-  def expand(cases):
+  def expand(cases, names=None, ids=None, serial=False):
     cases = list(cases)
 
     if not cases:
-      return lambda func: pytest.mark.skip("no parameterized cases")(func)
+      return lambda func: unittest.skip("no parameterized cases")(func)
 
-    def decorator(func):
-      params = [p for p in inspect.signature(func).parameters if p != 'self']
-      normalized = [c if isinstance(c, tuple) else (c,) for c in cases]
-      # Infer arg count from first case so extra params (e.g. from @given) are left untouched
-      expand_params = params[: len(normalized[0])]
-      if len(expand_params) == 1:
-        return pytest.mark.parametrize(expand_params[0], [c[0] for c in normalized])(func)
-      return pytest.mark.parametrize(', '.join(expand_params), normalized)(func)
+    if serial:
+      def decorator(func):
+        normalized = [case if isinstance(case, tuple) else (case,) for case in cases]
 
-    return decorator
+        def wrapper(self):
+          for case in normalized:
+            with self.subTest():
+              func(self, *case)
+
+        wrapper.__name__ = func.__name__
+        wrapper.__doc__ = func.__doc__
+        return wrapper
+
+      return decorator
+
+    return lambda func: _Expanded(func, cases, names, ids)
+
+
+class _Expanded:
+  """Descriptor that turns every parameter case into a real unittest method."""
+
+  def __init__(self, func, cases, names, ids):
+    self.func = func
+    self.cases = [c if isinstance(c, tuple) else (c,) for c in cases]
+    self.names = names
+    self.ids = ids
+
+  def __set_name__(self, owner, name):
+    params = [p for p in inspect.signature(self.func).parameters if p != "self"]
+
+    for index, case in enumerate(self.cases):
+      label = self.ids(*case) if self.ids is not None else None
+      method_name = f"{name}_{index}" + (f"_{_to_safe_name(label)}" if label is not None else "")
+
+      def test_method(test_case, current_case=case):
+        if self.names is None:
+          self.func(test_case, *current_case)
+        else:
+          values = dict(zip(self.names, current_case, strict=True))
+          values.update({param: test_case._fixture(param) for param in params if param not in values})
+          self.func(test_case, **values)
+
+      test_method.__name__ = method_name
+      test_method.__doc__ = self.func.__doc__
+      setattr(owner, method_name, test_method)
+
+    # The descriptor itself is only a method factory, not a test.
+    setattr(owner, name, None)
 
 
 def parameterized_class(attrs, input_list=None):
@@ -39,16 +77,16 @@ def parameterized_class(attrs, input_list=None):
   def decorator(cls):
     globs = sys._getframe(1).f_globals
     for i, params in enumerate(params_list):
-      # append sanitized string param values so pytest -k can filter by them
+      # Append sanitized values so unittest's -k can filter by them.
       suffix = "_".join(filter(None, (_to_safe_name(v) for v in params.values() if isinstance(v, str))))
       name = f"{cls.__name__}_{i}" + (f"_{suffix}" if suffix else "")
       new_cls = type(name, (cls,), dict(params))
       new_cls.__module__ = cls.__module__
-      new_cls.__test__ = True  # override inherited False so pytest collects this subclass
+      new_cls.__unittest_skip__ = False
       globs[name] = new_cls
-    # Don't collect the un-parametrised base, but return it so outer decorators
-    # (e.g. @pytest.mark.skip) land on it and propagate to subclasses via MRO.
-    cls.__test__ = False
+    # Don't collect the un-parametrised base.
+    cls.__unittest_skip__ = True
+    cls.__unittest_skip_why__ = "parameterized base class"
     return cls
 
   return decorator

openpilot/common/test.py

@@ -0,0 +1,172 @@
+import contextlib
+import gc
+import inspect
+import os
+import subprocess
+import unittest
+from unittest import mock
+
+from openpilot.common.hardware import HARDWARE, TICI
+from openpilot.common.prefix import OpenpilotPrefix
+from openpilot.system.manager import manager
+
+
+@contextlib.contextmanager
+def clean_env():
+  starting_env = dict(os.environ)
+  try:
+    yield
+  finally:
+    os.environ.clear()
+    os.environ.update(starting_env)
+
+
+class OpenpilotTestCase(unittest.TestCase):
+  """TestCase with openpilot's per-test isolation and legacy hook support."""
+
+  TICI_TEST = False
+  SKIP_TICI_SETUP = False
+  SHARED_DOWNLOAD_CACHE = False
+  SLOW_TEST = False
+
+  def __init_subclass__(cls, **kwargs):
+    super().__init_subclass__(**kwargs)
+    # Hide legacy pytest xunit hook names from pytest. unittest invokes the
+    # preserved hooks inside the OpenpilotPrefix boundary below.
+    for name in ("setup_method", "teardown_method"):
+      hook = cls.__dict__.get(name)
+      if hook is not None:
+        setattr(cls, f"openpilot_{name}", hook)
+        setattr(cls, name, None)
+
+  def _fixture(self, name):
+    if name == "mocker":
+      return Mocker(self.addCleanup)
+    if name == "monkeypatch":
+      return MonkeyPatch(self.addCleanup)
+    if name == "subtests":
+      return SubTests(self)
+
+    fixture = getattr(inspect.getmodule(type(self)), name)
+    kwargs = {p: self._fixture(p) for p in inspect.signature(fixture).parameters}
+    value = fixture(**kwargs)
+    if inspect.isgenerator(value):
+      generator = value
+      value = next(generator)
+      self.addCleanup(lambda: next(generator, None))
+    return value
+
+  def _callTestMethod(self, method):
+    params = [name for name, param in inspect.signature(method).parameters.items()
+              if param.default is inspect.Parameter.empty]
+    return method(**{name: self._fixture(name) for name in params})
+
+  def run(self, result=None):
+    # This boundary cannot live in setUp/tearDown: existing unittest classes
+    # are allowed to override those hooks without calling super().
+    if ((self.SLOW_TEST and os.environ.get("SKIP_SLOW")) or
+        (self.TICI_TEST and not TICI) or getattr(type(self), "__unittest_skip__", False)):
+      return super().run(result)
+    test_env = clean_env()
+    test_env.__enter__()
+    prefix = OpenpilotPrefix(shared_download_cache=self.SHARED_DOWNLOAD_CACHE)
+    prefix.__enter__()
+    try:
+      return super().run(result)
+    finally:
+      prefix.__exit__(None, None, None)
+      manager.manager_cleanup()
+      if not gc.isenabled():
+        gc.enable()
+        gc.collect()
+      test_env.__exit__(None, None, None)
+
+  @classmethod
+  def setUpClass(cls):
+    super().setUpClass()
+    if cls.SLOW_TEST and os.environ.get("SKIP_SLOW"):
+      raise unittest.SkipTest("slow test")
+    if cls.TICI_TEST and not TICI:
+      raise unittest.SkipTest("Skipping tici test on PC")
+    cls._class_env = clean_env()
+    cls._class_env.__enter__()
+    setup_class = getattr(cls, "setup_class", None)
+    if setup_class is not None:
+      setup_class()
+
+  @classmethod
+  def tearDownClass(cls):
+    try:
+      teardown_class = getattr(cls, "teardown_class", None)
+      if teardown_class is not None:
+        teardown_class()
+    finally:
+      cls._class_env.__exit__(None, None, None)
+      super().tearDownClass()
+
+  def setUp(self):
+    super().setUp()
+    if self.TICI_TEST and not TICI:
+      self.skipTest("Skipping tici test on PC")
+
+    if self.TICI_TEST and not self.SKIP_TICI_SETUP:
+      HARDWARE.initialize_hardware()
+      HARDWARE.set_power_save(False)
+      subprocess.run(["pkill", "-9", "-f", "athena"], check=False)
+
+    setup_method = getattr(self, "openpilot_setup_method", None)
+    if setup_method is not None:
+      setup_method()
+
+  def tearDown(self):
+    try:
+      teardown_method = getattr(self, "openpilot_teardown_method", None)
+      if teardown_method is not None:
+        teardown_method()
+    finally:
+      super().tearDown()
+
+
+class Mocker:
+  Mock = mock.Mock
+  MagicMock = mock.MagicMock
+  call = mock.call
+  ANY = mock.ANY
+
+  def __init__(self, add_cleanup):
+    self._add_cleanup = add_cleanup
+    self.patch = Patch(self._start)
+
+  def _start(self, patcher):
+    value = patcher.start()
+    self._add_cleanup(patcher.stop)
+    return value
+
+
+class Patch:
+  def __init__(self, start):
+    self._start = start
+
+  def __call__(self, *args, **kwargs):
+    return self._start(mock.patch(*args, **kwargs))
+
+  def object(self, *args, **kwargs):
+    return self._start(mock.patch.object(*args, **kwargs))
+
+
+class MonkeyPatch:
+  def __init__(self, add_cleanup):
+    self._add_cleanup = add_cleanup
+
+  def setattr(self, target, name, value):
+    patcher = mock.patch.object(target, name, value)
+    patcher.start()
+    self._add_cleanup(patcher.stop)
+
+
+class SubTests:
+  def __init__(self, test_case):
+    self._test_case = test_case
+
+  def test(self, label=None, **kwargs):
+    return self._test_case.subTest(**kwargs) if label is None else self._test_case.subTest(label, **kwargs)

openpilot/common/tests/test_file_helpers.py

@@ -1,10 +1,11 @@
 import os
 from uuid import uuid4
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.utils import atomic_write
 
 
-class TestFileHelpers:
+class TestFileHelpers(OpenpilotTestCase):
   def run_atomic_write_func(self, atomic_write_func):
     path = f"/tmp/tmp{uuid4()}"
     with atomic_write_func(path) as f:

openpilot/common/tests/test_markdown.py

@@ -1,10 +1,11 @@
 import os
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.basedir import BASEDIR
 from openpilot.common.markdown import parse_markdown
 
 
-class TestMarkdown:
+class TestMarkdown(OpenpilotTestCase):
   def test_all_release_notes(self):
     with open(os.path.join(BASEDIR, "RELEASES.md")) as f:
       release_notes = f.read().split("\n\n")

openpilot/common/tests/test_params.py

@@ -1,13 +1,13 @@
-import pytest
 import datetime
 import os
 import threading
 import time
 import uuid
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params, ParamKeyFlag, UnknownKeyName
 
-class TestParams:
+class TestParams(OpenpilotTestCase):
   def setup_method(self):
     self.params = Params()
 
@@ -50,16 +50,16 @@ class TestParams:
     assert self.params.get("CarParams", block=True) == b"test"
 
   def test_params_unknown_key_fails(self):
-    with pytest.raises(UnknownKeyName):
+    with self.assertRaises(UnknownKeyName):
       self.params.get("swag")
 
-    with pytest.raises(UnknownKeyName):
+    with self.assertRaises(UnknownKeyName):
       self.params.get_bool("swag")
 
-    with pytest.raises(UnknownKeyName):
+    with self.assertRaises(UnknownKeyName):
       self.params.put("swag", "abc", block=True)
 
-    with pytest.raises(UnknownKeyName):
+    with self.assertRaises(UnknownKeyName):
       self.params.put_bool("swag", True, block=True)
 
   def test_remove_not_there(self):

openpilot/common/tests/test_simple_kalman.py

@@ -1,7 +1,8 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.simple_kalman import KF1D
 
 
-class TestSimpleKalman:
+class TestSimpleKalman(OpenpilotTestCase):
   def setup_method(self):
     dt = 0.01
     x0_0 = 0.0

openpilot/common/transformations/tests/test_coordinates.py

@@ -1,5 +1,6 @@
 import numpy as np
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.common.transformations.coordinates as coord
 
 geodetic_positions = np.array([[37.7610403, -122.4778699, 115],
@@ -41,7 +42,7 @@ ned_offsets_batch = np.array([[  53.88103168,   43.83445935,  -46.27488057],
                               [  78.56272609,   18.53100158,  -43.25290759]])
 
 
-class TestNED:
+class TestNED(OpenpilotTestCase):
   def test_small_distances(self):
     start_geodetic = np.array([33.8042184, -117.888593, 0.0])
     local_coord = coord.LocalCoord.from_geodetic(start_geodetic)

openpilot/common/transformations/tests/test_orientation.py

@@ -1,6 +1,6 @@
 import numpy as np
-import pytest
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.transformations.orientation import euler2quat, quat2euler, euler2rot, rot2euler, \
                                                rot2quat, quat2rot, \
                                                ned_euler_from_ecef
@@ -30,7 +30,7 @@ ned_eulers = np.array([[ 0.46806039, -0.4881889 ,  1.65697808],
                     [ 2.50450101,  0.36304151,  0.33136365]])
 
 
-class TestOrientation:
+class TestOrientation(OpenpilotTestCase):
   def test_quat_euler(self):
     for i, eul in enumerate(eulers):
       np.testing.assert_allclose(quats[i], euler2quat(eul), rtol=1e-7)
@@ -62,13 +62,13 @@ class TestOrientation:
     # np.testing.assert_allclose(ned_eulers, ned_euler_from_ecef(ecef_positions, eulers), rtol=1e-7)
 
   def test_inputs(self):
-    with pytest.raises(ValueError):
+    with self.assertRaises(ValueError):
       euler2quat([1, 2])
 
-    with pytest.raises(ValueError):
+    with self.assertRaises(ValueError):
       quat2rot([1, 2, 3])
 
-    with pytest.raises(IndexError):
+    with self.assertRaises(IndexError):
       rot2quat(np.zeros((2, 2)))
 
   def test_euler_rot_consistency(self):

openpilot/selfdrive/car/tests/big_cars_test.sh

@@ -8,4 +8,4 @@ export MAX_EXAMPLES=300
 export INTERNAL_SEG_CNT=300
 export INTERNAL_SEG_LIST=openpilot/selfdrive/car/tests/test_models_segs.txt
 
-cd openpilot/selfdrive/car/tests && pytest test_models.py test_car_interfaces.py
+pytest -n logical --dist worksteal openpilot/selfdrive/car/tests/test_models.py openpilot/selfdrive/car/tests/test_car_interfaces.py

openpilot/selfdrive/car/tests/test_car_interfaces.py

@@ -1,6 +1,7 @@
 import os
 import hypothesis.strategies as st
 from hypothesis import Phase, given, settings
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from opendbc.car.structs import car
@@ -18,7 +19,7 @@ from openpilot.selfdrive.test.fuzzy_generation import FuzzyGenerator
 MAX_EXAMPLES = int(os.environ.get('MAX_EXAMPLES', '60'))
 
 
-class TestCarInterfaces:
+class TestCarInterfaces(OpenpilotTestCase):
   # FIXME: Due to the lists used in carParams, Phase.target is very slow and will cause
   #  many generated examples to overrun when max_examples > ~20, don't use it
   @parameterized.expand([(car,) for car in sorted(PLATFORMS)] + [MOCK.MOCK])

openpilot/selfdrive/car/tests/test_cruise_speed.py

@@ -1,7 +1,7 @@
-import pytest
 import itertools
 import numpy as np
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized_class
 from openpilot.cereal import log
 from openpilot.selfdrive.car.cruise import VCruiseHelper, V_CRUISE_MIN, V_CRUISE_MAX, V_CRUISE_INITIAL, IMPERIAL_INCREMENT
@@ -35,18 +35,18 @@ def run_cruise_simulation(cruise, e2e, personality, t_end=20.):
                       [True, False], # e2e
                       log.LongitudinalPersonality.schema.enumerants, # personality
                       [5,35])) # speed
-class TestCruiseSpeed:
+class TestCruiseSpeed(OpenpilotTestCase):
   def test_cruise_speed(self):
     print(f'Testing {self.speed} m/s')
     cruise_speed = float(self.speed)
 
     simulation_steady_state = run_cruise_simulation(cruise_speed, self.e2e, self.personality)
-    assert simulation_steady_state == pytest.approx(cruise_speed, abs=.01), f'Did not reach {self.speed} m/s'
+    self.assertAlmostEqual(simulation_steady_state, cruise_speed, delta=.01, msg=f'Did not reach {self.speed} m/s')
 
 
 # TODO: test pcmCruise
 @parameterized_class(('pcm_cruise',), [(False,)])
-class TestVCruiseHelper:
+class TestVCruiseHelper(OpenpilotTestCase):
   def setup_method(self):
     self.CP = car.CarParams(pcmCruise=self.pcm_cruise)
     self.v_cruise_helper = VCruiseHelper(self.CP)

openpilot/selfdrive/car/tests/test_docs.py

@@ -1,8 +1,9 @@
 from opendbc.car.docs import generate_cars_md, get_all_car_docs
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.car.docs import CARS_MD_TEMPLATE
 
 
-class TestCarDocs:
+class TestCarDocs(OpenpilotTestCase):
   @classmethod
   def setup_class(cls):
     cls.all_cars = get_all_car_docs()

openpilot/selfdrive/car/tests/test_models.py

@@ -1,12 +1,12 @@
 import time
 import os
-import pytest
 import random
 import unittest
 from collections import defaultdict, Counter
 import hypothesis.strategies as st
 from hypothesis import Phase, given, settings
 from openpilot.common.parameterized import parameterized_class
+from openpilot.common.test import OpenpilotTestCase
 from opendbc.car import DT_CTRL, gen_empty_fingerprint, structs
 from opendbc.car.can_definitions import CanData
 from opendbc.car.car_helpers import FRAME_FINGERPRINT, interfaces
@@ -66,9 +66,9 @@ def get_test_cases() -> list[tuple[str, CarTestRoute | None]]:
   return test_cases
 
 
-@pytest.mark.slow
-@pytest.mark.shared_download_cache
-class TestCarModelBase(unittest.TestCase):
+class TestCarModelBase(OpenpilotTestCase):
+  SLOW_TEST = True
+  SHARED_DOWNLOAD_CACHE = True
   platform: Platform | None = None
   test_route: CarTestRoute | None = None
 
@@ -302,8 +302,7 @@ class TestCarModelBase(unittest.TestCase):
     CC = structs.CarControl(cruiseControl=structs.CarControl.CruiseControl(resume=True))
     test_car_controller(CC.as_reader())
 
-  # Skip stdout/stderr capture with pytest, causes elevated memory usage
-  @pytest.mark.nocapture
+  # Capturing stdout/stderr here causes elevated memory usage.
   @settings(max_examples=MAX_EXAMPLES, deadline=None,
             phases=(Phase.reuse, Phase.generate, Phase.shrink))
   @given(data=st.data())
@@ -471,7 +470,6 @@ class TestCarModelBase(unittest.TestCase):
 
 
 @parameterized_class(('platform', 'test_route'), get_test_cases())
-@pytest.mark.xdist_group_class_property('test_route')
 class TestCarModel(TestCarModelBase):
   pass
 

openpilot/selfdrive/controls/tests/test_following_distance.py

@@ -1,5 +1,5 @@
-import pytest
 import itertools
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized_class
 
 from openpilot.cereal import log
@@ -36,11 +36,12 @@ def run_following_distance_simulation(v_lead, t_end=100.0, e2e=False, personalit
                        log.LongitudinalPersonality.standard,
                        log.LongitudinalPersonality.aggressive],
                       [0,10,35])) # speed
-class TestFollowingDistance:
+class TestFollowingDistance(OpenpilotTestCase):
   def test_following_distance(self):
     v_lead = float(self.speed)
     simulation_steady_state = run_following_distance_simulation(v_lead, e2e=self.e2e, personality=self.personality)
     correct_steady_state = desired_follow_distance(v_lead, v_lead, get_T_FOLLOW(self.personality))
     err_ratio = 0.2 if self.e2e else 0.1
     abs_err_margin = 0.5 if v_lead > 0.0 else 1.15
-    assert simulation_steady_state == pytest.approx(correct_steady_state, abs=err_ratio * correct_steady_state + abs_err_margin)
+    self.assertAlmostEqual(simulation_steady_state, correct_steady_state,
+                           delta=err_ratio * correct_steady_state + abs_err_margin)

openpilot/selfdrive/controls/tests/test_latcontrol.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.cereal import log
@@ -14,7 +15,7 @@ from openpilot.selfdrive.controls.lib.latcontrol_torque import LatControlTorque
 from openpilot.selfdrive.controls.lib.latcontrol_angle import LatControlAngle
 
 
-class TestLatControl:
+class TestLatControl(OpenpilotTestCase):
 
   @parameterized.expand([(HONDA.HONDA_CIVIC, LatControlPID), (TOYOTA.TOYOTA_RAV4, LatControlTorque),
                          (NISSAN.NISSAN_LEAF, LatControlAngle), (GM.CHEVROLET_BOLT_EUV, LatControlTorque)])

openpilot/selfdrive/controls/tests/test_latcontrol_torque_buffer.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.cereal import log
@@ -16,7 +17,7 @@ def get_controller(car_name):
   controller = LatControlTorque(CP.as_reader(), CI, DT_CTRL)
   return controller, VM
 
-class TestLatControlTorqueBuffer:
+class TestLatControlTorqueBuffer(OpenpilotTestCase):
 
   @parameterized.expand([(TOYOTA.TOYOTA_COROLLA_TSS2,)])
   def test_request_buffer_consistency(self, car_name):

openpilot/selfdrive/controls/tests/test_leads.py

@@ -1,10 +1,11 @@
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 
 from opendbc.car.toyota.values import CAR as TOYOTA
 from openpilot.selfdrive.test.process_replay import replay_process_with_name
 
 
-class TestLeads:
+class TestLeads(OpenpilotTestCase):
   def test_radar_fault(self):
     # if there's no radar-related can traffic, radard should either not respond or respond with an error
     # this is tightly coupled with underlying car radar_interface implementation, but it's a good sanity check

openpilot/selfdrive/controls/tests/test_longcontrol.py

@@ -1,7 +1,8 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.controls.lib.longcontrol import LongCtrlState, long_control_state_trans
 
 
-class TestLongControlStateTransition:
+class TestLongControlStateTransition(OpenpilotTestCase):
 
   def test_stay_stopped(self):
     active = True
@@ -23,18 +24,18 @@ class TestLongControlStateTransition:
                              should_stop=False, brake_pressed=False, cruise_standstill=False)
     assert next_state == LongCtrlState.off
 
-def test_engage():
-  active = True
-  current_state = LongCtrlState.off
-  next_state = long_control_state_trans(active, current_state,
+  def test_engage(self):
+    active = True
+    current_state = LongCtrlState.off
+    next_state = long_control_state_trans(active, current_state,
                              should_stop=True, brake_pressed=False, cruise_standstill=False)
-  assert next_state == LongCtrlState.stopping
-  next_state = long_control_state_trans(active, current_state,
+    assert next_state == LongCtrlState.stopping
+    next_state = long_control_state_trans(active, current_state,
                              should_stop=False, brake_pressed=True, cruise_standstill=False)
-  assert next_state == LongCtrlState.stopping
-  next_state = long_control_state_trans(active, current_state,
+    assert next_state == LongCtrlState.stopping
+    next_state = long_control_state_trans(active, current_state,
                              should_stop=False, brake_pressed=False, cruise_standstill=True)
-  assert next_state == LongCtrlState.stopping
-  next_state = long_control_state_trans(active, current_state,
+    assert next_state == LongCtrlState.stopping
+    next_state = long_control_state_trans(active, current_state,
                              should_stop=False, brake_pressed=False, cruise_standstill=False)
-  assert next_state == LongCtrlState.pid
+    assert next_state == LongCtrlState.pid

openpilot/selfdrive/controls/tests/test_torqued_lat_accel_offset.py

@@ -1,4 +1,5 @@
 import numpy as np
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging
 from opendbc.car.structs import car
 from opendbc.car import ACCELERATION_DUE_TO_GRAVITY
@@ -56,7 +57,7 @@ def simulate_straight_road_msgs(est):
     for which, msg in (('carControl', carControl), ('carOutput', carOutput), ('carState', carState), ('livePose', livePose)):
       est.handle_log(t, which, msg)
 
-class TestTorquedLatAccelOffset:
+class TestTorquedLatAccelOffset(OpenpilotTestCase):
   def test_estimated_offset(self):
     steer_torques, lat_accels = generate_inputs(TORQUE_TUNE_BIASED, la_err_std=LA_ERR_STD, input_noise_std=INPUT_NOISE_STD)
     est = get_warmed_up_estimator(steer_torques, lat_accels)

openpilot/selfdrive/locationd/test/test_calibrationd.py

@@ -2,6 +2,7 @@ import random
 
 import numpy as np
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal import log
 from openpilot.common.params import Params
@@ -29,7 +30,7 @@ def process_messages(c, cam_odo_calib, cycles,
                         [0.0, 0.0, HEIGHT_INIT.item()],
                         [cam_odo_height_std, cam_odo_height_std, cam_odo_height_std])
 
-class TestCalibrationd:
+class TestCalibrationd(OpenpilotTestCase):
 
   def test_read_saved_params(self):
     msg = messaging.new_message('liveCalibration')

openpilot/selfdrive/locationd/test/test_lagd.py

@@ -1,8 +1,9 @@
 import random
 import numpy as np
 import time
-import pytest
+import unittest
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging, log
 from opendbc.car.structs import car
 from openpilot.selfdrive.locationd.lagd import LateralLagEstimator, retrieve_initial_lag, masked_normalized_cross_correlation, \
@@ -45,7 +46,7 @@ def process_messages(estimator, lag_frames, n_frames, vego=25.0, rejection_thres
     estimator.update_estimate()
 
 
-class TestLagd:
+class TestLagd(OpenpilotTestCase):
   def test_read_saved_params(self):
     params = Params()
 
@@ -137,7 +138,7 @@ class TestLagd:
     assert np.allclose(msg.liveDelay.lateralDelayEstimateStd, 0.0, atol=0.01)
     assert msg.liveDelay.calPerc == 100
 
-  @pytest.mark.skipif(PC, reason="only on device")
+  @unittest.skipIf(PC, "only on device")
   def test_estimator_performance(self):
     mocked_CP = car.CarParams(steerActuatorDelay=0.5)
     estimator = LateralLagEstimator(mocked_CP, DT)

openpilot/selfdrive/locationd/test/test_locationd_scenarios.py

@@ -1,7 +1,11 @@
+import fcntl
 import numpy as np
+import os
+import tempfile
 from collections import defaultdict
 from enum import Enum
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.tools.lib.logreader import LogReader
 from openpilot.selfdrive.locationd.lagd import masked_symmetric_moving_average
 from openpilot.selfdrive.test.process_replay.migration import migrate_all
@@ -96,7 +100,7 @@ def run_scenarios(scenario, logs):
   return get_select_fields_data(logs), get_select_fields_data(replayed_logs)
 
 
-class TestLocationdScenarios:
+class TestLocationdScenarios(OpenpilotTestCase):
   """
   Test locationd with different scenarios. In all these scenarios, we expect the following:
     - locationd kalman filter should never go unstable (we care mostly about yaw_rate, roll, gpsOK, inputsOK, sensorsOK)
@@ -105,7 +109,20 @@ class TestLocationdScenarios:
 
   @classmethod
   def setup_class(cls):
-    cls.logs = migrate_all(LogReader(TEST_ROUTE))
+    # xdist can initialize this class in several workers at once. URLFile's
+    # cache writes are atomic, but cache misses are not locked, so every worker
+    # otherwise downloads the same route concurrently.
+    lock_path = os.path.join(tempfile.gettempdir(), "openpilot-locationd-scenarios.lock")
+    ready_path = f"{lock_path}.ready"
+    logs = None
+    with open(lock_path, "w") as lock:
+      fcntl.flock(lock, fcntl.LOCK_EX)
+      if not os.path.exists(ready_path):
+        logs = list(LogReader(TEST_ROUTE))
+        open(ready_path, "w").close()
+    if logs is None:
+      logs = list(LogReader(TEST_ROUTE))
+    cls.logs = migrate_all(logs)
 
   def test_base(self):
     """

openpilot/selfdrive/locationd/test/test_paramsd.py

@@ -1,6 +1,7 @@
 import random
 import numpy as np
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging
 from openpilot.selfdrive.locationd.paramsd import retrieve_initial_vehicle_params
 from openpilot.selfdrive.locationd.models.car_kf import CarKalman
@@ -19,7 +20,7 @@ def get_random_live_parameters(CP):
   return msg
 
 
-class TestParamsd:
+class TestParamsd(OpenpilotTestCase):
   def test_read_saved_params(self):
     params = Params()
 

openpilot/selfdrive/locationd/test/test_torqued.py

@@ -1,8 +1,9 @@
 from opendbc.car.structs import car
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.locationd.torqued import TorqueEstimator
 
 
-class TestTorqued:
+class TestTorqued(OpenpilotTestCase):
   def test_cal_percent(self):
     est = TorqueEstimator(car.CarParams())
     msg = est.get_msg()

openpilot/selfdrive/monitoring/test_monitoring.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import log
 from openpilot.common.realtime import DT_DMON
 from openpilot.selfdrive.monitoring.policy import DriverMonitoring, DRIVER_MONITOR_SETTINGS
@@ -46,7 +47,7 @@ always_distracted = [msg_DISTRACTED] * int(TEST_TIMESPAN / DT_DMON)
 always_true = [True] * int(TEST_TIMESPAN / DT_DMON)
 always_false = [False] * int(TEST_TIMESPAN / DT_DMON)
 
-class TestMonitoring:
+class TestMonitoring(OpenpilotTestCase):
   def _run_seq(self, msgs, interaction, engaged, lowspeed):
     DM = DriverMonitoring()
     alert_lvls = []

openpilot/selfdrive/pandad/tests/test_pandad.py

@@ -1,7 +1,7 @@
 import os
-import pytest
 import time
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal import log
 from openpilot.common.gpio import gpio_set, gpio_init
@@ -13,8 +13,8 @@ from openpilot.common.hardware.tici.pins import GPIO
 HERE = os.path.dirname(os.path.realpath(__file__))
 
 
-@pytest.mark.tici
-class TestPandad:
+class TestPandad(OpenpilotTestCase):
+  TICI_TEST = True
   def teardown_method(self):
     managed_processes['pandad'].stop()
 

openpilot/selfdrive/pandad/tests/test_pandad_loopback.py

@@ -2,10 +2,10 @@ import os
 import copy
 import random
 import time
-import pytest
 from collections import defaultdict
 from pprint import pprint
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal import log
 from opendbc.car.structs import car
@@ -69,8 +69,8 @@ def send_random_can_messages(sendcan, count):
   return sent_msgs
 
 
-@pytest.mark.tici
-class TestBoarddLoopback:
+class TestBoarddLoopback(OpenpilotTestCase):
+  TICI_TEST = True
   @classmethod
   def setup_class(cls):
     os.environ['STARTED'] = '1'

openpilot/selfdrive/pandad/tests/test_pandad_spi.py

@@ -1,9 +1,9 @@
 import os
 import time
 import numpy as np
-import pytest
 import random
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.common.timeout import Timeout
@@ -12,8 +12,8 @@ from openpilot.selfdrive.pandad.tests.test_pandad_loopback import setup_pandad,
 
 JUNGLE_SPAM = "JUNGLE_SPAM" in os.environ
 
-@pytest.mark.tici
-class TestBoarddSpi:
+class TestBoarddSpi(OpenpilotTestCase):
+  TICI_TEST = True
   @classmethod
   def setup_class(cls):
     os.environ['STARTED'] = '1'

openpilot/selfdrive/selfdrived/tests/test_alertmanager.py

@@ -1,10 +1,11 @@
 import random
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.selfdrived.events import Alert, EmptyAlert, EVENTS
 from openpilot.selfdrive.selfdrived.alertmanager import AlertManager
 
 
-class TestAlertManager:
+class TestAlertManager(OpenpilotTestCase):
 
   def test_duration(self):
     """

openpilot/selfdrive/selfdrived/tests/test_alerts.py

@@ -4,6 +4,7 @@ import os
 import random
 from PIL import Image, ImageDraw, ImageFont
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import log
 from opendbc.car.structs import car
 from openpilot.cereal.messaging import SubMaster
@@ -24,7 +25,7 @@ for event_types in EVENTS.values():
     ALERTS.append(alert)
 
 
-class TestAlerts:
+class TestAlerts(OpenpilotTestCase):
 
   @classmethod
   def setup_class(cls):

openpilot/selfdrive/selfdrived/tests/test_state_machine.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import log
 from openpilot.common.realtime import DT_CTRL
 from openpilot.selfdrive.selfdrived.state import StateMachine, SOFT_DISABLE_TIME
@@ -21,7 +22,7 @@ def make_event(event_types: list[str | None]):
   return 0
 
 
-class TestStateMachine:
+class TestStateMachine(OpenpilotTestCase):
   def setup_method(self):
     self.events = Events()
     self.state_machine = StateMachine()

openpilot/selfdrive/test/cpp_harness.py

@@ -1,10 +0,0 @@
-#!/usr/bin/env python3
-import subprocess
-import sys
-
-from openpilot.common.prefix import OpenpilotPrefix
-
-with OpenpilotPrefix():
-  ret = subprocess.call(sys.argv[1:])
-
-sys.exit(ret)

openpilot/selfdrive/test/helpers.py

@@ -3,7 +3,6 @@ import http.server
 import os
 import threading
 import time
-import pytest
 
 from functools import wraps
 
@@ -32,7 +31,7 @@ def release_only(f):
   @wraps(f)
   def wrap(self, *args, **kwargs):
     if "RELEASE" not in os.environ:
-      pytest.skip("This test is only for release branches")
+      self.skipTest("This test is only for release branches")
     f(self, *args, **kwargs)
   return wrap
 

openpilot/selfdrive/test/longitudinal_maneuvers/test_longitudinal.py

@@ -1,4 +1,5 @@
 import itertools
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized_class
 
 from openpilot.selfdrive.controls.lib.longitudinal_mpc_lib.long_mpc import STOP_DISTANCE
@@ -179,7 +180,7 @@ def create_maneuvers(kwargs):
 
 
 @parameterized_class(("e2e", "force_decel"), itertools.product([True, False], repeat=2))
-class TestLongitudinalControl:
+class TestLongitudinalControl(OpenpilotTestCase):
   e2e: bool
   force_decel: bool
 

openpilot/selfdrive/test/process_replay/test_fuzzy.py

@@ -2,6 +2,7 @@ import copy
 import os
 from hypothesis import given, HealthCheck, Phase, settings
 import hypothesis.strategies as st
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.cereal import log
@@ -17,7 +18,7 @@ NOT_TESTED = ['selfdrived', 'controlsd', 'card', 'plannerd', 'calibrationd', 'dm
 TEST_CASES = [(cfg.proc_name, copy.deepcopy(cfg)) for cfg in pr.CONFIGS if cfg.proc_name not in NOT_TESTED]
 MAX_EXAMPLES = int(os.environ.get("MAX_EXAMPLES", "10"))
 
-class TestFuzzProcesses:
+class TestFuzzProcesses(OpenpilotTestCase):
 
   # TODO: make this faster and increase examples
   @parameterized.expand(TEST_CASES)

openpilot/selfdrive/test/process_replay/test_regen.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.selfdrive.test.process_replay.regen import regen_segment
@@ -26,7 +27,9 @@ def ci_setup_data_readers(route, sidx):
   return lr, frs
 
 
-class TestRegen:
+class TestRegen(OpenpilotTestCase):
+  SLOW_TEST = True
+
   @parameterized.expand(TESTED_SEGMENTS)
   def test_engaged(self, case_name, segment):
     route, sidx = segment.rsplit("--", 1)

openpilot/selfdrive/test/test_onroad.py

@@ -1,13 +1,13 @@
 import math
 import json
 import os
-import pytest
 import shutil
 import subprocess
 import time
 import numpy as np
 from collections import Counter, defaultdict
 from pathlib import Path
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.utils import tabulate
 
 from openpilot.cereal import log
@@ -102,9 +102,9 @@ def cputime_total(ct):
   return ct.cpuUser + ct.cpuSystem + ct.cpuChildrenUser + ct.cpuChildrenSystem
 
 
-@pytest.mark.tici
-@pytest.mark.skip_tici_setup
-class TestOnroad:
+class TestOnroad(OpenpilotTestCase):
+  TICI_TEST = True
+  SKIP_TICI_SETUP = True
 
   @classmethod
   def setup_class(cls):

openpilot/selfdrive/test/test_power_draw.py

@@ -1,8 +1,8 @@
 from collections import defaultdict, deque
-import pytest
 import time
 import numpy as np
 from dataclasses import dataclass
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.utils import tabulate
 
 import openpilot.cereal.messaging as messaging
@@ -38,8 +38,8 @@ PROCS = [
 ]
 
 
-@pytest.mark.tici
-class TestPowerDraw:
+class TestPowerDraw(OpenpilotTestCase):
+  TICI_TEST = True
 
   def setup_method(self):
     Params().put("CarParams", get_demo_car_params().to_bytes(), block=True)

openpilot/selfdrive/ui/mici/tests/test_widget_leaks.py

@@ -1,6 +1,6 @@
 import gc
 import weakref
-import pytest
+import unittest
 
 # FIXME: known small leaks not worth worrying about at the moment
 KNOWN_LEAKS = {
@@ -41,8 +41,9 @@ def get_child_widgets(widget) -> list:
   return children
 
 
-class TestWidgetLeaks:
-  @pytest.mark.skip(reason="segfaults")
+from openpilot.common.test import OpenpilotTestCase
+class TestWidgetLeaks(OpenpilotTestCase):
+  @unittest.skip("segfaults")
   def test_dialogs_do_not_leak(self):
     import pyray as rl
     rl.set_config_flags(rl.ConfigFlags.FLAG_WINDOW_HIDDEN)

openpilot/selfdrive/ui/tests/test_feedbackd.py

@@ -1,12 +1,14 @@
-import pytest
+import unittest
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from opendbc.car.structs import car
 from openpilot.common.params import Params
 from openpilot.system.manager.process_config import managed_processes
 
 
-@pytest.mark.skip("tmp disabled")
-class TestFeedbackd:
+@unittest.skip("tmp disabled")
+class TestFeedbackd(OpenpilotTestCase):
   def setup_method(self):
     self.pm = messaging.PubMaster(['carState', 'rawAudioData'])
     self.sm = messaging.SubMaster(['audioFeedback'])
@@ -25,7 +27,7 @@ class TestFeedbackd:
       self.pm.send('rawAudioData', audio_msg)
       self.sm.update(timeout=100)
 
-  @pytest.mark.parametrize("record_feedback", [False, True])
+  @parameterized.expand([False, True])
   def test_audio_feedback(self, record_feedback):
     Params().put_bool("RecordAudioFeedback", record_feedback, block=True)
 

openpilot/selfdrive/ui/tests/test_raylib_ui.py

@@ -1,8 +1,9 @@
 import time
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.test.helpers import with_processes
 
 
-class TestRaylibUi:
+class TestRaylibUi(OpenpilotTestCase):
   @with_processes(["ui"])
   def test_raylib_ui(self):
     """Test initialization of the UI widgets is successful."""

openpilot/selfdrive/ui/tests/test_soundd.py

@@ -1,3 +1,4 @@
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import log, messaging
 from openpilot.cereal.messaging import SubMaster, PubMaster
 from openpilot.selfdrive.ui.soundd import SELFDRIVE_STATE_TIMEOUT, check_selfdrive_timeout_alert
@@ -7,7 +8,7 @@ import time
 AudibleAlert = log.SelfdriveState.AudibleAlert
 
 
-class TestSoundd:
+class TestSoundd(OpenpilotTestCase):
   def test_check_selfdrive_timeout_alert(self):
     sm = SubMaster(['selfdriveState'])
     pm = PubMaster(['selfdriveState'])
@@ -31,4 +32,3 @@ class TestSoundd:
     assert check_selfdrive_timeout_alert(sm)
 
   # TODO: add test with micd for checking that soundd actually outputs sounds
-

openpilot/selfdrive/ui/tests/test_translations.py

@@ -3,8 +3,9 @@ import re
 import string
 from pathlib import Path
 
-import pytest
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.ui.translations.potools import parse_po
 from openpilot.system.ui.lib.multilang import LANGUAGES_FILE, TRANSLATIONS_DIR
 
@@ -46,13 +47,13 @@ def load_po_text(po_path: Path) -> str:
   return po_path.read_text(encoding='utf-8')
 
 
-class TestTranslations:
-  @pytest.mark.parametrize("language_code", sorted(TRANSLATION_LANGUAGES.values()))
+class TestTranslations(OpenpilotTestCase):
+  @parameterized.expand(sorted(TRANSLATION_LANGUAGES.values()))
   def test_translation_file_exists(self, language_code: str):
     po_path = PO_DIR / f"app_{language_code}.po"
     assert po_path.exists(), f"missing translation file: {po_path}"
 
-  @pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+  @parameterized.expand(sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
   def test_translation_placeholders_are_preserved(self, po_path: Path):
     _, entries = parse_po(po_path)
     language = po_path.stem.removeprefix("app_")
@@ -89,14 +90,14 @@ class TestTranslations:
         )
         assert translated_placeholders == source_placeholders, message
 
-  @pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+  @parameterized.expand(sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
   def test_translation_refs_do_not_include_line_numbers(self, po_path: Path):
     for line in load_po_text(po_path).splitlines():
       assert not LINE_NUMBER_REF_RE.match(line), (
         f"{po_path.name}: line-number source reference found: {line}"
       )
 
-  @pytest.mark.parametrize("po_path", sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
+  @parameterized.expand(sorted(PO_DIR.glob("app_*.po")), ids=lambda p: p.name)
   def test_translation_entities_are_valid(self, po_path: Path):
     matches = BAD_ENTITY_RE.findall(load_po_text(po_path))
     assert not matches, (

openpilot/system/athena/tests/test_athenad.py

@@ -1,4 +1,3 @@
-import pytest
 from functools import wraps
 import json
 import multiprocessing
@@ -14,6 +13,8 @@ from datetime import datetime, timedelta
 from websocket import ABNF
 from websocket._exceptions import WebSocketConnectionClosedException
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging
 
 from openpilot.common.params import Params
@@ -47,16 +48,14 @@ def with_upload_handler(func):
       thread.join()
   return wrapper
 
-@pytest.fixture
 def mock_create_connection(mocker):
     return mocker.patch('openpilot.system.athena.athenad.create_connection')
 
-@pytest.fixture
 def host():
   with http_server_context(handler=HTTPRequestHandler, setup=seed_athena_server) as (host, port):
     yield f"http://{host}:{port}"
 
-class TestAthenadMethods:
+class TestAthenadMethods(OpenpilotTestCase):
   @classmethod
   def setup_class(cls):
     cls.SOCKET_PORT = 45454
@@ -111,7 +110,7 @@ class TestAthenadMethods:
     assert dispatcher["echo"]("bob") == "bob"
 
   def test_get_message(self):
-    with pytest.raises(TimeoutError) as _:
+    with self.assertRaises(TimeoutError) as _:
       dispatcher["getMessage"]("controlsState")
 
     end_event = multiprocessing.Event()
@@ -184,14 +183,14 @@ class TestAthenadMethods:
     if fn.endswith('.zst'):
       assert athenad.strip_zst_extension(fn) == fn[:-4]
 
-  @pytest.mark.parametrize("compress", [True, False])
+  @parameterized.expand([True, False], names=("compress",))
   def test_do_upload(self, host, compress):
     # random bytes to ensure rather large object post-compression
     fn = self._create_file('qlog', data=os.urandom(10000 * 1024))
 
     upload_fn = fn + ('.zst' if compress else '')
     item = athenad.UploadItem(path=upload_fn, url="http://localhost:1238", headers={}, created_at=int(time.time()*1000), id='')  # noqa: TID251
-    with pytest.raises(requests.exceptions.ConnectionError):
+    with self.assertRaises(requests.exceptions.ConnectionError):
       athenad._do_upload(item)
 
     item = athenad.UploadItem(path=upload_fn, url=f"{host}/qlog.zst", headers={}, created_at=int(time.time()*1000), id='')  # noqa: TID251
@@ -236,7 +235,7 @@ class TestAthenadMethods:
     # TODO: also check that end_event and metered network raises AbortTransferException
     assert athenad.upload_queue.qsize() == 0
 
-  @pytest.mark.parametrize("status,retry", [(500,True), (412,False)])
+  @parameterized.expand([(500,True), (412,False)], names=("status", "retry"))
   @with_upload_handler
   def test_upload_handler_retry(self, mocker, host, status, retry):
     mock_put = mocker.patch('openpilot.system.athena.athenad.UPLOAD_SESS.put')

openpilot/system/athena/tests/test_athenad_ping.py

@@ -1,9 +1,10 @@
-import pytest
+import unittest
 import subprocess
 import threading
 import time
 from typing import cast
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params
 from openpilot.common.timeout import Timeout
 from openpilot.system.athena import athenad
@@ -19,7 +20,7 @@ def wifi_radio(on: bool) -> None:
   subprocess.run(["nmcli", "radio", "wifi", "on" if on else "off"], check=True)
 
 
-class TestAthenadPing:
+class TestAthenadPing(OpenpilotTestCase):
   params: Params
   dongle_id: str
 
@@ -90,12 +91,12 @@ class TestAthenadPing:
         time.sleep(0.1)
       print("ping received")
 
-  @pytest.mark.skipif(not TICI, reason="only run on desk")
+  @unittest.skipIf(not TICI, "only run on desk")
   def test_offroad(self, subtests, mocker) -> None:
     self.params.put_bool("IsOffroad", True, block=True)
     self.assertTimeout(60 + TIMEOUT_TOLERANCE, subtests, mocker)  # based using TCP keepalive settings
 
-  @pytest.mark.skipif(not TICI, reason="only run on desk")
+  @unittest.skipIf(not TICI, "only run on desk")
   def test_onroad(self, subtests, mocker) -> None:
     self.params.put_bool("IsOffroad", False, block=True)
     self.assertTimeout(21 + TIMEOUT_TOLERANCE, subtests, mocker)

openpilot/system/athena/tests/test_registration.py

@@ -2,13 +2,14 @@ import json
 from Crypto.PublicKey import RSA
 from pathlib import Path
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params
 from openpilot.system.athena.registration import register, UNREGISTERED_DONGLE_ID
 from openpilot.system.athena.tests.helpers import MockResponse
 from openpilot.common.hardware.hw import Paths
 
 
-class TestRegistration:
+class TestRegistration(OpenpilotTestCase):
 
   def setup_method(self):
     # clear params and setup key paths

openpilot/system/camerad/test/test_camerad.py

@@ -1,8 +1,9 @@
 import os
 import time
-import pytest
 import numpy as np
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.tools.lib.log_time_series import msgs_to_time_series
 from openpilot.system.camerad.snapshot import get_snapshots
@@ -38,7 +39,6 @@ def run_and_log(procs, services, duration):
   with processes_context(procs):
     return collect_logs(services, duration)
 
-@pytest.fixture(scope="module")
 def _camera_session():
   """Single camerad session that collects logs and exposure data.
      Runs until exposure stabilizes (min TEST_TIMESPAN seconds for enough log data)."""
@@ -69,20 +69,18 @@ def _camera_session():
 
   return ts, exposure
 
-@pytest.fixture(scope="module")
-def logs(_camera_session):
-  return _camera_session[0]
+class TestCamerad(OpenpilotTestCase):
+  TICI_TEST = True
 
-@pytest.fixture(scope="module")
-def exposure_data(_camera_session):
-  return _camera_session[1]
+  @classmethod
+  def setUpClass(cls):
+    super().setUpClass()
+    cls.logs, cls.exposure_data = _camera_session()
 
-@pytest.mark.tici
-class TestCamerad:
-  @pytest.mark.parametrize("cam", CAMERAS)
-  def test_camera_exposure(self, exposure_data, cam):
+  @parameterized.expand(CAMERAS, names=("cam",))
+  def test_camera_exposure(self, cam):
     lo, hi = EXPOSURE_RANGE
-    checks = exposure_data[cam]
+    checks = self.exposure_data[cam]
     assert len(checks) >= EXPOSURE_STABLE_COUNT, f"{cam}: only got {len(checks)} samples"
 
     # check that exposure converges into the valid range
@@ -96,34 +94,34 @@ class TestCamerad:
     for i, (median, mean) in enumerate(checks):
       ok = _in_range(median, mean)
       if in_range and not ok:
-        pytest.fail(f"{cam}: exposure regressed on sample {i+1} " +
+        self.fail(f"{cam}: exposure regressed on sample {i+1} " +
                     f"(median={median:.4f}, mean={mean:.4f}, expected: ({lo}, {hi}))")
       in_range = ok
 
-  def test_frame_skips(self, logs):
+  def test_frame_skips(self):
     for c in CAMERAS:
-      assert set(np.diff(logs[c]['frameId'])) == {1, }, f"{c} has frame skips"
+      assert set(np.diff(self.logs[c]['frameId'])) == {1, }, f"{c} has frame skips"
 
-  def test_frame_sync(self, logs):
+  def test_frame_sync(self):
     SYNCED_CAMS = ('roadCameraState', 'wideRoadCameraState')
-    n = range(len(logs['roadCameraState']['t'][:-10]))
+    n = range(len(self.logs['roadCameraState']['t'][:-10]))
 
-    frame_ids = {i: [logs[cam]['frameId'][i] for cam in CAMERAS] for i in n}
+    frame_ids = {i: [self.logs[cam]['frameId'][i] for cam in CAMERAS] for i in n}
     assert all(len(set(v)) == 1 for v in frame_ids.values()), "frame IDs not aligned"
 
     # road and wide cameras should be synced within 1.1ms
-    synced_times = {i: [logs[cam]['timestampSof'][i] for cam in SYNCED_CAMS] for i in n}
+    synced_times = {i: [self.logs[cam]['timestampSof'][i] for cam in SYNCED_CAMS] for i in n}
     diffs = {i: (max(ts) - min(ts))/1e6 for i, ts in synced_times.items()}
     laggy_frames = {k: v for k, v in diffs.items() if v > 1.1}
     assert len(laggy_frames) == 0, f"Frames not synced properly: {laggy_frames=}"
 
     # driver camera should be staggered ~25ms from road camera
     for i in n:
-      offset_ms = abs(logs['driverCameraState']['timestampSof'][i] - logs['roadCameraState']['timestampSof'][i]) / 1e6
+      offset_ms = abs(self.logs['driverCameraState']['timestampSof'][i] - self.logs['roadCameraState']['timestampSof'][i]) / 1e6
       assert 20 < offset_ms < 30, f"driver camera stagger out of range at frame {i}: {offset_ms:.1f}ms (expected ~25ms)"
 
-  def test_sanity_checks(self, logs):
-    self._sanity_checks(logs)
+  def test_sanity_checks(self):
+    self._sanity_checks(self.logs)
 
   def _sanity_checks(self, ts):
     for c in CAMERAS:

openpilot/system/hardware/tests/test_fan_controller.py

@@ -1,10 +1,11 @@
-import pytest
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.system.hardware.fan_controller import FanController
 
 ALL_CONTROLLERS = [FanController]
 
-class TestFanController:
+class TestFanController(OpenpilotTestCase):
   def wind_up(self, controller, ignition=True):
     for _ in range(1000):
       controller.update(100, ignition)
@@ -13,31 +14,31 @@ class TestFanController:
     for _ in range(1000):
       controller.update(10, ignition)
 
-  @pytest.mark.parametrize("controller_class", ALL_CONTROLLERS)
+  @parameterized.expand(ALL_CONTROLLERS)
   def test_hot_onroad(self, controller_class):
     controller = controller_class(2)
     self.wind_up(controller)
     assert controller.update(100, True) >= 70
 
-  @pytest.mark.parametrize("controller_class", ALL_CONTROLLERS)
+  @parameterized.expand(ALL_CONTROLLERS)
   def test_offroad_limits(self, controller_class):
     controller = controller_class(2)
     self.wind_up(controller)
     assert controller.update(100, False) <= 30
 
-  @pytest.mark.parametrize("controller_class", ALL_CONTROLLERS)
+  @parameterized.expand(ALL_CONTROLLERS)
   def test_no_fan_wear(self, controller_class):
     controller = controller_class(2)
     self.wind_down(controller)
     assert controller.update(10, False) == 0
 
-  @pytest.mark.parametrize("controller_class", ALL_CONTROLLERS)
+  @parameterized.expand(ALL_CONTROLLERS)
   def test_limited(self, controller_class):
     controller = controller_class(2)
     self.wind_up(controller, True)
     assert controller.update(100, True) == 100
 
-  @pytest.mark.parametrize("controller_class", ALL_CONTROLLERS)
+  @parameterized.expand(ALL_CONTROLLERS)
   def test_windup_speed(self, controller_class):
     controller = controller_class(2)
     self.wind_down(controller, True)

openpilot/system/hardware/tests/test_power_monitoring.py

@@ -1,5 +1,5 @@
-import pytest
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params
 from openpilot.system.hardware.power_monitoring import PowerMonitoring, CAR_BATTERY_CAPACITY_uWh, \
                                                 CAR_CHARGING_RATE_W, VBATT_PAUSE_CHARGING, DELAY_SHUTDOWN_TIME_S
@@ -22,13 +22,9 @@ def pm_patch(mocker, name, value, constant=False):
     mocker.patch(f"openpilot.system.hardware.power_monitoring.{name}", return_value=value)
 
 
-@pytest.fixture(autouse=True)
-def mock_time(mocker):
-  mocker.patch("time.monotonic", mock_time_monotonic)
-
-
-class TestPowerMonitoring:
+class TestPowerMonitoring(OpenpilotTestCase):
   def setup_method(self):
+    self._fixture("mocker").patch("time.monotonic", mock_time_monotonic)
     self.params = Params()
 
   # Test to see that it doesn't do anything when pandaState is None

openpilot/system/loggerd/tests/loggerd_tests_common.py

@@ -3,6 +3,7 @@ import random
 from pathlib import Path
 
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.system.loggerd.deleter as deleter
 import openpilot.system.loggerd.uploader as uploader
 from openpilot.common.params import Params
@@ -53,7 +54,7 @@ class MockApiIgnore:
   def get_token(self):
     return "fake-token"
 
-class UploaderTestCase:
+class UploaderTestCase(OpenpilotTestCase):
   f_type = "UNKNOWN"
 
   root: Path

openpilot/system/loggerd/tests/test_deleter.py

@@ -17,7 +17,7 @@ class TestDeleter(UploaderTestCase):
 
   def setup_method(self):
     self.f_type = "fcamera.hevc"
-    super().setup_method()
+    super().openpilot_setup_method()
     self.fake_stats = Stats(f_bavail=0, f_blocks=10, f_frsize=4096)
     deleter.os.statvfs = self.fake_statvfs  # ty: ignore[invalid-assignment]  # test double
 

openpilot/system/loggerd/tests/test_encoder.py

@@ -1,6 +1,5 @@
 import math
 import os
-import pytest
 import shutil
 import subprocess
 import time
@@ -8,6 +7,7 @@ from pathlib import Path
 
 from tqdm import trange
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params
 from openpilot.common.timeout import Timeout
 from openpilot.common.hardware import TICI
@@ -29,8 +29,8 @@ CAMERAS = [
 FILE_SIZE_TOLERANCE = 0.7
 
 
-@pytest.mark.tici # TODO: all of loggerd should work on PC
-class TestEncoder:
+class TestEncoder(OpenpilotTestCase):
+  TICI_TEST = True
 
   def setup_method(self):
     self._clear_logs()

openpilot/system/loggerd/tests/test_loggerd.py

@@ -8,8 +8,9 @@ import time
 from collections.abc import Collection
 from collections import defaultdict
 from pathlib import Path
-import pytest
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal import log
 from openpilot.cereal.services import SERVICE_LIST
@@ -32,7 +33,7 @@ CEREAL_SERVICES = [f for f in log.Event.schema.union_fields if f in SERVICE_LIST
                    and SERVICE_LIST[f].should_log and "encode" not in f.lower()]
 
 
-class TestLoggerd:
+class TestLoggerd(OpenpilotTestCase):
   def _get_latest_log_dir(self):
     log_dirs = sorted(Path(Paths.log_root()).iterdir(), key=lambda f: f.stat().st_mtime)
     return log_dirs[-1]
@@ -193,7 +194,6 @@ class TestLoggerd:
       assert getattr(initData, initData_key) == v
       assert logged_params[param_key].decode() == v
 
-  @pytest.mark.xdist_group("camera_encoder_tests")  # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
   def test_rotation(self):
     Params().put("RecordFront", True, block=True)
 
@@ -314,8 +314,7 @@ class TestLoggerd:
     segment_dir = self._get_latest_log_dir()
     assert getxattr(segment_dir, PRESERVE_ATTR_NAME) is None
 
-  @pytest.mark.xdist_group("camera_encoder_tests")  # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
-  @pytest.mark.parametrize("record_front", [True, False])
+  @parameterized.expand([True, False])
   def test_record_front(self, record_front):
     params = Params()
     params.put_bool("RecordFront", record_front, block=True)
@@ -325,8 +324,7 @@ class TestLoggerd:
     dcamera_hevc_exists = os.path.exists(os.path.join(self._get_latest_log_dir(), 'dcamera.hevc'))
     assert dcamera_hevc_exists == record_front
 
-  @pytest.mark.xdist_group("camera_encoder_tests")  # setting xdist group ensures tests are run in same worker, prevents encoderd from crashing
-  @pytest.mark.parametrize("record_audio", [True, False])
+  @parameterized.expand([True, False])
   def test_record_audio(self, record_audio):
     params = Params()
     params.put_bool("RecordAudio", record_audio, block=True)

openpilot/system/loggerd/tests/test_uploader.py

@@ -37,7 +37,7 @@ cloudlog.addHandler(log_handler)
 
 class TestUploader(UploaderTestCase):
   def setup_method(self):
-    super().setup_method()
+    super().openpilot_setup_method()
     log_handler.reset()
 
   def start_thread(self):

openpilot/system/manager/test/test_manager.py

@@ -1,9 +1,10 @@
 import os
-import pytest
+import unittest
 import signal
 import time
 
 from opendbc.car.structs import car
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.params import Params
 import openpilot.system.manager.manager as manager
 from openpilot.system.manager.process import ensure_running
@@ -16,7 +17,7 @@ MAX_STARTUP_TIME = 3
 BLACKLIST_PROCS = ['manage_athenad', 'pandad', 'pigeond']
 
 
-class TestManager:
+class TestManager(OpenpilotTestCase):
   def setup_method(self):
     HARDWARE.set_power_save(False)
 
@@ -47,7 +48,7 @@ class TestManager:
     assert params.get("OpenpilotEnabledToggle")
     assert params.get("RouteCount") == 0
 
-  @pytest.mark.skip("this test is flaky the way it's currently written, should be moved to test_onroad")
+  @unittest.skip("this test is flaky the way it's currently written, should be moved to test_onroad")
   def test_clean_exit(self, subtests):
     """
       Ensure all processes exit cleanly when stopped.

openpilot/system/sensord/tests/test_sensord.py

@@ -1,10 +1,10 @@
 import os
 import subprocess
-import pytest
 import time
 import numpy as np
 from collections import namedtuple, defaultdict
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.common.gpio import get_irqs_for_action
@@ -54,8 +54,8 @@ def iter_measurements(events):
     for measurement in msgs:
       yield measurement, getattr(measurement, measurement.which())
 
-@pytest.mark.tici
-class TestSensord:
+class TestSensord(OpenpilotTestCase):
+  TICI_TEST = True
   @classmethod
   def setup_class(cls):
     # enable LSM self test

openpilot/system/tests/test_logmessaged.py

@@ -2,13 +2,14 @@ import glob
 import os
 import time
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.system.manager.process_config import managed_processes
 from openpilot.common.hardware.hw import Paths
 from openpilot.common.swaglog import cloudlog, ipchandler
 
 
-class TestLogmessaged:
+class TestLogmessaged(OpenpilotTestCase):
   def setup_method(self):
     # clear the IPC buffer in case some other tests used cloudlog and filled it
     ipchandler.close()
@@ -52,4 +53,3 @@ class TestLogmessaged:
 
     logsize = sum([os.path.getsize(f) for f in self._get_log_files()])
     assert (n*len(msg)) < logsize < (n*(len(msg)+1024))
-

openpilot/system/ubloxd/tests/test_pigeond.py

@@ -1,6 +1,6 @@
-import pytest
 import time
 
+from openpilot.common.test import OpenpilotTestCase
 import openpilot.cereal.messaging as messaging
 from openpilot.cereal.services import SERVICE_LIST
 from openpilot.common.gpio import gpio_read
@@ -10,8 +10,8 @@ from openpilot.common.hardware.tici.pins import GPIO
 
 
 # TODO: test TTFF when we have good A-GNSS
-@pytest.mark.tici
-class TestPigeond:
+class TestPigeond(OpenpilotTestCase):
+  TICI_TEST = True
 
   def teardown_method(self):
     managed_processes['pigeond'].stop()

openpilot/system/ui/lib/tests/test_handle_state_change.py

@@ -3,15 +3,16 @@
 Tests the state machine in isolation by constructing a WifiManager with mocked
 DBus, then calling _handle_state_change directly with NM state transitions.
 """
-import pytest
+import unittest
 from jeepney.low_level import MessageType
-from pytest_mock import MockerFixture
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import Mocker, OpenpilotTestCase
 from openpilot.system.ui.lib.networkmanager import NMDeviceState, NMDeviceStateReason
 from openpilot.system.ui.lib.wifi_manager import WifiManager, WifiState, ConnectStatus
 
 
-def _make_wm(mocker: MockerFixture, connections=None):
+def _make_wm(mocker: Mocker, connections=None):
   """Create a WifiManager with only the fields _handle_state_change touches."""
   mocker.patch.object(WifiManager, '_initialize')
   wm = WifiManager.__new__(WifiManager)
@@ -50,7 +51,7 @@ def fire_wpa_connect(wm: WifiManager) -> None:
 # Basic transitions
 # ---------------------------------------------------------------------------
 
-class TestDisconnected:
+class TestDisconnected(OpenpilotTestCase):
   def test_generic_disconnect_clears_state(self, mocker):
     wm = _make_wm(mocker)
     wm._wifi_state = WifiState(ssid="Net", status=ConnectStatus.CONNECTED)
@@ -92,7 +93,7 @@ class TestDisconnected:
     assert wm._wifi_state.status == ConnectStatus.DISCONNECTED
 
 
-class TestDeactivating:
+class TestDeactivating(OpenpilotTestCase):
   def test_deactivating_noop_for_non_connection_removed(self, mocker):
     """DEACTIVATING with non-CONNECTION_REMOVED reason is a no-op."""
     wm = _make_wm(mocker)
@@ -103,10 +104,10 @@ class TestDeactivating:
     assert wm._wifi_state.ssid == "Net"
     assert wm._wifi_state.status == ConnectStatus.CONNECTED
 
-  @pytest.mark.parametrize("status, expected_clears", [
+  @parameterized.expand([
     (ConnectStatus.CONNECTED, True),
     (ConnectStatus.CONNECTING, False),
-  ])
+  ], names=("status", "expected_clears"))
   def test_deactivating_connection_removed(self, mocker, status, expected_clears):
     """DEACTIVATING(CONNECTION_REMOVED) clears CONNECTED but preserves CONNECTING.
 
@@ -130,7 +131,7 @@ class TestDeactivating:
       assert wm._wifi_state.status == ConnectStatus.CONNECTING
 
 
-class TestPrepareConfig:
+class TestPrepareConfig(OpenpilotTestCase):
   def test_user_initiated_skips_dbus_lookup(self, mocker):
     """User called _set_connecting('B') — PREPARE must not overwrite via DBus.
 
@@ -148,7 +149,7 @@ class TestPrepareConfig:
     assert wm._wifi_state.status == ConnectStatus.CONNECTING
     wm._get_active_wifi_connection.assert_not_called()
 
-  @pytest.mark.parametrize("state", [NMDeviceState.PREPARE, NMDeviceState.CONFIG])
+  @parameterized.expand([NMDeviceState.PREPARE, NMDeviceState.CONFIG], names=("state",))
   def test_auto_connect_looks_up_ssid(self, mocker, state):
     """Auto-connection (ssid=None): PREPARE and CONFIG must look up ssid from NM."""
     wm = _make_wm(mocker, connections={"AutoNet": "/path/auto"})
@@ -179,7 +180,7 @@ class TestPrepareConfig:
     assert wm._wifi_state.status == ConnectStatus.CONNECTING
 
 
-class TestNeedAuth:
+class TestNeedAuth(OpenpilotTestCase):
   def test_wrong_password_fires_callback(self, mocker):
     """NEED_AUTH+SUPPLICANT_DISCONNECT from CONFIG = real wrong password."""
     wm = _make_wm(mocker)
@@ -272,16 +273,16 @@ class TestNeedAuth:
     assert len(wm._callback_queue) == 0
 
 
-class TestPassthroughStates:
+class TestPassthroughStates(OpenpilotTestCase):
   """NEED_AUTH (generic), IP_CONFIG, IP_CHECK, SECONDARIES, FAILED (generic) are no-ops."""
 
-  @pytest.mark.parametrize("state", [
+  @parameterized.expand([
     NMDeviceState.NEED_AUTH,
     NMDeviceState.IP_CONFIG,
     NMDeviceState.IP_CHECK,
     NMDeviceState.SECONDARIES,
     NMDeviceState.FAILED,
-  ])
+  ], names=("state",))
   def test_passthrough_is_noop(self, mocker, state):
     wm = _make_wm(mocker)
     wm._set_connecting("Net")
@@ -293,7 +294,7 @@ class TestPassthroughStates:
     assert len(wm._callback_queue) == 0
 
 
-class TestActivated:
+class TestActivated(OpenpilotTestCase):
   def test_sets_connected(self, mocker):
     """ACTIVATED sets status to CONNECTED and fires callback."""
     wm = _make_wm(mocker, connections={"MyNet": "/path/mynet"})
@@ -344,7 +345,7 @@ class TestActivated:
 # guard) shrink these race windows significantly. The epoch counter closes the
 # remaining gaps.
 
-class TestThreadRaces:
+class TestThreadRaces(OpenpilotTestCase):
   def test_prepare_race_user_tap_during_dbus(self, mocker):
     """User taps B while PREPARE's DBus call is in flight for auto-connect.
 
@@ -416,7 +417,7 @@ class TestThreadRaces:
 # Full sequences (NM signal order from real devices)
 # ---------------------------------------------------------------------------
 
-class TestFullSequences:
+class TestFullSequences(OpenpilotTestCase):
   def test_normal_connect(self, mocker):
     """User connects to saved network: full happy path.
 
@@ -771,7 +772,7 @@ class TestFullSequences:
     wm.process_callbacks()
     cb.assert_called_once_with("Hotspot")
 
-  @pytest.mark.xfail(reason="TODO: FAILED(SSID_NOT_FOUND) should emit error for UI")
+  @unittest.expectedFailure  # "TODO: FAILED(SSID_NOT_FOUND) should emit error for UI"
   def test_ssid_not_found(self, mocker):
     """Network drops off while connected — hotspot turned off.
 
@@ -843,7 +844,7 @@ class TestFullSequences:
 # Verified on device: when ActivateConnection returns UnknownConnection error,
 # NM emits no state signals. The worker error path is the only recovery point.
 
-class TestWorkerErrorRecovery:
+class TestWorkerErrorRecovery(OpenpilotTestCase):
   """Worker threads re-sync with NM via _init_wifi_state on DBus errors,
   preserving actual NM state instead of blindly clearing to DISCONNECTED."""
 

openpilot/system/webrtc/tests/test_stream_session.py

@@ -3,6 +3,7 @@ import json
 import time
 
 import capnp
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging, log
 from teleoprtc.tracks import VIDEO_CLOCK_RATE
 
@@ -10,7 +11,7 @@ from openpilot.system.webrtc.webrtcd import CerealOutgoingMessageProxy, CerealIn
 from openpilot.system.webrtc.device.video import LiveStreamVideoStreamTrack
 
 
-class TestStreamSession:
+class TestStreamSession(OpenpilotTestCase):
   def setup_method(self):
     self.loop = asyncio.new_event_loop()
 

openpilot/test_native.py

@@ -0,0 +1,25 @@
+import os
+import subprocess
+
+from openpilot.common.basedir import BASEDIR
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
+
+
+NATIVE_TESTS = (
+  "openpilot/common/tests/test_common",
+  "openpilot/selfdrive/pandad/tests/test_pandad_canprotocol",
+  "openpilot/system/loggerd/tests/test_logger",
+  "openpilot/tools/cabana/tests/test_cabana",
+  "openpilot/tools/cabana/tests/test_dbc_core",
+  "openpilot/tools/replay/tests/test_replay",
+)
+
+
+class TestNative(OpenpilotTestCase):
+  @parameterized.expand(NATIVE_TESTS)
+  def test_native(self, executable):
+    path = os.path.join(BASEDIR, executable)
+    if not os.path.exists(path):
+      self.skipTest(f"optional native test was not built: {executable}")
+    subprocess.run([path], check=True)

openpilot/tools/jotpluggler/test_jotpluggler.py

@@ -5,7 +5,8 @@ from pathlib import Path
 JOTPLUGGLER_DIR = Path(__file__).parent
 
 
-class TestJotpluggler:
+from openpilot.common.test import OpenpilotTestCase
+class TestJotpluggler(OpenpilotTestCase):
   def test_help(self):
     result = subprocess.run(["./jotpluggler", "-h"], cwd=JOTPLUGGLER_DIR, capture_output=True, text=True)
     assert result.returncode == 0, result.stderr

openpilot/tools/lib/tests/test_caching.py

@@ -3,8 +3,9 @@ import os
 import shutil
 import socket
 import tempfile
-import pytest
 
+from openpilot.common.parameterized import parameterized
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.selfdrive.test.helpers import http_server_context
 from openpilot.common.hardware.hw import Paths
 from openpilot.tools.lib.url_file import URLFile, prune_cache
@@ -30,12 +31,11 @@ class CachingTestRequestHandler(http.server.BaseHTTPRequestHandler):
     self.end_headers()
 
 
-@pytest.fixture
 def host():
   with http_server_context(handler=CachingTestRequestHandler) as (host, port):
     yield f"http://{host}:{port}"
 
-class TestFileDownload:
+class TestFileDownload(OpenpilotTestCase):
 
   def test_pipeline_defaults(self, host):
     # TODO: parameterize the defaults so we don't rely on hard-coded values in xx
@@ -59,7 +59,7 @@ class TestFileDownload:
     # ensure caching on by default and cache dir gets created
     os.environ.pop("DISABLE_FILEREADER_CACHE", None)
     if os.path.exists(Paths.download_cache_root()):
-      shutil.rmtree(Paths.download_cache_root())
+      shutil.rmtree(Paths.download_cache_root(), ignore_errors=True)
     URLFile(f"{host}/test.txt").get_length()
     URLFile(f"{host}/test.txt").read()
     assert os.path.exists(Paths.download_cache_root())
@@ -117,7 +117,7 @@ class TestFileDownload:
     self.compare_loads(large_file_url, length - 100, 100)
     self.compare_loads(large_file_url)
 
-  @pytest.mark.parametrize("cache_enabled", [True, False])
+  @parameterized.expand([True, False], names=("cache_enabled",))
   def test_recover_from_missing_file(self, host, cache_enabled):
     if cache_enabled:
       os.environ.pop("DISABLE_FILEREADER_CACHE", None)
@@ -135,7 +135,7 @@ class TestFileDownload:
     assert length == 4
 
 
-class TestCache:
+class TestCache(OpenpilotTestCase):
   def test_prune_cache(self, monkeypatch):
     with tempfile.TemporaryDirectory() as tmpdir:
       monkeypatch.setattr(Paths, 'download_cache_root', staticmethod(lambda: tmpdir + "/"))

openpilot/tools/lib/tests/test_comma_car_segments.py

@@ -1,13 +1,14 @@
-import pytest
+import unittest
 import requests
 from opendbc.car.fingerprints import MIGRATION
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.tools.lib.comma_car_segments import get_comma_car_segments_database, get_url
 from openpilot.tools.lib.logreader import LogReader
 from openpilot.tools.lib.route import SegmentRange
 
 
-@pytest.mark.skip(reason="huggingface is flaky, run this test manually to check for issues")
-class TestCommaCarSegments:
+@unittest.skip("huggingface is flaky, run this test manually to check for issues")
+class TestCommaCarSegments(OpenpilotTestCase):
   def test_database(self):
     database = get_comma_car_segments_database()
 

openpilot/tools/lib/tests/test_logreader.py

@@ -4,9 +4,10 @@ import io
 import shutil
 import tempfile
 import os
-import pytest
+import unittest
 import requests
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.parameterized import parameterized
 
 from openpilot.cereal import log as capnp_log
@@ -47,7 +48,7 @@ def setup_source_scenario(mocker, is_internal=False):
   yield
 
 
-class TestLogReader:
+class TestLogReader(OpenpilotTestCase):
   @parameterized.expand([
     (f"{TEST_ROUTE}", ALL_SEGS),
     (f"{TEST_ROUTE.replace('/', '|')}", ALL_SEGS),
@@ -72,7 +73,7 @@ class TestLogReader:
     (f"https://useradmin.comma.ai/?onebox={TEST_ROUTE.replace('/', '|')}", ALL_SEGS),
     (f"https://useradmin.comma.ai/?onebox={TEST_ROUTE.replace('/', '%7C')}", ALL_SEGS),
   ])
-  @pytest.mark.skip("this got flaky. internet tests are stupid.")
+  @unittest.skip("this got flaky. internet tests are stupid.")
   def test_indirect_parsing(self, identifier, expected):
     parsed = parse_indirect(identifier)
     sr = SegmentRange(parsed)
@@ -90,7 +91,7 @@ class TestLogReader:
     sr = SegmentRange(identifier)
     assert str(sr) == expected
 
-  @pytest.mark.parametrize("cache_enabled", [True, False])
+  @parameterized.expand([True, False], names=("cache_enabled",))
   def test_direct_parsing(self, mocker, cache_enabled):
     file_exists_mock = mocker.patch("openpilot.tools.lib.filereader.file_exists")
     if cache_enabled:
@@ -107,7 +108,7 @@ class TestLogReader:
       l = len(list(LogReader(f)))
       assert l > 100
 
-    with pytest.raises(URLFileException) if not cache_enabled else pytest.raises(AssertionError):
+    with self.assertRaises(URLFileException if not cache_enabled else AssertionError):
       l = len(list(LogReader(QLOG_FILE.replace("/3/", "/200/"))))
 
     # file_exists should not be called for direct files
@@ -126,44 +127,44 @@ class TestLogReader:
     (f"{TEST_ROUTE}--3a",),
   ])
   def test_bad_ranges(self, segment_range):
-    with pytest.raises(AssertionError):
+    with self.assertRaises(AssertionError):
       _ = SegmentRange(segment_range).seg_idxs
 
-  @pytest.mark.parametrize("segment_range, api_call", [
+  @parameterized.expand([
     (f"{TEST_ROUTE}/0", False),
     (f"{TEST_ROUTE}/:2", False),
     (f"{TEST_ROUTE}/0:", True),
     (f"{TEST_ROUTE}/-1", True),
     (f"{TEST_ROUTE}", True),
-  ])
+  ], names=("segment_range", "api_call"))
   def test_slicing_api_call(self, mocker, segment_range, api_call):
     max_seg_mock = mocker.patch("openpilot.tools.lib.route.get_max_seg_number_cached")
     max_seg_mock.return_value = NUM_SEGS
     _ = SegmentRange(segment_range).seg_idxs
     assert api_call == max_seg_mock.called
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_modes(self):
     qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0", ReadMode.QLOG)))
     rlog_len = len(list(LogReader(f"{TEST_ROUTE}/0", ReadMode.RLOG)))
 
     assert qlog_len * 6 < rlog_len
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_modes_from_name(self):
     qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/q")))
     rlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/r")))
 
     assert qlog_len * 6 < rlog_len
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_list(self):
     qlog_len = len(list(LogReader(f"{TEST_ROUTE}/0/q")))
     qlog_len_2 = len(list(LogReader([f"{TEST_ROUTE}/0/q", f"{TEST_ROUTE}/0/q"])))
 
     assert qlog_len * 2 == qlog_len_2
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_multiple_iterations(self, mocker):
     init_mock = mocker.patch("openpilot.tools.lib.logreader._LogFileReader")
     lr = LogReader(f"{TEST_ROUTE}/0/q")
@@ -175,14 +176,14 @@ class TestLogReader:
 
     assert qlog_len1 == qlog_len2
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_helpers(self):
     lr = LogReader(f"{TEST_ROUTE}/0/q")
     assert lr.first("carParams").carFingerprint == "SUBARU OUTBACK 6TH GEN"
     assert 0 < len(list(lr.filter("carParams"))) < len(list(lr))
 
   @parameterized.expand([(True,), (False,)])
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_run_across_segments(self, cache_enabled):
     if cache_enabled:
       os.environ.pop("DISABLE_FILEREADER_CACHE", None)
@@ -191,7 +192,7 @@ class TestLogReader:
     lr = LogReader(f"{TEST_ROUTE}/0:4")
     assert len(lr.run_across_segments(4, noop)) == len(list(lr))
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_auto_mode(self, subtests, mocker):
     lr = LogReader(f"{TEST_ROUTE}/0/q")
     qlog_len = len(list(lr))
@@ -207,7 +208,7 @@ class TestLogReader:
 
     with subtests.test("interactive_no"):
       mocker.patch("sys.stdin", new=io.StringIO("n\n"))
-      with pytest.raises(LogsUnavailable):
+      with self.assertRaises(LogsUnavailable):
         lr = LogReader(f"{TEST_ROUTE}/0", default_mode=ReadMode.AUTO_INTERACTIVE, sources=[comma_api_source])
 
     with subtests.test("non_interactive"):
@@ -215,7 +216,7 @@ class TestLogReader:
       log_len = len(list(lr))
       assert qlog_len == log_len
 
-  @pytest.mark.parametrize("is_internal", [True, False])
+  @parameterized.expand([True, False], names=("is_internal",))
   def test_auto_source_scenarios(self, mocker, is_internal):
     lr = LogReader(QLOG_FILE)
     qlog_len = len(list(lr))
@@ -225,7 +226,7 @@ class TestLogReader:
       log_len = len(list(lr))
       assert qlog_len == log_len
 
-  @pytest.mark.slow
+  @unittest.skipIf(os.environ.get("SKIP_SLOW"), "slow test")
   def test_sort_by_time(self):
     msgs = list(LogReader(f"{TEST_ROUTE}/0/q"))
     assert msgs != sorted(msgs, key=lambda m: m.logMonoTime)
@@ -254,7 +255,7 @@ class TestLogReader:
       # ensure new message is added, but is not a union type
       msgs = list(LogReader(qlog.name))
       assert len(msgs) == num_msgs + 1
-      with pytest.raises(capnp.KjException):
+      with self.assertRaises(capnp.KjException):
         [m.which() for m in msgs]
 
       # should not be added when only_union_types=True

openpilot/tools/lib/tests/test_route_library.py

@@ -1,8 +1,9 @@
 from collections import namedtuple
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.tools.lib.route import SegmentName
 
-class TestRouteLibrary:
+class TestRouteLibrary(OpenpilotTestCase):
   def test_segment_name_formats(self):
     Case = namedtuple('Case', ['input', 'expected_route', 'expected_segment_num', 'expected_data_dir'])
 

openpilot/tools/plotjuggler/test_plotjuggler.py

@@ -5,17 +5,18 @@ import signal
 import subprocess
 import time
 
-import pytest
+import unittest
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.common.basedir import BASEDIR
 from openpilot.common.timeout import Timeout
 from openpilot.tools.plotjuggler.juggle import DEMO_ROUTE, install
 
 PJ_DIR = os.path.join(BASEDIR, "openpilot/tools/plotjuggler")
 
-class TestPlotJuggler:
+class TestPlotJuggler(OpenpilotTestCase):
 
-  @pytest.mark.skipif(not shutil.which('qmake'), reason="Qt not installed")
+  @unittest.skipIf(not shutil.which('qmake'), "Qt not installed")
   def test_demo(self):
     install()
 

openpilot/tools/sim/tests/conftest.py

@@ -1,8 +0,0 @@
-import pytest
-
-def pytest_addoption(parser):
-  parser.addoption("--test_duration", action="store", default=60, type=int, help="Seconds to run metadrive drive")
-
-@pytest.fixture
-def test_duration(request):
-  return request.config.getoption("--test_duration")

openpilot/tools/sim/tests/test_metadrive_bridge.py

@@ -1,17 +1,22 @@
-import pytest
 import warnings
+import unittest
+import importlib
 
 # Since metadrive depends on pkg_resources, and pkg_resources is deprecated as an API
 warnings.filterwarnings("ignore", category=DeprecationWarning)
 
-from openpilot.tools.sim.bridge.metadrive.metadrive_bridge import MetaDriveBridge
+try:
+  MetaDriveBridge = importlib.import_module("openpilot.tools.sim.bridge.metadrive.metadrive_bridge").MetaDriveBridge
+except ModuleNotFoundError:
+  MetaDriveBridge = None
 from openpilot.tools.sim.tests.test_sim_bridge import TestSimBridgeBase
 
-@pytest.mark.slow
+@unittest.skipIf(MetaDriveBridge is None, "metadrive is not installed")
 class TestMetaDriveBridge(TestSimBridgeBase):
-  @pytest.fixture(autouse=True)
-  def setup_create_bridge(self, test_duration):
+  def setup_method(self):
+    super().openpilot_setup_method()
     self.test_duration = 30
 
   def create_bridge(self):
+    assert MetaDriveBridge is not None
     return MetaDriveBridge(False, False, self.test_duration, True)

openpilot/tools/sim/tests/test_sim_bridge.py

@@ -1,21 +1,23 @@
 import os
 import subprocess
 import time
-import pytest
+import unittest
 
 from multiprocessing import Queue
 
+from openpilot.common.test import OpenpilotTestCase
 from openpilot.cereal import messaging
 from openpilot.common.basedir import BASEDIR
 from openpilot.tools.sim.bridge.common import QueueMessageType
 
 SIM_DIR = os.path.join(BASEDIR, "openpilot/tools/sim")
 
-class TestSimBridgeBase:
+class TestSimBridgeBase(OpenpilotTestCase):
+  SLOW_TEST = True
   @classmethod
   def setup_class(cls):
     if cls is TestSimBridgeBase:
-      raise pytest.skip("Don't run this base class, run test_metadrive_bridge.py instead")
+      raise unittest.SkipTest("Don't run this base class, run test_metadrive_bridge.py instead")
 
   def setup_method(self):
     self.processes = []

pyproject.toml

@@ -65,11 +65,7 @@ testing = [
   "hypothesis ==6.47.*",
   "ty",
   "pytest",
-  "pytest-cpp",
-  "pytest-subtests",
-  # https://github.com/pytest-dev/pytest-xdist/pull/1229
-  "pytest-xdist @ git+https://github.com/sshane/pytest-xdist@2b4372bd62699fb412c4fe2f95bf9f01bd2018da",
-  "pytest-mock",
+  "pytest-xdist",
   "ruff",
   "codespell",
 ]
@@ -120,18 +116,8 @@ allow-direct-references = true
 
 [tool.pytest.ini_options]
 minversion = "6.0"
-addopts = "-Werror --strict-config --strict-markers --durations=10 -n auto --dist=loadgroup"
-cpp_files = "test_*"
-cpp_harness = "openpilot/selfdrive/test/cpp_harness.py"
+addopts = "-Werror --strict-config --durations=10"
 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",
 ]
@@ -168,6 +154,7 @@ line-length = 160
 lint.flake8-implicit-str-concat.allow-multiline = false
 
 [tool.ruff.lint.flake8-tidy-imports.banned-api]
+"pytest".msg = "Use unittest"
 "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
 

tools/op.sh

@@ -327,7 +327,7 @@ function op_lint() {
 
 function op_test() {
   op_before_cmd
-  op_run_command pytest "$@"
+  op_run_command pytest -n logical --dist worksteal "$@"
 }
 
 function op_replay() {
@@ -431,7 +431,7 @@ function op_default() {
   echo -e "  ${BOLD}sim${NC}          Run openpilot in a simulator"
   echo -e "  ${BOLD}lint${NC}         Run the linter"
   echo -e "  ${BOLD}post-commit${NC}  Install the linter as a post-commit hook"
-  echo -e "  ${BOLD}test${NC}         Run all unit tests from pytest"
+  echo -e "  ${BOLD}test${NC}         Run all unit tests"
   echo ""
   echo -e "${BOLD}${UNDERLINE}Options:${NC}"
   echo -e "  ${BOLD}-d, --dir${NC}"

uv.lock

@@ -749,9 +749,6 @@ testing = [
     { name = "coverage" },
     { name = "hypothesis" },
     { name = "pytest" },
-    { name = "pytest-cpp" },
-    { name = "pytest-mock" },
-    { name = "pytest-subtests" },
     { name = "pytest-xdist" },
     { name = "ruff" },
     { name = "ty" },
@@ -802,10 +799,7 @@ requires-dist = [
     { 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 = "pytest-xdist", marker = "extra == 'testing'" },
     { name = "pyzmq" },
     { name = "qrcode" },
     { name = "rednose", marker = "extra == 'submodules'", editable = "rednose_repo" },
@@ -1004,51 +998,18 @@ 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+g2b4372bd6"
-source = { git = "https://github.com/sshane/pytest-xdist?rev=2b4372bd62699fb412c4fe2f95bf9f01bd2018da#2b4372bd62699fb412c4fe2f95bf9f01bd2018da" }
+version = "3.8.0"
+source = { registry = "https://pypi.org/simple" }
 dependencies = [
     { name = "execnet" },
     { name = "pytest" },
 ]
+sdist = { url = "https://files.pythonhosted.org/packages/78/b4/439b179d1ff526791eb921115fca8e44e596a13efeda518b9d845a619450/pytest_xdist-3.8.0.tar.gz", hash = "sha256:7e578125ec9bc6050861aa93f2d59f1d8d085595d6551c2c90b6f4fad8d3a9f1", size = 88069, upload-time = "2025-07-01T13:30:59.346Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/ca/31/d4e37e9e550c2b92a9cbc2e4d0b7420a27224968580b5a447f420847c975/pytest_xdist-3.8.0-py3-none-any.whl", hash = "sha256:202ca578cfeb7370784a8c33d6d05bc6e13b4f25b5053c30a152269fd10f0b88", size = 46396, upload-time = "2025-07-01T13:30:56.632Z" },
+]
 
 [[package]]
 name = "python-dateutil"