Test Fixes

openpilot/starpilot/galaxy/server.py

@@ -97,7 +97,8 @@ def make_server(*, port=8082, host='127.0.0.1', monitor=None, owner=None, crashe
                 sounds=None, software_operations=None, drive_stats=None, layout_preview_socket=None, controllers_socket=None,
                 remote_pairing=None, parked=None, camera_snapshot=None, clock=time.monotonic, starport_setup=None,
                 starport_client=None, navigation=None, offline_maps=None, drive_state=None, cloud_provider=None, cloud_offroad=None, projection_layout=None,
-                local_access=None, tmux_live=None, starport_logs=None, starport_log_sender=None, hotspot=None, toyota_tools=None, replay_client=None, replay_root=None,
+                local_access=None, tmux_live=None, starport_logs=None, starport_log_sender=None, hotspot=None, toyota_tools=None,
+                replay_client=None, replay_root=None,
                 spotify_setup=None, companion=None, phone_location=None, reverse_proxy=None):
   from openpilot.starpilot.galaxy.companion import CompanionRegistry, APK_PATH
   from openpilot.starpilot.gps.phone import PhoneGpsPublisher

openpilot/starpilot/galaxy/starport_send.py

@@ -279,7 +279,7 @@ class StarPortLogSender:
         left_out = " The device snapshot was left out to fit Discord's 10 MB limit."
       self._set(message="Sending logs")
       warning = self._sender(data, name, note, summary)
-      self._set(state="sent", message=f"Logs sent.{left_out} {warning or 'Thanks!'}".strip(), sentAt=int(time.time()))
+      self._set(state="sent", message=f"Logs sent.{left_out} {warning or 'Thanks!'}".strip(), sentAt=int(time.time()))  # noqa: TID251 shown as a wall-clock time
     except SendRejected as error:
       self._set(state="failed", message=str(error)[:400])
     except Exception as error:

openpilot/starpilot/galaxy/web/css/galaxy.css

@@ -2064,7 +2064,7 @@ progress { max-width: 100%; accent-color: var(--primary); }
 .gx-guided .gx-starport-pill { border: 0; background: var(--g-fill-hover); color: var(--g-muted); font-size: 13px; font-weight: var(--fw-bold); }
 .gx-guided .gx-starport-pill--ready { background: color-mix(in srgb, var(--g-done) 20%, transparent); color: var(--g-done); }
 
-/* Connection options: a radio row with a coloured icon tile, as in iOS Settings. */
+/* Connection options: a radio row with a colored icon tile, as in iOS Settings. */
 .gx-guided__options { --g-inset: 70px; }
 .gx-guided__options > li { display: block; }
 .gx-guided__option {

openpilot/starpilot/system/starport/bugsink_report.py

@@ -170,6 +170,11 @@ MAX_OFFERED_MODES = 16
 SETUP_TYPES = ("builtin", "adapter", "usb", "app")
 
 
+def _config_field(config: dict, number: int, default=None):
+  values = config.get(str(number), config.get(number))
+  return values[0] if isinstance(values, list) and values and isinstance(values[0], int) else default
+
+
 def offered_modes(events: list[dict]) -> list[str]:
   """Every video mode the head unit offered, e.g. "1280x720 H.264 margins 0x240", from its service discovery.
 
@@ -182,14 +187,11 @@ def offered_modes(events: list[dict]) -> list[str]:
       if not isinstance(config, dict):
         continue
 
-      def field(number, default=None):
-        values = config.get(str(number), config.get(number))
-        return values[0] if isinstance(values, list) and values and isinstance(values[0], int) else default
-
-      resolution, codec = field(1), field(10, CODEC_H264_BP)
+      resolution, codec = _config_field(config, 1), _config_field(config, 10, CODEC_H264_BP)
       text = f"{RESOLUTION_NAMES.get(resolution, f'resolution {resolution}')} {CODEC_NAMES.get(codec, f'codec {codec}')}"
-      if field(3, 0) or field(4, 0):
-        text += f" margins {field(3, 0)}x{field(4, 0)}"
+      margins = _config_field(config, 3, 0), _config_field(config, 4, 0)
+      if any(margins):
+        text += f" margins {margins[0]}x{margins[1]}"
       modes.append(text)
   return modes[:MAX_OFFERED_MODES]
 

openpilot/starpilot/system/starport/new_setup.py

@@ -11,7 +11,7 @@ FLAG_PATH = DATA_DIR / "new_setup"
 
 
 def enabled(path: Path | None = None) -> bool:
-  """Absent, unreadable or unrecognised means on; only an explicit 0 turns the new flow off."""
+  """Absent, unreadable or unrecognized means on; only an explicit 0 turns the new flow off."""
   try:
     return (path or FLAG_PATH).read_text().strip() != "0"
   except (OSError, UnicodeError):

openpilot/starpilot/system/starport/session.py

@@ -662,11 +662,9 @@ class ProjectionSession(Session):
   grace_network = ""
   grace_focus_lost = False
 
-  @staticmethod
-  def radio_change() -> str | None:
-    """Replaced by the supervisor: the comma's other Wi-Fi network's name if it joined or roamed in the last few
-    seconds ("" when unnamed), else None."""
-    return None
+  # Replaced by the supervisor: the comma's other Wi-Fi network's name if it joined or roamed in the last few
+  # seconds ("" when unnamed), else None.
+  radio_change: Callable[..., str | None] = staticmethod(lambda: None)
 
   # HeadunitPad acknowledges every fragment (codec config included) rather than every frame.
   fragment_acks = False

openpilot/starpilot/system/starport/setup_ladder.py

@@ -133,7 +133,7 @@ def _clean(entry) -> dict:
 
 
 class SetupLadder:
-  def __init__(self, path: Path | None = None, boot_id=_boot_id, clock=time.time):
+  def __init__(self, path: Path | None = None, boot_id=_boot_id, clock=time.time):  # noqa: TID251 persisted timestamps
     self.path = path or STATE_PATH
     self.boot_id, self.clock = boot_id, clock
     self.cars: dict[str, dict] = {}

openpilot/starpilot/system/starport/supervisor.py

@@ -435,7 +435,7 @@ class Supervisor:
     self._attempt_ended_at: float | None = None
     self._hfp_link = threading.Event()  # set when the chosen car opens the hands-free link
     self._modern_reply: dict[str, bool] = {}  # car address -> answer its v4+ version request like a current phone
-    self._ladder = setup_ladder.SetupLadder(identity_store.CONFIG_PATH.with_name("setup_state.json"))   # which handshake to try for each car, and which one last worked
+    self._ladder = setup_ladder.SetupLadder(identity_store.CONFIG_PATH.with_name("setup_state.json"))  # handshake to try per car
     self._strategy: setup_ladder.Strategy | None = None   # the one this attempt runs
     self._video_seen = False                    # video was acknowledged in this session: later failures say nothing about the handshake
     self._run_started = 0.0
@@ -1455,7 +1455,6 @@ class Supervisor:
     bluez.connect_device(address)
     self._wait(1.0)
 
-    from openpilot.starpilot.system.starport import bt_sockets
     channel = int(config.get("rfcomm_channel") or 0)
     source = "config"
     if not channel:

openpilot/starpilot/system/starport/tests/test_bugsink_report.py

@@ -450,7 +450,7 @@ OFFER = {"channels": [{"id": 3, "video_configs": [
 
 
 def adapter_session():
-  events = [list(item) for item in STREAMED]
+  events: list = [list(item) for item in STREAMED]
   events[0] = (0, 0, "session_start", {"receiver": "AAWireless", "trigger": "auto",
                                        "setup": {"type": "adapter", "profile": "aawireless", "profile_source": "override"}})
   events[5] = (0, 10, "discovered", {**OFFER, "head_unit": STREAMED[5][3]["head_unit"]})
@@ -479,7 +479,7 @@ def test_report_keeps_setup_offered_modes_and_the_working_strategy(tmp_path):
 
 
 def test_report_records_that_the_address_was_not_sent_and_old_logs_still_report(tmp_path):
-  events = [list(item) for item in STREAMED]
+  events: list = [list(item) for item in STREAMED]
   events.insert(6, (0, 10, "bluetooth_opened", {"channel": 6, "phone_address": ""}))
   assert br.drive_event([br.session_record(write_session(tmp_path, 1, events))])["tags"]["aa.phone_address_sent"] == "no"
   old = br.session_record(write_session(tmp_path, 2, STREAMED))  # a log written before these fields existed
@@ -489,14 +489,14 @@ def test_report_records_that_the_address_was_not_sent_and_old_logs_still_report(
 
 
 def test_unknown_setup_values_are_dropped(tmp_path):
-  events = [list(item) for item in STREAMED]
+  events: list = [list(item) for item in STREAMED]
   events[0] = (0, 0, "session_start", {"setup": {"type": "<script>", "profile": "x" * 90, "profile_source": "guess"}})
   setup = br.session_record(write_session(tmp_path, 1, events))["setup"]
   assert "type" not in setup and "profile_source" not in setup and len(setup["profile"]) <= 40
 
 
 def test_report_shows_what_the_ladder_tried_and_what_worked(tmp_path):
-  events = [list(item) for item in STREAMED]
+  events: list = [list(item) for item in STREAMED]
   events.insert(1, (0, 1, "ladder_strategy", {"rung": "standard", "learned": False}))
   events.insert(2, (0, 5, "ladder_failed", {"outcome": "advanced", "rung": "standard", "next": "other_reply"}))
   events.insert(3, (0, 6, "ladder_strategy", {"rung": "other_reply", "learned": False}))
@@ -510,7 +510,7 @@ def test_report_shows_what_the_ladder_tried_and_what_worked(tmp_path):
 
 
 def test_report_says_how_the_rfcomm_channel_was_found(tmp_path):
-  events = [list(item) for item in STREAMED]
+  events: list = [list(item) for item in STREAMED]
   events.insert(1, (0, 1, "sdp_retry", {"error": "x"}))
   events.insert(2, (0, 2, "rfcomm_channel", {"channel": 8, "source": "sdp"}))
   assert br.session_record(write_session(tmp_path, 1, events))["sdp"] == {"channel": 8, "source": "sdp", "retries": 1}

openpilot/starpilot/system/starport/tests/test_car_adoption.py

@@ -44,7 +44,8 @@ def test_the_only_candidate_is_chosen_trusted_and_the_projection_role_given_back
   sup._adopt_only_car(1000.0)
   assert sup.config["receiver_address"] == CAR and sup.config["receiver_name"] == "Civic"
   assert identity_store.load_config()["receiver_address"] == CAR
-  assert bluez.trusted[-1] == CAR and bluez.acquired == sup.baseline + 1 and bluez.class_restored >= 1  # the phone look is given back; the standby gateway stays
+  # the phone look is given back; the standby gateway stays
+  assert bluez.trusted[-1] == CAR and bluez.acquired == sup.baseline + 1 and bluez.class_restored >= 1
   assert "car_adopted" in names(sup) and sup._candidates == []
 
 

openpilot/starpilot/system/starport/tests/test_car_never_answered.py

@@ -40,6 +40,7 @@ class TestChooseOnlyCar(unittest.TestCase):
   def test_exactly_one_paired_android_auto_device_is_adopted(self):
     from openpilot.starpilot.system.starport.supervisor import choose_only_car
     only, candidates = choose_only_car([self.car("A"), self.car("B", starport=False), self.car("C", paired=False)])
+    assert only is not None
     self.assertEqual((only["address"], [c["address"] for c in candidates]), ("A", ["A"]))
 
   def test_two_candidates_or_none_are_never_guessed(self):

openpilot/starpilot/system/starport/tests/test_reachability.py

@@ -61,7 +61,8 @@ def test_stream_losses_are_told_apart_by_what_the_log_says(tmp_path):
   assert stream_loss_kind("projecting: Head unit disconnected after 4/8 bytes") == "disconnected"
   assert stream_loss_kind("something else") == "other"
   for text, kind in (("Video acknowledgement older than 2.5 s", "stalled"), ("Head unit disconnected after 4/8 bytes", "disconnected")):
-    events = [(0, 0, "session_start", {}), (0, 1, "stage", {"state": "streaming"}), (0, 9, "attempt_failed", {"stage": "streaming", "kind": "EOFError", "error": text})]
+    events = [(0, 0, "session_start", {}), (0, 1, "stage", {"state": "streaming"}),
+              (0, 9, "attempt_failed", {"stage": "streaming", "kind": "EOFError", "error": text})]
     event = br.drive_event([br.session_record(write_session(tmp_path, 3 if kind == "stalled" else 4, events))])
     assert event["tags"]["aa.cause"] == "projection_interrupted" and event["tags"]["aa.loss_kind"] == kind
 

openpilot/starpilot/system/starport/tests/test_setup_ladder.py

@@ -18,7 +18,7 @@ def ladder(tmp_path):
 
 
 def refuse(ladder, strategy, **overrides):
-  values = dict(ladder_failure=True, video_seen=False, version_major=4, adapter=False)
+  values = {"ladder_failure": True, "video_seen": False, "version_major": 4, "adapter": False}
   values.update(overrides)
   return ladder.failed(CAR, strategy, **values)
 
@@ -157,11 +157,11 @@ def test_only_sixteen_cars_are_kept_newest_first(tmp_path):
 
 
 # ---- the supervisor's use of it -------------------------------------------------------------------------------------
-import time as _time  # noqa: E402
-from types import SimpleNamespace  # noqa: E402
+import time as _time
+from types import SimpleNamespace
 
-from openpilot.starpilot.system.starport.tests.fake_head_unit import make_identity  # noqa: E402
-from openpilot.starpilot.system.starport.tests.test_starport import make_supervisor  # noqa: E402
+from openpilot.starpilot.system.starport.tests.fake_head_unit import make_identity
+from openpilot.starpilot.system.starport.tests.test_starport import make_supervisor
 
 RECEIVER = "AA:BB:CC:DD:EE:01"  # the car make_supervisor selects
 

openpilot/starpilot/system/starport/tests/test_shared_radio_grace.py

@@ -102,6 +102,7 @@ def test_watch_reports_a_carrier_change_with_the_network_name(tmp_path, monkeypa
       assert time.monotonic() < deadline
       time.sleep(0.01)
     assert logged == [("other_wifi_changed", {"interface": "wlan0", "network": "Huckleberrys House"})]
+    assert watch.changed_at is not None
     assert watch.recent(now=watch.changed_at + supervisor_module.RADIO_CHANGE_WINDOW + 0.1) is None
   finally:
     watch.stop()