feat(vehicle): add navigation/messaging/media signed commands

tesla_fleet_api/tesla/vehicle/commands.py

@@ -180,6 +180,11 @@
     TeslaAuthResponseAction,
     SetupCloudProfileWithLocalProfileUuidAction,
     GetLocalProfilesForVaultUuidAction,
+    UiSetUpcomingCalendarEntries,
+    TakeDrivenoteAction,
+    VideoRequestAction,
+    NavigationRouteAction,
+    GetMessagesAction,
 )
 from google.protobuf.timestamp_pb2 import Timestamp
 from tesla_protocol.command.vehicle_pb2 import (
@@ -2515,3 +2520,58 @@ async def get_local_profiles_for_vault_uuid(
             ),
             mutating=False,
         )
+
+    async def upcoming_calendar_entries(self, calendar_data: str) -> dict[str, Any]:
+        """Sends upcoming calendar entries to the vehicle.
+
+        Signed-command sibling of the REST-only ``VehicleFleet.upcoming_calendar_entries``.
+        """
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    uiSetUpcomingCalendarEntries=UiSetUpcomingCalendarEntries(
+                        calendar_data=calendar_data
+                    )
+                )
+            )
+        )
+
+    async def take_drivenote(self, note: str) -> dict[str, Any]:
+        """Records a drive note.
+
+        Signed-command sibling of the REST-only ``VehicleFleet.take_drivenote``.
+        """
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    takeDrivenoteAction=TakeDrivenoteAction(note=note)
+                )
+            )
+        )
+
+    async def video_request(self, url: str) -> dict[str, Any]:
+        """Requests the vehicle open a video stream from the given URL (e.g. sentry/dashcam viewer)."""
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    videoRequestAction=VideoRequestAction(url=url)
+                )
+            )
+        )
+
+    async def navigation_route(self) -> dict[str, Any]:
+        """Triggers vehicle route navigation."""
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    navigationRouteAction=NavigationRouteAction()
+                )
+            )
+        )
+
+    async def get_messages(self) -> dict[str, Any]:
+        """Gets vehicle in-car messages."""
+        return await self._sendInfotainment(
+            Action(vehicleAction=VehicleAction(getMessagesAction=GetMessagesAction())),
+            mutating=False,
+        )

tests/test_ble_nav_messaging_media_commands.py

@@ -0,0 +1,87 @@
+"""Navigation/messaging/media commands over the mocked BLE transport.
+
+``upcoming_calendar_entries``/``take_drivenote`` are signed-command siblings of
+the existing REST-only ``VehicleFleet`` methods of the same name - cross-
+transport parity for those two is covered in ``test_cross_transport_parity.py``.
+"""
+
+from tesla_protocol.command.car_server_pb2 import Action
+from tesla_protocol.command.universal_message_pb2 import Domain
+
+from ble_mocked_transport import (
+    MockedBleTransportTestCase,
+    decrypt_sent_command,
+    infotainment_action_ok_reply,
+)
+
+
+def _decode_vehicle_action(vehicle, sent_msg):
+    plaintext = decrypt_sent_command(vehicle, sent_msg)
+    action = Action.FromString(plaintext)
+    assert sent_msg.to_destination.domain == Domain.DOMAIN_INFOTAINMENT
+    return action.vehicleAction
+
+
+class UpcomingCalendarEntriesTests(MockedBleTransportTestCase):
+    async def test_sends_calendar_data(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        result = await vehicle.upcoming_calendar_entries("some-ics-payload")
+
+        self.assertEqual(result, {"response": {"result": True, "reason": ""}})
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertEqual(
+            vehicle_action.uiSetUpcomingCalendarEntries.calendar_data,
+            "some-ics-payload",
+        )
+
+
+class TakeDrivenoteTests(MockedBleTransportTestCase):
+    async def test_sends_note(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.take_drivenote("check the noise near the front left wheel")
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertEqual(
+            vehicle_action.takeDrivenoteAction.note,
+            "check the noise near the front left wheel",
+        )
+
+
+class VideoRequestTests(MockedBleTransportTestCase):
+    async def test_sends_url(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.video_request("https://example.com/stream.m3u8")
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertEqual(
+            vehicle_action.videoRequestAction.url, "https://example.com/stream.m3u8"
+        )
+
+
+class NavigationRouteTests(MockedBleTransportTestCase):
+    async def test_sends_navigation_route_action(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        result = await vehicle.navigation_route()
+
+        self.assertEqual(result, {"response": {"result": True, "reason": ""}})
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertTrue(vehicle_action.HasField("navigationRouteAction"))
+
+
+class GetMessagesTests(MockedBleTransportTestCase):
+    async def test_sends_get_messages_action(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.get_messages()
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertTrue(vehicle_action.HasField("getMessagesAction"))

tests/test_cross_transport_parity.py

@@ -195,3 +195,43 @@ async def test_explicit_order_passes_through_unchanged_on_both_transports(
         await ble.navigation_gps_request(37.3230, -122.0322, order=2)
         action = _sent_vehicle_action(ble, send)
         self.assertEqual(action.navigationGpsRequest.order, 2)
+
+
+class UpcomingCalendarEntriesParityTests(MockedBleTransportTestCase):
+    """``upcoming_calendar_entries`` must carry the same ``calendar_data`` string
+    on both the REST-only cloud path and its new signed-command BLE sibling."""
+
+    async def test_calendar_data_survives_on_both_transports(self) -> None:
+        cloud, request = _make_fleet_vehicle(self.VIN)
+        await cloud.upcoming_calendar_entries("some-ics-payload")
+        assert request.await_args is not None
+        self.assertEqual(
+            request.await_args.kwargs["json"], {"calendar_data": "some-ics-payload"}
+        )
+
+        ble, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+        await ble.upcoming_calendar_entries("some-ics-payload")
+        action = _sent_vehicle_action(ble, send)
+        self.assertEqual(
+            action.uiSetUpcomingCalendarEntries.calendar_data, "some-ics-payload"
+        )
+
+
+class TakeDrivenoteParityTests(MockedBleTransportTestCase):
+    """``take_drivenote`` must carry the same ``note`` string on both the
+    REST-only cloud path and its new signed-command BLE sibling."""
+
+    async def test_note_survives_on_both_transports(self) -> None:
+        cloud, request = _make_fleet_vehicle(self.VIN)
+        await cloud.take_drivenote("brake noise up front")
+        assert request.await_args is not None
+        self.assertEqual(
+            request.await_args.kwargs["json"], {"note": "brake noise up front"}
+        )
+
+        ble, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+        await ble.take_drivenote("brake noise up front")
+        action = _sent_vehicle_action(ble, send)
+        self.assertEqual(action.takeDrivenoteAction.note, "brake noise up front")