mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-08-06 00:36:25 +08:00
firestar5683
91 lines
2.9 KiB
Python
91 lines
2.9 KiB
Python
import threading
|
|
import time
|
|
from types import SimpleNamespace
|
|
|
|
from openpilot.selfdrive.ui.lib.ui_param_cache import UIParamCache
|
|
from openpilot.selfdrive.ui import ui_state as ui_state_module
|
|
|
|
|
|
def test_raylib_ui_uses_read_through_param_cache():
|
|
assert isinstance(ui_state_module.ui_state.ui_params, UIParamCache)
|
|
assert ui_state_module.ui_state.ui_params is not ui_state_module.ui_state.params
|
|
|
|
|
|
def test_usbgpu_poll_does_not_block_ui_thread(monkeypatch):
|
|
started = threading.Event()
|
|
release = threading.Event()
|
|
|
|
def poll():
|
|
started.set()
|
|
release.wait(timeout=1.0)
|
|
return True
|
|
|
|
monkeypatch.setattr(ui_state_module, "chestnut_present", poll)
|
|
state = object.__new__(ui_state_module.UIState)
|
|
state.usbgpu = False
|
|
state._usbgpu_update_time = 0.0
|
|
state._usbgpu_poll_thread = None
|
|
|
|
state._schedule_usbgpu_poll(now=1.0, force=True)
|
|
assert started.wait(timeout=0.2)
|
|
polling_thread = state._usbgpu_poll_thread
|
|
state._schedule_usbgpu_poll(now=2.0, force=True)
|
|
assert state._usbgpu_poll_thread is polling_thread
|
|
|
|
release.set()
|
|
polling_thread.join(timeout=1.0)
|
|
assert state.usbgpu is True
|
|
|
|
|
|
def test_ui_update_reports_subphases(monkeypatch):
|
|
phases = []
|
|
calls = []
|
|
state = object.__new__(ui_state_module.UIState)
|
|
state.prime_state = SimpleNamespace(start=lambda: calls.append("prime"))
|
|
state.sm = SimpleNamespace(update=lambda timeout: calls.append(("submaster", timeout)))
|
|
def update_state(progress_hook):
|
|
progress_hook("ui.update.before_state_params")
|
|
calls.append("state")
|
|
progress_hook("ui.update.after_state_params")
|
|
|
|
state._update_state = update_state
|
|
state._update_status = lambda progress_hook: calls.append("status")
|
|
state._param_update_time = time.monotonic()
|
|
monkeypatch.setattr(ui_state_module, "device", SimpleNamespace(update=lambda: calls.append("device")))
|
|
|
|
state.update(progress_hook=phases.append)
|
|
|
|
assert calls == ["prime", ("submaster", 0), "state", "status", "device"]
|
|
assert phases == [
|
|
"ui.update.before_prime_state",
|
|
"ui.update.before_submaster",
|
|
"ui.update.before_state",
|
|
"ui.update.before_state_params",
|
|
"ui.update.after_state_params",
|
|
"ui.update.before_status",
|
|
"ui.update.before_params",
|
|
"ui.update.before_device",
|
|
"ui.update.after_device",
|
|
]
|
|
|
|
|
|
def test_ui_update_reports_offroad_callback(monkeypatch):
|
|
phases = []
|
|
callback_calls = []
|
|
state = object.__new__(ui_state_module.UIState)
|
|
state.started = False
|
|
state._started_prev = True
|
|
state._engaged_prev = False
|
|
state.status = ui_state_module.UIStatus.DISENGAGED
|
|
state.sm = SimpleNamespace(frame=2)
|
|
state._offroad_transition_callbacks = [lambda: callback_calls.append("offroad")]
|
|
state._engaged_transition_callbacks = []
|
|
|
|
state._update_status(phases.append)
|
|
|
|
assert callback_calls == ["offroad"]
|
|
assert phases == [
|
|
"ui.update.before_offroad_callback.<lambda>",
|
|
"ui.update.after_offroad_callback.<lambda>",
|
|
]
|