feat(vehicle): add niche low-certainty-value signed commands

tesla_fleet_api/tesla/vehicle/commands.py

@@ -174,7 +174,14 @@
     # Group 12: Power modes
     SetLowPowerModeAction,
     SetKeepAccessoryPowerModeAction,
+    # Group 18: Niche/low-certainty-value commands (captain full-coverage directive)
+    PiiKeyRequest,
+    PseudonymSyncRequest,
+    TeslaAuthResponseAction,
+    SetupCloudProfileWithLocalProfileUuidAction,
+    GetLocalProfilesForVaultUuidAction,
 )
+from google.protobuf.timestamp_pb2 import Timestamp
 from tesla_protocol.command.vehicle_pb2 import (
     VehicleData,
     VehicleState,
@@ -2404,3 +2411,107 @@ async def auto_secure_vehicle(self) -> dict[str, Any]:
         return await self._sendVehicleSecurity(
             UnsignedMessage(RKEAction=RKEAction_E.RKE_ACTION_AUTO_SECURE_VEHICLE)
         )
+
+    # Group 18: Niche/low-certainty-value commands
+    #
+    # Included per the captain's full-proto-coverage directive; each has no
+    # known third-party consumer use case today, unlike every other command
+    # in this file. See AGENTS.md for detail on why each is niche.
+
+    async def pii_key_request(
+        self, subscriber_public_key: str, pii_key_expiration: int
+    ) -> dict[str, Any]:
+        """Requests a PII (privacy/telemetry pseudonymization) key.
+
+        ``pii_key_expiration`` is a Unix timestamp in seconds.
+        """
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    piiKeyRequest=PiiKeyRequest(
+                        subscriber_public_key=subscriber_public_key,
+                        pii_key_expiration=Timestamp(seconds=pii_key_expiration),
+                    )
+                )
+            )
+        )
+
+    async def pseudonym_sync_request(
+        self, last_known_pseudonym_hashed: bytes
+    ) -> dict[str, Any]:
+        """Syncs the vehicle's privacy/telemetry pseudonym with a previously known hashed value."""
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    pseudonymSyncRequest=PseudonymSyncRequest(
+                        last_known_pseudonym_hashed=last_known_pseudonym_hashed
+                    )
+                )
+            )
+        )
+
+    async def tesla_auth_response(
+        self,
+        client_id: str,
+        scope: str,
+        access_token: str,
+        refresh_token: str,
+        expiry_timestamp: int,
+        error: str = "",
+        scoped_token: str = "",
+    ) -> dict[str, Any]:
+        """Passes a completed "Sign in with Tesla" OAuth response to the vehicle screen.
+
+        This is a pass-through: the caller must already have a completed OAuth
+        response from elsewhere in their own auth flow; the library cannot
+        originate these fields itself.
+        """
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    teslaAuthResponseAction=TeslaAuthResponseAction(
+                        client_id=client_id,
+                        scope=scope,
+                        access_token=access_token,
+                        refresh_token=refresh_token,
+                        expiry_timestamp=expiry_timestamp,
+                        error=error,
+                        scoped_token=scoped_token,
+                    )
+                )
+            )
+        )
+
+    async def setup_cloud_profile_with_local_profile_uuid(
+        self,
+        cloud_vault_uuid: str,
+        local_profile_uuid: str,
+        delete_local_profile_after_setup: bool = False,
+    ) -> dict[str, Any]:
+        """Links a local Tesla profile to a cloud vault profile UUID."""
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    setupCloudProfileWithLocalProfileUuidAction=SetupCloudProfileWithLocalProfileUuidAction(
+                        cloud_vault_uuid=cloud_vault_uuid,
+                        local_profile_uuid=local_profile_uuid,
+                        delete_local_profile_after_setup=delete_local_profile_after_setup,
+                    )
+                )
+            )
+        )
+
+    async def get_local_profiles_for_vault_uuid(
+        self, vault_uuid: str
+    ) -> dict[str, Any]:
+        """Gets local Tesla profiles associated with a cloud vault UUID."""
+        return await self._sendInfotainment(
+            Action(
+                vehicleAction=VehicleAction(
+                    getLocalProfilesForVaultUuidAction=GetLocalProfilesForVaultUuidAction(
+                        vault_uuid=vault_uuid
+                    )
+                )
+            ),
+            mutating=False,
+        )

tests/test_ble_niche_commands.py

@@ -0,0 +1,106 @@
+"""Niche/low-certainty-value command group over the mocked BLE transport.
+
+pii_key_request, pseudonym_sync_request, tesla_auth_response,
+setup_cloud_profile_with_local_profile_uuid, and
+get_local_profiles_for_vault_uuid - included per the captain's
+full-proto-coverage directive despite no known third-party consumer use
+case (see commands.py's Group 18 docstring).
+"""
+
+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 PiiKeyRequestTests(MockedBleTransportTestCase):
+    async def test_sends_public_key_and_expiration(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.pii_key_request("subscriber-pubkey", 1893456000)
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        req = vehicle_action.piiKeyRequest
+        self.assertEqual(req.subscriber_public_key, "subscriber-pubkey")
+        self.assertEqual(req.pii_key_expiration.seconds, 1893456000)
+
+
+class PseudonymSyncRequestTests(MockedBleTransportTestCase):
+    async def test_sends_last_known_pseudonym_hashed(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.pseudonym_sync_request(b"\xaa\xbb\xcc")
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertEqual(
+            vehicle_action.pseudonymSyncRequest.last_known_pseudonym_hashed,
+            b"\xaa\xbb\xcc",
+        )
+
+
+class TeslaAuthResponseTests(MockedBleTransportTestCase):
+    async def test_sends_full_oauth_response(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.tesla_auth_response(
+            client_id="my-client",
+            scope="openid",
+            access_token="access-tok",
+            refresh_token="refresh-tok",
+            expiry_timestamp=1893456000,
+            scoped_token="scoped-tok",
+        )
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        action = vehicle_action.teslaAuthResponseAction
+        self.assertEqual(action.client_id, "my-client")
+        self.assertEqual(action.scope, "openid")
+        self.assertEqual(action.access_token, "access-tok")
+        self.assertEqual(action.refresh_token, "refresh-tok")
+        self.assertEqual(action.expiry_timestamp, 1893456000)
+        self.assertEqual(action.error, "")
+        self.assertEqual(action.scoped_token, "scoped-tok")
+
+
+class SetupCloudProfileWithLocalProfileUuidTests(MockedBleTransportTestCase):
+    async def test_sends_uuids_and_delete_flag(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.setup_cloud_profile_with_local_profile_uuid(
+            "cloud-uuid", "local-uuid", delete_local_profile_after_setup=True
+        )
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        action = vehicle_action.setupCloudProfileWithLocalProfileUuidAction
+        self.assertEqual(action.cloud_vault_uuid, "cloud-uuid")
+        self.assertEqual(action.local_profile_uuid, "local-uuid")
+        self.assertTrue(action.delete_local_profile_after_setup)
+
+
+class GetLocalProfilesForVaultUuidTests(MockedBleTransportTestCase):
+    async def test_sends_vault_uuid(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        await vehicle.get_local_profiles_for_vault_uuid("vault-uuid")
+
+        vehicle_action = _decode_vehicle_action(vehicle, send.await_args.args[0])
+        self.assertEqual(
+            vehicle_action.getLocalProfilesForVaultUuidAction.vault_uuid,
+            "vault-uuid",
+        )