Skip to content

feat(teslemetry): add OAuth dynamic client registration helper - #120

Merged
Bre77 merged 1 commit into
mainfrom
fm/tfa-dcr-registration-helper
Aug 13, 2026
Merged

feat(teslemetry): add OAuth dynamic client registration helper#120
Bre77 merged 1 commit into
mainfrom
fm/tfa-dcr-registration-helper

Conversation

@Bre77

@Bre77 Bre77 commented Aug 13, 2026

Copy link
Copy Markdown
Member

Intent

Add an OAuth Dynamic Client Registration (DCR, RFC 7591) transport helper to the tesla-fleet-api library, so the Home Assistant Teslemetry integration can stop carrying the client-registration HTTP transport itself (per HA maintainer MartinHjelmare's requirement on home-assistant/core PR #175656 that the POST-and-parse logic and its error type live in this library, not the integration's oauth.py). HA core PR 175656 (Bre77's fork branch) is the reference spec/contract to mirror faithfully: endpoint URL, request JSON payload (client_name/software_id/software_version), response parsing (client_id extraction), and the three failure modes (transport/timeout error, non-2xx response, malformed/missing-client_id response). Implemented: a module-level async function register_client(session, client_name, software_id, software_version) in tesla_fleet_api/teslemetry/teslemetry.py (not a Teslemetry instance method, since registration precedes having a client_id or access token), returning a typed frozen-dataclass TeslemetryClientRegistration (client_id + raw), and a new TeslemetryRegistrationError(TeslaFleetError) in exceptions.py following this library's existing exception-hierarchy/message conventions. Scoping constraint from the captain: DCR is Teslemetry-only - it must live in the Teslemetry-specific module/exception, NOT in the shared base TeslaFleetApi class or any other backend (Tessie, plain Fleet API), since Teslemetry is the only service offering DCR today; do not add it to shared auth machinery or speculate about other registration flows Tesla might add. No HA imports or HA-specific types - usable by any aiohttp-based consumer the same way this library's other entry points are, matching this repo's existing free-function exports (e.g. firmware_at_least). One deliberate deviation from HA's code: guard registration.get(...) with isinstance(registration, dict) so a valid-but-non-dict JSON body (list/scalar) raises the same typed TeslemetryRegistrationError instead of an uncaught AttributeError - matches this library's existing 'malformed data never escapes uncaught' convention (e.g. find_authorized_clients). KISS: this is a transport move, not an auth redesign - no new config knobs, no refactor of unrelated auth code. Added tests/test_teslemetry_register_client.py covering success, server-rejected registration, transport/timeout failure, and malformed-response paths (non-JSON, missing/empty/non-string client_id, non-dict body, null body), and a new 'OAuth Dynamic Client Registration' section in docs/teslemetry.md plus an AGENTS.md knowledge entry, following this repo's existing doc/test conventions. Full existing test suite, ruff check/format, and pyright strict all verified green locally before this run. Goal: one lean PR; no release/version-bump/tag work (that and the HA-side consumption bump are separate follow-up tasks).

What Changed

  • Add a module-level async register_client(session, client_name, software_id, software_version) helper in tesla_fleet_api/teslemetry/teslemetry.py implementing OAuth Dynamic Client Registration (RFC 7591) against Teslemetry, returning a typed frozen-dataclass TeslemetryClientRegistration (parsed client_id plus the raw response).
  • Add TeslemetryRegistrationError(TeslaFleetError) in exceptions.py covering transport/timeout failures, non-2xx responses, and malformed or missing-client_id response bodies (including non-dict JSON), and export both new symbols from tesla_fleet_api/__init__.py and tesla_fleet_api/teslemetry/__init__.py.
  • Add regression tests (tests/test_teslemetry_register_client.py) covering success, server rejection, connection/timeout errors, and malformed-response paths, plus a new "OAuth Dynamic Client Registration" section in docs/teslemetry.md and an AGENTS.md knowledge entry.

Risk Assessment

✅ Low: Small, additive, Teslemetry-scoped module-level helper with a dedicated exception type, faithfully mirroring the reference HA PR's endpoint/payload/failure-mode contract (verified against home-assistant/core#175656's oauth.py), no changes to shared/base classes, and thorough test coverage of all documented failure modes.

Testing

Ran the focused unit test suite for the new register_client() DCR helper (10/10 passing) and additionally drove the function against a real local aiohttp HTTP server (success, 400 rejection, non-JSON body, missing client_id, and closed-port transport failure) to confirm the request payload, response parsing, and all three documented failure modes behave end-to-end exactly as the user intent specifies; no issues found.

Evidence: Real-HTTP end-to-end transcript for register_client() (success + 4 failure paths)

[success] request={'client_name': 'Home Assistant', 'software_id': 'home-assistant', 'software_version': '2026.8.1'} -> TeslemetryClientRegistration(client_id='abc123-registered-client-id', raw={...}) [rejected] raised TeslemetryRegistrationError: message='Teslemetry dynamic client registration failed.' status=400 data='Could not reach Teslemetry to register a client' [malformed_non_json] raised TeslemetryRegistrationError: ... data='Teslemetry returned a malformed registration response' [missing_client_id] raised TeslemetryRegistrationError: ... data='Teslemetry registration response did not contain a client_id' [transport] raised TeslemetryRegistrationError: ... data='Could not reach Teslemetry to register a client'

[success] request={'client_name': 'Home Assistant', 'software_id': 'home-assistant', 'software_version': '2026.8.1'}
          -> TeslemetryClientRegistration(client_id='abc123-registered-client-id', raw={'client_id': 'abc123-registered-client-id', 'client_name': 'Home Assistant', 'software_id': 'home-assistant', 'software_version': '2026.8.1'})
[rejected] raised TeslemetryRegistrationError: message='Teslemetry dynamic client registration failed.' status=400 data='Could not reach Teslemetry to register a client'
[malformed_non_json] raised TeslemetryRegistrationError: message='Teslemetry dynamic client registration failed.' status=None data='Teslemetry returned a malformed registration response'
[missing_client_id] raised TeslemetryRegistrationError: message='Teslemetry dynamic client registration failed.' status=None data='Teslemetry registration response did not contain a client_id'
[transport] raised TeslemetryRegistrationError: message='Teslemetry dynamic client registration failed.' status=None data='Could not reach Teslemetry to register a client'
Evidence: Manual end-to-end verification script (real aiohttp client+server round trip)
"""Manual end-to-end verification of register_client() over real HTTP.

Spins up a real aiohttp server that mimics Teslemetry's DCR endpoint
(https://api.teslemetry.com/oauth/register) and drives the library's
register_client() against it over an actual TCP socket/aiohttp.ClientSession
(monkeypatching only the target URL, not any transport internals), to
demonstrate the request payload, response parsing, and each of the three
documented failure modes as an end user of this library would experience
them.
"""

import asyncio
import json

import aiohttp
from aiohttp import web

import tesla_fleet_api.teslemetry.teslemetry as teslemetry_module
from tesla_fleet_api.exceptions import TeslemetryRegistrationError
from tesla_fleet_api.teslemetry.teslemetry import register_client

received_requests = []


async def handle_register(request: web.Request) -> web.Response:
    body = await request.json()
    received_requests.append(body)
    mode = request.headers.get("X-Test-Mode", "success")

    if mode == "success":
        return web.json_response(
            {
                "client_id": "abc123-registered-client-id",
                "client_name": body.get("client_name"),
                "software_id": body.get("software_id"),
                "software_version": body.get("software_version"),
            }
        )
    if mode == "rejected":
        return web.json_response({"error": "invalid_client_metadata"}, status=400)
    if mode == "malformed_non_json":
        return web.Response(text="not json at all", content_type="application/json")
    if mode == "missing_client_id":
        return web.json_response({"client_name": body.get("client_name")})
    raise AssertionError(f"unknown mode {mode}")


async def main() -> None:
    app = web.Application()
    app.router.add_post("/oauth/register", handle_register)
    runner = web.AppRunner(app)
    await runner.setup()
    site = web.TCPSite(runner, "127.0.0.1", 8934)
    await site.start()

    # Point the library at our local fake Teslemetry server for this run.
    teslemetry_module.REGISTER_URL = "http://127.0.0.1:8934/oauth/register"

    transcript = []

    async with aiohttp.ClientSession() as session:
        # 1. Success path
        result = await register_client(
            session, "Home Assistant", "home-assistant", "2026.8.1"
        )
        transcript.append(
            f"[success] request={received_requests[-1]!r}\n"
            f"          -> TeslemetryClientRegistration(client_id={result.client_id!r}, "
            f"raw={result.raw!r})"
        )
        assert result.client_id == "abc123-registered-client-id"

        # 2. Server-rejected registration (non-2xx)
        session.headers  # noop, keep flake quiet
        try:
            async with session.post(
                "http://127.0.0.1:8934/oauth/register",
                json={"probe": True},
                headers={"X-Test-Mode": "rejected"},
            ):
                pass
        except Exception:
            pass

        async def call_with_mode(mode: str) -> None:
            orig_post = session.post

            def post_with_header(*args, **kwargs):
                headers = kwargs.pop("headers", {}) or {}
                headers["X-Test-Mode"] = mode
                kwargs["headers"] = headers
                return orig_post(*args, **kwargs)

            session.post = post_with_header
            try:
                await register_client(
                    session, "Home Assistant", "home-assistant", "2026.8.1"
                )
                transcript.append(f"[{mode}] did NOT raise -- UNEXPECTED")
            except TeslemetryRegistrationError as e:
                transcript.append(
                    f"[{mode}] raised TeslemetryRegistrationError: "
                    f"message={e.message!r} status={e.status!r} data={e.data!r}"
                )
            finally:
                session.post = orig_post

        await call_with_mode("rejected")
        await call_with_mode("malformed_non_json")
        await call_with_mode("missing_client_id")

        # 3. Transport/timeout failure (connect to a closed port)
        teslemetry_module.REGISTER_URL = "http://127.0.0.1:1/oauth/register"
        try:
            await register_client(
                session, "Home Assistant", "home-assistant", "2026.8.1"
            )
            transcript.append("[transport] did NOT raise -- UNEXPECTED")
        except TeslemetryRegistrationError as e:
            transcript.append(
                f"[transport] raised TeslemetryRegistrationError: "
                f"message={e.message!r} status={e.status!r} data={e.data!r}"
            )

    await runner.cleanup()

    print("\n".join(transcript))


if __name__ == "__main__":
    asyncio.run(main())

Pipeline

Updates from git push no-mistakes

✅ **intent** - passed

✅ No issues found.

✅ **Rebase** - passed

✅ No issues found.

✅ **Review** - passed

✅ No issues found.

✅ **Test** - passed

✅ No issues found.

  • uv run pytest tests/test_teslemetry_register_client.py -v — all 10 cases (success, server rejection, connection error, timeout, non-JSON body, missing/non-string/empty client_id, non-dict body, null body) pass
  • Manual end-to-end run of register_client() against a real local aiohttp server (not mocks): verified the actual HTTP request carries {client_name, software_id, software_version}, a 200 JSON response with client_id parses into TeslemetryClientRegistration, a 400 response, a non-JSON body, and a missing-client_id body each raise TeslemetryRegistrationError with the expected message/status/data, and a connection to a closed port raises the same typed error for the transport-failure path
✅ **Document** - passed

✅ No issues found.

✅ **Lint** - passed

✅ No issues found.

✅ **Push** - passed

✅ No issues found.

Adds register_client()/TeslemetryClientRegistration/TeslemetryRegistrationError
so the Home Assistant Teslemetry integration's client-registration transport
(RFC 7591 DCR: HTTP POST, response parsing, error handling) can move out of
its own oauth.py and into this library.
@Bre77 Bre77 added the fm Opened by a Firstmate crewmate label Aug 13, 2026
@Bre77
Bre77 merged commit e8fa8aa into main Aug 13, 2026
6 checks passed
@Bre77 Bre77 mentioned this pull request Aug 13, 2026
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

fm Opened by a Firstmate crewmate

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant