diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 390a0cf..d47e52d 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -180,6 +180,8 @@ TeslaAuthResponseAction, SetupCloudProfileWithLocalProfileUuidAction, GetLocalProfilesForVaultUuidAction, + DeleteDashcamClipsAction, + FormatUSBAction, ) from google.protobuf.timestamp_pb2 import Timestamp from tesla_protocol.command.vehicle_pb2 import ( @@ -2515,3 +2517,27 @@ async def get_local_profiles_for_vault_uuid( ), mutating=False, ) + + # No confirmation guard, matching the existing erase_user_data() + # precedent: this library exposes the signed command as-is and leaves + # any confirmation UX to the caller. + + async def format_usb(self) -> dict[str, Any]: + """Formats the USB drive connected to the vehicle, erasing all files on it. Irreversible.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + formatUsbAction=FormatUSBAction(format_usb=True) + ) + ) + ) + + async def delete_dashcam_clips(self) -> dict[str, Any]: + """Deletes all dashcam clips stored on the vehicle. Irreversible.""" + return await self._sendInfotainment( + Action( + vehicleAction=VehicleAction( + deleteDashcamClipsAction=DeleteDashcamClipsAction(delete_clips=True) + ) + ) + ) diff --git a/tests/test_ble_destructive_commands.py b/tests/test_ble_destructive_commands.py new file mode 100644 index 0000000..0bddda2 --- /dev/null +++ b/tests/test_ble_destructive_commands.py @@ -0,0 +1,47 @@ +"""Destructive data-clearing commands over the mocked BLE transport. + +``format_usb``/``delete_dashcam_clips`` are unguarded no-confirmation +wrappers, matching the existing ``erase_user_data()`` precedent +(``tests/test_ble_mocked_commands.py`` has no dedicated test for that one +either - it is exercised indirectly via the cross-transport suite). +""" + +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 FormatUsbTests(MockedBleTransportTestCase): + async def test_sends_format_usb_true(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.format_usb() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.formatUsbAction.format_usb) + + +class DeleteDashcamClipsTests(MockedBleTransportTestCase): + async def test_sends_delete_clips_true(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = infotainment_action_ok_reply() + + result = await vehicle.delete_dashcam_clips() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0]) + self.assertTrue(vehicle_action.deleteDashcamClipsAction.delete_clips)