feat(teslemetry): add typed authorized clients accessor

AGENTS.md

@@ -142,6 +142,7 @@ Source protobuf definitions live in `proto/`; generated Python files live in `te
 - **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=<name> transport=<t> 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=<ClassName> 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.
 - **`_log_request_result` (`fleet.py`) must tolerate any JSON-legal REST body, not just dicts**: it runs after the HTTP request already succeeded, so it's a logging convenience only - a non-dict body (`null`, a list, a bare scalar; live case: Teslemetry's `list_authorized_clients` returning `null`) must never raise there. It guards with `isinstance(data, dict)` before calling `.get()`, logging `result=success` and returning for anything else. Regression tests in `tests/test_command_logging.py` (`test_null_json_body_returns_none_without_raising` etc.) cover null/list/scalar bodies.
+- **Typed accessor pattern for undocumented raw-dict responses**: `TeslemetryEnergySite.find_authorized_clients()` (`teslemetry/energysite.py`) is the library's first frozen-dataclass typed wrapper over a raw `dict[str, Any]`/`None`/`list` REST response, added so API-parsing logic (envelope unwrap, field lookup, null-body handling, enum typing) lives in the library instead of each consumer (e.g. Home Assistant's config flow) reimplementing it. Deliberately scoped tight rather than defensively broad, since no site has been observed with a populated client list to confirm the per-entry shape against: `_authorized_clients_list()` does one precise, non-recursive envelope unwrap (`{"response": {"authorized_clients": [...]}}`, or a bare list with no envelope) rather than a multi-key/depth-first search, and `AuthorizedClient` models only the two fields a pairing flow actually reads (`public_key`, `state`), each accepting only the specific key-name variant pairs empirically evidenced, not speculative extras. Two rules any future typed accessor over an undocumented response shape must keep, demonstrated here even though `AuthorizedClientState` (`const.py`) has no falsy member: (1) field lookup must check key presence (`key in payload`), never `payload.get(key) or default` - a legal falsy value is not "missing", and a present-but-unrecognized enum value (`_normalize_state()`) is returned raw rather than coerced to `None`; (2) a `None` body, an unrecognized shape, and an explicit empty list are NOT distinguished here - they all mean "no clients" for this endpoint and collapse to `AuthorizedClients.clients == []` (never `None`), a narrower policy than the general absent-vs-empty distinction principle because both known real inputs currently only produce "empty". Tesla has not published an OpenAPI schema for pairing endpoints, so `const.py`'s enums are the schema of record; a live sample with an actual paired client is a follow-up to confirm the populated-entry shape and may require widening `AuthorizedClient`. The raw `list_authorized_clients()` method is kept alongside as the untyped escape hatch. Tests: `tests/test_teslemetry_authorized_clients.py`.
 
 ## Maintaining this file
 

docs/teslemetry.md

@@ -540,6 +540,42 @@ async def main():
 asyncio.run(main())
 ```
 
+## Energy Site Authorized Clients
+
+Teslemetry energy sites support the same raw `list_authorized_clients` command
+as Fleet API energy sites, plus a typed `find_authorized_clients` helper for
+consumers that need to inspect the client list. The helper returns an
+`AuthorizedClients` result. Tesla has not published a schema for this pairing
+endpoint, so the typed helper only unwraps the one envelope shape and models
+the two client fields (`public_key`, `state`) confirmed by the endpoint's own
+known consumer; `clients` is always a list - a null response body, an
+unrecognized response shape, and an explicitly empty client list all mean "no
+authorized clients" and return `[]`. `state` is typed as
+`AuthorizedClientState`. The raw response is still available on `raw` for
+anything not modeled.
+
+```python
+async def main():
+    async with aiohttp.ClientSession() as session:
+        teslemetry = Teslemetry(
+            session=session,
+            access_token="<access_token>",
+        )
+
+        energy_site = teslemetry.energySites.create(12345)
+
+        result = await energy_site.find_authorized_clients()
+        for client in result.clients:
+            print(client.public_key, client.state)
+
+        # The untyped response is still available when callers need the exact
+        # Teslemetry payload.
+        raw = await energy_site.list_authorized_clients()
+        print(raw)
+
+asyncio.run(main())
+```
+
 ## Migrate to OAuth
 
 The `migrate_to_oauth` method migrates from an access token to OAuth.

tesla_fleet_api/teslemetry/energysite.py

@@ -1,16 +1,135 @@
 from __future__ import annotations
 
 import base64
-from typing import Any
+from dataclasses import dataclass
+from typing import Any, cast
 
 from tesla_fleet_api.const import (
     AuthorizedClientKeyType,
+    AuthorizedClientState,
     AuthorizedClientType,
     Method,
 )
 from tesla_fleet_api.tesla.energysite import EnergySite, EnergySites
 
 
+def _field(payload: dict[str, Any], *keys: str) -> Any:
+    """Return the first present key's value.
+
+    Checks key presence rather than truthiness, so a legal falsy value
+    (``0``, ``False``, ``""``) is never mistaken for a missing field.
+    """
+    for key in keys:
+        if key in payload:
+            return payload[key]
+    return None
+
+
+def _normalize_state(value: Any) -> AuthorizedClientState | int | str | None:
+    """Type a raw ``state`` value against ``AuthorizedClientState``.
+
+    Tesla has not published an OpenAPI schema for this pairing endpoint, so
+    ``const.py``'s enum is the schema of record. A recognized int or member
+    name (case-insensitive) becomes the enum member; a present-but-
+    unrecognized value is returned unchanged rather than dropped to
+    ``None``, since only a genuinely absent field means ``None``. ``bool``
+    is excluded from the int branch since it subclasses ``int`` but is
+    never a legal state code.
+    """
+    if (
+        value is None
+        or isinstance(value, AuthorizedClientState)
+        or isinstance(value, bool)
+    ):
+        return value
+    if isinstance(value, int):
+        try:
+            return AuthorizedClientState(value)
+        except ValueError:
+            return value
+    if isinstance(value, str):
+        try:
+            return AuthorizedClientState[value.strip().upper()]
+        except KeyError:
+            return value
+    return value
+
+
+def _parse_client(payload: dict[str, Any]) -> AuthorizedClient:
+    return AuthorizedClient(
+        public_key=_field(payload, "public_key", "publicKey"),
+        state=_normalize_state(_field(payload, "state", "authorized_client_state")),
+        raw=payload,
+    )
+
+
+@dataclass(frozen=True, slots=True)
+class AuthorizedClient:
+    """One entry from a Teslemetry ``list_authorized_clients`` response.
+
+    Only ``public_key`` and ``state`` are modeled - the two fields a
+    pairing flow needs to confirm a registered key. Tesla has not
+    published this response's schema, so anything else on an entry is
+    available via ``raw`` rather than guessed at. Each field accepts the
+    two key-name variants observed for it (``public_key``/``publicKey``,
+    ``state``/``authorized_client_state``).
+    """
+
+    public_key: str | None
+    state: AuthorizedClientState | int | str | None
+    raw: dict[str, Any]
+
+
+@dataclass(frozen=True, slots=True)
+class AuthorizedClients:
+    """Parsed result of :meth:`TeslemetryEnergySite.find_authorized_clients`.
+
+    ``clients`` is always a list: a ``None`` response body, an unrecognized
+    response shape, and an explicitly empty ``authorized_clients`` list all
+    mean "no authorized clients to report" for this endpoint and collapse
+    to ``[]`` rather than being distinguished.
+
+    The envelope this unwraps (``{"response": {"authorized_clients": [...]}}``)
+    is pinned from the pairing flow's own defensive handling of this
+    undocumented endpoint. No site has been observed with a populated
+    client list yet, so that per-entry shape is unconfirmed by a live
+    sample - see :class:`AuthorizedClient`.
+    """
+
+    clients: list[AuthorizedClient]
+    raw: Any
+
+
+def _authorized_clients_list(payload: Any) -> list[Any]:
+    """Return the raw authorized-clients list from a command response, or [].
+
+    A precise, single-path unwrap of the one confirmed envelope - not a
+    search across candidate wrapper keys for this undocumented endpoint.
+    """
+    if payload is None:
+        return []
+    if isinstance(payload, list):
+        return cast("list[Any]", payload)
+    if not isinstance(payload, dict):
+        return []
+    body = cast("dict[str, Any]", payload)
+    response = body.get("response")
+    if isinstance(response, dict):
+        body = cast("dict[str, Any]", response)
+    value = body.get("authorized_clients")
+    return cast("list[Any]", value) if isinstance(value, list) else []
+
+
+def _parse_authorized_clients(payload: Any) -> AuthorizedClients:
+    """Parse a raw ``list_authorized_clients()`` response into typed clients."""
+    clients = [
+        _parse_client(cast("dict[str, Any]", entry))
+        for entry in _authorized_clients_list(payload)
+        if isinstance(entry, dict)
+    ]
+    return AuthorizedClients(clients=clients, raw=payload)
+
+
 class TeslemetryEnergySite(EnergySite):
     """Teslemetry specific energy site."""
 
@@ -77,6 +196,17 @@ async def list_authorized_clients(self) -> dict[str, Any]:
             f"api/1/energy_sites/{self.energy_site_id}/command/authorized_clients",
         )
 
+    async def find_authorized_clients(self) -> AuthorizedClients:
+        """List authorized clients on the energy gateway, parsed into a typed result.
+
+        Prefer this over :meth:`list_authorized_clients` for consumers that
+        need to inspect the client list - it centralizes the response
+        parsing (envelope unwrap, null-body handling, ``state`` typing)
+        here instead of in the caller. See :class:`AuthorizedClients` for
+        the exact semantics.
+        """
+        return _parse_authorized_clients(await self.list_authorized_clients())
+
     async def remove_authorized_client(
         self, params: dict[str, Any] | None = None
     ) -> dict[str, Any]:

tests/test_teslemetry_authorized_clients.py

@@ -0,0 +1,177 @@
+"""Tests for TeslemetryEnergySite.find_authorized_clients(), the typed accessor
+over the Teslemetry ``command/authorized_clients`` endpoint.
+
+This endpoint's schema is undocumented; the wire-shape variants covered here
+(null body, bare list, wrapper envelope, key-name casing) are pinned from the
+Home Assistant Teslemetry integration's own defensive parsing of it. No site
+has been observed with a populated client list yet, so the per-entry shape
+is not live-sample-confirmed - see ``AuthorizedClient`` in
+``tesla_fleet_api/teslemetry/energysite.py``.
+"""
+
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+from typing import Any
+from unittest import IsolatedAsyncioTestCase
+from unittest.mock import AsyncMock, MagicMock
+
+from tesla_fleet_api.const import AuthorizedClientState
+from tesla_fleet_api.teslemetry.teslemetry import Teslemetry
+
+_UNSET = object()
+
+PUBLIC_KEY_B64 = "MIIBCgKCAQEAsomeBase64EncodedRsaPublicKeyBytes=="
+
+
+def _fake_response(*, json_body: object = _UNSET) -> MagicMock:
+    resp = MagicMock()
+    resp.status = 200
+    resp.ok = True
+    resp.content_type = "application/json"
+    resp.url = "https://example.com/x"
+    resp.headers = {}
+    resp.json = AsyncMock(return_value={} if json_body is _UNSET else json_body)
+    resp.text = AsyncMock(return_value="")
+    return resp
+
+
+def _make_session(response: object) -> MagicMock:
+    session = MagicMock()
+
+    @asynccontextmanager
+    async def _ctx(*args: Any, **kwargs: Any):
+        yield response
+
+    session.request = MagicMock(side_effect=lambda *a, **k: _ctx(*a, **k))
+    return session
+
+
+def _make_site(json_body: object):
+    api = Teslemetry(
+        session=_make_session(_fake_response(json_body=json_body)),
+        access_token="token",
+    )
+    return api.energySites.create(12345)
+
+
+class GetAuthorizedClientsTests(IsolatedAsyncioTestCase):
+    async def test_normal_payload_round_trips(self) -> None:
+        site = _make_site(
+            {
+                "response": {
+                    "authorized_clients": [
+                        "not-a-dict",
+                        {
+                            "public_key": PUBLIC_KEY_B64,
+                            "state": 3,
+                        },
+                    ]
+                }
+            }
+        )
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(len(result.clients), 1)
+        matched = result.clients[0]
+        self.assertEqual(matched.public_key, PUBLIC_KEY_B64)
+        self.assertEqual(matched.state, AuthorizedClientState.VERIFIED)
+
+    async def test_bare_list_payload_with_no_envelope(self) -> None:
+        site = _make_site([{"public_key": PUBLIC_KEY_B64, "state": 1}])
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(len(result.clients), 1)
+        self.assertEqual(result.clients[0].state, AuthorizedClientState.PENDING)
+
+    async def test_camel_case_entry_fields_are_recognized(self) -> None:
+        site = _make_site(
+            {
+                "response": {
+                    "authorized_clients": [
+                        {
+                            "publicKey": PUBLIC_KEY_B64,
+                            "authorized_client_state": 3,
+                        }
+                    ]
+                }
+            }
+        )
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(len(result.clients), 1)
+        matched = result.clients[0]
+        self.assertEqual(matched.public_key, PUBLIC_KEY_B64)
+        self.assertEqual(matched.state, AuthorizedClientState.VERIFIED)
+
+    async def test_state_as_string_is_typed_via_enum(self) -> None:
+        site = _make_site(
+            {
+                "response": {
+                    "authorized_clients": [
+                        {"public_key": PUBLIC_KEY_B64, "state": "verified"}
+                    ]
+                }
+            }
+        )
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(result.clients[0].state, AuthorizedClientState.VERIFIED)
+
+    async def test_unrecognized_present_state_is_preserved_not_dropped(self) -> None:
+        site = _make_site(
+            {
+                "response": {
+                    "authorized_clients": [{"public_key": PUBLIC_KEY_B64, "state": 0}]
+                }
+            }
+        )
+
+        result = await site.find_authorized_clients()
+
+        # 0 is not a member of AuthorizedClientState, but it is a present
+        # value - it must not be coerced to None (which means "absent").
+        self.assertEqual(result.clients[0].state, 0)
+        self.assertIsNotNone(result.clients[0].state)
+
+    async def test_explicitly_empty_list_returns_typed_empty_list(self) -> None:
+        site = _make_site({"response": {"authorized_clients": []}})
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(result.clients, [])
+
+    async def test_absent_field_returns_typed_empty_list(self) -> None:
+        site = _make_site({"response": {"foo": "bar"}})
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(result.clients, [])
+
+    async def test_null_body_returns_typed_empty_list_without_raising(self) -> None:
+        site = _make_site(None)
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(result.clients, [])
+        self.assertIsNone(result.raw)
+
+    async def test_non_dict_entries_are_skipped(self) -> None:
+        site = _make_site(
+            {
+                "response": {
+                    "authorized_clients": [
+                        "not-a-dict",
+                        {"public_key": PUBLIC_KEY_B64, "state": 3},
+                    ]
+                }
+            }
+        )
+
+        result = await site.find_authorized_clients()
+
+        self.assertEqual(len(result.clients), 1)