From 36b70b1f9fb4189495bea37faf86abca118cbdc7 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 26 Aug 2026 08:59:53 +1000 Subject: [PATCH] fix: raise on 401/403 instead of retrying, and fix README's undocumented async-with usage TeslemetryStream never implemented __aenter__/__aexit__ - the connection lifecycle is deliberately listener-driven (async_add_listener connects on the first listener, disconnects on the last removed), with connect()/ close()/listen() as the manual alternative. The README's async-with examples didn't match either path and raised TypeError on first use, so they're rewritten to the listener-driven pattern the class actually supports. Separately, aiohttp.ClientResponseError is a subtype of ClientError, so a 401/403 from a bad access token was being retried forever as if it were a transient network blip - the caller saw no events and no error. __anext__ now treats 401/403 as terminal and raises TeslemetryStreamAuthenticationError instead of retrying; every other ClientError (including other response statuses) keeps the existing backoff-and-reconnect behavior unchanged. --- AGENTS.md | 2 + CLAUDE.md | 3 +- README.md | 46 ++++---- teslemetry_stream/__init__.py | 2 + teslemetry_stream/exception.py | 6 + teslemetry_stream/stream.py | 12 +- tests/test_auth_failure.py | 194 +++++++++++++++++++++++++++++++++ 7 files changed, 240 insertions(+), 25 deletions(-) mode change 120000 => 100644 CLAUDE.md create mode 100644 tests/test_auth_failure.py diff --git a/AGENTS.md b/AGENTS.md index c1189b3..06b1e43 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,6 +20,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`. - `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. A failed populating fetch (`aiohttp.ClientError`/timeout, or a non-200/404 status via `raise_for_status()`) is not authoritative like a 404: `_ensure_populated()` catches it, logs, and leaves the vehicle unpopulated rather than letting it propagate - every `listen_*` method reaches `add_field()` through `_enable_field()`'s fire-and-forget `asyncio.create_task()`, where an uncaught exception would just silently abandon the field request instead of sending the PATCH. - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). +- `TeslemetryStream` has no `__aenter__`/`__aexit__` - do not reintroduce `async with TeslemetryStream(...)` in the README or examples. Connection lifecycle is entirely listener-driven: `async_add_listener` connects on the first (public) listener and disconnects on the last one removed; `connect()`/`close()`/`listen()` exist for callers who want to manage the connection themselves instead. +- `__anext__` treats a `aiohttp.ClientResponseError` with `status` 401 or 403 as terminal, not transient: it sets `active = False` and raises `TeslemetryStreamAuthenticationError` (chaining the original error) rather than retrying, since a rejected token can never succeed on retry. Every other `aiohttp.ClientError` (including other response statuses) keeps the pre-existing backoff-and-reconnect behavior. `tests/test_auth_failure.py` covers both the 401/403 surfacing and that a genuine transient `ClientError` still retries and reconnects. - `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch. ## Maintaining this file diff --git a/CLAUDE.md b/CLAUDE.md deleted file mode 120000 index 47dc3e3..0000000 --- a/CLAUDE.md +++ /dev/null @@ -1 +0,0 @@ -AGENTS.md \ No newline at end of file diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..a9d4d26 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,2 @@ + +@AGENTS.md diff --git a/README.md b/README.md index 56baf5e..6c4fe1e 100644 --- a/README.md +++ b/README.md @@ -25,30 +25,30 @@ The TeslemetryStream class requires: The full list of possible values are provided in `TelemetryFields` and `TelemetryAlerts` -To connect, either use `async with` on the instance, call `connect()`, or register an callback with `async_add_listener`, which will connect when added and disconnect when removed. +The simplest way to connect is to register a callback with `async_add_listener`: it connects automatically when the first listener is added, and disconnects when the last one is removed - no explicit `connect()`/`close()` call needed. -Using `connect()` or `listen()` will require you to close the session manually using `close()`. +If you want to manage the connection yourself instead, call `connect()` and drive the stream with `listen()`; in that case you are responsible for calling `close()` when done. ## Example The following example puts the listening loop in the background, then stopping after 20 seconds. ``` async def main(): async with aiohttp.ClientSession() as session: - async with TeslemetryStream( + stream = TeslemetryStream( access_token="", vin="", # for single vehicles server="na.teslemetry.com" # or "eu.teslemetry.com" session=session, - ) as stream: + ) - def callback(event): - print(event["data"]) + def callback(event): + print(event["data"]) - remove = stream.async_add_listener(callback) + remove = stream.async_add_listener(callback) - print("Running") - await asyncio.sleep(60) - remove() + print("Running") + await asyncio.sleep(60) + remove() ``` ## Using Typed Listen Methods @@ -58,27 +58,27 @@ The library provides typed listen methods for various telemetry signals. These m ```python async def main(): async with aiohttp.ClientSession() as session: - async with TeslemetryStream( + stream = TeslemetryStream( access_token="", vin="", # for single vehicles session=session, - ) as stream: + ) - vehicle = stream.get_vehicle("") + vehicle = stream.get_vehicle("") - def battery_level_callback(battery_level): - print(f"Battery Level: {battery_level}") + def battery_level_callback(battery_level): + print(f"Battery Level: {battery_level}") - def vehicle_speed_callback(vehicle_speed): - print(f"Vehicle Speed: {vehicle_speed}") + def vehicle_speed_callback(vehicle_speed): + print(f"Vehicle Speed: {vehicle_speed}") - remove_battery_level_listener = vehicle.listen_BatteryLevel(battery_level_callback) - remove_vehicle_speed_listener = vehicle.listen_VehicleSpeed(vehicle_speed_callback) + remove_battery_level_listener = vehicle.listen_BatteryLevel(battery_level_callback) + remove_vehicle_speed_listener = vehicle.listen_VehicleSpeed(vehicle_speed_callback) - print("Running") - await asyncio.sleep(60) - remove_battery_level_listener() - remove_vehicle_speed_listener() + print("Running") + await asyncio.sleep(60) + remove_battery_level_listener() + remove_vehicle_speed_listener() ``` ## Writing Your Own Listener with Multiple Signals diff --git a/teslemetry_stream/__init__.py b/teslemetry_stream/__init__.py index b1c7d5e..0c69208 100644 --- a/teslemetry_stream/__init__.py +++ b/teslemetry_stream/__init__.py @@ -8,6 +8,7 @@ ) from .energysite import TeslemetryStreamEnergySite from .exception import ( + TeslemetryStreamAuthenticationError, TeslemetryStreamConnectionError, TeslemetryStreamEnded, TeslemetryStreamError, @@ -24,6 +25,7 @@ "Signal", "SseTopic", "TeslemetryStream", + "TeslemetryStreamAuthenticationError", "TeslemetryStreamConnectionError", "TeslemetryStreamEnded", "TeslemetryStreamEnergySite", diff --git a/teslemetry_stream/exception.py b/teslemetry_stream/exception.py index d8e79e6..9921d38 100644 --- a/teslemetry_stream/exception.py +++ b/teslemetry_stream/exception.py @@ -23,3 +23,9 @@ class TeslemetryStreamEnded(TeslemetryStreamError): """Teslemetry Stream Connection Error""" message = "The stream was ended by the server." + + +class TeslemetryStreamAuthenticationError(TeslemetryStreamError): + """Teslemetry Stream Authentication Error""" + + message = "The access token was rejected (401/403) and will not be retried." diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index a067dca..d3d7cdc 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -11,7 +11,7 @@ from .const import CreditsEvent from .energysite import TeslemetryStreamEnergySite -from .exception import TeslemetryStreamEnded +from .exception import TeslemetryStreamAuthenticationError, TeslemetryStreamEnded from .vehicle import TeslemetryStreamVehicle LOGGER = logging.getLogger(__package__) @@ -325,6 +325,8 @@ async def __anext__(self) -> dict[str, Any]: :return: Next event as a dictionary. :raises StopAsyncIteration: If the stream is stopped. :raises TeslemetryStreamEnded: If the stream is ended by the server. + :raises TeslemetryStreamAuthenticationError: If the access token is + rejected with a 401 or 403. """ while self.active: try: @@ -353,6 +355,14 @@ async def __anext__(self) -> dict[str, Any]: LOGGER.warning("Stream ended by server") self._close_response() except aiohttp.ClientError as error: + if isinstance(error, aiohttp.ClientResponseError) and error.status in (401, 403): + # A rejected token is a definitive answer, not a transient + # blip - retrying it can never succeed, and doing so masks + # a bad credential as an indefinitely quiet stream. + LOGGER.error("Authentication failed, not retrying: %s", repr(error)) + self.active = False + self._close_response() + raise TeslemetryStreamAuthenticationError() from error LOGGER.warning("Client error: %s", repr(error)) self._close_response() delay = min(2**self.retries, 600) diff --git a/tests/test_auth_failure.py b/tests/test_auth_failure.py new file mode 100644 index 0000000..f7995dd --- /dev/null +++ b/tests/test_auth_failure.py @@ -0,0 +1,194 @@ +"""Regression tests for __anext__'s auth-failure handling: a 401/403 +aiohttp.ClientResponseError is a subtype of aiohttp.ClientError, so without +special-casing it, an invalid token is swallowed and retried exactly like a +transient network blip - the caller sees no events and no error. These tests +cover the fix (auth failure surfaces as TeslemetryStreamAuthenticationError +and is not retried) and its safety rail (a genuine transient ClientError +still retries and reconnects, unchanged). +""" +from __future__ import annotations + +import asyncio +import contextlib +from typing import Any + +import aiohttp + +from teslemetry_stream.exception import TeslemetryStreamAuthenticationError +from teslemetry_stream.stream import TeslemetryStream + +REQUEST_INFO = aiohttp.RequestInfo( + url="https://fake.teslemetry.com/sse", + method="GET", + headers={}, # type: ignore[arg-type] + real_url="https://fake.teslemetry.com/sse", # type: ignore[arg-type] +) + + +class FakeContent: + """Async-iterable response body that blocks until failed or cancelled.""" + + def __init__(self) -> None: + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeContent: + return self + + async def __anext__(self) -> bytes: + await self._blocker + raise AssertionError("unreachable - blocker only resolves via an exception") + + def fail(self, exc: BaseException) -> None: + if not self._blocker.done(): + self._blocker.set_exception(exc) + + +class FakeResponse: + """Minimal stand-in for the aiohttp response `connect()` awaits.""" + + def __init__(self) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + self.content = FakeContent() + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeSession: + """Captures every `get()` call; a queue of `get_results` (each either an + exception to raise or a FakeResponse to return) drives each call in + order, mirroring how a real 401 raises straight out of the connect GET + when `raise_for_status=True`.""" + + def __init__(self, get_results: list[Any]) -> None: + self.calls = 0 + self._get_results = list(get_results) + self.responses: list[FakeResponse] = [] + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls += 1 + result = self._get_results.pop(0) + if isinstance(result, BaseException): + raise result + self.responses.append(result) + return result + + +def make_stream(session: FakeSession) -> TeslemetryStream: + return TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token="bad-token", + server="api.teslemetry.com", + manual=True, + ) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def drain_cancelled(task: asyncio.Task[Any]) -> None: + with contextlib.suppress(asyncio.CancelledError): + await task + + +async def test_401_surfaces_and_does_not_retry(results: list[bool]) -> None: + error = aiohttp.ClientResponseError( + request_info=REQUEST_INFO, history=(), status=401, message="Unauthorized" + ) + session = FakeSession([error]) + stream = make_stream(session) + + task = asyncio.create_task(stream.listen()) + + raised: Exception | None = None + try: + await task + except TeslemetryStreamAuthenticationError as exc: + raised = exc + + results.append( + check("a 401 surfaces as TeslemetryStreamAuthenticationError", raised is not None) + ) + results.append( + check( + "the original 401 is chained as the cause", + raised is not None and raised.__cause__ is error, + ) + ) + results.append( + check("no retry is attempted after a 401", session.calls == 1, f"got {session.calls}") + ) + results.append(check("the stream stops rather than looping forever", not stream.active)) + + +async def test_403_surfaces_and_does_not_retry(results: list[bool]) -> None: + error = aiohttp.ClientResponseError( + request_info=REQUEST_INFO, history=(), status=403, message="Forbidden" + ) + session = FakeSession([error]) + stream = make_stream(session) + + task = asyncio.create_task(stream.listen()) + + raised: Exception | None = None + try: + await task + except TeslemetryStreamAuthenticationError as exc: + raised = exc + + results.append( + check("a 403 surfaces as TeslemetryStreamAuthenticationError", raised is not None) + ) + results.append( + check("no retry is attempted after a 403", session.calls == 1, f"got {session.calls}") + ) + + +async def test_transient_client_error_still_retries_and_reconnects(results: list[bool]) -> None: + """A genuine transient failure (connection reset mid-stream, not a 401/403 + response) must keep retrying and reconnecting exactly as before - the + auth-failure special-case must not touch this path.""" + session = FakeSession([FakeResponse(), FakeResponse()]) + stream = make_stream(session) + + task = asyncio.create_task(stream.listen()) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append(check("initial connect happened", session.calls == 1, f"got {session.calls}")) + + session.responses[0].content.fail(aiohttp.ClientError("boom")) + # retries starts at 0, so the first backoff delay is 2**0 == 1 second. + await asyncio.sleep(1.2) + + results.append(check("the stream is still active after a transient error", stream.active)) + results.append( + check( + "the stream reconnected instead of surfacing an error", + session.calls == 2, + f"got {session.calls}", + ) + ) + results.append(check("the listen task is still running", not task.done())) + + stream.close() + await drain_cancelled(task) + + +async def main() -> None: + results: list[bool] = [] + await test_401_surfaces_and_does_not_retry(results) + await test_403_surfaces_and_does_not_retry(results) + await test_transient_client_error_still_retries_and_reconnects(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())