From d99f0c9b10846491e3878ac90d4b548aeb3c9760 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sat, 11 Jul 2026 10:42:02 +1000 Subject: [PATCH 1/4] Add DEBUG-level command/transport/outcome logging Adds terse, grep-friendly debug logging so a live BLE-primary Router deployment can be diagnosed: which transport (bluetooth/fleet/teslemetry/tessie) served each command, and why it failed. Logging only, no behavior changes. - Commands._sendVehicleSecurity/_getVehicleSecurity/_sendInfotainment/_getInfotainment (commands.py) log command= transport= result=..., covering both BLE and Fleet-signed commands from one choke point each. - TeslaFleetApi._request (fleet.py) logs the same shape for Fleet/Teslemetry/Tessie REST commands. - VehicleBluetooth's verify_commands timeout resolution logs a second line (verify_commands=resolved/unresolved). - Router._dispatch logs which backend served each call and any failover hop. --- AGENTS.md | 1 + docs/bluetooth_vehicles.md | 29 +++ tesla_fleet_api/router/base.py | 12 +- tesla_fleet_api/tesla/fleet.py | 117 ++++++---- tesla_fleet_api/tesla/vehicle/bluetooth.py | 45 +++- tesla_fleet_api/tesla/vehicle/commands.py | 97 ++++++-- tesla_fleet_api/tesla/vehicle/signed.py | 1 + tesla_fleet_api/teslemetry/teslemetry.py | 1 + tesla_fleet_api/tessie/tessie.py | 1 + tests/test_command_logging.py | 251 +++++++++++++++++++++ uv.lock | 2 +- 11 files changed, 490 insertions(+), 67 deletions(-) create mode 100644 tests/test_command_logging.py diff --git a/AGENTS.md b/AGENTS.md index d60f075..fbdd635 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -136,6 +136,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te - **`pair()` confirms whitelisting two ways: one-shot reply OR verify-by-state poll**: the whitelist-op success is a single VCSEC frame, and it lands on a dead session (lost forever) if the BLE link cycles while the user walks to the car to approve - live-observed on HA/macOS CoreBluetooth, where `tesla_fleet_api` logged "Reconnecting to S..." every ~100s during a pending pair, and the plain one-shot wait hung despite car-side completion. `pair()` (`bluetooth.py`) keeps the reply as the fast path (waits one `poll_interval` for it) but, on a lost reply, polls `_pair_probe()` every `poll_interval` until an overall `timeout` (default 300s) elapses. The probe is a VCSEC `_handshake` with our own public key: it succeeds only once the key is whitelisted and faults `NotOnWhitelistFault` until then - `_pair_probe` maps any `TeslaFleetError` (incl. transport failures from a mid-wait reconnect) to "not yet", so polling survives reconnects. The whitelist op is written **exactly once** - never re-sent - because a re-send re-prompts the user (and the retry/double-execute hazard documented above applies). Deadline with neither path confirming raises a typed `BluetoothTimeout`. Default behavior, no new knob (`poll_interval` is a defaulted param). - **Idle BLE keepalive (`keepalive_interval`)**: an idle held BLE link to the vehicle drops at ~42s mean lifetime (link supervision timeout ~720ms underneath, bluetoothd-verified on macOS CoreBluetooth); a single trivial passive GATT read every ~20s extends lifetime ~10x (~400s observed). `VehicleBluetooth.__init__`'s `keepalive_interval` (default `DEFAULT_KEEPALIVE_INTERVAL` = 20.0, `None`/`0` disables; threaded through `Vehicles`/`VehiclesBluetooth.create*`) starts one asyncio task per connection (`_keepalive_loop`, `bluetooth.py`) that reads `VERSION_UUID` only after `keepalive_interval` seconds of genuine GATT idleness. **Idle-triggered, not periodic**: `_last_activity` is bumped on every `_send` write and every `_on_notify` frame, so an active session never gets extra traffic; the loop recomputes the wait each pass and fires only when idle. The read is **bounded** (`_keepalive_timeout`, 2s) and **best-effort** - a prior un-timed RSSI read hung indefinitely against a sleeping car, so every attempt carries a timeout and swallows all failures (`_keepalive_read` catches `Exception`, never `CancelledError`); a failed keepalive never raises into user code, never triggers reconnect (the existing `connect_if_needed` machinery owns recovery), and never wakes the car. Task lifecycle is tied to the connection: started at the end of `connect()` (after `start_notify`), cancelled-and-awaited in `disconnect()` and restarted cleanly on reconnect (`_start_keepalive`/`_stop_keepalive`). **Sleep tradeoff**: these reads keep an *awake* car awake and defer vehicle sleep - consumers wanting the car to sleep should disable keepalive or disconnect when idle. Tests in `tests/test_ble_keepalive.py`. - **Cross-transport parity (cloud REST `VehicleFleet` vs BLE `Commands`)**: the same-named command on both paths should build a semantically equivalent instruction from identical args - a divergence there is a bug, but response *bodies* legitimately differ (REST JSON dict vs decoded protobuf) and are not. `tests/test_cross_transport_parity.py` locks the equivalence in with mocked-both-transports tests. Known **non-bug FORM differences** (do not "fix"): `set_scheduled_departure`'s `preconditioning_enabled`/`off_peak_charging_enabled` (no proto fields), `window_control` lat/lon and `navigation_sc_request` `id` (no proto fields), `navigation_request`'s `type`/`locale`/`timestamp_ms` (REST share-intent framing), and `media_volume_up` (no Tesla REST endpoint - BLE-only; cloud raises volume via `adjust_volume`). Two **open divergences left unfixed** pending live verification: `clear_pin_to_drive_admin` builds `DrivingClearSpeedLimitPinAction` (speed-limit PIN, not PIN-to-Drive - suspected mismapping, security-sensitive), and `navigation_gps_request`'s `order` is required on BLE but optional on cloud (signature mismatch; null-order wire semantics undecided). +- **Per-command debug logging chokepoints and the `command=` name it derives**: `LOGGER.debug` lines of the form `command= transport= result=...` are emitted from exactly four places, not per-method - `Commands._sendVehicleSecurity`/`_getVehicleSecurity`/`_sendInfotainment`/`_getInfotainment` (`commands.py`, covers both BLE and Fleet-signed) and `TeslaFleetApi._request` (`fleet.py`, covers Fleet/Teslemetry/Tessie REST). `transport` comes from a `_transport_name` `ClassVar` set per concrete class (`"bluetooth"`/`"fleet"`/`"teslemetry"`/`"tessie"`), mirroring the existing `_auth_method` pattern - add that ClassVar to any new `Commands`/`TeslaFleetApi` subclass. For BLE/Fleet-signed, `command` is **not** the Python method name; it's derived from the populated protobuf oneof field (`vcsec_command_name`/`infotainment_command_name` in `commands.py`), e.g. `door_lock()` logs as `RKE_ACTION_LOCK` and `set_charge_limit()` as `chargingSetLimitAction` - deliberately robust to call-site changes since it reads the message being sent, not the call stack. `VehicleBluetooth`'s `verify_commands` resolution logs a second, separate line (`verify_commands=resolved`/`unresolved`) rather than duplicating the base class's raw-attempt line. `Router._dispatch` (`router/base.py`) logs `command=... backend= result=...` per backend tried, independent of the above. See `docs/bluetooth_vehicles.md`'s "Troubleshooting: Enable Debug Logging" section for the user-facing format; `tests/test_command_logging.py` locks in the exact line shapes. ## Maintaining this file diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 01c6fb6..d4f9ffb 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -398,3 +398,32 @@ enough to prove either failure or success. `remote_boombox(sound)` also uses the INFO-domain signed-command transport and plays through the vehicle external speaker. Use it only when someone is present to confirm the sound is appropriate for the vehicle's surroundings. + +## Troubleshooting: Enable Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to see, for every command, which +transport served it and how it ended: + +```python +import logging + +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + +Each command logs one terse, grep-friendly line: + +``` +command=RKE_ACTION_LOCK transport=bluetooth result=True reason= +command=set_charge_limit transport=teslemetry result=success +command=mediaPlayAction transport=bluetooth result=error error=BluetoothTimeout: Bluetooth command timed out waiting for vehicle response. +``` + +`transport` is `bluetooth`, `fleet`, `teslemetry`, or `tessie`. For BLE signed +commands, `command` is the underlying VCSEC/infotainment field name (e.g. +`RKE_ACTION_LOCK`, `chargingSetLimitAction`), not the Python method name; for +REST commands it is the endpoint's final path segment (e.g. `set_charge_limit`). +For BLE commands run with `verify_commands=True`, a resolved timeout logs a +second line with `verify_commands=resolved` and the confirmed result. `Router` +additionally logs `command=... backend= result=...` for each +backend it tries, so a BLE-primary/cloud-fallback setup shows exactly which +backend served each call and why a failover happened. diff --git a/tesla_fleet_api/router/base.py b/tesla_fleet_api/router/base.py index 993bdab..27196fc 100644 --- a/tesla_fleet_api/router/base.py +++ b/tesla_fleet_api/router/base.py @@ -148,15 +148,21 @@ async def _routed(*args: Any, **kwargs: Any) -> Any: last_exc: BaseException | None = None for backend, attr in targets[start:]: try: - return await _maybe_await(attr(*args, **kwargs)) + result = await _maybe_await(attr(*args, **kwargs)) except (Exception, TeslaFleetError) as e: # noqa: BLE001 - any failure -> next backend last_exc = e LOGGER.debug( - "Backend %s call %r failed, routing to next backend: %s", - type(backend).__name__, + "command=%s backend=%s result=error error=%s: %s", name, + type(backend).__name__, + type(e).__name__, e, ) + continue + LOGGER.debug( + "command=%s backend=%s result=success", name, type(backend).__name__ + ) + return result # The loop always runs at least once (``start`` only advances past # the primary when a later backend remains), so a failure here means # every applicable backend raised. diff --git a/tesla_fleet_api/tesla/fleet.py b/tesla_fleet_api/tesla/fleet.py index 4d1b1d3..6a82e77 100644 --- a/tesla_fleet_api/tesla/fleet.py +++ b/tesla_fleet_api/tesla/fleet.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Callable from json import dumps -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING, Any, ClassVar import aiohttp @@ -13,6 +13,7 @@ LibraryError, MissingToken, ResponseError, + TeslaFleetError, raise_for_status, ) from tesla_fleet_api.tesla.tesla import Tesla @@ -45,6 +46,8 @@ class TeslaFleetApi(Tesla): user: "User" partner: "Partner" vehicles: "Vehicles[TeslaFleetApi]" + # Transport identity for debug logging; Teslemetry/Tessie override this. + _transport_name: ClassVar[str] = "fleet" def __init__( self, @@ -119,57 +122,79 @@ async def _request( ) -> dict[str, Any]: """Send a request to the Tesla Fleet API.""" - if not self.server: - raise ValueError("Server was not set at init. Call find_server() first.") + # Trailing path segment (e.g. "door_lock", "vehicle_data") as the + # debug-log command name; the full path (with VIN) is already logged + # below via resp.url, so this adds no new exposure. + command = path.rsplit("/", 1)[-1] - if method == Method.GET: - json = None + try: + if not self.server: + raise ValueError( + "Server was not set at init. Call find_server() first." + ) - access_token = await self.access_token() + if method == Method.GET: + json = None - headers = { - "Authorization": f"Bearer {access_token}", - "X-Library": f"python tesla_fleet_api {VERSION}", - } + access_token = await self.access_token() - # Remove None values from params and json - if params: - params = { - k: _normalize_query_value(v) for k, v in params.items() if v is not None + headers = { + "Authorization": f"Bearer {access_token}", + "X-Library": f"python tesla_fleet_api {VERSION}", } - LOGGER.debug("Parameters: %s", params) - if json: - json = {k: v for k, v in json.items() if v is not None} - LOGGER.debug("Body: %s", dumps(json)) - headers["Content-Type"] = "application/json" - - async with self.session.request( - method, - f"{self.server}/{path}", - headers=headers, - json=json, - params=params, - ) as resp: - LOGGER.debug("Status %s from %s", resp.status, resp.url) - if "x-txid" in resp.headers: - LOGGER.debug("Response TXID: %s", resp.headers["x-txid"]) - if "RateLimit-Reset" in resp.headers: - LOGGER.debug( - "Rate limit reset: %s", resp.headers.get("RateLimit-Reset") - ) - if "Retry-After" in resp.headers: - LOGGER.debug("Retry after: %s", resp.headers.get("Retry-After")) - if not resp.ok: - await raise_for_status(resp) - - if not resp.content_type.lower().startswith("application/json"): - LOGGER.debug("Response type is: %s", resp.content_type) - raise ResponseError(status=resp.status, data=await resp.text()) - - data = await resp.json() - LOGGER.debug("Response JSON: %s", data) - return data + # Remove None values from params and json + if params: + params = { + k: _normalize_query_value(v) + for k, v in params.items() + if v is not None + } + LOGGER.debug("Parameters: %s", params) + if json: + json = {k: v for k, v in json.items() if v is not None} + LOGGER.debug("Body: %s", dumps(json)) + headers["Content-Type"] = "application/json" + + async with self.session.request( + method, + f"{self.server}/{path}", + headers=headers, + json=json, + params=params, + ) as resp: + LOGGER.debug("Status %s from %s", resp.status, resp.url) + if "x-txid" in resp.headers: + LOGGER.debug("Response TXID: %s", resp.headers["x-txid"]) + if "RateLimit-Reset" in resp.headers: + LOGGER.debug( + "Rate limit reset: %s", resp.headers.get("RateLimit-Reset") + ) + if "Retry-After" in resp.headers: + LOGGER.debug("Retry after: %s", resp.headers.get("Retry-After")) + + if not resp.ok: + await raise_for_status(resp) + + if not resp.content_type.lower().startswith("application/json"): + LOGGER.debug("Response type is: %s", resp.content_type) + raise ResponseError(status=resp.status, data=await resp.text()) + + data = await resp.json() + LOGGER.debug("Response JSON: %s", data) + except (Exception, TeslaFleetError) as e: + LOGGER.debug( + "command=%s transport=%s result=error error=%s: %s", + command, + self._transport_name, + type(e).__name__, + e, + ) + raise + LOGGER.debug( + "command=%s transport=%s result=success", command, self._transport_name + ) + return data async def status(self) -> str: """This endpoint returns the string "ok" if the API is operating normally. No HTTP headers are required.""" diff --git a/tesla_fleet_api/tesla/vehicle/bluetooth.py b/tesla_fleet_api/tesla/vehicle/bluetooth.py index c7a2b65..e9100ab 100644 --- a/tesla_fleet_api/tesla/vehicle/bluetooth.py +++ b/tesla_fleet_api/tesla/vehicle/bluetooth.py @@ -23,7 +23,11 @@ TeslaFleetError, WhitelistOperationStatus, ) -from tesla_fleet_api.tesla.vehicle.commands import Commands +from tesla_fleet_api.tesla.vehicle.commands import ( + Commands, + infotainment_command_name, + vcsec_command_name, +) # Protocol from tesla_fleet_api.tesla.vehicle.proto.car_server_pb2 import ( @@ -320,6 +324,7 @@ class VehicleBluetooth(Commands[BluetoothParentT], Generic[BluetoothParentT]): _ekey: ec.EllipticCurvePublicKey _buffer: ReassemblingBuffer _auth_method = "aes" + _transport_name = "bluetooth" _ack_followup_timeout: float = 2 _default_timeout: float = 5 # A lost actuation ack is inconclusive; the contract is verify-by-state, so @@ -750,7 +755,25 @@ async def _sendVehicleSecurity(self, command: UnsignedMessage) -> dict[str, Any] try: return await super()._sendVehicleSecurity(command) except BluetoothTimeout as timeout: - return await self._resolve_timeout(_vcsec_verify_plan(command), timeout) + name = vcsec_command_name(command) + try: + result = await self._resolve_timeout( + _vcsec_verify_plan(command), timeout + ) + except BluetoothTimeout: + LOGGER.debug( + "command=%s transport=%s verify_commands=unresolved", + name, + self._transport_name, + ) + raise + LOGGER.debug( + "command=%s transport=%s verify_commands=resolved result=%s", + name, + self._transport_name, + result.get("response", {}).get("result"), + ) + return result async def _sendInfotainment(self, command: Action) -> dict[str, Any]: """Send an infotainment command, optionally resolving a lost-ack timeout by state.""" @@ -759,11 +782,27 @@ async def _sendInfotainment(self, command: Action) -> dict[str, Any]: try: return await super()._sendInfotainment(command) except BluetoothTimeout as timeout: + name = infotainment_command_name(command) resolver = _INFOTAINMENT_VERIFY_PLANS.get( command.vehicleAction.WhichOneof("vehicle_action_msg") ) plan = resolver(command.vehicleAction) if resolver else None - return await self._resolve_timeout(plan, timeout) + try: + result = await self._resolve_timeout(plan, timeout) + except BluetoothTimeout: + LOGGER.debug( + "command=%s transport=%s verify_commands=unresolved", + name, + self._transport_name, + ) + raise + LOGGER.debug( + "command=%s transport=%s verify_commands=resolved result=%s", + name, + self._transport_name, + result.get("response", {}).get("result"), + ) + return result async def _resolve_timeout( self, plan: VerifyPlan | None, timeout: BluetoothTimeout diff --git a/tesla_fleet_api/tesla/vehicle/commands.py b/tesla_fleet_api/tesla/vehicle/commands.py index 6c73080..aa2d624 100644 --- a/tesla_fleet_api/tesla/vehicle/commands.py +++ b/tesla_fleet_api/tesla/vehicle/commands.py @@ -17,6 +17,7 @@ MESSAGE_FAULTS, SIGNED_MESSAGE_INFORMATION_FAULTS, NotOnWhitelistFault, + TeslaFleetError, # TeslaFleetMessageFaultInvalidSignature, TeslaFleetMessageFaultIncorrectEpoch, TeslaFleetMessageFaultInvalidTokenOrCounter, @@ -227,6 +228,46 @@ ) +def vcsec_command_name(command: UnsignedMessage) -> str: + """Derive a short debug-log command name from a VCSEC UnsignedMessage's populated field.""" + field = command.WhichOneof("sub_message") + if field == "RKEAction": + return RKEAction_E.Name(command.RKEAction) + return field or "unknown" + + +def infotainment_command_name(command: Action) -> str: + """Derive a short debug-log command name from an infotainment Action's populated field.""" + return command.vehicleAction.WhichOneof("vehicle_action_msg") or "unknown" + + +def _log_command_result(name: str, transport: str, result: dict[str, Any]) -> None: + """Log the terminal outcome of one signed command at debug level.""" + response = result.get("response") + if isinstance(response, dict): + response_dict = cast("dict[str, Any]", response) + LOGGER.debug( + "command=%s transport=%s result=%s reason=%s", + name, + transport, + response_dict.get("result"), + response_dict.get("reason"), + ) + else: + LOGGER.debug("command=%s transport=%s result=success", name, transport) + + +def _log_command_error(name: str, transport: str, exc: BaseException) -> None: + """Log a signed command failure at debug level; the exception type carries the ack-vs-timeout distinction.""" + LOGGER.debug( + "command=%s transport=%s result=error error=%s: %s", + name, + transport, + type(exc).__name__, + exc, + ) + + class Session(Generic[CommandParentT]): """A connect to a domain""" @@ -293,6 +334,8 @@ class Commands(ABC, Vehicle[CommandParentT], Generic[CommandParentT]): _from_destination: bytes _sessions: dict[int, Session[CommandParentT]] _auth_method: ClassVar[Literal["hmac", "aes"]] + # Transport identity for debug logging; set per concrete subclass. + _transport_name: ClassVar[str] def __init__( self, @@ -679,30 +722,56 @@ async def _sendVehicleSecurity(self, command: UnsignedMessage) -> dict[str, Any] A VCSEC actuation replies with a single terminal ack and no data frame, so ``expects_data=False`` lets ``_send`` return on that ack. """ - return await self._command( - Domain.DOMAIN_VEHICLE_SECURITY, - command.SerializeToString(), - expects_data=False, - ) + name = vcsec_command_name(command) + try: + result = await self._command( + Domain.DOMAIN_VEHICLE_SECURITY, + command.SerializeToString(), + expects_data=False, + ) + except (Exception, TeslaFleetError) as e: + _log_command_error(name, self._transport_name, e) + raise + _log_command_result(name, self._transport_name, result) + return result async def _getVehicleSecurity(self, command: UnsignedMessage) -> VehicleStatus: """Sign and send a read request to the Vehicle Security computer.""" - reply = await self._command( - Domain.DOMAIN_VEHICLE_SECURITY, command.SerializeToString() - ) + name = vcsec_command_name(command) + try: + reply = await self._command( + Domain.DOMAIN_VEHICLE_SECURITY, command.SerializeToString() + ) + except (Exception, TeslaFleetError) as e: + _log_command_error(name, self._transport_name, e) + raise + _log_command_result(name, self._transport_name, reply) return reply["response"] async def _sendInfotainment(self, command: Action) -> dict[str, Any]: """Sign and send a message to Infotainment computer.""" - return await self._command( - Domain.DOMAIN_INFOTAINMENT, command.SerializeToString() - ) + name = infotainment_command_name(command) + try: + result = await self._command( + Domain.DOMAIN_INFOTAINMENT, command.SerializeToString() + ) + except (Exception, TeslaFleetError) as e: + _log_command_error(name, self._transport_name, e) + raise + _log_command_result(name, self._transport_name, result) + return result async def _getInfotainment(self, command: Action) -> VehicleData: """Sign and send a message to Infotainment computer.""" - reply = await self._command( - Domain.DOMAIN_INFOTAINMENT, command.SerializeToString() - ) + name = infotainment_command_name(command) + try: + reply = await self._command( + Domain.DOMAIN_INFOTAINMENT, command.SerializeToString() + ) + except (Exception, TeslaFleetError) as e: + _log_command_error(name, self._transport_name, e) + raise + _log_command_result(name, self._transport_name, reply) return reply["response"] async def handshakeVehicleSecurity(self) -> None: diff --git a/tesla_fleet_api/tesla/vehicle/signed.py b/tesla_fleet_api/tesla/vehicle/signed.py index 8092fdf..e58338a 100644 --- a/tesla_fleet_api/tesla/vehicle/signed.py +++ b/tesla_fleet_api/tesla/vehicle/signed.py @@ -21,6 +21,7 @@ class VehicleSigned( # pyright: ignore[reportIncompatibleMethodOverride] """Class describing the Tesla Fleet API vehicle endpoints and commands for a specific vehicle with command signing.""" _auth_method = "hmac" + _transport_name = "fleet" def __init__(self, parent: SignedParentT, vin: str): """Initialize the VehicleSigned class.""" diff --git a/tesla_fleet_api/teslemetry/teslemetry.py b/tesla_fleet_api/teslemetry/teslemetry.py index 97c541b..e72d10a 100644 --- a/tesla_fleet_api/teslemetry/teslemetry.py +++ b/tesla_fleet_api/teslemetry/teslemetry.py @@ -14,6 +14,7 @@ class Teslemetry(TeslaFleetApi): vehicles: TeslemetryVehicles Vehicles = TeslemetryVehicles EnergySites = TeslemetryEnergySites + _transport_name = "teslemetry" def __init__( self, diff --git a/tesla_fleet_api/tessie/tessie.py b/tesla_fleet_api/tessie/tessie.py index 01a17ca..f33e2f2 100644 --- a/tesla_fleet_api/tessie/tessie.py +++ b/tesla_fleet_api/tessie/tessie.py @@ -11,6 +11,7 @@ class Tessie(TeslaFleetApi): server = "https://api.tessie.com" vehicles: TessieVehicles Vehicles = TessieVehicles + _transport_name = "tessie" def __init__( self, diff --git a/tests/test_command_logging.py b/tests/test_command_logging.py new file mode 100644 index 0000000..6ae3827 --- /dev/null +++ b/tests/test_command_logging.py @@ -0,0 +1,251 @@ +"""Tests for the DEBUG-level command/transport/outcome logging added for +troubleshooting whether a command went over Bluetooth, Fleet API, Teslemetry, +or Tessie, and why it failed. + +Covers the three logging choke points: ``Commands._sendVehicleSecurity`` / +``_sendInfotainment`` (BLE and Fleet-signed commands), ``TeslaFleetApi._request`` +(Fleet/Teslemetry/Tessie REST commands), and ``Router._dispatch`` (backend +selection and failover). +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from unittest import IsolatedAsyncioTestCase +from unittest.mock import AsyncMock, MagicMock + +from tesla_fleet_api.const import Method +from tesla_fleet_api.exceptions import BluetoothTimeout, NotFound +from tesla_fleet_api.router import Router +from tesla_fleet_api.tesla.fleet import TeslaFleetApi +from tesla_fleet_api.teslemetry.teslemetry import Teslemetry +from tesla_fleet_api.tessie.tessie import Tessie + +from ble_mocked_transport import MockedBleTransportTestCase, vcsec_ok_reply + +LOGGER_NAME = "tesla_fleet_api" + + +class BleCommandLoggingTests(MockedBleTransportTestCase): + """BLE commands log their command name, ``transport=bluetooth``, and outcome.""" + + async def test_success_logs_command_transport_and_result(self) -> None: + vehicle, send = self.make_vehicle() + send.return_value = vcsec_ok_reply() + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await vehicle.door_lock() + + self.assertTrue( + any( + "command=RKE_ACTION_LOCK" in line + and "transport=bluetooth" in line + and "result=True" in line + for line in captured.output + ), + captured.output, + ) + + async def test_timeout_logs_error_with_exception_type(self) -> None: + vehicle, send = self.make_vehicle() + send.side_effect = BluetoothTimeout() + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + with self.assertRaises(BluetoothTimeout): + await vehicle.door_lock() + + self.assertTrue( + any( + "command=RKE_ACTION_LOCK" in line + and "transport=bluetooth" in line + and "result=error" in line + and "BluetoothTimeout" in line + for line in captured.output + ), + captured.output, + ) + + async def test_verify_commands_resolution_logs_verified_outcome(self) -> None: + from tesla_fleet_api.tesla.vehicle.proto.vcsec_pb2 import ( + VehicleLockState_E, + VehicleStatus, + ) + + from ble_mocked_transport import vcsec_vehicle_status_reply + + vehicle, send = self.make_vehicle(verify_commands=True) + send.side_effect = [ + BluetoothTimeout(), + vcsec_vehicle_status_reply( + VehicleStatus( + vehicleLockState=VehicleLockState_E.VEHICLELOCKSTATE_LOCKED + ) + ), + ] + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + result = await vehicle.door_lock() + + self.assertEqual(result, {"response": {"result": True, "reason": ""}}) + self.assertTrue( + any( + "command=RKE_ACTION_LOCK" in line + and "transport=bluetooth" in line + and "verify_commands=resolved" in line + for line in captured.output + ), + captured.output, + ) + + +def _fake_response( + *, + status: int = 200, + ok: bool = True, + content_type: str = "application/json", + json_body: object = None, +): + resp = MagicMock() + resp.status = status + resp.ok = ok + resp.content_type = content_type + resp.url = "https://example.com/x" + resp.headers = {} + resp.json = AsyncMock(return_value=json_body if json_body is not None else {}) + resp.text = AsyncMock(return_value="") + return resp + + +def _make_session(response: object) -> MagicMock: + session = MagicMock() + + @asynccontextmanager + async def _ctx(*args, **kwargs): + yield response + + session.request = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k)) + return session + + +class RestCommandLoggingTests(IsolatedAsyncioTestCase): + """REST commands (Fleet/Teslemetry/Tessie) log endpoint name, transport, outcome.""" + + async def test_fleet_success_logs_command_transport_and_result(self) -> None: + resp = _fake_response(json_body={"response": {"result": True}}) + api = TeslaFleetApi( + session=_make_session(resp), + access_token="token", + server="https://fleet.example.com", + ) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request(Method.POST, "api/1/vehicles/VIN123/command/door_lock") + + self.assertTrue( + any( + "command=door_lock" in line + and "transport=fleet" in line + and "result=success" in line + for line in captured.output + ), + captured.output, + ) + + async def test_fleet_failure_logs_command_transport_and_error(self) -> None: + resp = _fake_response(status=404, ok=False, json_body={"error": "not found"}) + api = TeslaFleetApi( + session=_make_session(resp), + access_token="token", + server="https://fleet.example.com", + ) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + with self.assertRaises(NotFound): + await api._request( + Method.POST, "api/1/vehicles/VIN123/command/door_lock" + ) + + self.assertTrue( + any( + "command=door_lock" in line + and "transport=fleet" in line + and "result=error" in line + and "NotFound" in line + for line in captured.output + ), + captured.output, + ) + + async def test_teslemetry_success_logs_transport_teslemetry(self) -> None: + resp = _fake_response(json_body={"response": {"result": True}}) + api = Teslemetry(session=_make_session(resp), access_token="token") + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request(Method.POST, "api/1/vehicles/VIN123/command/door_lock") + + self.assertTrue( + any("transport=teslemetry" in line for line in captured.output), + captured.output, + ) + + async def test_tessie_success_logs_transport_tessie(self) -> None: + resp = _fake_response(json_body={"response": {"result": True}}) + api = Tessie(session=_make_session(resp), access_token="token") + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request(Method.POST, "vehicles/VIN123/command/door_lock") + + self.assertTrue( + any("transport=tessie" in line for line in captured.output), + captured.output, + ) + + +class _FakePrimary: + def __init__(self, *, fail: bool = False): + self.fail = fail + + async def shared(self, value: int) -> str: + if self.fail: + raise ConnectionError("primary failed") + return f"primary:{value}" + + +class _FakeFallback: + async def shared(self, value: int) -> str: + return f"fallback:{value}" + + +class RouterCommandLoggingTests(IsolatedAsyncioTestCase): + """Router logs which backend served a call, and any failover hop.""" + + async def test_primary_success_logs_backend_and_result(self) -> None: + router = Router(_FakePrimary(), _FakeFallback()) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + result = await router.shared(1) + + self.assertEqual(result, "primary:1") + self.assertTrue( + any( + "command=shared" in line + and "backend=_FakePrimary" in line + and "result=success" in line + for line in captured.output + ), + captured.output, + ) + + async def test_primary_failure_fails_over_and_logs_both_hops(self) -> None: + router = Router(_FakePrimary(fail=True), _FakeFallback()) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + result = await router.shared(1) + + self.assertEqual(result, "fallback:1") + joined = "\n".join(captured.output) + self.assertIn("backend=_FakePrimary", joined) + self.assertIn("result=error", joined) + self.assertIn("ConnectionError", joined) + self.assertIn("backend=_FakeFallback", joined) + self.assertIn("result=success", joined) diff --git a/uv.lock b/uv.lock index 90e1eb5..25a5d28 100644 --- a/uv.lock +++ b/uv.lock @@ -728,7 +728,7 @@ wheels = [ [[package]] name = "tesla-fleet-api" -version = "1.6.2" +version = "1.6.3" source = { editable = "." } dependencies = [ { name = "aiofiles" }, From afe50e4df6d431bf83a68bef4bb4e1e268e2632a Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sat, 11 Jul 2026 10:50:34 +1000 Subject: [PATCH 2/4] no-mistakes(review): Log REST command body outcomes --- docs/bluetooth_vehicles.md | 2 +- tesla_fleet_api/tesla/fleet.py | 21 ++++++++++--- tests/test_command_logging.py | 56 +++++++++++++++++++++++++++++++++- 3 files changed, 73 insertions(+), 6 deletions(-) diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index d4f9ffb..339d0df 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -414,7 +414,7 @@ Each command logs one terse, grep-friendly line: ``` command=RKE_ACTION_LOCK transport=bluetooth result=True reason= -command=set_charge_limit transport=teslemetry result=success +command=set_charge_limit transport=teslemetry result=True reason= command=mediaPlayAction transport=bluetooth result=error error=BluetoothTimeout: Bluetooth command timed out waiting for vehicle response. ``` diff --git a/tesla_fleet_api/tesla/fleet.py b/tesla_fleet_api/tesla/fleet.py index 6a82e77..822abac 100644 --- a/tesla_fleet_api/tesla/fleet.py +++ b/tesla_fleet_api/tesla/fleet.py @@ -2,7 +2,7 @@ from collections.abc import Awaitable, Callable from json import dumps -from typing import TYPE_CHECKING, Any, ClassVar +from typing import TYPE_CHECKING, Any, ClassVar, cast import aiohttp @@ -33,6 +33,21 @@ def _normalize_query_value(value: Any) -> Any: return value +def _log_request_result(command: str, transport: str, data: dict[str, Any]) -> None: + response = data.get("response") + if isinstance(response, dict) and "result" in response: + response_dict = cast("dict[str, Any]", response) + LOGGER.debug( + "command=%s transport=%s result=%s reason=%s", + command, + transport, + response_dict.get("result"), + response_dict.get("reason"), + ) + else: + LOGGER.debug("command=%s transport=%s result=success", command, transport) + + # Based on https://developer.tesla.com/docs/fleet-api class TeslaFleetApi(Tesla): """Class describing the Tesla Fleet API.""" @@ -191,9 +206,7 @@ async def _request( e, ) raise - LOGGER.debug( - "command=%s transport=%s result=success", command, self._transport_name - ) + _log_request_result(command, self._transport_name, data) return data async def status(self) -> str: diff --git a/tests/test_command_logging.py b/tests/test_command_logging.py index 6ae3827..541c93b 100644 --- a/tests/test_command_logging.py +++ b/tests/test_command_logging.py @@ -131,7 +131,7 @@ class RestCommandLoggingTests(IsolatedAsyncioTestCase): """REST commands (Fleet/Teslemetry/Tessie) log endpoint name, transport, outcome.""" async def test_fleet_success_logs_command_transport_and_result(self) -> None: - resp = _fake_response(json_body={"response": {"result": True}}) + resp = _fake_response(json_body={"response": {"result": True, "reason": ""}}) api = TeslaFleetApi( session=_make_session(resp), access_token="token", @@ -145,6 +145,60 @@ async def test_fleet_success_logs_command_transport_and_result(self) -> None: any( "command=door_lock" in line and "transport=fleet" in line + and "result=True" in line + and "reason=" in line + for line in captured.output + ), + captured.output, + ) + + async def test_fleet_command_rejection_logs_response_result(self) -> None: + resp = _fake_response( + json_body={ + "response": { + "result": False, + "reason": "cabin comfort remote settings not enabled", + } + } + ) + api = TeslaFleetApi( + session=_make_session(resp), + access_token="token", + server="https://fleet.example.com", + ) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request( + Method.POST, + "api/1/vehicles/VIN123/command/remote_seat_heater_request", + ) + + self.assertTrue( + any( + "command=remote_seat_heater_request" in line + and "transport=fleet" in line + and "result=False" in line + and "reason=cabin comfort remote settings not enabled" in line + for line in captured.output + ), + captured.output, + ) + + async def test_fleet_non_command_json_logs_transport_success(self) -> None: + resp = _fake_response(json_body={"response": {"vehicle_id": 123}}) + api = TeslaFleetApi( + session=_make_session(resp), + access_token="token", + server="https://fleet.example.com", + ) + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request(Method.GET, "api/1/vehicles/VIN123/vehicle_data") + + self.assertTrue( + any( + "command=vehicle_data" in line + and "transport=fleet" in line and "result=success" in line for line in captured.output ), From 64a43b3c76f4790c215e2a0b810e0d2665191bf4 Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sat, 11 Jul 2026 10:53:59 +1000 Subject: [PATCH 3/4] no-mistakes(review): Log Tessie command outcomes correctly --- tesla_fleet_api/tesla/fleet.py | 20 ++++++++++++-------- tests/test_command_logging.py | 20 +++++++++++++++++++- 2 files changed, 31 insertions(+), 9 deletions(-) diff --git a/tesla_fleet_api/tesla/fleet.py b/tesla_fleet_api/tesla/fleet.py index 822abac..51f2987 100644 --- a/tesla_fleet_api/tesla/fleet.py +++ b/tesla_fleet_api/tesla/fleet.py @@ -36,16 +36,20 @@ def _normalize_query_value(value: Any) -> Any: def _log_request_result(command: str, transport: str, data: dict[str, Any]) -> None: response = data.get("response") if isinstance(response, dict) and "result" in response: - response_dict = cast("dict[str, Any]", response) - LOGGER.debug( - "command=%s transport=%s result=%s reason=%s", - command, - transport, - response_dict.get("result"), - response_dict.get("reason"), - ) + result_data = cast("dict[str, Any]", response) + elif "result" in data: + result_data = data else: LOGGER.debug("command=%s transport=%s result=success", command, transport) + return + + LOGGER.debug( + "command=%s transport=%s result=%s reason=%s", + command, + transport, + result_data.get("result"), + result_data.get("reason"), + ) # Based on https://developer.tesla.com/docs/fleet-api diff --git a/tests/test_command_logging.py b/tests/test_command_logging.py index 541c93b..3edf36f 100644 --- a/tests/test_command_logging.py +++ b/tests/test_command_logging.py @@ -243,7 +243,7 @@ async def test_teslemetry_success_logs_transport_teslemetry(self) -> None: ) async def test_tessie_success_logs_transport_tessie(self) -> None: - resp = _fake_response(json_body={"response": {"result": True}}) + resp = _fake_response(json_body={"result": True}) api = Tessie(session=_make_session(resp), access_token="token") with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: @@ -254,6 +254,24 @@ async def test_tessie_success_logs_transport_tessie(self) -> None: captured.output, ) + async def test_tessie_command_rejection_logs_top_level_result(self) -> None: + resp = _fake_response(json_body={"result": False, "reason": "already_locked"}) + api = Tessie(session=_make_session(resp), access_token="token") + + with self.assertLogs(LOGGER_NAME, level="DEBUG") as captured: + await api._request(Method.POST, "vehicles/VIN123/command/door_lock") + + self.assertTrue( + any( + "command=door_lock" in line + and "transport=tessie" in line + and "result=False" in line + and "reason=already_locked" in line + for line in captured.output + ), + captured.output, + ) + class _FakePrimary: def __init__(self, *, fail: bool = False): From b91c1b3ba6f8b36c4d0efb4157cb31fec5761e8d Mon Sep 17 00:00:00 2001 From: firstmate crewmate Date: Sat, 11 Jul 2026 10:59:43 +1000 Subject: [PATCH 4/4] no-mistakes(document): Document command debug logging --- README.md | 20 ++++++++++++++++++++ docs/bluetooth_vehicles.md | 1 + docs/fleet_api_energy_sites.md | 12 ++++++++++++ docs/fleet_api_signed_commands.md | 16 ++++++++++++++++ docs/fleet_api_vehicles.md | 19 +++++++++++++++++++ docs/teslemetry.md | 12 ++++++++++++ docs/tessie.md | 12 ++++++++++++ tesla_fleet_api/router/base.py | 2 ++ 8 files changed, 94 insertions(+) diff --git a/README.md b/README.md index 0c59639..c38e08f 100644 --- a/README.md +++ b/README.md @@ -226,10 +226,30 @@ await router.set_operation(...) # local first, cloud on failure `Router`, `VehicleRouter`, and `EnergySiteRouter` are all importable from `tesla_fleet_api.router` (and, for backward compatibility, from `tesla_fleet_api.tesla`). +Enable `DEBUG` logging for `tesla_fleet_api` to see which backend served a routed call and why failover happened. + > **Warning:** Because a failed call is replayed on the next backend, a non-idempotent command (e.g. `honk_horn`, `actuate_trunk`, `door_unlock`, `charge_start`) that fails _mid-flight_ — after a backend may have already partially applied it — can be **double-executed** (or executed more than once across a longer chain) when it is retried on the next backend. This is a deliberate tradeoff of per-command failover. When the primary is `VehicleBluetooth`, pass `verify_commands=True` to resolve supported mutating command timeouts by state before failover; callers needing exactly-once semantics for other commands should gate dispatch with an explicit `health` check or call the underlying backends directly. > > Dispatch is implemented via `__getattr__`, which does not proxy dunder methods, so `async with Router(...)` does **not** manage a backend's BLE connection lifecycle (`__aenter__`/`__aexit__`). Commands still auto-connect on send; for explicit connect/disconnect reach through `router.primary` (or `router.backends`). +### Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to see each command's command +name, transport/backend, and result. In standalone scripts, configure a handler +first: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + +Command log lines use `transport=bluetooth`, `fleet`, `teslemetry`, or `tessie`. +Routers also emit `backend=` lines for each backend tried. See +[Bluetooth for Vehicles](docs/bluetooth_vehicles.md#troubleshooting-enable-debug-logging) +for examples and the signed-command naming details. + ### Teslemetry The `Teslemetry` class provides methods to interact with the Teslemetry service. Here's a basic example: diff --git a/docs/bluetooth_vehicles.md b/docs/bluetooth_vehicles.md index 339d0df..5803c00 100644 --- a/docs/bluetooth_vehicles.md +++ b/docs/bluetooth_vehicles.md @@ -407,6 +407,7 @@ transport served it and how it ended: ```python import logging +logging.basicConfig(level=logging.DEBUG) logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) ``` diff --git a/docs/fleet_api_energy_sites.md b/docs/fleet_api_energy_sites.md index 126028f..593f207 100644 --- a/docs/fleet_api_energy_sites.md +++ b/docs/fleet_api_energy_sites.md @@ -30,6 +30,18 @@ async def main(): asyncio.run(main()) ``` +## Troubleshooting: Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to log each Fleet API request's +final path segment, `transport=fleet`, and result: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + ## Backup Reserve You can adjust the backup reserve for a specific energy site using its ID: diff --git a/docs/fleet_api_signed_commands.md b/docs/fleet_api_signed_commands.md index dafcc61..15ce55a 100644 --- a/docs/fleet_api_signed_commands.md +++ b/docs/fleet_api_signed_commands.md @@ -190,6 +190,22 @@ state after the call instead of relying on the number of send attempts. `adjust_volume(volume)` accepts absolute volume values from `0.0` through `11.0`, matching the Fleet API command validation. +## Troubleshooting: Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to log each signed command's +protobuf command name, `transport=fleet`, and result: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + +For signed commands, `command` is the underlying VCSEC or infotainment field +name (for example `RKE_ACTION_LOCK` or `chargingSetLimitAction`), not the Python +method name. + ## Flash Lights You can flash the lights of a specific vehicle using its VIN: diff --git a/docs/fleet_api_vehicles.md b/docs/fleet_api_vehicles.md index 66915a7..39fdfc5 100644 --- a/docs/fleet_api_vehicles.md +++ b/docs/fleet_api_vehicles.md @@ -29,6 +29,25 @@ async def main(): asyncio.run(main()) ``` +## Troubleshooting: Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to log each Fleet API request's +final path segment, `transport=fleet`, and result: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + +Example command lines: + +``` +command=vehicle_data transport=fleet result=success +command=set_charge_limit transport=fleet result=True reason= +``` + ## Create a Vehicle You can create a `VehicleFleet` instance for a specific vehicle using its VIN: diff --git a/docs/teslemetry.md b/docs/teslemetry.md index a4a3b65..0e5b109 100644 --- a/docs/teslemetry.md +++ b/docs/teslemetry.md @@ -21,6 +21,18 @@ async def main(): asyncio.run(main()) ``` +## Troubleshooting: Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to log each Teslemetry request's +final path segment, `transport=teslemetry`, and result: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + ## Ping The `ping` method sends a ping request to the Teslemetry server. diff --git a/docs/tessie.md b/docs/tessie.md index 914505c..c4b463d 100644 --- a/docs/tessie.md +++ b/docs/tessie.md @@ -28,6 +28,18 @@ async def main(): asyncio.run(main()) ``` +## Troubleshooting: Debug Logging + +Enable the `tesla_fleet_api` logger at `DEBUG` to log each Tessie request's +final path segment, `transport=tessie`, and result: + +```python +import logging + +logging.basicConfig(level=logging.DEBUG) +logging.getLogger("tesla_fleet_api").setLevel(logging.DEBUG) +``` + ## Top-Level Client Methods These methods exist on `Tessie` itself. diff --git a/tesla_fleet_api/router/base.py b/tesla_fleet_api/router/base.py index 27196fc..0ddd7e4 100644 --- a/tesla_fleet_api/router/base.py +++ b/tesla_fleet_api/router/base.py @@ -54,6 +54,8 @@ class Router(Generic[PrimaryT, SecondaryT]): failure or a disconnect), the same call is automatically retried on the next backend that has it, with the same arguments. The error only propagates when every applicable backend fails, in which case the last error is raised. + Each attempted backend emits a ``DEBUG`` log line with the routed command + name, backend class, and success/error result. .. warning::