diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 390a0cf..054a58d 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/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, + ) diff --git a/tests/test_ble_nav_messaging_media_commands.py b/tests/test_ble_nav_messaging_media_commands.py new file mode 100644 index 0000000..7c0242d --- /dev/null +++ b/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")) diff --git a/tests/test_cross_transport_parity.py b/tests/test_cross_transport_parity.py index 11a0405..349df69 100644 --- a/tests/test_cross_transport_parity.py +++ b/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")