mirror of
https://github.com/firestar5683/StarPilot.git
synced 2026-07-07 22:52:06 +08:00
firestar5683
442 lines
16 KiB
Python
442 lines
16 KiB
Python
import os
|
|
import pytest
|
|
import signal
|
|
import time
|
|
from pathlib import Path
|
|
import json
|
|
|
|
from cereal import car
|
|
from openpilot.common.params import Params, ParamKeyFlag
|
|
import openpilot.system.manager.manager as manager
|
|
from openpilot.system.manager.process import ensure_running
|
|
from openpilot.system.manager.process import PythonProcess
|
|
from openpilot.system.manager.process_config import (
|
|
managed_processes,
|
|
procs,
|
|
python_ui_enabled,
|
|
python_ui_process_start_method,
|
|
)
|
|
from openpilot.system.hardware import HARDWARE
|
|
|
|
os.environ['FAKEUPLOAD'] = "1"
|
|
|
|
MAX_STARTUP_TIME = 3
|
|
BLACKLIST_PROCS = ['manage_athenad', 'pandad', 'pigeond']
|
|
|
|
|
|
class FileBackedFakeParams:
|
|
def __init__(
|
|
self,
|
|
root: Path,
|
|
values: dict[str, object] | None = None,
|
|
keys: set[str] | None = None,
|
|
flags: dict[str, ParamKeyFlag] | None = None,
|
|
):
|
|
self.root = root
|
|
self.keys = set(keys or [])
|
|
self.flags = dict(flags or {})
|
|
self.root.mkdir(parents=True, exist_ok=True)
|
|
for key, value in (values or {}).items():
|
|
self.put(key, value)
|
|
|
|
def get_param_path(self, key=""):
|
|
return str(self.root / (key.decode() if isinstance(key, bytes) else str(key)))
|
|
|
|
def all_keys(self):
|
|
return sorted(self.keys)
|
|
|
|
def get_key_flag(self, key):
|
|
return self.flags.get(key.decode() if isinstance(key, bytes) else str(key), ParamKeyFlag.PERSISTENT)
|
|
|
|
def get(self, key):
|
|
path = Path(self.get_param_path(key))
|
|
if not path.is_file():
|
|
return None
|
|
|
|
raw = path.read_bytes()
|
|
try:
|
|
return raw.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return raw
|
|
|
|
def get_bool(self, key):
|
|
value = self.get(key)
|
|
if value is None:
|
|
return False
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8", errors="ignore")
|
|
return str(value).strip().lower() in ("1", "true", "yes", "on")
|
|
|
|
def put(self, key, value):
|
|
self.keys.add(key.decode() if isinstance(key, bytes) else str(key))
|
|
path = Path(self.get_param_path(key))
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
if isinstance(value, bytes):
|
|
raw = value
|
|
elif isinstance(value, bool):
|
|
raw = b"1" if value else b"0"
|
|
elif isinstance(value, float):
|
|
raw = str(float(value)).encode("utf-8")
|
|
elif isinstance(value, (dict, list)):
|
|
raw = json.dumps(value, separators=(",", ":")).encode("utf-8")
|
|
else:
|
|
raw = str(value).encode("utf-8")
|
|
|
|
path.write_bytes(raw)
|
|
|
|
def put_bool(self, key, value):
|
|
self.put(key, bool(value))
|
|
|
|
def put_float(self, key, value):
|
|
self.put(key, float(value))
|
|
|
|
|
|
class TestManager:
|
|
def setup_method(self):
|
|
HARDWARE.set_power_save(False)
|
|
|
|
# ensure clean CarParams
|
|
params = Params()
|
|
params.clear_all()
|
|
|
|
def teardown_method(self):
|
|
manager.manager_cleanup()
|
|
|
|
def test_manager_prepare(self):
|
|
os.environ['PREPAREONLY'] = '1'
|
|
manager.main()
|
|
|
|
def test_duplicate_procs(self):
|
|
assert len(procs) == len(managed_processes), "Duplicate process names"
|
|
|
|
def test_remote_access_procs_start_before_ui(self):
|
|
names = [p.name for p in procs]
|
|
ui_idx = names.index("ui")
|
|
|
|
assert names.index("the_galaxy") < ui_idx
|
|
assert names.index("galaxy") < ui_idx
|
|
|
|
def test_python_ui_process_start_method_follows_ui_implementation(self):
|
|
assert python_ui_process_start_method(False, False) == "fork"
|
|
assert python_ui_process_start_method(True, False) == "subprocess"
|
|
assert python_ui_process_start_method(True, True) == "fork"
|
|
|
|
def test_python_ui_subprocess_is_scoped_to_ui(self):
|
|
ui_proc = managed_processes["ui"]
|
|
uses_python_ui = python_ui_enabled(HARDWARE.get_device_type())
|
|
subprocess_scoped_procs = {"the_galaxy", "updated"}
|
|
|
|
assert isinstance(ui_proc, PythonProcess) == uses_python_ui
|
|
if uses_python_ui:
|
|
assert ui_proc.start_method == python_ui_process_start_method(uses_python_ui)
|
|
for proc in procs:
|
|
if isinstance(proc, PythonProcess) and proc.name not in subprocess_scoped_procs | {"ui"}:
|
|
assert proc.start_method is None
|
|
|
|
def test_python_ui_env_override(self, monkeypatch):
|
|
monkeypatch.setenv("USE_RAYLIB_UI", "1")
|
|
assert python_ui_enabled("tici") is True
|
|
|
|
monkeypatch.setenv("USE_RAYLIB_UI", "0")
|
|
assert python_ui_enabled("mici") is False
|
|
|
|
def test_manager_startup_toggles_use_params_only(self, tmp_path, monkeypatch):
|
|
monkeypatch.setenv("GIT_BRANCH", "StarPilot")
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"DeviceManagement": True,
|
|
"NoLogging": True,
|
|
"NoUploads": True,
|
|
"DisableOnroadUploads": True,
|
|
"SpeedLimitFiller": True,
|
|
"VisionSpeedLimitDetection": True,
|
|
"ForceOffroad": True,
|
|
"ForceOnroad": False,
|
|
})
|
|
|
|
toggles = manager._get_manager_startup_toggles(params)
|
|
|
|
assert toggles.no_logging is True
|
|
assert toggles.no_uploads is True
|
|
assert toggles.no_onroad_uploads is True
|
|
assert toggles.speed_limit_filler is True
|
|
assert toggles.vision_speed_limit_detection is True
|
|
assert toggles.force_offroad is True
|
|
assert toggles.force_onroad is False
|
|
|
|
def test_restore_missing_params_from_cache_preserves_live_values(self, tmp_path):
|
|
params = FileBackedFakeParams(
|
|
tmp_path / "params",
|
|
{"ExistingParam": "live"},
|
|
keys={"ExistingParam", "RestoredParam", "MissingParam", "TransientParam"},
|
|
flags={"TransientParam": ParamKeyFlag.CLEAR_ON_MANAGER_START},
|
|
)
|
|
params_cache = FileBackedFakeParams(
|
|
tmp_path / "cache",
|
|
{"ExistingParam": "cached", "RestoredParam": "restored", "TransientParam": "stale"},
|
|
)
|
|
|
|
restored_keys = manager._restore_missing_params_from_cache(params, params_cache)
|
|
|
|
assert restored_keys == ["RestoredParam"]
|
|
assert params.get("ExistingParam") == "live"
|
|
assert params.get("RestoredParam") == "restored"
|
|
assert params.get("MissingParam") is None
|
|
assert params.get("TransientParam") is None
|
|
|
|
def test_iter_param_store_keys_skips_lock_and_temp_files(self, tmp_path):
|
|
store_path = tmp_path / "params"
|
|
store_path.mkdir()
|
|
(store_path / "GoodParam").write_text("1")
|
|
(store_path / ".lock").write_text("")
|
|
(store_path / ".tmp_value_abc").write_text("stale")
|
|
(store_path / "nested").mkdir()
|
|
|
|
assert manager._iter_param_store_keys(store_path) == {"GoodParam"}
|
|
|
|
def test_blacklisted_procs(self):
|
|
# TODO: ensure there are blacklisted procs until we have a dedicated test
|
|
assert len(BLACKLIST_PROCS), "No blacklisted procs to test not_run"
|
|
|
|
def test_set_params_with_default_value(self):
|
|
params = Params()
|
|
params.clear_all()
|
|
|
|
os.environ['PREPAREONLY'] = '1'
|
|
manager.main()
|
|
for k in params.all_keys():
|
|
default_value = params.get_default_value(k)
|
|
if default_value is not None:
|
|
assert params.get(k) == default_value
|
|
assert params.get("OpenpilotEnabledToggle")
|
|
assert params.get("RouteCount") == 0
|
|
|
|
def test_migrate_legacy_experimental_longitudinal(self):
|
|
class FakeParams:
|
|
def __init__(self, values):
|
|
self.values = dict(values)
|
|
|
|
def get(self, key):
|
|
return self.values.get(key)
|
|
|
|
def get_bool(self, key):
|
|
value = self.values.get(key)
|
|
if value is None:
|
|
return False
|
|
if isinstance(value, bytes):
|
|
value = value.decode("utf-8", errors="ignore")
|
|
if isinstance(value, str):
|
|
return value.strip().lower() in ("1", "true", "yes", "on")
|
|
return bool(value)
|
|
|
|
def put_bool(self, key, value):
|
|
self.values[key] = b"1" if value else b"0"
|
|
|
|
def remove(self, key):
|
|
self.values.pop(key, None)
|
|
|
|
params = FakeParams({"ExperimentalLongitudinalEnabled": b"1"})
|
|
params_cache = FakeParams({})
|
|
|
|
manager.migrate_legacy_experimental_longitudinal(params, params_cache)
|
|
|
|
assert params.get_bool("AlphaLongitudinalEnabled")
|
|
assert params_cache.get_bool("AlphaLongitudinalEnabled")
|
|
assert params.get("ExperimentalLongitudinalEnabled") is None
|
|
assert params_cache.get("ExperimentalLongitudinalEnabled") is None
|
|
|
|
def test_migrate_starpilot_default_parity_preserves_existing_values(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_DEFAULTS_PARITY_MIGRATION_FLAG", tmp_path / "starpilot_defaults_parity_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"AdvancedLateralTune": False,
|
|
"ForceAutoTuneOff": False,
|
|
"HumanAcceleration": True,
|
|
"CEModelStopTime": 3.5,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {
|
|
"NNFF": True,
|
|
})
|
|
|
|
manager.migrate_starpilot_default_parity(params, params_cache)
|
|
|
|
assert not params.get_bool("AdvancedLateralTune")
|
|
assert not params.get_bool("ForceAutoTuneOff")
|
|
assert params.get_bool("HumanAcceleration")
|
|
assert params.get("CEModelStopTime") == "3.5"
|
|
assert params_cache.get_bool("NNFF")
|
|
|
|
def test_migrate_disable_humanlike_defaults(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_HUMANLIKE_DISABLE_MIGRATION_FLAG", tmp_path / "starpilot_humanlike_disable_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"HumanAcceleration": True,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {
|
|
"HumanLaneChanges": True,
|
|
})
|
|
|
|
manager.migrate_disable_humanlike_defaults(params, params_cache)
|
|
|
|
assert not params.get_bool("HumanAcceleration")
|
|
assert not params.get_bool("HumanLaneChanges")
|
|
assert not params_cache.get_bool("HumanAcceleration")
|
|
assert not params_cache.get_bool("HumanLaneChanges")
|
|
|
|
def test_cleanup_removed_starpilot_params(self, tmp_path):
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"HumanFollowing": True,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {
|
|
"HumanFollowing": False,
|
|
})
|
|
|
|
manager.cleanup_removed_starpilot_params(params, params_cache)
|
|
|
|
assert not Path(params.get_param_path("HumanFollowing")).exists()
|
|
assert not Path(params_cache.get_param_path("HumanFollowing")).exists()
|
|
|
|
def test_migrate_legacy_starpilot_params_cache_copies_marker_sources(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PARAMS_CACHE_MIGRATION_FLAG", tmp_path / "starpilot_params_cache_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params")
|
|
legacy_cache = tmp_path / "legacy_cache"
|
|
new_cache = tmp_path / "new_cache"
|
|
legacy_store = manager._params_store_path(legacy_cache)
|
|
legacy_store.mkdir(parents=True)
|
|
(legacy_store / "RemapCancelToDistance").write_text("0")
|
|
(legacy_store / "ClusterOffset").write_text("1.02")
|
|
|
|
manager.migrate_legacy_starpilot_params_cache(params, legacy_cache, new_cache)
|
|
|
|
new_store = manager._params_store_path(new_cache)
|
|
assert (new_store / "RemapCancelToDistance").read_text() == "0"
|
|
assert (new_store / "ClusterOffset").read_text() == "1.02"
|
|
assert manager.STARPILOT_PARAMS_CACHE_MIGRATION_FLAG.exists()
|
|
|
|
def test_migrate_legacy_starpilot_params_cache_skips_without_marker(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PARAMS_CACHE_MIGRATION_FLAG", tmp_path / "starpilot_params_cache_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params")
|
|
legacy_cache = tmp_path / "legacy_cache"
|
|
new_cache = tmp_path / "new_cache"
|
|
legacy_store = manager._params_store_path(legacy_cache)
|
|
legacy_store.mkdir(parents=True)
|
|
(legacy_store / "ClusterOffset").write_text("1.02")
|
|
|
|
manager.migrate_legacy_starpilot_params_cache(params, legacy_cache, new_cache)
|
|
|
|
assert not (manager._params_store_path(new_cache) / "ClusterOffset").exists()
|
|
assert manager.STARPILOT_PARAMS_CACHE_MIGRATION_FLAG.exists()
|
|
|
|
def test_migrate_legacy_starpilot_params_cache_does_not_overwrite_new_cache(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PARAMS_CACHE_MIGRATION_FLAG", tmp_path / "starpilot_params_cache_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params")
|
|
legacy_cache = tmp_path / "legacy_cache"
|
|
new_cache = tmp_path / "new_cache"
|
|
legacy_store = manager._params_store_path(legacy_cache)
|
|
new_store = manager._params_store_path(new_cache)
|
|
legacy_store.mkdir(parents=True)
|
|
new_store.mkdir(parents=True)
|
|
(legacy_store / "RemapCancelToDistance").write_text("0")
|
|
(legacy_store / "ClusterOffset").write_text("1.02")
|
|
(new_store / "ClusterOffset").write_text("1.0")
|
|
|
|
manager.migrate_legacy_starpilot_params_cache(params, legacy_cache, new_cache)
|
|
|
|
assert (new_store / "ClusterOffset").read_text() == "1.0"
|
|
assert (new_store / "RemapCancelToDistance").read_text() == "0"
|
|
|
|
def test_migrate_cluster_offset_default_resets_legacy_default_only(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_CLUSTER_OFFSET_MIGRATION_FLAG", tmp_path / "starpilot_cluster_offset_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"ClusterOffset": 1.015,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
|
|
|
|
manager.migrate_cluster_offset_default(params, params_cache)
|
|
|
|
assert params.get("ClusterOffset") == "1.0"
|
|
assert params_cache.get("ClusterOffset") == "1.0"
|
|
|
|
def test_migrate_cluster_offset_default_preserves_custom_values(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_CLUSTER_OFFSET_MIGRATION_FLAG", tmp_path / "starpilot_cluster_offset_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"ClusterOffset": 1.02,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
|
|
|
|
manager.migrate_cluster_offset_default(params, params_cache)
|
|
|
|
assert params.get("ClusterOffset") == "1.02"
|
|
assert params_cache.get("ClusterOffset") is None
|
|
|
|
def test_migrate_prioritize_smooth_following_default_seeds_disabled(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PRIORITIZE_SMOOTH_FOLLOWING_MIGRATION_FLAG", tmp_path / "starpilot_prioritize_smooth_following_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
|
|
|
|
manager.migrate_prioritize_smooth_following_default(params, params_cache)
|
|
|
|
assert not params.get_bool("PrioritizeSmoothFollowing")
|
|
assert not params_cache.get_bool("PrioritizeSmoothFollowing")
|
|
|
|
def test_migrate_prioritize_smooth_following_default_inverts_legacy_coast_toggle(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PRIORITIZE_SMOOTH_FOLLOWING_MIGRATION_FLAG", tmp_path / "starpilot_prioritize_smooth_following_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"CoastUpToLeads": False,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
|
|
|
|
manager.migrate_prioritize_smooth_following_default(params, params_cache)
|
|
|
|
assert params.get_bool("PrioritizeSmoothFollowing")
|
|
assert params_cache.get_bool("PrioritizeSmoothFollowing")
|
|
|
|
def test_migrate_prioritize_smooth_following_default_preserves_existing_values(self, tmp_path, monkeypatch):
|
|
monkeypatch.setattr(manager, "STARPILOT_PRIORITIZE_SMOOTH_FOLLOWING_MIGRATION_FLAG", tmp_path / "starpilot_prioritize_smooth_following_v1")
|
|
|
|
params = FileBackedFakeParams(tmp_path / "params", {
|
|
"PrioritizeSmoothFollowing": True,
|
|
"CoastUpToLeads": True,
|
|
})
|
|
params_cache = FileBackedFakeParams(tmp_path / "cache", {})
|
|
|
|
manager.migrate_prioritize_smooth_following_default(params, params_cache)
|
|
|
|
assert params.get_bool("PrioritizeSmoothFollowing")
|
|
|
|
@pytest.mark.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.
|
|
"""
|
|
HARDWARE.set_power_save(False)
|
|
manager.manager_init()
|
|
|
|
CP = car.CarParams.new_message()
|
|
procs = ensure_running(managed_processes.values(), True, Params(), CP, not_run=BLACKLIST_PROCS)
|
|
|
|
time.sleep(10)
|
|
|
|
for p in procs:
|
|
with subtests.test(proc=p.name):
|
|
state = p.get_process_state_msg()
|
|
assert state.running, f"{p.name} not running"
|
|
exit_code = p.stop(retry=False)
|
|
|
|
assert p.name not in BLACKLIST_PROCS, f"{p.name} was started"
|
|
|
|
assert exit_code is not None, f"{p.name} failed to exit"
|
|
|
|
# TODO: interrupted blocking read exits with 1 in cereal. use a more unique return code
|
|
exit_codes = [0, 1]
|
|
if p.sigkill:
|
|
exit_codes = [-signal.SIGKILL]
|
|
assert exit_code in exit_codes, f"{p.name} died with {exit_code}"
|