test(ble): add mocked transport coverage for BLE commands

AGENTS.md

@@ -22,6 +22,14 @@ uv run pytest tests
 Tests live in `tests/` and use `unittest.IsolatedAsyncioTestCase` (collected and
 run natively by pytest — `pytest-asyncio` is not required).
 
+BLE command tests over a mocked transport build on `tests/ble_mocked_transport.py`
+(`MockedBleTransportTestCase`): it patches `VehicleBluetooth._send` and
+pre-marks both signed-command sessions ready, so a test drives any inherited
+`Commands` method with no real BLE/GATT connection and asserts on the signed
+`RoutableMessage` built (`decrypt_sent_command`) and on canned replies
+(`vcsec_ok_reply`/`infotainment_action_ok_reply`/`infotainment_vehicle_data_reply`).
+See `tests/test_ble_mocked_commands.py` for worked examples.
+
 ## API References
 
 - Tesla Fleet: https://developer.tesla.com/docs/fleet-api/endpoints/vehicle-endpoints

tests/ble_mocked_transport.py

@@ -0,0 +1,159 @@
+"""Reusable base for BLE command tests: a VehicleBluetooth with a mocked ``_send``.
+
+Feeds canned, already-decrypted ``RoutableMessage`` replies straight past the
+real GATT/encryption layer so command-to-proto construction and reply-decoding
+are unit-testable with no car and no BLE/GATT connection whatsoever. Both
+signed-command sessions are pre-marked ready so ``_command`` never attempts a
+real handshake.
+"""
+
+from __future__ import annotations
+
+import struct
+from typing import Any, cast
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import AsyncMock, MagicMock
+
+from cryptography.hazmat.primitives.asymmetric import ec
+from cryptography.hazmat.primitives.ciphers.aead import AESGCM
+from cryptography.hazmat.primitives.hashes import SHA256, Hash
+
+from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth
+from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import (
+    ActionStatus,
+    Response,
+)
+from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import (
+    OperationStatus_E as InfotainmentOperationStatus_E,
+)
+from tesla_fleet_api.tesla.vehicle.proto.signatures_pb2 import SignatureType, Tag
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import (
+    Destination,
+    Domain,
+    RoutableMessage,
+)
+from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
+    CommandStatus,
+    FromVCSECMessage,
+)
+from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import (
+    OperationStatus_E as VcsecOperationStatus_E,
+)
+from tesla_fleet_api.tesla.vehicle.proto.vehicle_pb2 import VehicleData
+
+VIN = "5YJXCAE43LF123456"
+
+
+def vcsec_ok_reply() -> RoutableMessage:
+    """A canned VCSEC reply as returned by a signed command that succeeded."""
+    body = FromVCSECMessage(
+        commandStatus=CommandStatus(
+            operationStatus=VcsecOperationStatus_E.OPERATIONSTATUS_OK
+        )
+    )
+    return RoutableMessage(
+        from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY),
+        protobuf_message_as_bytes=body.SerializeToString(),
+    )
+
+
+def infotainment_action_ok_reply() -> RoutableMessage:
+    """A canned infotainment reply as returned by a signed action that succeeded."""
+    body = Response(
+        actionStatus=ActionStatus(
+            result=InfotainmentOperationStatus_E.OPERATIONSTATUS_OK
+        )
+    )
+    return RoutableMessage(
+        from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT),
+        protobuf_message_as_bytes=body.SerializeToString(),
+    )
+
+
+def infotainment_vehicle_data_reply(vehicle_data: VehicleData) -> RoutableMessage:
+    """A canned infotainment ``vehicleData`` reply carrying the given proto."""
+    body = Response(vehicleData=vehicle_data)
+    return RoutableMessage(
+        from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT),
+        protobuf_message_as_bytes=body.SerializeToString(),
+    )
+
+
+class MockedBleTransportTestCase(IsolatedAsyncioTestCase):
+    """Base test case providing a ``VehicleBluetooth`` with ``_send`` mocked."""
+
+    VIN = VIN
+
+    def make_vehicle(self) -> tuple[VehicleBluetooth[Any], AsyncMock]:
+        """Build a VehicleBluetooth whose ``_send`` and connection are fully mocked.
+
+        Returns the vehicle plus the ``AsyncMock`` standing in for ``_send`` -
+        set ``send.return_value``/``side_effect`` to script replies.
+        """
+        parent = MagicMock()
+        parent.private_key = ec.generate_private_key(ec.SECP256R1())
+        vehicle = VehicleBluetooth(parent, self.VIN)
+
+        # Mark both signed-command sessions ready so _command skips the
+        # handshake round-trip (which would otherwise also go through _send).
+        sessions = cast("dict[int, Any]", getattr(vehicle, "_sessions"))
+        for session in sessions.values():
+            session.epoch = b"\x00" * 16
+            session.hmac = b"\x00" * 32
+            session.delta = 0
+            session.sharedKey = b"\x00" * 16
+
+        send = AsyncMock()
+        setattr(vehicle, "_send", send)
+        # connect_if_needed would otherwise attempt a real BLE connection.
+        setattr(vehicle, "connect_if_needed", AsyncMock())
+
+        return vehicle, send
+
+
+def decrypt_sent_command(vehicle: VehicleBluetooth[Any], msg: RoutableMessage) -> bytes:
+    """Decrypt the AES-GCM command payload of a ``RoutableMessage`` built by ``_commandAes``.
+
+    Mirrors the AAD ``_commandAes`` builds when encrypting, using the fixed
+    ``sharedKey`` ``make_vehicle`` installs, so a test can assert on the
+    plaintext command proto that was actually about to be sent to the car.
+    """
+    domain = msg.to_destination.domain
+    sessions = cast("dict[int, Any]", getattr(vehicle, "_sessions"))
+    session = sessions[domain]
+    assert session.sharedKey is not None
+    sig = msg.signature_data.AES_GCM_Personalized_data
+
+    metadata = bytes(
+        [
+            Tag.TAG_SIGNATURE_TYPE,
+            1,
+            SignatureType.SIGNATURE_TYPE_AES_GCM_PERSONALIZED,
+            Tag.TAG_DOMAIN,
+            1,
+            domain,
+            Tag.TAG_PERSONALIZATION,
+            17,
+            *vehicle.vin.encode(),
+            Tag.TAG_EPOCH,
+            len(sig.epoch),
+            *sig.epoch,
+            Tag.TAG_EXPIRES_AT,
+            4,
+            *struct.pack(">I", sig.expires_at),
+            Tag.TAG_COUNTER,
+            4,
+            *struct.pack(">I", sig.counter),
+            Tag.TAG_FLAGS,
+            4,
+            *struct.pack(">I", msg.flags),
+            Tag.TAG_END,
+        ]
+    )
+
+    aad = Hash(SHA256())
+    aad.update(metadata)
+    aesgcm = AESGCM(session.sharedKey)
+    return aesgcm.decrypt(
+        sig.nonce, msg.protobuf_message_as_bytes + sig.tag, aad.finalize()
+    )

tests/test_ble_mocked_commands.py

@@ -0,0 +1,83 @@
+"""BLE command tests over a mocked transport (no car, no BLE/GATT connection).
+
+Proves the mocked-transport pattern end to end: inherited ``Commands`` route
+through ``VehicleBluetooth``'s ``_send``, produce the expected signed
+``RoutableMessage``, and a canned reply decodes to the expected result. Every
+later BLE command-test PR builds on ``ble_mocked_transport``.
+"""
+
+from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import Action
+from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain
+from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import RKEAction_E, UnsignedMessage
+from tesla_fleet_api.tesla.vehicle.proto.vehicle_pb2 import ChargeState, VehicleData
+
+from ble_mocked_transport import (
+    MockedBleTransportTestCase,
+    decrypt_sent_command,
+    infotainment_action_ok_reply,
+    infotainment_vehicle_data_reply,
+    vcsec_ok_reply,
+)
+
+
+class DoorLockTests(MockedBleTransportTestCase):
+    """``door_lock`` (VCSEC, inherited from Commands) over the mocked BLE transport."""
+
+    async def test_sends_rke_lock_and_decodes_ok_reply(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = vcsec_ok_reply()
+
+        result = await vehicle.door_lock()
+
+        self.assertEqual(result, {"response": {"result": True, "reason": ""}})
+
+        send.assert_awaited_once()
+        await_args = send.await_args
+        assert await_args is not None
+        sent_msg = await_args.args[0]
+        self.assertEqual(sent_msg.to_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY)
+        self.assertTrue(sent_msg.signature_data.HasField("AES_GCM_Personalized_data"))
+
+        plaintext = decrypt_sent_command(vehicle, sent_msg)
+        unsigned = UnsignedMessage.FromString(plaintext)
+        self.assertEqual(unsigned.RKEAction, RKEAction_E.RKE_ACTION_LOCK)
+
+
+class SetTempsTests(MockedBleTransportTestCase):
+    """``set_temps`` (INFO, inherited from Commands) over the mocked BLE transport."""
+
+    async def test_sends_hvac_temperature_action_and_decodes_ok_reply(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_action_ok_reply()
+
+        result = await vehicle.set_temps(21.5, 19.0)
+
+        self.assertEqual(result, {"response": {"result": True, "reason": ""}})
+
+        send.assert_awaited_once()
+        await_args = send.await_args
+        assert await_args is not None
+        sent_msg = await_args.args[0]
+        self.assertEqual(sent_msg.to_destination.domain, Domain.DOMAIN_INFOTAINMENT)
+        self.assertTrue(sent_msg.signature_data.HasField("AES_GCM_Personalized_data"))
+
+        plaintext = decrypt_sent_command(vehicle, sent_msg)
+        action = Action.FromString(plaintext)
+        hvac = action.vehicleAction.hvacTemperatureAdjustmentAction
+        self.assertAlmostEqual(hvac.driver_temp_celsius, 21.5)
+        self.assertAlmostEqual(hvac.passenger_temp_celsius, 19.0)
+
+
+class ChargeStateTypedReplyTests(MockedBleTransportTestCase):
+    """``charge_state`` (INFO read, defined on VehicleBluetooth) decodes a typed reply."""
+
+    async def test_decodes_canned_vehicle_data_to_charge_state_proto(self) -> None:
+        vehicle, send = self.make_vehicle()
+        send.return_value = infotainment_vehicle_data_reply(
+            VehicleData(charge_state=ChargeState(battery_level=42))
+        )
+
+        charge_state = await vehicle.charge_state()
+
+        self.assertIsInstance(charge_state, ChargeState)
+        self.assertEqual(charge_state.battery_level, 42)