From ed35f6991ca6991c872689da6ac1f39f94098dab Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 9 Jul 2026 19:44:37 +1000 Subject: [PATCH 1/3] test(ble): add closures/locks regression coverage over mocked transport Covers door_unlock, auto_secure_vehicle, charge_port_door_open/close, and the 8 individual door commands (door_lock already had coverage). Live-verify against the test car confirmed door_lock/door_unlock/auto_secure_vehicle round-trip correctly; charge_port_door_* and 7 of 8 individual doors are deferred (see AGENTS.md gotcha) so this PR ships mocked-transport regression tests for the full group with live-verify status tracked separately. --- AGENTS.md | 1 + tests/test_ble_mocked_closures_locks.py | 168 ++++++++++++++++++++++++ 2 files changed, 169 insertions(+) create mode 100644 tests/test_ble_mocked_closures_locks.py diff --git a/AGENTS.md b/AGENTS.md index 2812445..59a8e8b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -119,6 +119,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **BLE domain-routing gotcha**: `Domain` (`proto/universal_message.proto`) has more values (`DOMAIN_BROADCAST`, `DOMAIN_AUTHD`, ...) than `VehicleBluetooth._queues` has keys (only `DOMAIN_VEHICLE_SECURITY`/`DOMAIN_INFOTAINMENT`). `_on_message` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) must look up `_queues` with `.get()` and drop unrecognized domains rather than indexing directly — indexing raises `KeyError` inside the `ReassemblingBuffer` callback, aborting reassembly of any further already-buffered messages in that notification. - **BLE infotainment boot-delay gotcha**: `wake_up()` (VCSEC) returns as soon as the vehicle-security computer acks it, well before the infotainment computer is ready to complete a signed-command handshake. An INFO-domain read/command issued immediately after `wake_up()` can raise `BluetoothTimeout` on the handshake through no fault of the command itself. Live-verified: waiting ~10s after `wake_up()` before the first INFO read is sufficient on the test rig; callers doing INFO work right after waking should retry-with-backoff rather than treat one timeout as failure. - **BLE `vehicle_data()` response-size cap**: the vehicle's signed-command implementation enforces its own response-size limit independent of the BLE transport's packet reassembly. Live-verified: a single-endpoint `vehicle_data()` call (or any of the dedicated per-substate readers like `charge_state()`) succeeds, but requesting as few as two `BluetoothVehicleData` endpoints together reliably raises `TeslaFleetMessageFaultResponseSizeExceedsMTU` (`exceptions.py`). This is why `vehicle_data()`'s `endpoints` arg has no all-endpoints default (unlike the cloud method) - prefer the per-substate readers, or a single-endpoint `vehicle_data()` call, over a multi-endpoint composite. Auto-chunking (split under the cap, merge replies) would fix this properly but is not implemented. +- **BLE individual-door powered-close gotcha**: `open_*_door()` unlatches a door over VCSEC; on a Model 3 there is no reliable powered close. Live-verified: `close_rear_passenger_door()` returned an OK ack (`{"result": True}`) but `closures_state().door_open_passenger_rear` stayed `True` - the ack only means the car accepted the command, not that the door physically re-latched (a human has to push it shut). Never chain an automated snapshot→act→verify→restore cycle across an individual door-open command; treat the 8 door commands as ack-verified only, or require a human to confirm the physical re-close before trusting `closures_state()` again. ## Maintaining this file diff --git a/tests/test_ble_mocked_closures_locks.py b/tests/test_ble_mocked_closures_locks.py new file mode 100644 index 0000000..277bec0 --- /dev/null +++ b/tests/test_ble_mocked_closures_locks.py @@ -0,0 +1,168 @@ +"""Regression tests for the closures/locks command group over the mocked BLE transport. + +Covers ``door_unlock``, ``auto_secure_vehicle``, ``charge_port_door_open/close`` +(all VCSEC, inherited from ``Commands``) and the 8 individual door open/close +commands (VCSEC, defined on ``VehicleBluetooth`` - no cloud REST equivalent). +``door_lock`` is already covered in ``test_ble_mocked_commands.py``. +""" + +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain +from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ( + ClosureMoveType_E, + RKEAction_E, + UnsignedMessage, +) + +from ble_mocked_transport import ( + MockedBleTransportTestCase, + decrypt_sent_command, + vcsec_ok_reply, +) + + +class DoorUnlockTests(MockedBleTransportTestCase): + async def test_sends_rke_unlock_and_decodes_ok_reply(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + result = await vehicle.door_unlock() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + sent_msg = send.await_args.args[0] + self.assertEqual(sent_msg.to_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY) + + plaintext = decrypt_sent_command(vehicle, sent_msg) + unsigned = UnsignedMessage.FromString(plaintext) + self.assertEqual(unsigned.RKEAction, RKEAction_E.RKE_ACTION_UNLOCK) + + +class AutoSecureVehicleTests(MockedBleTransportTestCase): + async def test_sends_rke_auto_secure_and_decodes_ok_reply(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + result = await vehicle.auto_secure_vehicle() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + sent_msg = send.await_args.args[0] + self.assertEqual(sent_msg.to_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY) + + plaintext = decrypt_sent_command(vehicle, sent_msg) + unsigned = UnsignedMessage.FromString(plaintext) + self.assertEqual(unsigned.RKEAction, RKEAction_E.RKE_ACTION_AUTO_SECURE_VEHICLE) + + +class ChargePortDoorTests(MockedBleTransportTestCase): + async def test_open_sends_closure_move_open(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + result = await vehicle.charge_port_door_open() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + sent_msg = send.await_args.args[0] + plaintext = decrypt_sent_command(vehicle, sent_msg) + unsigned = UnsignedMessage.FromString(plaintext) + self.assertEqual( + unsigned.closureMoveRequest.chargePort, + ClosureMoveType_E.CLOSURE_MOVE_TYPE_OPEN, + ) + + async def test_close_sends_closure_move_close(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + result = await vehicle.charge_port_door_close() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + sent_msg = send.await_args.args[0] + plaintext = decrypt_sent_command(vehicle, sent_msg) + unsigned = UnsignedMessage.FromString(plaintext) + self.assertEqual( + unsigned.closureMoveRequest.chargePort, + ClosureMoveType_E.CLOSURE_MOVE_TYPE_CLOSE, + ) + + +class IndividualDoorTests(MockedBleTransportTestCase): + """The 8 individual door commands - each sets exactly one ``ClosureMoveRequest`` field.""" + + DOORS = [ + ( + "open_front_driver_door", + "frontDriverDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_OPEN, + ), + ( + "close_front_driver_door", + "frontDriverDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_CLOSE, + ), + ( + "open_front_passenger_door", + "frontPassengerDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_OPEN, + ), + ( + "close_front_passenger_door", + "frontPassengerDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_CLOSE, + ), + ( + "open_rear_driver_door", + "rearDriverDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_OPEN, + ), + ( + "close_rear_driver_door", + "rearDriverDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_CLOSE, + ), + ( + "open_rear_passenger_door", + "rearPassengerDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_OPEN, + ), + ( + "close_rear_passenger_door", + "rearPassengerDoor", + ClosureMoveType_E.CLOSURE_MOVE_TYPE_CLOSE, + ), + ] + + async def test_each_door_command_sets_only_its_own_field(self) -> None: + for method_name, field_name, expected_move in self.DOORS: + with self.subTest(method=method_name): + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + result = await getattr(vehicle, method_name)() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + sent_msg = send.await_args.args[0] + self.assertEqual( + sent_msg.to_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY + ) + + plaintext = decrypt_sent_command(vehicle, sent_msg) + unsigned = UnsignedMessage.FromString(plaintext) + request = unsigned.closureMoveRequest + self.assertEqual(getattr(request, field_name), expected_move) + + for other_field in ( + "frontDriverDoor", + "frontPassengerDoor", + "rearDriverDoor", + "rearPassengerDoor", + "rearTrunk", + "frontTrunk", + "chargePort", + "tonneau", + ): + if other_field == field_name: + continue + self.assertEqual( + getattr(request, other_field), + ClosureMoveType_E.CLOSURE_MOVE_TYPE_NONE, + f"{method_name} unexpectedly set {other_field}", + ) From ea09eadd31ad2ec42daa3b15aba32a39f6b979f2 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 9 Jul 2026 19:47:12 +1000 Subject: [PATCH 2/3] docs(ble): document live-verify status in closures/locks test module Records which commands are live-verified (locks) vs unit-test-only pending a captain-present session (charge port cable was engaged; individual doors leave a door physically ajar with no powered close on this Model 3). --- tests/test_ble_mocked_closures_locks.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/tests/test_ble_mocked_closures_locks.py b/tests/test_ble_mocked_closures_locks.py index 277bec0..bb999fb 100644 --- a/tests/test_ble_mocked_closures_locks.py +++ b/tests/test_ble_mocked_closures_locks.py @@ -4,6 +4,18 @@ (all VCSEC, inherited from ``Commands``) and the 8 individual door open/close commands (VCSEC, defined on ``VehicleBluetooth`` - no cloud REST equivalent). ``door_lock`` is already covered in ``test_ble_mocked_commands.py``. + +Live-verify status: ``door_lock``/``door_unlock``/``auto_secure_vehicle`` are +live-verified (full snapshot->act->verify->restore->confirm cycle against the +test car). ``charge_port_door_open/close`` and the 8 individual door commands +are unit-test-only here - mocked-transport coverage only, no live actuation. +``charge_port_door_*`` is deferred because the test car had a charge cable +physically engaged (closing over an engaged latch is unsafe to test remotely). +The individual doors are deferred because an ``open_*_door()`` unlatches the +door but there is no reliable powered close on this Model 3 - one live probe +left a door physically ajar needing a human push to close (see the BLE +individual-door powered-close gotcha in AGENTS.md). Treat these the same as +the CAPTAIN-PRESENT group: live-verify only in a captain-present session. """ from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import Domain From 758fd39a89b576ea398321faccb3cb0e7ac2b8c1 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Thu, 9 Jul 2026 19:52:08 +1000 Subject: [PATCH 3/3] no-mistakes(document): Document BLE door close caveat --- docs/bluetooth_vehicles.md | 6 +++++- tesla_fleet_api/tesla/vehicle/bluetooth.py | 16 ++++++++-------- 2 files changed, 13 insertions(+), 9 deletions(-) diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 65b3a64..b0538c1 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -85,6 +85,11 @@ INFO-domain reads immediately after waking, such as `charge_state()` or ## Open/Close Individual Doors (Bluetooth Only) The individual door closure commands are Bluetooth-only and are not available via Fleet API signed commands. +`open_*_door()` unlatches the selected door. `close_*_door()` only means the +vehicle accepted the close request; on vehicles without reliable powered door +close support, the door may remain physically ajar until someone pushes it +shut. Do not rely on an automated open-then-close cycle without checking +`closures_state()` or getting physical confirmation. Available commands: @@ -109,7 +114,6 @@ async def main(): vehicle = tesla_bluetooth.vehicles.create("") await vehicle.connect() await vehicle.open_front_driver_door() - await vehicle.close_front_driver_door() await vehicle.disconnect() asyncio.run(main()) diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 2df19e0..78d7272 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -369,7 +369,7 @@ async def _send( # Group 12: VCSEC closures (Bluetooth-only for individual doors) async def open_front_driver_door(self) -> dict[str, Any]: - """Opens the front driver door.""" + """Unlatches/opens the front driver door.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -379,7 +379,7 @@ async def open_front_driver_door(self) -> dict[str, Any]: ) async def close_front_driver_door(self) -> dict[str, Any]: - """Closes the front driver door.""" + """Requests front driver door close; OK ack may not mean physically latched.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -389,7 +389,7 @@ async def close_front_driver_door(self) -> dict[str, Any]: ) async def open_front_passenger_door(self) -> dict[str, Any]: - """Opens the front passenger door.""" + """Unlatches/opens the front passenger door.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -399,7 +399,7 @@ async def open_front_passenger_door(self) -> dict[str, Any]: ) async def close_front_passenger_door(self) -> dict[str, Any]: - """Closes the front passenger door.""" + """Requests front passenger door close; OK ack may not mean physically latched.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -409,7 +409,7 @@ async def close_front_passenger_door(self) -> dict[str, Any]: ) async def open_rear_driver_door(self) -> dict[str, Any]: - """Opens the rear driver door.""" + """Unlatches/opens the rear driver door.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -419,7 +419,7 @@ async def open_rear_driver_door(self) -> dict[str, Any]: ) async def close_rear_driver_door(self) -> dict[str, Any]: - """Closes the rear driver door.""" + """Requests rear driver door close; OK ack may not mean physically latched.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -429,7 +429,7 @@ async def close_rear_driver_door(self) -> dict[str, Any]: ) async def open_rear_passenger_door(self) -> dict[str, Any]: - """Opens the rear passenger door.""" + """Unlatches/opens the rear passenger door.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest( @@ -439,7 +439,7 @@ async def open_rear_passenger_door(self) -> dict[str, Any]: ) async def close_rear_passenger_door(self) -> dict[str, Any]: - """Closes the rear passenger door.""" + """Requests rear passenger door close; OK ack may not mean physically latched.""" return await self._sendVehicleSecurity( UnsignedMessage( closureMoveRequest=ClosureMoveRequest(