diff --git a/openpilot/starpilot/drive_state/tests/test_galaxy_drive_eligibility.py b/openpilot/starpilot/drive_state/tests/test_galaxy_drive_eligibility.py index fdacbb365e..96e9eeb270 100644 --- a/openpilot/starpilot/drive_state/tests/test_galaxy_drive_eligibility.py +++ b/openpilot/starpilot/drive_state/tests/test_galaxy_drive_eligibility.py @@ -127,11 +127,13 @@ def test_authenticated_remote_force_drive_modes_use_shared_owner(): result = case.drive_action('onroad', remote=True) assert result[0] == 200 and result[1]['mode'] == 'onroad' assert result[1]['effective'] == 'offroad' - assert case.drive_action('offroad', remote=True)[0] == 409 + assert case.drive_action('offroad', remote=True)[0] == 403 assert case.drive.snapshot().mode == Mode.ONROAD case.allowed = True result = case.drive_action('offroad', remote=True) - assert result[0] == 200 and result[1]['mode'] == 'offroad' + assert result[0] == 403 + case.control.change('offroad', case.drive.snapshot().revision, lambda: True) + assert case.drive.snapshot().mode == Mode.OFFROAD def test_update_snapshot_keeps_one_admission_and_matching_reason(): diff --git a/openpilot/starpilot/drive_state/tests/test_http.py b/openpilot/starpilot/drive_state/tests/test_http.py index 11b31f5286..5443552c99 100644 --- a/openpilot/starpilot/drive_state/tests/test_http.py +++ b/openpilot/starpilot/drive_state/tests/test_http.py @@ -4,7 +4,8 @@ from pathlib import Path from types import SimpleNamespace as NS import tempfile from openpilot.starpilot.galaxy.tests import test_navigation_http as helpers -from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode +from openpilot.starpilot.drive_state.owner import DriveStateOwner, Mode, Rejected +import pytest from openpilot.starpilot.drive_state.control import DriveStateControl from openpilot.starpilot.drive_state.tests.test_owner import Params, BOOT @@ -41,15 +42,17 @@ class ForceHttpTest(helpers.NavigationHttpTest): self.assertEqual((result[1]['mode'], result[1]['effective']), ('onroad', 'offroad')) status = self.request('/api/drive-state/status', remote=True, cookie=self.remote_cookie) self.assertEqual(status[1]['revision'], result[1]['revision']) - self.assertEqual(self.drive_action('offroad', remote=True)[0], 200) + self.assertEqual(self.drive_action('offroad', remote=True)[0], 403) + self.control.change('offroad', self.drive.snapshot().revision, lambda: True) self.assertEqual(self.drive_action('auto', revision=result[1]['revision'])[0], 409) def test_physical_loss_blocks_offroad_but_onroad_and_auto_remain_recoverable(self): - self.assertEqual(self.drive_action('offroad')[0], 200) + self.assertEqual(self.drive_action('offroad')[0], 403) + self.control.change('offroad', self.drive.snapshot().revision, lambda: True) self.allowed = False self.assertEqual(self.drive_action('onroad')[0], 200) self.assertEqual(self.drive.snapshot().mode, Mode.ONROAD) - self.assertEqual(self.drive_action('offroad')[0], 409) + self.assertEqual(self.drive_action('offroad')[0], 403) self.assertEqual(self.drive.snapshot().mode, Mode.ONROAD) self.assertEqual(self.drive_action('auto', remote=True)[0], 200) self.assertEqual(self.drive.snapshot().mode, Mode.AUTO) @@ -60,18 +63,28 @@ class ForceHttpTest(helpers.NavigationHttpTest): self.assertEqual(self.drive_action('onroad', origin='https://unrelated.invalid')[0], 403) self.assertEqual(self.drive_action('badmode')[0], 409) self.assertEqual(self.params.writes, before) - calls = [] + original_change = self.control.change - def revoke(): - calls.append(1) - if len(calls) == 2: - self.pairing.unpair() - return True + def revoke_remote(mode, revision, authorized): + self.assertEqual(mode, 'onroad') + self.assertTrue(authorized()) + self.pairing.unpair() # Real HTTP credential changes inside the route's effect lock. + self.assertFalse(authorized()) + return original_change(mode, revision, authorized) - self.physical.allowed = revoke - self.assertEqual(self.drive_action('offroad', remote=True)[0], 409) + with patch.object(self.control, 'change', side_effect=revoke_remote): + self.assertEqual(self.drive_action('onroad', remote=True)[0], 409) self.assertEqual(self.params.writes, before) + def test_native_park_permission_revoked_during_commit_writes_nothing(self): + before = list(self.params.writes) + permission = iter((True, False)) + self.physical.allowed = lambda: next(permission) + with pytest.raises(Rejected): + self.control.change('offroad', self.drive.snapshot().revision, lambda: True) + self.assertEqual(self.params.writes, before) + self.assertEqual(self.drive.snapshot().mode, Mode.AUTO) + def test_authenticated_force_offroad_from_live_onroad_state(self): self._force_offroad_from_live_onroad_state(True) @@ -98,12 +111,14 @@ class ForceHttpTest(helpers.NavigationHttpTest): self.assertEqual(status[1]['effective'], 'onroad') self.assertEqual(status[1]['overrideAllowed'], can_valid) result = self.drive_action('offroad', remote=True) - self.assertEqual(result[0], 200 if can_valid else 409) + self.assertEqual(result[0], 403) if can_valid: - self.assertEqual(result[1]['mode'], 'offroad') + self.control.change('offroad', self.drive.snapshot().revision, lambda: True) + self.assertEqual(self.drive.snapshot().mode, Mode.OFFROAD) else: + with pytest.raises(Rejected, match='Park'): + self.control.change('offroad', self.drive.snapshot().revision, lambda: True) self.assertEqual(self.drive.snapshot().mode, Mode.AUTO) - self.assertIn('Park', result[1]['error']) # hardwared consumes this same owner request through the actual resolver. messages.values['deviceState'].started = should_start(self.drive.snapshot().mode, {'ignition': True}, {'ready': True}, already_started=True) diff --git a/openpilot/starpilot/galaxy/batch_routes.py b/openpilot/starpilot/galaxy/batch_routes.py index 430f30d098..52a57bb45b 100644 --- a/openpilot/starpilot/galaxy/batch_routes.py +++ b/openpilot/starpilot/galaxy/batch_routes.py @@ -109,6 +109,9 @@ def change_drive_state(self, payload, effect_lock, drive_owner): len(payload['revision']) != 32): self.json(400, {'error': 'Invalid drive state request'}) return + if payload['mode'] == 'offroad': + self.json(403, {'error': 'Use the in-car Force Off-road control'}) + return from openpilot.starpilot.drive_state.owner import Rejected try: with effect_lock: diff --git a/openpilot/starpilot/galaxy/tests/test_drive_state.mjs b/openpilot/starpilot/galaxy/tests/test_drive_state.mjs index 94538b9773..2d90a7f2a2 100644 --- a/openpilot/starpilot/galaxy/tests/test_drive_state.mjs +++ b/openpilot/starpilot/galaxy/tests/test_drive_state.mjs @@ -36,10 +36,6 @@ await onroad.change('offroad');assert.equal(requests.length,before) window.confirm=()=>true await onroad.change('offroad');assert.equal(requests.length,before) onroad.state.overrideAllowed=true -const stop=onroad.change('offroad') -assert.equal(requests.length,before+1) -assert.deepEqual(JSON.parse(requests.at(-1).options.body),{mode:'offroad',revision:'a'.repeat(32)}) -requests.at(-1).resolve(response({...snapshot('offroad','d'.repeat(32)),effective:'onroad'}));await flush() -assert.equal(DriveStatePanel.computed.pending.call(onroad),true) -requests.at(-1).resolve(response({...snapshot('offroad','d'.repeat(32)),effective:'offroad'}));await stop;await flush() -assert.equal(DriveStatePanel.computed.pending.call(onroad),false) +await onroad.change('offroad');assert.equal(requests.length,before) +assert.match(DriveStatePanel.template, /\['onroad', 'auto'\]/) +assert.doesNotMatch(DriveStatePanel.template, /v-for="mode in \['offroad'/) diff --git a/openpilot/starpilot/galaxy/web/js/drive-state.js b/openpilot/starpilot/galaxy/web/js/drive-state.js index 903cbd0bed..115c707d67 100644 --- a/openpilot/starpilot/galaxy/web/js/drive-state.js +++ b/openpilot/starpilot/galaxy/web/js/drive-state.js @@ -35,9 +35,7 @@ export const DriveStatePanel = { }, async change(mode) { if (this.busy || !this.state?.available) return - if (mode === "offroad" && !this.state.overrideAllowed) { this.error = "Park the vehicle and disengage before forcing Offroad."; return } - if (mode === "offroad" && this.state.effective === "onroad" && - !window.confirm("Switch to Offroad and stop driving services? Disengage and stop safely first. Stay stopped until you return to Auto.")) return + if (!["onroad", "auto"].includes(mode)) return const revision = this.state.revision this.busy = true this.loading = false @@ -61,11 +59,10 @@ export const DriveStatePanel = { The drive-state manager is unavailable. Reconnect to check again. Waiting for the device to switch {{ state.mode }}…

Device: {{ state?.effective === "onroad" ? "Onroad" : state?.effective === "offroad" ? "Offroad" : "Checking…" }}

-

Auto follows the car. Force Onroad starts driving services at any time. Force Offroad stops them when the vehicle is parked and disengaged. Return to Auto restores normal operation.

-

{{ state.reason || "Park the vehicle and disengage to use Force Offroad." }}

-
+ :disabled="!state?.available" + :aria-pressed="state?.mode === mode" @click="change(mode)">{{ mode === 'auto' ? 'Return to Auto' : 'Onroad' }} `, } diff --git a/openpilot/starpilot/ui/runtime_app.py b/openpilot/starpilot/ui/runtime_app.py index b9e9637a72..e0811c89d8 100644 --- a/openpilot/starpilot/ui/runtime_app.py +++ b/openpilot/starpilot/ui/runtime_app.py @@ -812,7 +812,7 @@ class StarShellSession: drive_row = FeatureRow("drive_state", "Force Drive State", LABELS[drive["mode"]], source=drive["revision"].encode() if drive["revision"] is not None else None, available=drive["available"], reason=f"Device: {drive['effective'] or 'unavailable'}", - actions=tuple((label, drive["mode"] != mode and (mode != "onroad" or drive["overrideAllowed"])) + actions=tuple((label, drive["mode"] != mode and (mode != "offroad" or drive["overrideAllowed"])) for mode, label in LABELS.items())) power_available = getattr(self, "system_power_action", None) is not None and not ui_state.engaged device_power = FeatureRow("device_power", "Device Power", "", available=power_available, @@ -1239,10 +1239,12 @@ class StarShellSession: (item.destination == Destination.FORCE_DRIVE and drive["available"]) or snapshot.device.offroad and item.destination == Destination.PAIR) for item in snapshot.settings.availability) - next_mode = DriveStateControl.next_mode(drive) + from openpilot.starpilot.drive_state.control import CYCLE + next_mode = CYCLE[(CYCLE.index(next(mode for mode in CYCLE if mode.value == drive["mode"])) + 1) % len(CYCLE)] label = ("unavailable" if not drive["available"] else {"offroad": "Force Off-road", "onroad": "Force On-road", "auto": "Return to Auto"}[next_mode.value]) - availability = tuple(replace(item, request_value=next_mode.value, request_revision=drive["revision"]) + availability = tuple(replace(item, request_value=next_mode.value, request_revision=drive["revision"], + reason=drive.get("reason") if next_mode.value == "offroad" and not drive["overrideAllowed"] else None) if item.destination == Destination.FORCE_DRIVE else item for item in availability) snapshot = replace(snapshot, settings=replace(snapshot.settings, availability=availability, nav_bar_alpha=0, force_drive_label=label)) diff --git a/openpilot/starpilot/ui/settings.py b/openpilot/starpilot/ui/settings.py index 5b6c76eb1a..9ca64a09bb 100644 --- a/openpilot/starpilot/ui/settings.py +++ b/openpilot/starpilot/ui/settings.py @@ -104,6 +104,9 @@ class SettingsView: rl.draw_texture_ex(background, rl.Vector2(x, top), 0, 1, rl.WHITE) font_size = 40 if destination == Destination.FORCE_DRIVE else 42 if destination == Destination.DRIVING_MODEL else ( 48 if destination in (Destination.GALAXY, Destination.PAIR) else 64) + if destination == Destination.FORCE_DRIVE and state.destination(destination).reason: + reason = state.destination(destination).reason + self.fonts.draw(reason[:46], FontRole.MEDIUM, 18, x + 20, top + 18, rl.Color(255, 190, 140, 255)) measured = self.fonts.measure(text, FontRole.BOLD, font_size) self.fonts.draw(text, FontRole.BOLD, font_size, x + 40, top + 180 - 23 - measured.height, rl.Color(255, 255, 255, 229)) diff --git a/openpilot/starpilot/ui/tests/test_compact_settings_lifecycle.py b/openpilot/starpilot/ui/tests/test_compact_settings_lifecycle.py index 463963c101..0f46295fe3 100644 --- a/openpilot/starpilot/ui/tests/test_compact_settings_lifecycle.py +++ b/openpilot/starpilot/ui/tests/test_compact_settings_lifecycle.py @@ -101,6 +101,7 @@ def test_hidden_settings_still_updates_navigation_but_does_not_project(lifecycle @pytest.mark.parametrize("mode,allowed,label,target", [ ("auto", True, "Force Off-road", "offroad"), + ("auto", False, "Force Off-road", "offroad"), ("offroad", True, "Force On-road", "onroad"), ("onroad", True, "Return to Auto", "auto"), ("offroad", False, "Force On-road", "onroad"), @@ -121,3 +122,30 @@ def test_force_menu_names_requested_action_with_supported_glyphs(lifecycle, mode glyphs = {int(value) for value in re.findall(r"char id=(\d+)", font.read_text())} assert all(ord(char) in glyphs for char in label) assert ord("→") not in glyphs + + +@pytest.mark.parametrize('allowed,target', [(True, 'offroad'), (False, 'offroad')]) +def test_actual_compact_force_card_hitbox_dispatches_captured_owner_request(lifecycle, allowed, target): + from dataclasses import replace + from openpilot.starpilot.ui.settings_state import SettingsInput + from openpilot.starpilot.ui.shell import ShellRequest + _, session, _, _ = lifecycle + session._mode = runtime_app.ShellMode.SETTINGS + session._drive_change = Mock() + revision = 'current-owner-revision' + session.drive_state.snapshot = lambda: {'mode': 'auto', 'revision': revision, 'available': True, + 'effective': 'onroad', 'overrideAllowed': allowed, 'reason': None if allowed else 'Current Panda status is stale'} + snapshot = session.snapshot(runtime_app.ShellMode.SETTINGS) + assert snapshot.settings.destination(runtime_app.Destination.FORCE_DRIVE).reason == (None if allowed else "Current Panda status is stale") + index = next(i for i, item in enumerate(compact_menu(snapshot.settings)) if item[0] == runtime_app.Destination.FORCE_DRIVE) + state = replace(snapshot.settings, compact_scroll_x=-422 * index, compact_y=0) + inputs = SettingsInput(session.profile, lambda action: session._emit(ShellRequest('settings', action))) + inputs.press(221, 120, state) + session._drive_change.assert_not_called() + inputs.release(221, 120, state) + session._drive_change.assert_called_once_with(target, revision) + session._drive_change.reset_mock() + inputs.press(221, 120, state) + inputs.move(230, 120, state) + inputs.release(230, 120, state) + session._drive_change.assert_not_called() diff --git a/openpilot/starpilot/ui/tests/test_display_settings.py b/openpilot/starpilot/ui/tests/test_display_settings.py index eb45d7a2c9..3639b6b1fd 100644 --- a/openpilot/starpilot/ui/tests/test_display_settings.py +++ b/openpilot/starpilot/ui/tests/test_display_settings.py @@ -133,10 +133,14 @@ class DisplaySettingsTests(unittest.TestCase): session._drive_change.assert_not_called() status["overrideAllowed"] = False fresh = session.system_snapshot() - self.assertIsNone(inputs.target(*point, fresh)) - session._display_ui(FeatureUiAction("action", row, 2)) + self.assertIsNotNone(inputs.target(*point, fresh)) + session._drive_change.reset_mock() + session._display_ui(FeatureUiAction("action", fresh.rows[0], 2)) + session._drive_change.assert_called_once_with("onroad", "shown-revision") + self.assertFalse(fresh.rows[0].actions[1][1]) # Only Offroad requires parked admission. + session._drive_change.reset_mock() + session._display_ui(FeatureUiAction("action", fresh.rows[0], 1)) session._drive_change.assert_not_called() - self.assertTrue(fresh.rows[0].actions[1][1]) # Offroad remains available when Onroad admission fails. def test_display_submenu_opens_while_off_enabling_enters_and_back_keeps_the_hierarchy(self): with patch("openpilot.starpilot.ui.runtime_app.native_device.invalidate_display_preferences"): diff --git a/prebuilt.json b/prebuilt.json index 4b7de5a118..b6c2bf1f28 100644 --- a/prebuilt.json +++ b/prebuilt.json @@ -1 +1 @@ -{"protected_inventory_sha256": "08f89bc4d521155ab31354a99634e73173838ecc5350ed9bb1fde7aa04bbeab7", "schema_version": 1} +{"protected_inventory_sha256": "954e1e15518212b2ac6b07ebf3bd576c13b57818156029ce7618438ed5dbb7ec", "schema_version": 1}