From 427d3ac6ce0079cba7dafd7b5ec4697ba9e1a151 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 23 Jul 2026 08:45:21 +1000 Subject: [PATCH] feat(vehicle): add remaining signed commands (UI/unit, nav/messaging/media, charging/utility, connectivity/diagnostics) Consolidates four previously-separate command-group PRs into one to avoid the same-file merge conflicts serial merging kept re-creating. Wraps 20 previously-unimplemented VehicleAction fields: - UI/unit preferences: set_temperature_unit, set_distance_unit, set_time_display_format, set_tire_pressure_unit, set_energy_display_format, set_phone_setting_preferences. - Navigation/messaging/media: upcoming_calendar_entries and take_drivenote (signed-command siblings of the existing REST-only VehicleFleet methods of the same name), video_request, navigation_route, get_messages. - Charging/utility: set_rate_tariff, get_rate_tariff, add_managed_charging_site, remove_managed_charging_site, get_managed_charging_sites, set_discharge_limit (a distinct feature from the existing set_powershare_discharge_limit). - Connectivity/diagnostics: bluetooth_classic_pairing_request, bandwidth_test, fetch_keys_info. set_rate_tariff takes tesla_protocol message types directly for its deeply-nested tariff schedule rather than a parallel flattened API. Both Fleet-signed and BLE transports get every command for free via the shared Commands ABC. Purely additive - no existing signatures touched. --- tesla_fleet_api/const.py | 42 +++ tesla_fleet_api/tesla/vehicle/commands.py | 293 ++++++++++++++++++ tests/test_ble_charging_utility_commands.py | 114 +++++++ ...t_ble_connectivity_diagnostics_commands.py | 54 ++++ .../test_ble_nav_messaging_media_commands.py | 87 ++++++ tests/test_ble_ui_unit_preference_commands.py | 125 ++++++++ tests/test_cross_transport_parity.py | 40 +++ 7 files changed, 755 insertions(+) create mode 100644 tests/test_ble_charging_utility_commands.py create mode 100644 tests/test_ble_connectivity_diagnostics_commands.py create mode 100644 tests/test_ble_nav_messaging_media_commands.py create mode 100644 tests/test_ble_ui_unit_preference_commands.py diff --git a/tesla_fleet_api/const.py b/tesla_fleet_api/const.py index ec0ea8d..cf121bf 100644 --- a/tesla_fleet_api/const.py +++ b/tesla_fleet_api/const.py @@ -268,6 +268,48 @@ class SeatHeaterLevel(StrEnum): HIGH = "high" +class TemperatureUnit(IntEnum): + """Displayed temperature unit options""" + + FAHRENHEIT = 0 + CELSIUS = 1 + + +class DistanceUnit(IntEnum): + """Displayed distance unit options""" + + MILES = 0 + KILOMETERS = 1 + + +class TimeDisplayFormat(IntEnum): + """Displayed clock format options""" + + HOUR_12 = 0 + HOUR_24 = 1 + + +class TirePressureUnit(IntEnum): + """Displayed tire pressure unit options""" + + PSI = 0 + BAR = 1 + + +class EnergyDisplayFormat(IntEnum): + """Displayed energy/range unit options""" + + PERCENTAGE = 0 + DISTANCE = 1 + + +class PhoneFontSize(IntEnum): + """Paired-phone display font size options""" + + STANDARD = 0 + LARGE = 1 + + class BluetoothVehicleData(StrEnum): CHARGE_STATE = "GetChargeState" CLIMATE_STATE = "GetClimateState" diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index d47e52d..43252cf 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -43,6 +43,12 @@ CabinOverheatProtectionTemp, SunRoofCommand, WindowCommand, + TemperatureUnit, + DistanceUnit, + TimeDisplayFormat, + TirePressureUnit, + EnergyDisplayFormat, + PhoneFontSize, ) # Protocol @@ -182,6 +188,30 @@ GetLocalProfilesForVaultUuidAction, DeleteDashcamClipsAction, FormatUSBAction, + SetTemperatureUnitAction, + SetDistanceUnitAction, + SetTimeDisplayFormatAction, + SetTirePressureUnitAction, + SetEnergyDisplayFormatAction, + SetPhoneSettingPreferencesAction, + PhoneUnitPreferences, + UiSetUpcomingCalendarEntries, + TakeDrivenoteAction, + VideoRequestAction, + NavigationRouteAction, + GetMessagesAction, + SetRateTariffRequest, + GetRateTariffRequest, + AddManagedChargingSiteRequest, + RemoveManagedChargingSiteRequest, + GetManagedChargingSitesRequest, + SetDischargeLimitAction, + ManagedChargingSite, + ManagerType, + SiteController, + BluetoothClassicPairingRequest, + BandwidthTest, + FetchKeysInfoAction, ) from google.protobuf.timestamp_pb2 import Timestamp from tesla_protocol.command.vehicle_pb2 import ( @@ -2541,3 +2571,266 @@ async def delete_dashcam_clips(self) -> dict[str, Any]: ) ) ) + + async def set_temperature_unit(self, unit: TemperatureUnit | int) -> dict[str, Any]: + """Sets the vehicle's displayed temperature unit.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setTemperatureUnitAction=SetTemperatureUnitAction( + unit=unit # pyright: ignore[reportArgumentType] + ) + ) + ) + ) + + async def set_distance_unit(self, unit: DistanceUnit | int) -> dict[str, Any]: + """Sets the vehicle's displayed distance unit.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setDistanceUnitAction=SetDistanceUnitAction( + unit=unit # pyright: ignore[reportArgumentType] + ) + ) + ) + ) + + async def set_time_display_format( + self, format: TimeDisplayFormat | int + ) -> dict[str, Any]: + """Sets the vehicle's displayed clock format (12h/24h).""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setTimeDisplayFormatAction=SetTimeDisplayFormatAction( + format=format # pyright: ignore[reportArgumentType] + ) + ) + ) + ) + + async def set_tire_pressure_unit( + self, unit: TirePressureUnit | int + ) -> dict[str, Any]: + """Sets the vehicle's displayed tire pressure unit.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setTirePressureUnitAction=SetTirePressureUnitAction( + unit=unit # pyright: ignore[reportArgumentType] + ) + ) + ) + ) + + async def set_energy_display_format( + self, format: EnergyDisplayFormat | int + ) -> dict[str, Any]: + """Sets the vehicle's displayed energy/range unit (percentage/distance).""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setEnergyDisplayFormatAction=SetEnergyDisplayFormatAction( + format=format # pyright: ignore[reportArgumentType] + ) + ) + ) + ) + + async def set_phone_setting_preferences( + self, + font_size: PhoneFontSize | int = PhoneFontSize.STANDARD, + *, + distance_unit: DistanceUnit | int | None = None, + temperature_unit: TemperatureUnit | int | None = None, + ) -> dict[str, Any]: + """Sets paired-phone display preferences: font size and, optionally, distance/temperature units.""" + action = SetPhoneSettingPreferencesAction( + font_size=font_size # pyright: ignore[reportArgumentType] + ) + if distance_unit is not None or temperature_unit is not None: + unit_kwargs: dict[str, Any] = {} + if distance_unit is not None: + unit_kwargs["distance_unit"] = distance_unit + if temperature_unit is not None: + unit_kwargs["temperature_unit"] = temperature_unit + action.unit_preferences.CopyFrom(PhoneUnitPreferences(**unit_kwargs)) + return await self._sendInfotainment( + Action(vehicleAction=VehicleAction(setPhoneSettingPreferencesAction=action)) + ) + + 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, + ) + + async def set_rate_tariff( + self, + seasons: SetRateTariffRequest.Seasons, + tariff: SetRateTariffRequest.Tariff | None = None, + ) -> dict[str, Any]: + """Sets a time-of-use rate tariff schedule for charge-on-solar-style optimization. + + ``seasons``/``tariff`` are ``tesla_protocol`` message types + (``SetRateTariffRequest.Seasons``/``.Tariff``) - the tariff schedule is + deeply nested (up to 5 named seasons, each with 4 time-of-use period + types), so construct them directly rather than through a parallel + flattened API. + """ + action = SetRateTariffRequest(seasons=seasons) + if tariff is not None: + action.tariff.CopyFrom(tariff) + return await self._sendInfotainment( + Action(vehicleAction=VehicleAction(setRateTariffRequest=action)) + ) + + async def get_rate_tariff(self) -> dict[str, Any]: + """Gets the current time-of-use rate tariff schedule.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction(getRateTariffRequest=GetRateTariffRequest()) + ), + mutating=False, + ) + + async def add_managed_charging_site( + self, public_key: str, lat: float, lon: float + ) -> dict[str, Any]: + """Registers a managed charging site (utility managed-charging program) for this vehicle.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + addManagedChargingSiteRequest=AddManagedChargingSiteRequest( + site=ManagedChargingSite( + public_key=public_key, + manager_type=ManagerType(site_controller=SiteController()), + lat_lon=LatLong(latitude=lat, longitude=lon), + ) + ) + ) + ) + ) + + async def remove_managed_charging_site(self, public_key: str) -> dict[str, Any]: + """Removes a previously-registered managed charging site by its public key.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + removeManagedChargingSiteRequest=RemoveManagedChargingSiteRequest( + public_key=public_key + ) + ) + ) + ) + + async def get_managed_charging_sites(self) -> dict[str, Any]: + """Gets the list of registered managed charging sites.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + getManagedChargingSitesRequest=GetManagedChargingSitesRequest() + ) + ), + mutating=False, + ) + + async def set_discharge_limit(self, discharge_limit: int) -> dict[str, Any]: + """Sets the vehicle's general discharge limit. + + Not the same feature as ``set_powershare_discharge_limit`` + (``SetPowershareDischargeLimitAction``, a distinct proto message). + """ + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + setDischargeLimitAction=SetDischargeLimitAction( + discharge_limit=discharge_limit + ) + ) + ) + ) + + async def bluetooth_classic_pairing_request( + self, name: str, mac_address: bytes + ) -> dict[str, Any]: + """Requests Bluetooth Classic (not BLE) pairing with a phone for calls/audio.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + bluetoothClassicPairingRequest=BluetoothClassicPairingRequest( + utf8_name=name, + mac_address=mac_address, + ) + ) + ) + ) + + async def bandwidth_test(self, requested_size: int) -> dict[str, Any]: + """Runs a diagnostic bandwidth test of the given size in bytes.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + bandwidthTest=BandwidthTest(requested_size=requested_size) + ) + ) + ) + + async def fetch_keys_info(self) -> dict[str, Any]: + """Gets information about keys paired with the vehicle.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction(fetchKeysInfoAction=FetchKeysInfoAction()) + ), + mutating=False, + ) diff --git a/tests/test_ble_charging_utility_commands.py b/tests/test_ble_charging_utility_commands.py new file mode 100644 index 0000000..19ecdd4 --- /dev/null +++ b/tests/test_ble_charging_utility_commands.py @@ -0,0 +1,114 @@ +"""Charging/utility commands over the mocked BLE transport. + +``set_rate_tariff``/``add_managed_charging_site`` accept ``tesla_protocol`` +message types directly for their deeply-nested arguments rather than a +parallel flattened API - see ``commands.py`` docstrings. +""" + +from tesla_protocol.command.car_server_pb2 import Action, SetRateTariffRequest +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 SetRateTariffTests(MockedBleTransportTestCase): + async def test_sends_seasons_only(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + seasons = SetRateTariffRequest.Seasons( + Summer=SetRateTariffRequest.Season(from_month=6, to_month=8) + ) + await vehicle.set_rate_tariff(seasons) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setRateTariffRequest + self.assertEqual(action.seasons.Summer.from_month, 6) + self.assertEqual(action.seasons.Summer.to_month, 8) + self.assertFalse(action.HasField("tariff")) + + async def test_sends_tariff_when_given(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + seasons = SetRateTariffRequest.Seasons() + tariff = SetRateTariffRequest.Tariff(seasons=seasons) + await vehicle.set_rate_tariff(seasons, tariff) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setRateTariffRequest + self.assertTrue(action.HasField("tariff")) + + +class GetRateTariffTests(MockedBleTransportTestCase): + async def test_sends_get_rate_tariff_request(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.get_rate_tariff() + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("getRateTariffRequest")) + + +class AddManagedChargingSiteTests(MockedBleTransportTestCase): + async def test_sends_public_key_and_coordinates(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.add_managed_charging_site("pubkey-bytes", 37.3230, -122.0322) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + site = vehicle_action.addManagedChargingSiteRequest.site + self.assertEqual(site.public_key, "pubkey-bytes") + self.assertTrue(site.manager_type.HasField("site_controller")) + # LatLong lat/lon are 32-bit floats, so compare at reduced precision. + self.assertAlmostEqual(site.lat_lon.latitude, 37.3230, places=4) + self.assertAlmostEqual(site.lat_lon.longitude, -122.0322, places=4) + + +class RemoveManagedChargingSiteTests(MockedBleTransportTestCase): + async def test_sends_public_key(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.remove_managed_charging_site("pubkey-bytes") + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.removeManagedChargingSiteRequest.public_key, + "pubkey-bytes", + ) + + +class GetManagedChargingSitesTests(MockedBleTransportTestCase): + async def test_sends_get_managed_charging_sites_request(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.get_managed_charging_sites() + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("getManagedChargingSitesRequest")) + + +class SetDischargeLimitTests(MockedBleTransportTestCase): + async def test_sends_discharge_limit(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_discharge_limit(50) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual(vehicle_action.setDischargeLimitAction.discharge_limit, 50) diff --git a/tests/test_ble_connectivity_diagnostics_commands.py b/tests/test_ble_connectivity_diagnostics_commands.py new file mode 100644 index 0000000..025217d --- /dev/null +++ b/tests/test_ble_connectivity_diagnostics_commands.py @@ -0,0 +1,54 @@ +"""Connectivity/diagnostics commands over the mocked BLE transport.""" + +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 BluetoothClassicPairingRequestTests(MockedBleTransportTestCase): + async def test_sends_name_and_mac_address(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.bluetooth_classic_pairing_request( + "My Phone", b"\x00\x11\x22\x33\x44\x55" + ) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + req = vehicle_action.bluetoothClassicPairingRequest + self.assertEqual(req.utf8_name, "My Phone") + self.assertEqual(req.mac_address, b"\x00\x11\x22\x33\x44\x55") + + +class BandwidthTestTests(MockedBleTransportTestCase): + async def test_sends_requested_size(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.bandwidth_test(1024) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual(vehicle_action.bandwidthTest.requested_size, 1024) + + +class FetchKeysInfoTests(MockedBleTransportTestCase): + async def test_sends_fetch_keys_info_action(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.fetch_keys_info() + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.HasField("fetchKeysInfoAction")) 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_ble_ui_unit_preference_commands.py b/tests/test_ble_ui_unit_preference_commands.py new file mode 100644 index 0000000..8d4b988 --- /dev/null +++ b/tests/test_ble_ui_unit_preference_commands.py @@ -0,0 +1,125 @@ +"""UI/unit-preference commands over the mocked BLE transport. + +Trivial single-scalar/nested-message wrappers over +``Set*UnitAction``/``SetTimeDisplayFormatAction``/``SetEnergyDisplayFormatAction``/ +``SetPhoneSettingPreferencesAction``. +""" + +from tesla_protocol.command.car_server_pb2 import Action +from tesla_protocol.command.universal_message_pb2 import Domain + +from tesla_fleet_api.const import ( + DistanceUnit, + EnergyDisplayFormat, + PhoneFontSize, + TemperatureUnit, + TimeDisplayFormat, + TirePressureUnit, +) +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 SetTemperatureUnitTests(MockedBleTransportTestCase): + async def test_sends_celsius(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.set_temperature_unit(TemperatureUnit.CELSIUS) + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.setTemperatureUnitAction.unit, TemperatureUnit.CELSIUS + ) + + +class SetDistanceUnitTests(MockedBleTransportTestCase): + async def test_sends_kilometers(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_distance_unit(DistanceUnit.KILOMETERS) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.setDistanceUnitAction.unit, DistanceUnit.KILOMETERS + ) + + +class SetTimeDisplayFormatTests(MockedBleTransportTestCase): + async def test_sends_24_hour(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_time_display_format(TimeDisplayFormat.HOUR_24) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.setTimeDisplayFormatAction.format, + TimeDisplayFormat.HOUR_24, + ) + + +class SetTirePressureUnitTests(MockedBleTransportTestCase): + async def test_sends_bar(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_tire_pressure_unit(TirePressureUnit.BAR) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.setTirePressureUnitAction.unit, TirePressureUnit.BAR + ) + + +class SetEnergyDisplayFormatTests(MockedBleTransportTestCase): + async def test_sends_distance_format(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_energy_display_format(EnergyDisplayFormat.DISTANCE) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertEqual( + vehicle_action.setEnergyDisplayFormatAction.format, + EnergyDisplayFormat.DISTANCE, + ) + + +class SetPhoneSettingPreferencesTests(MockedBleTransportTestCase): + async def test_font_size_only_leaves_unit_preferences_unset(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_phone_setting_preferences(PhoneFontSize.LARGE) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + action = vehicle_action.setPhoneSettingPreferencesAction + self.assertEqual(action.font_size, PhoneFontSize.LARGE) + self.assertFalse(action.HasField("unit_preferences")) + + async def test_units_are_nested_under_unit_preferences(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + await vehicle.set_phone_setting_preferences( + distance_unit=DistanceUnit.KILOMETERS, + temperature_unit=TemperatureUnit.CELSIUS, + ) + + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + prefs = vehicle_action.setPhoneSettingPreferencesAction.unit_preferences + self.assertEqual(prefs.distance_unit, DistanceUnit.KILOMETERS) + self.assertEqual(prefs.temperature_unit, TemperatureUnit.CELSIUS) 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")