faster unit tests (#39202)

This commit is contained in:
Adeeb Shihadeh
2026-11-02 15:24:40 +00:00
committed by firestar5683
parent e5931a7321
commit c710aea5ce
10 changed files with 110 additions and 110 deletions
+14 -13
View File
@@ -5,6 +5,7 @@ import unittest
from functools import cache
from openpilot.common.test import OpenpilotTestCase
from openpilot.common.parameterized import parameterized
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, \
@@ -131,19 +132,19 @@ class TestLagd(OpenpilotTestCase):
assert msg.lateralDelay.validBlocks == 0
assert msg.lateralDelay.calPerc == 0
def test_estimator_basics(self, subtests):
for lag_frames in range(LAGD_MIN_LAG_FRAMES, LAGD_MAX_LAG_FRAMES - 1):
with subtests.test(msg=f"lag_frames={lag_frames}"):
mocked_CP = car.CarParams(steerActuatorDelay=0.5)
estimator = LateralLagEstimator(mocked_CP, DT, min_recovery_buffer_sec=0.0, min_yr=0.0)
process_messages(estimator, lag_frames, int(MIN_OKAY_WINDOW_SEC / DT) + BLOCK_NUM_NEEDED * BLOCK_SIZE)
msg = estimator.get_msg(True)
assert msg.lateralDelay.status == 'estimated'
assert np.allclose(msg.lateralDelay.lateralDelay, lag_frames * DT, atol=0.01)
assert np.allclose(msg.lateralDelay.lateralDelayEstimate, lag_frames * DT, atol=0.01)
assert np.allclose(msg.lateralDelay.lateralDelayEstimateStd, 0.0, atol=0.01)
assert msg.lateralDelay.validBlocks == BLOCK_NUM_NEEDED
assert msg.lateralDelay.calPerc == 100
@parameterized.expand(list(range(LAGD_MIN_LAG_FRAMES, LAGD_MAX_LAG_FRAMES - 1)),
ids=lambda lag_frames: f"lag_frames_{lag_frames}")
def test_estimator_basics(self, lag_frames):
mocked_CP = car.CarParams(steerActuatorDelay=0.5)
estimator = LateralLagEstimator(mocked_CP, DT, min_recovery_buffer_sec=0.0, min_yr=0.0)
process_messages(estimator, lag_frames, int(MIN_OKAY_WINDOW_SEC / DT) + BLOCK_NUM_NEEDED * BLOCK_SIZE)
msg = estimator.get_msg(True)
assert msg.lateralDelay.status == 'estimated'
assert np.allclose(msg.lateralDelay.lateralDelay, lag_frames * DT, atol=0.01)
assert np.allclose(msg.lateralDelay.lateralDelayEstimate, lag_frames * DT, atol=0.01)
assert np.allclose(msg.lateralDelay.lateralDelayEstimateStd, 0.0, atol=0.01)
assert msg.lateralDelay.validBlocks == BLOCK_NUM_NEEDED
assert msg.lateralDelay.calPerc == 100
def test_estimator_masking(self):
mocked_CP, lag_frames = car.CarParams(steerActuatorDelay=0.5), random.randint(LAGD_MIN_LAG_FRAMES, LAGD_MAX_LAG_FRAMES - 1)
@@ -1,15 +1,13 @@
import fcntl
import numpy as np
import os
import tempfile
from collections import defaultdict
from functools import cache
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
from openpilot.selfdrive.test.process_replay.process_replay import replay_process_with_name
from openpilot.selfdrive.test.process_replay.process_replay import get_process_config, replay_process_with_name
# TODO find a new segment to test
TEST_ROUTE = "4019fff6e54cf1c7|00000123--4bc0d95ef6/5"
@@ -68,7 +66,15 @@ def modify_logs_midway(logs, which, count, fn):
return sorted(non_which + which, key=lambda x: x.logMonoTime)
def run_scenarios(scenario, logs):
@cache
def get_logs():
# only keep what locationd consumes, plus carParams for replay setup and deviceMotion for comparison
services = set(get_process_config('locationd').pubs) | {'carParams', 'deviceMotion'}
return [m for m in migrate_all(LogReader(TEST_ROUTE)) if m.which() in services]
def run_scenarios(scenario):
logs = get_logs()
if scenario == Scenario.BASE:
pass
@@ -106,23 +112,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
- locationd kalman filter should never go unstable (we care mostly about yaw_rate, roll, gpsOK, inputsOK, sensorsOK)
- faulty values should be ignored, with appropriate flags set
"""
@classmethod
def setup_class(cls):
# 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)
SHARED_DOWNLOAD_CACHE = True # reuse the route across tests and runs
def test_base(self):
"""
@@ -131,7 +121,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
- yaw_rate: unchanged
- roll: unchanged
"""
orig_data, replayed_data = run_scenarios(Scenario.BASE, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.BASE)
assert np.allclose(orig_data['yaw_rate'], replayed_data['yaw_rate'], atol=np.radians(0.35))
assert np.allclose(orig_data['roll'], replayed_data['roll'], atol=np.radians(0.35))
@@ -143,7 +133,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
- roll: 0
- sensorsOK: False
"""
_, replayed_data = run_scenarios(Scenario.GYRO_OFF, self.logs)
_, replayed_data = run_scenarios(Scenario.GYRO_OFF)
assert np.allclose(replayed_data['yaw_rate'], 0.0)
assert np.allclose(replayed_data['roll'], 0.0)
assert np.all(replayed_data['sensors_flag'] == 0.0)
@@ -156,7 +146,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
- roll: unchanged
- inputsOK: False for some time after the spike, True for the rest
"""
orig_data, replayed_data = run_scenarios(Scenario.GYRO_SPIKE_MIDWAY, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.GYRO_SPIKE_MIDWAY)
assert np.allclose(orig_data['yaw_rate'], replayed_data['yaw_rate'], atol=np.radians(0.35))
assert np.allclose(orig_data['roll'], replayed_data['roll'], atol=np.radians(0.35))
assert np.all(replayed_data['inputs_flag'] == orig_data['inputs_flag'])
@@ -167,7 +157,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
Test: consistent timing spikes for N gyroscope messages in the middle of the segment
Expected Result: inputsOK becomes False after N of bad measurements
"""
orig_data, replayed_data = run_scenarios(Scenario.GYRO_CONSISTENT_SPIKES, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.GYRO_CONSISTENT_SPIKES)
assert np.diff(replayed_data['inputs_flag'])[501] == -1.0
assert np.diff(replayed_data['inputs_flag'])[708] == 1.0
@@ -179,7 +169,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
- roll: 0
- sensorsOK: False
"""
_, replayed_data = run_scenarios(Scenario.ACCEL_OFF, self.logs)
_, replayed_data = run_scenarios(Scenario.ACCEL_OFF)
assert np.allclose(replayed_data['yaw_rate'], 0.0)
assert np.allclose(replayed_data['roll'], 0.0)
assert np.all(replayed_data['sensors_flag'] == 0.0)
@@ -190,7 +180,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
Test: an accelerometer spike in the middle of the segment
Expected Result: Right now, the kalman filter is not robust to small spikes like it is to gyroscope spikes.
"""
orig_data, replayed_data = run_scenarios(Scenario.ACCEL_SPIKE_MIDWAY, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.ACCEL_SPIKE_MIDWAY)
assert np.allclose(orig_data['yaw_rate'], replayed_data['yaw_rate'], atol=np.radians(0.35))
assert np.allclose(orig_data['roll'], replayed_data['roll'], atol=np.radians(0.35))
@@ -199,7 +189,7 @@ class TestLocationdScenarios(OpenpilotTestCase):
Test: timing of 150ms off for the single accelerometer message in the middle of the segment
Expected Result: the message is ignored, and inputsOK is False for that time
"""
orig_data, replayed_data = run_scenarios(Scenario.SENSOR_TIMING_SPIKE_MIDWAY, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.SENSOR_TIMING_SPIKE_MIDWAY)
assert np.all(replayed_data['inputs_flag'] == orig_data['inputs_flag'])
assert np.all(replayed_data['sensors_flag'] == orig_data['sensors_flag'])
@@ -208,6 +198,6 @@ class TestLocationdScenarios(OpenpilotTestCase):
Test: consistent timing spikes for N accelerometer messages in the middle of the segment
Expected Result: inputsOK becomes False after N of bad measurements
"""
orig_data, replayed_data = run_scenarios(Scenario.SENSOR_TIMING_CONSISTENT_SPIKES, self.logs)
orig_data, replayed_data = run_scenarios(Scenario.SENSOR_TIMING_CONSISTENT_SPIKES)
assert np.diff(replayed_data['inputs_flag'])[501] == -1.0
assert np.diff(replayed_data['inputs_flag'])[707] == 1.0
@@ -1,30 +1,22 @@
#!/usr/bin/env python3
import time
import numpy as np
from openpilot.cereal import log
import openpilot.cereal.messaging as messaging
from openpilot.common.realtime import Ratekeeper, DT_MDL
from openpilot.common.realtime import DT_MDL
from openpilot.selfdrive.controls.lib.longcontrol import LongCtrlState
from openpilot.selfdrive.modeld.constants import ModelConstants
from openpilot.selfdrive.controls.lib.longitudinal_planner import LongitudinalPlanner
from openpilot.selfdrive.controls.radard import _LEAD_ACCEL_TAU
T_IDXS = np.array(ModelConstants.T_IDXS)
class Plant:
messaging_initialized = False
def __init__(self, lead_relevancy=False, speed=0.0, distance_lead=2.0,
enabled=True, only_lead2=False, only_radar=False, e2e=False, personality=0, force_decel=False):
self.rate = 1. / DT_MDL
if not Plant.messaging_initialized:
Plant.radar = messaging.pub_sock('radarState')
Plant.controls_state = messaging.pub_sock('controlsState')
Plant.selfdrive_state = messaging.pub_sock('selfdriveState')
Plant.car_state = messaging.pub_sock('carState')
Plant.plan = messaging.sub_sock('longitudinalPlan')
Plant.messaging_initialized = True
self.frame = 0
self.v_lead_prev = 0.0
@@ -43,10 +35,7 @@ class Plant:
self.personality = personality
self.force_decel = force_decel
self.rk = Ratekeeper(self.rate, print_delay_threshold=100.0)
self.ts = 1. / self.rate
time.sleep(0.1)
self.sm = messaging.SubMaster(['longitudinalPlan'])
from opendbc.car.honda.values import CAR
from opendbc.car.honda.interface import CarInterface
@@ -55,7 +44,7 @@ class Plant:
@property
def current_time(self):
return float(self.rk.frame) / self.rate
return float(self.frame) / self.rate
def step(self, v_lead=0.0, prob_lead=1.0, v_cruise=50., pitch=0.0, prob_throttle=1.0):
# ******** publish a fake model going straight and fake calibration ********
@@ -103,19 +92,12 @@ class Plant:
# Simulate model predicting slightly faster speed
# this is to ensure lead policy is effective when model
# does not predict slowdown in e2e mode
position = log.XYZTData.new_message()
position.x = [float(x) for x in (self.speed + 0.5) * np.array(ModelConstants.T_IDXS)]
model.modelV2.position = position
# does not predict slowdown
model.modelV2.position.x = ((self.speed + 0.5) * T_IDXS).tolist()
model.modelV2.action.desiredAcceleration = float(self.acceleration + 0.5)
velocity = log.XYZTData.new_message()
velocity.x = [float(x) for x in (self.speed + 0.5) * np.ones_like(ModelConstants.T_IDXS)]
velocity.x[0] = float(self.speed) # always start at current speed
model.modelV2.velocity = velocity
acceleration = log.XYZTData.new_message()
acceleration.x = [float(x) for x in np.zeros_like(ModelConstants.T_IDXS)]
model.modelV2.acceleration = acceleration
model.modelV2.meta.disengagePredictions.gasPressProbs = [float(prob_throttle) for _ in range(6)]
# always start at current speed
model.modelV2.velocity.x = [float(self.speed)] + [float(self.speed + 0.5)] * (len(T_IDXS) - 1)
model.modelV2.acceleration.x = [0.0] * len(T_IDXS)
control.controlsState.longControlState = LongCtrlState.pid if self.enabled else LongCtrlState.off
ss.selfdriveState.experimentalMode = self.e2e
@@ -165,7 +147,7 @@ class Plant:
# ******** update prevs ********
self.rk.monitor_time()
self.frame += 1
return {
"distance": self.distance,
@@ -488,7 +488,10 @@ class TestAthenadMethods(OpenpilotTestCase):
assert isinstance(resp[k], str), f"{k} is not a string"
assert len(resp[k]) > 0, f"{k} has no value"
def test_jsonrpc_handler(self):
def test_jsonrpc_handler(self, mocker):
# private queues, so the shutdown wakeup below can't leak into other tests
for q in ('recv_queue', 'send_queue', 'log_recv_queue'):
mocker.patch.object(athenad, q, type(getattr(athenad, q))())
end_event = threading.Event()
thread = threading.Thread(target=athenad.jsonrpc_handler, args=(end_event,))
thread.daemon = True
@@ -508,6 +511,8 @@ class TestAthenadMethods(OpenpilotTestCase):
assert json.loads(resp) == {'result': {'success': 1}, 'id': 0, 'jsonrpc': '2.0'}
finally:
end_event.set()
# wake the handler from its blocking queue get
athenad.recv_queue.put_nowait(json.dumps({'result': None, 'id': 0, 'jsonrpc': '2.0'}))
thread.join()
def test_get_logs_to_send_sorted(self):
@@ -1,5 +1,6 @@
import json
from Crypto.PublicKey import RSA
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from pathlib import Path
from openpilot.common.test import OpenpilotTestCase
@@ -23,12 +24,11 @@ class TestRegistration(OpenpilotTestCase):
self.dongle_id = persist_dir / "dongle_id"
def _generate_keys(self):
self.pub_key.touch()
k = RSA.generate(2048)
k = rsa.generate_private_key(public_exponent=65537, key_size=2048)
with open(self.priv_key, "wb") as f:
f.write(k.export_key())
f.write(k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.TraditionalOpenSSL, serialization.NoEncryption()))
with open(self.pub_key, "wb") as f:
f.write(k.publickey().export_key())
f.write(k.public_key().public_bytes(serialization.Encoding.PEM, serialization.PublicFormat.SubjectPublicKeyInfo))
def test_valid_cache(self, mocker):
# if all params are written, return the cached dongle id.
@@ -85,7 +85,7 @@ class TestLoggerd(OpenpilotTestCase):
managed_processes["loggerd"].start()
for s in services:
assert pm.wait_for_readers_to_update(s, timeout=5)
assert pm.wait_for_readers_to_update(s, timeout=5, dt=0.001)
sent_msgs = defaultdict(list)
for i in range(random.randint(2, 10) * 100):
@@ -101,10 +101,10 @@ class TestLoggerd(OpenpilotTestCase):
# that loggerd logged every message we sent.
if (i + 1) % 100 == 0:
for s in services:
assert pm.wait_for_readers_to_update(s, timeout=5)
assert pm.wait_for_readers_to_update(s, timeout=5, dt=0.001)
for s in services:
assert pm.wait_for_readers_to_update(s, timeout=5)
assert pm.wait_for_readers_to_update(s, timeout=5, dt=0.001)
managed_processes["loggerd"].stop()
return sent_msgs
@@ -131,7 +131,7 @@ class TestLoggerd(OpenpilotTestCase):
os.environ["LOGGERD_SEGMENT_LENGTH"] = str(segment_length)
managed_processes["loggerd"].start()
managed_processes["encoderd"].start()
assert pm.wait_for_readers_to_update("narrowRoadCameraState", timeout=5)
assert pm.wait_for_readers_to_update("narrowRoadCameraState", timeout=5, dt=0.001)
fps = 20
for n in range(1, int(num_segs * segment_length * fps) + 1):
+35 -13
View File
@@ -1,6 +1,7 @@
import glob
import os
import time
import uuid
from openpilot.common.test import OpenpilotTestCase
import openpilot.cereal.messaging as messaging
@@ -8,17 +9,27 @@ from openpilot.system.manager.process_config import managed_processes
from openpilot.common.hardware.hw import Paths
from openpilot.common.swaglog import SWAGLOG_MAX_SIZE, cloudlog
TIMEOUT = 10.
class TestLogmessaged(OpenpilotTestCase):
def setup_method(self):
managed_processes['logmessaged'].start()
while not os.path.exists(Paths.swaglog_ipc()):
time.sleep(0.01)
self.sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)
self.error_sock = messaging.sub_sock("logMessage", timeout=1000, conflate=False)
self.sock = messaging.sub_sock("logMessage", timeout=100, conflate=False)
self.error_sock = messaging.sub_sock("logMessage", timeout=100, conflate=False)
# ensure sockets are connected
time.sleep(0.5)
# ensure sockets are connected: probe until each socket has received one.
# debug logs are published but not written to the log files.
# logmessaged handles logs in order, so after the final drain no probes are in flight
pending = {self.sock, self.error_sock}
deadline = time.monotonic() + TIMEOUT
while pending:
assert time.monotonic() < deadline, "timed out connecting to logmessaged"
cloudlog.debug("probe")
time.sleep(0.01)
pending = {s for s in pending if not messaging.drain_sock(s)}
messaging.drain_sock(self.sock)
messaging.drain_sock(self.error_sock)
@@ -30,12 +41,26 @@ class TestLogmessaged(OpenpilotTestCase):
def _get_log_files(self):
return list(glob.glob(os.path.join(Paths.swaglog_root(), "swaglog.*")))
def _recv_one(self):
deadline = time.monotonic() + TIMEOUT
while (m := messaging.recv_one(self.sock)) is None:
assert time.monotonic() < deadline, "timed out waiting for logMessage"
return m
def _drain(self):
# logmessaged handles logs in order, so everything logged before the sentinel is received before it
sentinel = f"sentinel {uuid.uuid4()}"
cloudlog.debug(sentinel)
msgs = []
while sentinel not in (m := self._recv_one()).logMessage:
msgs.append(m)
return msgs
def test_simple_log(self):
msgs = [f"abc {i}" for i in range(10)]
for m in msgs:
cloudlog.error(m)
time.sleep(0.5)
m = messaging.drain_sock(self.sock)
m = self._drain()
assert len(m) == len(msgs)
assert len(self._get_log_files()) >= 1
@@ -44,11 +69,9 @@ class TestLogmessaged(OpenpilotTestCase):
msg = "a"*300*1024
for _ in range(n):
cloudlog.info(msg)
time.sleep(0.05)
time.sleep(0.5)
msgs = messaging.drain_sock(self.sock)
assert len(msgs) == n
# wait for each one, the non-blocking sender drops logs that don't fit in the socket buffer
assert msg in self._recv_one().logMessage
assert len(self._drain()) == 0
logsize = sum([os.path.getsize(f) for f in self._get_log_files()])
assert (n*len(msg)) < logsize < (n*(len(msg)+1024))
@@ -56,7 +79,6 @@ class TestLogmessaged(OpenpilotTestCase):
def test_too_big_log(self):
cloudlog.info("a"*SWAGLOG_MAX_SIZE)
cloudlog.info("small")
time.sleep(0.5)
msgs = messaging.drain_sock(self.sock)
msgs = self._drain()
assert len(msgs) == 1
+6 -2
View File
@@ -104,9 +104,13 @@ class TestFileDownload(OpenpilotTestCase):
self.compare_loads(small_file_url, length - 100, 100)
self.compare_loads(small_file_url, 50, 100)
# Load small file 100 bytes at a time
# Load small file 100 bytes at a time from the cache, checked against one full download
data = URLFile(small_file_url, cache=False).read()
assert len(data) == length
for i in range(length // 100):
self.compare_loads(small_file_url, 100 * i, 100)
file_cached = URLFile(small_file_url, cache=True)
file_cached.seek(100 * i)
assert file_cached.read(ll=100) == data[100 * i:100 * (i + 1)]
def test_large_file(self):
large_file_url = "https://commadataci.blob.core.windows.net/openpilotci/0375fdf7b1ce594d/2019-06-13--08-32-25/3/qlog.bz2"
+8 -13
View File
@@ -49,19 +49,14 @@ def setup_source_scenario(mocker, is_internal=False):
class TestLogReader(OpenpilotTestCase):
@classmethod
def setUpClass(cls):
super().setUpClass()
cls.tmpdir = tempfile.TemporaryDirectory()
cls.qlog_path = os.path.join(cls.tmpdir.name, "qlog")
cls.rlog_path = os.path.join(cls.tmpdir.name, "rlog")
cls._write_test_log(cls.qlog_path, 10, include_car_params=True)
cls._write_test_log(cls.rlog_path, 100)
@classmethod
def tearDownClass(cls):
cls.tmpdir.cleanup()
super().tearDownClass()
def setUp(self):
super().setUp()
tmpdir = tempfile.TemporaryDirectory()
self.addCleanup(tmpdir.cleanup)
self.qlog_path = os.path.join(tmpdir.name, "qlog")
self.rlog_path = os.path.join(tmpdir.name, "rlog")
self._write_test_log(self.qlog_path, 10, include_car_params=True)
self._write_test_log(self.rlog_path, 100)
@staticmethod
def _write_test_log(path, count, include_car_params=False):
@@ -16,6 +16,7 @@ from openpilot.tools.plotjuggler.juggle import DEMO_ROUTE, install
PJ_DIR = os.path.join(BASEDIR, "openpilot/tools/plotjuggler")
class TestPlotJuggler(OpenpilotTestCase):
SHARED_DOWNLOAD_CACHE = True # reuse the demo route across runs; this tests PlotJuggler, not downloading
@unittest.skipIf(not shutil.which('qmake'), "Qt not installed")
def test_demo(self):