Stabilize native registry and asynchronous source fixtures

opendbc_repo/opendbc/car/hyundai/tests/test_classic_deferred_takeover.py

@@ -23,6 +23,19 @@ from openpilot.starpilot.vehicle_startup import VehicleStartupOwner
 CARS = (CAR.GENESIS_G90, CAR.HYUNDAI_SONATA, CAR.HYUNDAI_SONATA_HYBRID, CAR.GENESIS_G80)
 
 
+@pytest.fixture(scope='module', autouse=True)
+def native_classic_positive_path():
+  # Positive diagnostic takeover is a DEBUG-only registry capability. The
+  # paired test_classic_startup_release suite retains RELEASE denial checks.
+  previous = libsafety_py.__dict__.get('libpath')
+  libsafety_py.load(libsafety_py._build_libsafety(release=False))
+  try:
+    yield
+  finally:
+    if previous is not None:
+      libsafety_py.load(previous)
+
+
 @pytest.mark.parametrize('identity', CARS)
 @pytest.mark.parametrize('scenario', (
   'healthy', 'stale_control', 'active_control', 'missing_source', 'native_mode',

openpilot/starpilot/conditional_mode/tests/test_card_frame_derivations.py

@@ -1,5 +1,6 @@
 """Frame-local pure derivations; live settings/authority checks still run."""
 from dataclasses import fields, replace
+import fcntl
 from pathlib import Path
 from types import SimpleNamespace
 from unittest.mock import patch
@@ -206,9 +207,18 @@ def test_actual_card_adoption_and_multiple_consumers_keep_live_authority(full_op
       with patch.object(module, 'REPLAY', False):
         card.configure_conditional_settings_worker()
       worker = card.conditional_settings_worker
+      def refresh_snapshot(now):
+        # Constructor Params writes can still be queued on the real writer.
+        # Serialize this source fixture with that writer; Card itself remains
+        # nonblocking and the original read-start timestamp is unchanged.
+        lock_path = Path(card.params.get_param_path('ConditionalModeConfig')).parent.parent / '.lock'
+        with lock_path.open('a+') as source_lock:
+          fcntl.flock(source_lock, fcntl.LOCK_SH)
+          worker.refresh(now)
+        assert worker.published.status == 'verified'
       for tick in (2, 3):
         now = NOW + tick * 10_000_000
-        worker.refresh(now)
+        refresh_snapshot(now)
         axis, _, command = drive(tick)
         with patch.object(runtime_settings, 'saved_selection_for_drive', wraps=runtime_settings.saved_selection_for_drive) as selection, \
              patch.object(worker.view, 'affirm', wraps=worker.view.affirm) as affirm:
@@ -220,7 +230,7 @@ def test_actual_card_adoption_and_multiple_consumers_keep_live_authority(full_op
       # A fresh restrictive worker publication must revoke optional proposals.
       Path(card.params.get_param_path('ConditionalModeConfig')).write_bytes(b'{')
       now = NOW + 40_000_000
-      worker.refresh(now)
+      refresh_snapshot(now)
       axis, _, command = drive(4)
       assert receive(now, axis=sd.pm.messages['aolAxisState'], event=sd.pm.messages['onroadEvents'])
       assert worker.view.verdict(worker.view.current, now_mono_ns=now, drive_id=NOW - 1_000_000_000).status == 'unavailable_document'

openpilot/starpilot/soloist/tests/test_runtime.py

@@ -334,19 +334,32 @@ def test_control_socket_round_trip(layout, tmp_path):
   stop = threading.Event()
   thread = threading.Thread(target=supervisor.serve, args=(sup, path, stop), daemon=True)
   thread.start()
-  for _ in range(50):
-    if Path(path).exists():
-      break
+  try:
     import time
-    time.sleep(0.02)
-  result = SpotifyClient(path).call('status')
-  assert result['status']['state'] == 'starting'
-  with pytest.raises(RuntimeError):
-    SpotifyClient(path).call('control', action='pause', args={})
-  stop.set()
-  thread.join(timeout=2)
-  Path(path).unlink(missing_ok=True)
-  os.rmdir(folder)
+    client = SpotifyClient(path)
+    result = None
+    for attempt in range(50):
+      if not Path(path).exists():
+        time.sleep(0.02)
+        continue
+      try:
+        result = client.call('status')
+        break
+      except (FileNotFoundError, ConnectionRefusedError):
+        if attempt == 49:
+          raise
+        time.sleep(0.02)
+    assert result is not None
+    assert result['status']['state'] == 'starting'
+    with pytest.raises(RuntimeError):
+      client.call('control', action='pause', args={})
+  finally:
+    stop.set()
+    thread.join(timeout=2)
+    Path(path).unlink(missing_ok=True)
+    os.rmdir(folder)
+  assert not thread.is_alive()
+
 
 
 def test_reap_orphans_only_targets_feature_binaries(layout, tmp_path, monkeypatch):

prebuilt.json

@@ -1 +1 @@
-{"protected_inventory_sha256": "83e9cc537ef292cc228b0a7fcabd7828406cb4cad1c767b591bccdfc4c61f6fa", "schema_version": 1}
+{"protected_inventory_sha256": "ec6bea42f0aa43a2767f792e9d09e7bc3d5ed61e062fa296a4b2ce69294818f5", "schema_version": 1}