StarPilot/system/manager/test/test_process_config.py

from types import SimpleNamespace

import pytest

from cereal import car
from opendbc.car.ford.values import CAR as FORD_CAR
from openpilot.system.manager.process_config import (
  allow_uploads,
  bluetooth_enabled,
  camera_run,
  managed_processes,
  sentry_mode,
  soundd_run,
  ublox,
  wheel_controls_enabled,
)


class FakeParams:
  def __init__(self, always_allow_uploads: bool = False):
    self.always_allow_uploads = always_allow_uploads

  def get_bool(self, key: str) -> bool:
    assert key == "AlwaysAllowUploads"
    return self.always_allow_uploads


@pytest.mark.parametrize(
  "started,no_uploads,no_onroad_uploads,always_allow_uploads,expected",
  [
    (True, False, False, False, True),
    (False, False, False, False, True),
    (True, True, False, False, False),
    (False, True, False, False, False),
    (True, True, True, False, False),
    (False, True, True, False, True),
    (True, True, False, True, True),
  ],
)
def test_allow_uploads(started, no_uploads, no_onroad_uploads, always_allow_uploads, expected):
  params = FakeParams(always_allow_uploads)
  toggles = SimpleNamespace(no_uploads=no_uploads, no_onroad_uploads=no_onroad_uploads)

  assert allow_uploads(started, params, car.CarParams.new_message(), toggles) is expected


def test_uploader_runs_at_background_priority():
  assert managed_processes["uploader"].nice == 19


@pytest.mark.parametrize("enabled", [False, True])
def test_bluetooth_process_is_param_gated(enabled):
  params = SimpleNamespace(get_bool=lambda key: enabled if key == "BluetoothEnabled" else False)
  assert bluetooth_enabled(False, params, car.CarParams.new_message(), SimpleNamespace()) is enabled


@pytest.mark.parametrize(
  "started,driver_view,audio_test,expected",
  [(True, False, False, True), (False, True, False, True), (False, False, True, True), (False, False, False, False)],
)
def test_soundd_runs_for_driving_and_bluetooth_audio_test(started, driver_view, audio_test, expected):
  values = {"IsDriverViewEnabled": driver_view, "BluetoothAudioTestActive": audio_test}
  params = SimpleNamespace(get_bool=lambda key: values.get(key, False))
  assert soundd_run(started, params, car.CarParams.new_message(), SimpleNamespace()) is expected


def test_wheel_controls_process_runs_on_supported_devices_at_background_priority():
  process = managed_processes["wheel_controlsd"]
  assert process.nice == 19


@pytest.mark.parametrize("enabled", [False, True])
def test_wheel_controls_process_is_mapping_gated(enabled):
  params = SimpleNamespace(get_bool=lambda key: enabled if key == "WheelControlsEnabled" else False)
  assert wheel_controls_enabled(False, params, car.CarParams.new_message(), SimpleNamespace()) is enabled


class CameraParams:
  def __init__(self, capture: bool):
    self.capture = capture

  def get_bool(self, key: str) -> bool:
    assert key in {"IsDriverViewEnabled", "SentryModeCapture"}
    return self.capture if key == "SentryModeCapture" else False


@pytest.mark.parametrize(
  "started,capture,expected",
  [
    (False, True, True),
    (False, False, False),
    (True, False, True),
  ],
)
def test_camera_run_preserves_onroad_camera_and_offroad_sentry_capture(started, capture, expected):
  assert camera_run(started, CameraParams(capture), car.CarParams.new_message(), SimpleNamespace()) is expected


class SentryParams:
  def __init__(self, enabled: bool):
    self.enabled = enabled

  def get_bool(self, key: str) -> bool:
    assert key == "SentryModeEnabled"
    return self.enabled


@pytest.mark.parametrize("started,enabled,expected", [(True, True, False), (False, True, True), (False, False, False)])
def test_sentry_process_is_offroad_only(started, enabled, expected):
  assert sentry_mode(started, SentryParams(enabled), car.CarParams.new_message(), SimpleNamespace()) is expected


class GpsParams:
  def __init__(self, CP=None):
    self.values = {}
    if CP is not None:
      self.values["CarParams"] = CP.to_bytes()

  def get(self, key: str):
    return self.values.get(key)

  def get_bool(self, key: str) -> bool:
    return bool(self.values.get(key, False))

  def put_bool(self, key: str, value: bool):
    self.values[key] = value


def test_ublox_waits_for_current_carparams(monkeypatch):
  monkeypatch.setattr("openpilot.system.manager.process_config.ublox_available", lambda: True)
  params = GpsParams()

  assert not ublox(True, params, car.CarParams.new_message(), SimpleNamespace())
  assert params.get_bool("UbloxAvailable")


@pytest.mark.parametrize("car_gps,expected", [(False, True), (True, False)])
def test_ublox_has_single_external_gps_publisher(monkeypatch, car_gps, expected):
  monkeypatch.setattr("openpilot.system.manager.process_config.ublox_available", lambda: True)
  CP = car.CarParams.new_message()
  if car_gps:
    CP.brand = "ford"
    CP.carFingerprint = FORD_CAR.FORD_MUSTANG_MACH_E_MK1
  else:
    CP.brand = "mock"
    CP.carFingerprint = "mock"
  params = GpsParams(CP)

  assert ublox(True, params, car.CarParams.new_message(), SimpleNamespace()) is expected
  assert params.get_bool("CarGpsAvailable") is car_gps