diff --git a/AGENTS.md b/AGENTS.md index 5b83564..4902b5e 100644 --- a/AGENTS.md +++ b/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 diff --git a/tests/ble_mocked_transport.py b/tests/ble_mocked_transport.py new file mode 100644 index 0000000..8287565 --- /dev/null +++ b/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() + ) diff --git a/tests/test_ble_mocked_commands.py b/tests/test_ble_mocked_commands.py new file mode 100644 index 0000000..9d0f1ac --- /dev/null +++ b/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) diff --git a/uv.lock b/uv.lock index 6cbffaf..e39b0d9 100644 --- a/uv.lock +++ b/uv.lock @@ -728,7 +728,7 @@ wheels = [ [[package]] name = "tesla-fleet-api" -version = "1.5.2" +version = "1.5.4" source = { editable = "." } dependencies = [ { name = "aiofiles" },