diff --git a/AGENTS.md b/AGENTS.md index 7ed81a1..b6fdd71 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -116,6 +116,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **Seat indexing gotcha**: two distinct seat enums with different conventions. `Seat` is **0-indexed** (`FRONT_LEFT=0`) and is for the manual seat heater/cooler paths (`remote_seat_heater_request`, `remote_seat_cooler_request`). `AutoSeat` is **1-indexed** (`FRONT_LEFT=1`, `FRONT_RIGHT=2`) and is the correct type for `remote_auto_seat_climate_request` on **both** backends — its values equal Tesla's REST wire values and the proto `AutoSeatPosition_*` enum. Don't mix them; passing a `Seat` to the auto-climate command is off-by-one (issue #11). - **Naming**: camelCase for class instance attributes that mirror API structure (`energySites`, `createFleet`). Snake_case for method names that are API endpoints. - **BLE discovery gotcha**: a Tesla vehicle advertises no 128-bit service UUID pre-connect — only its VIN-derived local name (`^S[a-f0-9]{16}[CDRP]$`), and only in the scan response, not the `ADV_IND`. `SERVICE_UUID` (`tesla_fleet_api/tesla/vehicle/bluetooth.py`) exists only as a GATT service after connecting. Never pass `service_uuids=[SERVICE_UUID]` as a `BleakScanner` discovery-time filter — it hides the vehicle on a direct BlueZ adapter (an ESPHome proxy doesn't enforce that filter the same way, which can mask the bug in testing). Scan unfiltered with active scanning and match by name; keep `SERVICE_UUID` for post-connect GATT use only. +- **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. ## Maintaining this file diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index 78660e2..2f96b7a 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -96,9 +96,12 @@ def prependLength(message: bytes) -> bytearray: class ReassemblingBuffer: """ - Reassembles bytearray streams where the first two bytes indicate the length of the message. - Handles potential packet corruption by discarding *entire* packets and retrying. - Uses a callback to process parsed messages. + Reassembles BLE notification chunks into length-prefixed RoutableMessages. + + Each message starts with a 2-byte length. One notification can contain part + of a message, exactly one message, or multiple messages. If a message cannot + be decoded, the buffer drops the current physical packet and resynchronizes + at the next recorded packet boundary. """ def __init__(self, callback: Callable[[RoutableMessage], None]): @@ -106,7 +109,6 @@ def __init__(self, callback: Callable[[RoutableMessage], None]): Initializes the buffer. Args: - message_type: The protobuf message type (e.g., RoutableMessage) to parse the assembled data. callback: A function that will be called with each parsed message. """ self.buffer: bytearray = bytearray() @@ -116,7 +118,7 @@ def __init__(self, callback: Callable[[RoutableMessage], None]): def receive_data(self, data: bytearray): """ - Receives a chunk of bytearray data and attempts to assemble a complete message. + Receive one BLE notification chunk and emit any complete messages. Args: data: The received bytearray data. @@ -181,6 +183,7 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]): _ekey: ec.EllipticCurvePublicKey _buffer: ReassemblingBuffer _auth_method = "aes" + _ack_followup_timeout: float = 2 def __init__( self, @@ -282,14 +285,24 @@ def _on_notify(self, sender: BleakGATTCharacteristic, data: bytearray) -> None: self._buffer.receive_data(data) def _on_message(self, msg: RoutableMessage) -> None: - """Receive messages from the Tesla BLE data.""" + """Route addressed BLE replies into the per-domain response queue.""" if msg.to_destination.routing_address != self._from_destination: LOGGER.debug("Ignoring broadcast message (not addressed to us)") return + queue = self._queues.get(msg.from_destination.domain) + if queue is None: + # Domain enum has values (e.g. DOMAIN_BROADCAST, DOMAIN_AUTHD) with + # no session/queue of our own; indexing _queues directly would + # raise KeyError and abort the reassembly loop mid-buffer. + LOGGER.debug( + f"Ignoring message from unhandled domain {msg.from_destination.domain}" + ) + return + LOGGER.debug(f"Received response: {msg}") - self._queues[msg.from_destination.domain].put_nowait(msg) + queue.put_nowait(msg) async def _send( self, msg: RoutableMessage, requires: str, timeout: int = 5 @@ -332,7 +345,7 @@ async def _send( "Received ACK for our request, waiting briefly for data follow-up" ) try: - async with asyncio.timeout(2): + async with asyncio.timeout(self._ack_followup_timeout): while True: resp2 = await self._queues[domain].get() LOGGER.debug( diff --git a/tests/test_ble_message_routing.py b/tests/test_ble_message_routing.py new file mode 100644 index 0000000..543bafd --- /dev/null +++ b/tests/test_ble_message_routing.py @@ -0,0 +1,95 @@ +"""Regression tests for VehicleBluetooth._on_message routing and filtering. + +Covers VCSEC-vs-INFO domain routing, dropping messages not addressed to us +(broadcast filtering), and the domain KeyError crash fixed in bluetooth.py: +``Domain`` has values (DOMAIN_BROADCAST, DOMAIN_AUTHD, ...) with no entry in +``_queues``, so a message addressed to us but from one of those domains must +be dropped rather than raising and aborting the reassembly loop mid-buffer. +""" + +from unittest import TestCase +from unittest.mock import MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) + +VIN = "5YJXCAE43LF123456" + + +def _make_vehicle() -> VehicleBluetooth: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + return VehicleBluetooth(parent, VIN) + + +class OnMessageRoutingTests(TestCase): + def test_vcsec_reply_routed_to_vehicle_security_queue(self) -> None: + vehicle = _make_vehicle() + msg = RoutableMessage( + to_destination=Destination(routing_address=vehicle._from_destination), + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY), + ) + + vehicle._on_message(msg) + + self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 1) + self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0) + + def test_infotainment_reply_routed_to_infotainment_queue(self) -> None: + vehicle = _make_vehicle() + msg = RoutableMessage( + to_destination=Destination(routing_address=vehicle._from_destination), + from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT), + ) + + vehicle._on_message(msg) + + self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 1) + self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0) + + def test_message_not_addressed_to_us_is_dropped(self) -> None: + vehicle = _make_vehicle() + msg = RoutableMessage( + to_destination=Destination(routing_address=b"someone-elses-address"), + from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT), + ) + + vehicle._on_message(msg) + + self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0) + self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0) + + def test_message_addressed_to_us_from_unhandled_domain_is_dropped_not_raised( + self, + ) -> None: + vehicle = _make_vehicle() + msg = RoutableMessage( + to_destination=Destination(routing_address=vehicle._from_destination), + from_destination=Destination(domain=Domain.DOMAIN_BROADCAST), + ) + + # Must not raise KeyError - DOMAIN_BROADCAST has no entry in _queues. + vehicle._on_message(msg) + + self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0) + self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0) + + def test_message_addressed_to_us_from_authd_domain_is_dropped_not_raised( + self, + ) -> None: + vehicle = _make_vehicle() + msg = RoutableMessage( + to_destination=Destination(routing_address=vehicle._from_destination), + from_destination=Destination(domain=Domain.DOMAIN_AUTHD), + ) + + vehicle._on_message(msg) + + self.assertEqual(vehicle._queues[Domain.DOMAIN_VEHICLE_SECURITY].qsize(), 0) + self.assertEqual(vehicle._queues[Domain.DOMAIN_INFOTAINMENT].qsize(), 0) diff --git a/tests/test_ble_reassembling_buffer.py b/tests/test_ble_reassembling_buffer.py new file mode 100644 index 0000000..d47c51b --- /dev/null +++ b/tests/test_ble_reassembling_buffer.py @@ -0,0 +1,116 @@ +"""Regression tests for ``ReassemblingBuffer`` framing and corruption resync. + +Exercises the length-prefixed reassembly directly (no VehicleBluetooth, no +GATT) to lock down: a message split across multiple BLE notification +chunks, multiple complete messages delivered in a single chunk, and +resynchronization after a corrupted/oversized packet using the +``packet_starts`` boundary tracking in ``discard_packet``. +""" + +from unittest import TestCase + +from tesla_fleet_api.tesla.vehicle.bluetooth import ReassemblingBuffer, prependLength +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) + + +def framed(msg: RoutableMessage) -> bytearray: + """Length-prefix a RoutableMessage the way the vehicle would frame it.""" + return prependLength(msg.SerializeToString()) + + +class ReassemblingBufferTests(TestCase): + def setUp(self) -> None: + self.received: list[RoutableMessage] = [] + self.buffer = ReassemblingBuffer(self.received.append) + + def test_single_message_single_chunk(self) -> None: + msg = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT) + ) + + self.buffer.receive_data(framed(msg)) + + self.assertEqual(len(self.received), 1) + self.assertEqual( + self.received[0].from_destination.domain, Domain.DOMAIN_INFOTAINMENT + ) + + def test_message_split_across_multiple_chunks(self) -> None: + msg = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY), + request_uuid=b"0123456789abcdef", + ) + payload = framed(msg) + + # Simulate GATT MTU fragmentation: deliver a few bytes at a time. + chunk_size = 5 + for i in range(0, len(payload), chunk_size): + self.buffer.receive_data(payload[i : i + chunk_size]) + if i + chunk_size < len(payload): + self.assertEqual( + self.received, [], "must not fire callback before full message" + ) + + self.assertEqual(len(self.received), 1) + self.assertEqual( + self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY + ) + self.assertEqual(self.received[0].request_uuid, b"0123456789abcdef") + + def test_multiple_messages_in_one_chunk(self) -> None: + first = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY) + ) + second = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT) + ) + + self.buffer.receive_data(framed(first) + framed(second)) + + self.assertEqual(len(self.received), 2) + self.assertEqual( + self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY + ) + self.assertEqual( + self.received[1].from_destination.domain, Domain.DOMAIN_INFOTAINMENT + ) + + def test_corrupt_packet_resyncs_to_next_packet_boundary(self) -> None: + good = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_INFOTAINMENT) + ) + + # A plausible-length header (so it's not caught by the oversized + # check) whose payload is not a valid RoutableMessage. + corrupt = prependLength(b"\xff" * 20) + + # Each call to receive_data is one physical BLE notification, so + # this records two distinct packet_starts entries for resync. + self.buffer.receive_data(corrupt) + self.buffer.receive_data(framed(good)) + + self.assertEqual(len(self.received), 1) + self.assertEqual( + self.received[0].from_destination.domain, Domain.DOMAIN_INFOTAINMENT + ) + + def test_oversized_length_header_discards_and_resyncs(self) -> None: + good = RoutableMessage( + from_destination=Destination(domain=Domain.DOMAIN_VEHICLE_SECURITY) + ) + + # Length header claims a message > 1024 bytes - must be treated as + # corrupt rather than stalling forever waiting for more data. + oversized_header = bytearray([0xFF, 0xFF]) + b"\x00" * 10 + + self.buffer.receive_data(oversized_header) + self.buffer.receive_data(framed(good)) + + self.assertEqual(len(self.received), 1) + self.assertEqual( + self.received[0].from_destination.domain, Domain.DOMAIN_VEHICLE_SECURITY + ) diff --git a/tests/test_ble_send_transport.py b/tests/test_ble_send_transport.py new file mode 100644 index 0000000..df74a2a --- /dev/null +++ b/tests/test_ble_send_transport.py @@ -0,0 +1,160 @@ +"""Regression tests for VehicleBluetooth._send: ACK-then-data follow-up, +stale-queue flushing, and the total-timeout path. + +Drives _send() directly with a mocked GATT client - write_gatt_char's +side_effect pushes canned replies onto the domain queue, standing in for +what _on_message would enqueue from a real notification - so these tests +exercise the real wait/ACK-follow-up state machine without a BLE connection. +""" + +from __future__ import annotations + +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.exceptions import BluetoothTimeout +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_fleet_api.tesla.vehicle.proto.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) + +VIN = "5YJXCAE43LF123456" +DOMAIN = Domain.DOMAIN_VEHICLE_SECURITY + + +def _make_vehicle() -> VehicleBluetooth: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle = VehicleBluetooth(parent, VIN) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + return vehicle + + +def _outgoing() -> RoutableMessage: + return RoutableMessage( + to_destination=Destination(domain=DOMAIN), + uuid=b"0123456789abcdef", + ) + + +class SendImmediateResponseTests(IsolatedAsyncioTestCase): + async def test_returns_response_that_already_has_required_field(self) -> None: + vehicle = _make_vehicle() + msg = _outgoing() + reply = RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"data", + ) + + async def fake_write(_uuid, _payload, _response): + vehicle._queues[DOMAIN].put_nowait(reply) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=fake_write) + + result = await vehicle._send(msg, "protobuf_message_as_bytes") + + self.assertEqual(result, reply) + vehicle.client.write_gatt_char.assert_awaited_once() + + +class SendAckFollowupTests(IsolatedAsyncioTestCase): + async def test_waits_past_ack_for_data_followup(self) -> None: + vehicle = _make_vehicle() + msg = _outgoing() + ack = RoutableMessage( + from_destination=Destination(domain=DOMAIN), request_uuid=msg.uuid + ) + data = RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"payload", + ) + + async def fake_write(_uuid, _payload, _response): + vehicle._queues[DOMAIN].put_nowait(ack) + vehicle._queues[DOMAIN].put_nowait(data) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=fake_write) + + result = await vehicle._send(msg, "protobuf_message_as_bytes") + + self.assertEqual(result, data) + + async def test_ack_without_followup_times_out_and_returns_ack(self) -> None: + vehicle = _make_vehicle() + vehicle._ack_followup_timeout = 0.05 + msg = _outgoing() + ack = RoutableMessage( + from_destination=Destination(domain=DOMAIN), request_uuid=msg.uuid + ) + + async def fake_write(_uuid, _payload, _response): + vehicle._queues[DOMAIN].put_nowait(ack) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=fake_write) + + result = await vehicle._send(msg, "protobuf_message_as_bytes") + + self.assertEqual(result, ack) + + async def test_unrelated_message_is_ignored_until_matching_data_arrives( + self, + ) -> None: + vehicle = _make_vehicle() + msg = _outgoing() + stray = RoutableMessage( + from_destination=Destination(domain=DOMAIN), request_uuid=b"not-ours" + ) + data = RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"payload", + ) + + async def fake_write(_uuid, _payload, _response): + vehicle._queues[DOMAIN].put_nowait(stray) + vehicle._queues[DOMAIN].put_nowait(data) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=fake_write) + + result = await vehicle._send(msg, "protobuf_message_as_bytes") + + self.assertEqual(result, data) + + +class SendQueueHygieneTests(IsolatedAsyncioTestCase): + async def test_stale_queued_message_is_discarded_before_send(self) -> None: + vehicle = _make_vehicle() + stale = RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"stale-leftover", + ) + vehicle._queues[DOMAIN].put_nowait(stale) + + msg = _outgoing() + fresh = RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=b"fresh", + ) + + async def fake_write(_uuid, _payload, _response): + vehicle._queues[DOMAIN].put_nowait(fresh) + + vehicle.client.write_gatt_char = AsyncMock(side_effect=fake_write) + + result = await vehicle._send(msg, "protobuf_message_as_bytes") + + self.assertEqual(result, fresh) + + +class SendTimeoutTests(IsolatedAsyncioTestCase): + async def test_no_response_raises_bluetooth_timeout(self) -> None: + vehicle = _make_vehicle() + msg = _outgoing() + vehicle.client.write_gatt_char = AsyncMock() + + with self.assertRaises(BluetoothTimeout): + await vehicle._send(msg, "protobuf_message_as_bytes", timeout=0.05)