From f6f09dde739b30de2c06233fcb26780e594b6a16 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 21:37:10 +1000 Subject: [PATCH 1/9] Sync vehicle config record from server-pushed config events The internal fields/preferTyped record only reflected what this client had itself requested or observed at connect, so a config change applied elsewhere (another client, the console, a server-side migration) left it stale - causing add_field/prefer_typed's no-op check to either send redundant requests or wrongly skip real ones. TeslemetryStreamVehicle now listens for the config SSE topic and merges well-typed pieces of the pushed record in, leaving a missing piece untouched and logging (without corrupting the record) a malformed one. --- AGENTS.md | 1 + teslemetry_stream/const.py | 1 + teslemetry_stream/vehicle.py | 43 +++++++ tests/test_batch_retry_storm.py | 5 + tests/test_config_events.py | 186 ++++++++++++++++++++++++++++++ tests/test_config_update.py | 5 + tests/test_field_type_coercion.py | 9 +- 7 files changed, 247 insertions(+), 3 deletions(-) create mode 100644 tests/test_config_events.py diff --git a/AGENTS.md b/AGENTS.md index 3e8600b..336104e 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,6 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs..environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery. - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. +- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history: `__init__` unconditionally registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field; a missing piece is left untouched, and a malformed piece (wrong type) is logged and skipped without touching the other piece or the pending `_config`. Because this listener is registered at construction, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener`. `tests/test_config_events.py` covers the record update, the resulting no-op/send behavior in `add_field`, and malformed/partial events. - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py index 967744a..1fa300a 100644 --- a/teslemetry_stream/const.py +++ b/teslemetry_stream/const.py @@ -23,6 +23,7 @@ class Key(StrEnum): ERRORS = "errors" VEHICLE_DATA = "vehicle_data" STATE = "state" + CONFIG = "config" STATUS = "status" NETWORK_INTERFACE = "networkInterface" SITE_ID = "site_id" diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 6bf9e7e..dfd7a82 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -86,6 +86,13 @@ def __init__(self, stream: TeslemetryStream, vin: str): # Callers that arrive while it is running merge into `_config` and # await it instead of starting their own PATCH. self._flight = None + # Keeps `fields`/`preferTyped` current when the server applies a + # config change outside this client (another client, the console, + # a Teslemetry-side migration), so add_field/prefer_typed's no-op + # checks don't act on stale history. + self.stream.async_add_listener( + self._on_config_event, {Key.VIN: self.vin, Key.CONFIG: None} + ) @property def config(self) -> dict[str, Any]: @@ -115,6 +122,42 @@ async def get_config(self) -> None: req.raise_for_status() + def _on_config_event(self, event: dict[str, Any]) -> None: + """Sync the record from a server-pushed config event. + + Only well-typed pieces are applied; a bad piece is logged and + skipped so it can't corrupt the last-known-good record, and the + other piece (if well-typed) still applies. + """ + config = event.get(Key.CONFIG) + if not isinstance(config, dict): + LOGGER.warning( + "Ignoring malformed config event for %s: %r", self.vin, config + ) + return + + if "fields" in config: + fields = config["fields"] + if isinstance(fields, dict): + self.fields = fields + else: + LOGGER.warning( + "Ignoring malformed fields in config event for %s: %r", + self.vin, + fields, + ) + + if "prefer_typed" in config: + prefer_typed = config["prefer_typed"] + if isinstance(prefer_typed, bool): + self.preferTyped = prefer_typed + else: + LOGGER.warning( + "Ignoring malformed prefer_typed in config event for %s: %r", + self.vin, + prefer_typed, + ) + async def update_config(self, config: dict[str, Any]) -> None: """Request a configuration update for the vehicle. diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index 116a4d4..2120317 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -31,6 +31,11 @@ class FakeStream: manual = True + def async_add_listener( + self, callback: Any, filters: dict[str, Any] | None = None + ) -> Any: + return lambda: None + def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle: """Build a vehicle that records payloads and replays canned responses.""" diff --git a/tests/test_config_events.py b/tests/test_config_events.py new file mode 100644 index 0000000..bf467e3 --- /dev/null +++ b/tests/test_config_events.py @@ -0,0 +1,186 @@ +"""Config-update SSE events keep the internal config record fresh. + +The server pushes a ``config`` event (``Key.CONFIG`` / ``SseTopic.CONFIG``) +shaped like ``{"vin": ..., "config": {"fields": {...}, "prefer_typed": bool}}``, +mirroring the REST ``get_config`` response body. ``TeslemetryStreamVehicle`` +registers an internal listener for it at construction so ``fields``/ +``preferTyped`` - and therefore the ``add_field``/``prefer_typed`` no-op +checks - reflect current server truth rather than only what this client has +itself requested or observed at connect. +""" +from __future__ import annotations + +import asyncio +import logging +from typing import Any, Callable + +from teslemetry_stream.const import Key +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + + +class FakeStream: + """Minimal stand-in for TeslemetryStream that captures the config listener.""" + + manual = True + + def __init__(self) -> None: + self.config_listener: Callable[[dict[str, Any]], None] | None = None + + def async_add_listener( + self, + callback: Callable[[dict[str, Any]], None], + filters: dict[str, Any] | None = None, + ) -> Callable[[], None]: + assert filters is not None + if Key.CONFIG in filters: + self.config_listener = callback + return lambda: None + + +class CaptureWarnings(logging.Handler): + """Collect formatted WARNING records emitted by the library.""" + + def __init__(self) -> None: + super().__init__(level=logging.WARNING) + self.messages: list[str] = [] + + def emit(self, record: logging.LogRecord) -> None: + self.messages.append(record.getMessage()) + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<56} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +def make_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]: + """Build a vehicle that records PATCH payloads instead of sending them.""" + stream = FakeStream() + vehicle = TeslemetryStreamVehicle(stream, VIN) # type: ignore[arg-type] + vehicle.sent = [] # type: ignore[attr-defined] + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + vehicle.sent.append(dict(config)) # type: ignore[attr-defined] + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + return vehicle, stream + + +async def main() -> None: + results = [] + + # A config event replaces the record with current server truth. + vehicle, stream = make_vehicle() + assert stream.config_listener is not None + stream.config_listener( + { + "vin": VIN, + "config": { + "fields": {"BatteryLevel": {"interval_seconds": 60}}, + "prefer_typed": True, + }, + } + ) + results.append( + check( + "a config event updates fields and prefer_typed", + vehicle.fields == {"BatteryLevel": {"interval_seconds": 60}} + and vehicle.preferTyped is True, + f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}", + ) + ) + + # A request that now matches the updated record is skipped - no PATCH sent. + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "a request matching the updated record is skipped", + vehicle.sent == [], + f"sent {vehicle.sent}", + ) + ) + + # A request that differs from the updated record is still sent. + await vehicle.add_field("BatteryLevel", 30) + results.append( + check( + "a request differing from the updated record is sent", + len(vehicle.sent) == 1 + and vehicle.sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 30}, + f"sent {vehicle.sent}", + ) + ) + + # A malformed/partial config event never corrupts the record. + handler = CaptureWarnings() + logger = logging.getLogger("teslemetry_stream") + logger.addHandler(handler) + try: + vehicle, stream = make_vehicle() + assert stream.config_listener is not None + stream.config_listener( + { + "vin": VIN, + "config": { + "fields": {"BatteryLevel": {}}, + "prefer_typed": False, + }, + } + ) + good_fields, good_typed = dict(vehicle.fields), vehicle.preferTyped + + # A non-dict "config" body is entirely rejected and logged. + stream.config_listener({"vin": VIN, "config": "not-a-dict"}) + results.append( + check( + "a non-dict config event is ignored, logged, and keeps last-good", + vehicle.fields == good_fields + and vehicle.preferTyped == good_typed + and any("malformed" in m.lower() for m in handler.messages), + f"fields {vehicle.fields}, warnings {handler.messages}", + ) + ) + + # A partially malformed body applies the well-typed piece and keeps + # the last-good value for the malformed piece. + handler.messages.clear() + stream.config_listener( + {"vin": VIN, "config": {"fields": "not-a-dict", "prefer_typed": True}} + ) + results.append( + check( + "a partial config event applies the good field, keeps the bad one", + vehicle.fields == good_fields + and vehicle.preferTyped is True + and any("malformed" in m.lower() for m in handler.messages), + f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}, " + f"warnings {handler.messages}", + ) + ) + + # A config event missing a key entirely leaves that piece untouched. + handler.messages.clear() + stream.config_listener( + {"vin": VIN, "config": {"fields": {"CarType": {}}}} + ) + results.append( + check( + "a config event omitting prefer_typed leaves it unchanged", + vehicle.fields == {"CarType": {}} and vehicle.preferTyped is True, + f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}", + ) + ) + finally: + logger.removeHandler(handler) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 62a3320..2ed2193 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -30,6 +30,11 @@ class FakeStream: manual = True + def async_add_listener( + self, callback: Any, filters: dict[str, Any] | None = None + ) -> Any: + return lambda: None + def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle: """Build a vehicle that records payloads and replays canned responses.""" diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py index 42921f2..8fc8846 100644 --- a/tests/test_field_type_coercion.py +++ b/tests/test_field_type_coercion.py @@ -31,10 +31,13 @@ def async_add_listener( callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None, ) -> Callable[[], None]: - # filters carries {"vin": ..., "data": {Signal: None}} — grab the field + # filters carries {"vin": ..., "data": {Signal: None}} — grab the field. + # The vehicle's own internal config-sync listener has no "data" key; + # it's not under test here, so just ignore it. assert filters is not None - signal = next(iter(filters["data"])) - self.captured[signal] = callback + if "data" in filters: + signal = next(iter(filters["data"])) + self.captured[signal] = callback return lambda: None From 9d07e96ea7df3b306992bad77d80ec386a27d9f8 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 21:53:18 +1000 Subject: [PATCH 2/9] Defer config-listener registration and exempt it from auto-close Registering the config-sync listener eagerly in the vehicle constructor called async_add_listener -> asyncio.create_task for the stream's first listener, which needs a running loop - breaking synchronous TeslemetryStream(vin=...)/vehicle construction. Registration is now lazy, on the first add_field/prefer_typed/update_config call. The listener also permanently occupied stream._listeners, so removing every public listener could never satisfy the "no more listeners" auto-close check and the SSE connection leaked open. async_add_listener gained an internal flag excluding bookkeeping-only listeners from that check. --- AGENTS.md | 2 +- teslemetry_stream/stream.py | 19 ++- teslemetry_stream/vehicle.py | 33 +++-- tests/test_batch_retry_storm.py | 2 +- tests/test_config_events.py | 19 ++- tests/test_config_listener_lifecycle.py | 155 ++++++++++++++++++++++++ tests/test_config_update.py | 2 +- tests/test_energysite_events.py | 2 +- tests/test_field_type_coercion.py | 1 + tests/test_sse_topics.py | 2 +- 10 files changed, 215 insertions(+), 22 deletions(-) create mode 100644 tests/test_config_listener_lifecycle.py diff --git a/AGENTS.md b/AGENTS.md index 336104e..02c66da 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs..environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery. - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. -- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history: `__init__` unconditionally registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field; a missing piece is left untouched, and a malformed piece (wrong type) is logged and skipped without touching the other piece or the pending `_config`. Because this listener is registered at construction, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener`. `tests/test_config_events.py` covers the record update, the resulting no-op/send behavior in `add_field`, and malformed/partial events. +- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history: `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field; a missing piece is left untouched, and a malformed piece (wrong type) is logged and skipped without touching the other piece or the pending `_config`. Registration is deferred to the first `add_field`/`prefer_typed`/`update_config` call (not `__init__`) because `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop - eager construction-time registration broke the previously-synchronous `TeslemetryStream(vin=...)`/vehicle construction. `async_add_listener(..., internal=True)` marks it so it's excluded from the "last listener removed" auto-close check in `TeslemetryStream` - otherwise a permanently-registered internal listener would keep `_listeners` non-empty forever and the SSE connection would never close once every real listener unsubscribed. Because of this, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)`. `tests/test_config_events.py` covers the record merge; `tests/test_config_listener_lifecycle.py` covers the lazy registration and the auto-close exclusion. - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index c96fbc6..f09664b 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -66,7 +66,8 @@ def __init__( else: self.topics = None self._listeners: dict[ - Callable[..., Any], tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None] + Callable[..., Any], + tuple[Callable[[dict[str, Any]], None], dict[str, Any] | None, bool], ] = {} self._connection_listeners: dict[Callable[..., Any], Callable[[bool], None]] = {} self._listen_task: asyncio.Task[None] | None = None @@ -361,13 +362,21 @@ async def __anext__(self) -> dict[str, Any]: raise StopAsyncIteration def async_add_listener( - self, callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None + self, + callback: Callable[[dict[str, Any]], None], + filters: dict[str, Any] | None = None, + internal: bool = False, ) -> Callable[[], None]: """ Listen for data updates. :param callback: Callback function to handle updates. :param filters: Filters to apply to the updates. + :param internal: True for a listener that keeps the client's own + state fresh (e.g. a vehicle's config-sync listener) rather than + serving a consumer callback. Excluded from the "last listener + removed" auto-close check, so a bookkeeping-only listener can't + pin the connection open forever once every real listener is gone. :return: Function to remove the listener. """ schedule_refresh = not self._listeners @@ -377,11 +386,11 @@ def remove_listener() -> None: Remove update listener. """ self._listeners.pop(remove_listener) - if not self._listeners: + if not any(not is_internal for _, _, is_internal in self._listeners.values()): LOGGER.info("Shutting down stream as there are no more listeners") self.close() - self._listeners[remove_listener] = (callback, filters) + self._listeners[remove_listener] = (callback, filters, internal) # This is the first listener - start the owned listen task, unless # one is already running or manual mode delegates that to the caller. @@ -415,7 +424,7 @@ async def listen(self) -> None: try: async for event in self: if event: - for listener, filters in self._listeners.values(): + for listener, filters, _internal in self._listeners.values(): if recursive_match(filters, event): try: listener(event) diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index dfd7a82..eaae98b 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -86,13 +86,11 @@ def __init__(self, stream: TeslemetryStream, vin: str): # Callers that arrive while it is running merge into `_config` and # await it instead of starting their own PATCH. self._flight = None - # Keeps `fields`/`preferTyped` current when the server applies a - # config change outside this client (another client, the console, - # a Teslemetry-side migration), so add_field/prefer_typed's no-op - # checks don't act on stale history. - self.stream.async_add_listener( - self._on_config_event, {Key.VIN: self.vin, Key.CONFIG: None} - ) + # Registration is deferred to first use (see _ensure_config_listener) - + # eagerly calling async_add_listener here would create the stream's + # owned task via asyncio.create_task before a loop necessarily exists, + # breaking a synchronous `TeslemetryStream(vin=...)`/vehicle construction. + self._config_listener_registered = False @property def config(self) -> dict[str, Any]: @@ -122,6 +120,24 @@ async def get_config(self) -> None: req.raise_for_status() + def _ensure_config_listener(self) -> None: + """Register the internal config-sync listener once, lazily. + + Keeps `fields`/`preferTyped` current when the server applies a + config change outside this client (another client, the console, + a Teslemetry-side migration), so add_field/prefer_typed's no-op + checks don't act on stale history. Marked `internal` so it doesn't + pin the stream open once every public listener is removed. + """ + if self._config_listener_registered: + return + self._config_listener_registered = True + self.stream.async_add_listener( + self._on_config_event, + {Key.VIN: self.vin, Key.CONFIG: None}, + internal=True, + ) + def _on_config_event(self, event: dict[str, Any]) -> None: """Sync the record from a server-pushed config event. @@ -165,6 +181,7 @@ async def update_config(self, config: dict[str, Any]) -> None: flush for this vehicle rather than starting a new one, so that a batch of listeners scheduled at the same time produces one PATCH. """ + self._ensure_config_listener() async with self.lock: self._config = merge(config, self._config) @@ -274,6 +291,7 @@ async def post_config(self, config: dict[str, Any]) -> dict[str, Any]: async def add_field(self, field: Signal | str, interval: int | None = None) -> None: """Handle vehicle data from the stream.""" + self._ensure_config_listener() if isinstance(field, Signal): field = field.value @@ -292,6 +310,7 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N async def prefer_typed(self, prefer_typed: bool) -> None: """Set prefer typed.""" + self._ensure_config_listener() if self.preferTyped == prefer_typed: return await self.update_config({"prefer_typed": prefer_typed}) diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index 2120317..2035544 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -32,7 +32,7 @@ class FakeStream: manual = True def async_add_listener( - self, callback: Any, filters: dict[str, Any] | None = None + self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False ) -> Any: return lambda: None diff --git a/tests/test_config_events.py b/tests/test_config_events.py index bf467e3..70f24e1 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -3,10 +3,12 @@ The server pushes a ``config`` event (``Key.CONFIG`` / ``SseTopic.CONFIG``) shaped like ``{"vin": ..., "config": {"fields": {...}, "prefer_typed": bool}}``, mirroring the REST ``get_config`` response body. ``TeslemetryStreamVehicle`` -registers an internal listener for it at construction so ``fields``/ -``preferTyped`` - and therefore the ``add_field``/``prefer_typed`` no-op -checks - reflect current server truth rather than only what this client has -itself requested or observed at connect. +lazily registers an internal listener for it (on first ``add_field``/ +``prefer_typed``/``update_config`` call, not at construction - see +``test_config_listener_lifecycle.py`` for why) so ``fields``/``preferTyped`` +- and therefore the ``add_field``/``prefer_typed`` no-op checks - reflect +current server truth rather than only what this client has itself +requested or observed at connect. """ from __future__ import annotations @@ -32,6 +34,7 @@ def async_add_listener( self, callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None, + internal: bool = False, ) -> Callable[[], None]: assert filters is not None if Key.CONFIG in filters: @@ -56,10 +59,16 @@ def check(label: str, ok: bool, detail: str = "") -> bool: def make_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]: - """Build a vehicle that records PATCH payloads instead of sending them.""" + """Build a vehicle that records PATCH payloads instead of sending them. + + Registration is lazy (see ``test_config_listener_lifecycle.py``), so + force it here the same way a real first ``add_field``/``prefer_typed``/ + ``update_config`` call would, to exercise the merge logic in isolation. + """ stream = FakeStream() vehicle = TeslemetryStreamVehicle(stream, VIN) # type: ignore[arg-type] vehicle.sent = [] # type: ignore[attr-defined] + vehicle._ensure_config_listener() async def patch_config(config: dict[str, Any]) -> dict[str, Any]: vehicle.sent.append(dict(config)) # type: ignore[attr-defined] diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py new file mode 100644 index 0000000..dee007d --- /dev/null +++ b/tests/test_config_listener_lifecycle.py @@ -0,0 +1,155 @@ +"""Regression tests for the vehicle config-sync listener's lifecycle. + +Two defects flagged in review of the config-consume feature: + +- Eagerly registering the internal config listener in + ``TeslemetryStreamVehicle.__init__`` called ``TeslemetryStream. + async_add_listener()`` -> ``asyncio.create_task()`` for the very first + listener, which requires a running event loop. That broke the + previously-synchronous ``TeslemetryStream(vin=...)``/vehicle + construction. Registration is now deferred to first use (inside + ``add_field``/``prefer_typed``/``update_config``, all async). +- The internal listener, once registered, stayed in + ``TeslemetryStream._listeners`` forever, so the "last listener removed" + auto-close check (which fires on an empty ``_listeners``) could never + trigger once every real/public listener was gone - the SSE connection + leaked open. ``async_add_listener(..., internal=True)`` now excludes it + from that check. +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from teslemetry_stream.stream import TeslemetryStream +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + + +class FakeSession: + """A session whose get() is never expected to be called in these tests.""" + + async def get(self, url: str, **kwargs: Any) -> Any: + raise AssertionError(f"unexpected session.get({url!r}) - these tests must not connect") + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +def make_stream(**kwargs: Any) -> TeslemetryStream: + # manual=True: these tests exercise listener bookkeeping, not the real + # connect/listen loop - FakeSession.get() intentionally isn't a working + # SSE endpoint. + kwargs.setdefault("manual", True) + return TeslemetryStream( + session=FakeSession(), # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + **kwargs, + ) + + +def test_sync_construction_without_a_loop() -> bool: + """Must run before any event loop exists - constructing a stream+vehicle + synchronously (the library's documented pre-async-context usage) must not + require or start one.""" + label = "TeslemetryStream(vin=...) construction outside a running loop does not raise" + try: + stream = make_stream(vin=VIN) + TeslemetryStreamVehicle(stream, VIN) + return check(label, True) + except RuntimeError as error: + return check(label, False, f"raised {error!r}") + + +async def test_lazy_registration_happens_on_first_use(results: list[bool]) -> None: + stream = make_stream() + vehicle = TeslemetryStreamVehicle(stream, VIN) + + results.append( + check( + "no listener is registered at construction", + len(stream._listeners) == 0, + f"listeners {len(stream._listeners)}", + ) + ) + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + await vehicle.add_field("BatteryLevel") + + results.append( + check( + "the internal config listener is registered by the first add_field call", + len(stream._listeners) == 1, + f"listeners {len(stream._listeners)}", + ) + ) + results.append( + check( + "the registered listener is marked internal", + all(is_internal for _, _, is_internal in stream._listeners.values()), + ) + ) + + +async def test_auto_close_after_last_public_listener_removed(results: list[bool]) -> None: + stream = make_stream() + vehicle = TeslemetryStreamVehicle(stream, VIN) + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + # Registers the internal (excluded) listener. + await vehicle.add_field("BatteryLevel") + + # A real/public listener on top of the internal one. + remove_public = stream.async_add_listener(lambda event: None) + + results.append( + check( + "two listeners are registered: one internal, one public", + len(stream._listeners) == 2, + f"listeners {len(stream._listeners)}", + ) + ) + + stream.active = True # simulate a live connection to observe close() flip it back + remove_public() + + results.append( + check( + "removing the last public listener still auto-closes the stream", + stream.active is False, + ) + ) + results.append( + check( + "the internal listener remains registered after auto-close", + len(stream._listeners) == 1, + f"listeners {len(stream._listeners)}", + ) + ) + + +async def main(pre_loop_results: list[bool]) -> None: + results: list[bool] = list(pre_loop_results) + await test_lazy_registration_happens_on_first_use(results) + await test_auto_close_after_last_public_listener_removed(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + # Must run before asyncio.run() starts a loop - that's the entire point. + pre_loop_results = [test_sync_construction_without_a_loop()] + asyncio.run(main(pre_loop_results)) diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 2ed2193..9acf03b 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -31,7 +31,7 @@ class FakeStream: manual = True def async_add_listener( - self, callback: Any, filters: dict[str, Any] | None = None + self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False ) -> Any: return lambda: None diff --git a/tests/test_energysite_events.py b/tests/test_energysite_events.py index 9c8749e..f5bd827 100644 --- a/tests/test_energysite_events.py +++ b/tests/test_energysite_events.py @@ -108,7 +108,7 @@ def make_stream() -> TeslemetryStream: def dispatch(stream: TeslemetryStream, event: dict[str, Any]) -> None: """Replicate stream.listen()'s per-event dispatch without a live connection.""" - for listener, filters in list(stream._listeners.values()): + for listener, filters, _internal in list(stream._listeners.values()): if recursive_match(filters, event): listener(event) diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py index 8fc8846..84a33b0 100644 --- a/tests/test_field_type_coercion.py +++ b/tests/test_field_type_coercion.py @@ -30,6 +30,7 @@ def async_add_listener( self, callback: Callable[[dict[str, Any]], None], filters: dict[str, Any] | None = None, + internal: bool = False, ) -> Callable[[], None]: # filters carries {"vin": ..., "data": {Signal: None}} — grab the field. # The vehicle's own internal config-sync listener has no "data" key; diff --git a/tests/test_sse_topics.py b/tests/test_sse_topics.py index 1d898da..af1b430 100644 --- a/tests/test_sse_topics.py +++ b/tests/test_sse_topics.py @@ -79,7 +79,7 @@ def make_stream(topics: Any = None) -> tuple[TeslemetryStream, FakeSession]: def dispatch(stream: TeslemetryStream, event: dict[str, Any]) -> None: """Replicate stream.listen()'s per-event dispatch without a live connection.""" - for listener, filters in list(stream._listeners.values()): + for listener, filters, _internal in list(stream._listeners.values()): if recursive_match(filters, event): listener(event) From 08080ff00000d433ebf2401a311025ac175e4cd3 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:06:56 +1000 Subject: [PATCH 3/9] Fix listener-restart regression and validate nested field entries async_add_listener's start check still used whole-registry emptiness, so an internal listener surviving auto-close made a later public listener's zero-to-one transition invisible and the owned task never restarted. Both the start and stop checks now count only public (non-internal) listeners. _on_config_event also accepted a "fields" dict whose entries weren't themselves dicts (e.g. a null), which add_field's no-op check then dereferenced with .get() and crashed on. Each entry is now validated; a bad one rejects the whole fields piece and keeps the prior record. --- teslemetry_stream/stream.py | 26 ++++++++++---- teslemetry_stream/vehicle.py | 8 ++++- tests/test_config_events.py | 37 ++++++++++++++++++++ tests/test_stream_lifecycle.py | 62 ++++++++++++++++++++++++++++++++++ 4 files changed, 125 insertions(+), 8 deletions(-) diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index f09664b..931d5d4 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -374,26 +374,38 @@ def async_add_listener( :param filters: Filters to apply to the updates. :param internal: True for a listener that keeps the client's own state fresh (e.g. a vehicle's config-sync listener) rather than - serving a consumer callback. Excluded from the "last listener - removed" auto-close check, so a bookkeeping-only listener can't - pin the connection open forever once every real listener is gone. + serving a consumer callback. Excluded from both the "first + listener" start check and the "last listener removed" auto-close + check - on either side of the registry, only public listeners + count - so a bookkeeping-only listener can neither pin the + connection open forever nor, by itself, block a later public + listener from restarting a closed one. :return: Function to remove the listener. """ - schedule_refresh = not self._listeners + + def has_public_listener() -> bool: + return any(not is_internal for _, _, is_internal in self._listeners.values()) + + # A transition from zero to one *public* listeners, not merely a + # non-empty registry - an internal listener surviving a prior + # auto-close must not block a later public listener from restarting + # the owned task. + schedule_refresh = not internal and not has_public_listener() def remove_listener() -> None: """ Remove update listener. """ self._listeners.pop(remove_listener) - if not any(not is_internal for _, _, is_internal in self._listeners.values()): + if not has_public_listener(): LOGGER.info("Shutting down stream as there are no more listeners") self.close() self._listeners[remove_listener] = (callback, filters, internal) - # This is the first listener - start the owned listen task, unless - # one is already running or manual mode delegates that to the caller. + # This is the first public listener - start the owned listen task, + # unless one is already running or manual mode delegates that to the + # caller. if ( schedule_refresh and not self.manual diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index eaae98b..c8af59e 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -154,7 +154,13 @@ def _on_config_event(self, event: dict[str, Any]) -> None: if "fields" in config: fields = config["fields"] - if isinstance(fields, dict): + # Every entry must itself be a dict (e.g. {"interval_seconds": 60} + # or {}) - `fields: dict[str, dict[str, int]]` - so a downstream + # `self.fields[field].get(...)` (add_field's no-op check) can't + # raise AttributeError on a null/scalar entry that snuck in. + if isinstance(fields, dict) and all( + isinstance(value, dict) for value in fields.values() + ): self.fields = fields else: LOGGER.warning( diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 70f24e1..1b3fb87 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -182,6 +182,43 @@ async def main() -> None: f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}", ) ) + + # A field entry that isn't itself a dict (e.g. null) is malformed + # shape too, even though the outer "fields" value is a dict - the + # whole "fields" piece is rejected, not just the bad entry, so it + # can't leave a null in self.fields that later crashes add_field. + good_fields = dict(vehicle.fields) + handler.messages.clear() + stream.config_listener( + { + "vin": VIN, + "config": {"fields": {"CarType": {}, "BatteryLevel": None}}, + } + ) + results.append( + check( + "a null nested field entry rejects the whole fields piece", + vehicle.fields == good_fields + and any("malformed" in m.lower() for m in handler.messages), + f"fields {vehicle.fields}, warnings {handler.messages}", + ) + ) + + # And, concretely, a later add_field for an unrelated field must not + # raise trying to .get() off the (rejected, never-stored) null entry. + try: + await vehicle.add_field("BatteryLevel", 60) + add_field_ok = True + except AttributeError as error: + add_field_ok = False + add_field_error = repr(error) + results.append( + check( + "add_field after a rejected null entry does not raise", + add_field_ok, + "" if add_field_ok else add_field_error, + ) + ) finally: logger.removeHandler(handler) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 2b0cf47..84ca7d5 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -243,6 +243,67 @@ async def test_close_prevents_reconnect_after_backoff(results: list[bool]) -> No ) +async def test_restart_after_public_readd_with_internal_listener_present( + results: list[bool], +) -> None: + """An internal (bookkeeping-only) listener surviving auto-close must not + block a later public listener from restarting the owned task.""" + session = FakeSession() + stream = make_stream(session) + + # An internal listener alone must not itself start the task - only + # public listeners drive connect/disconnect. + remove_internal = stream.async_add_listener(lambda event: None, internal=True) + results.append( + check( + "an internal-only listener does not start the owned task", + stream._listen_task is None, + ) + ) + + remove_public = stream.async_add_listener(lambda event: None) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append( + check( + "adding the first public listener connects", + session.calls == 1, + f"got {session.calls}", + ) + ) + + # Removing the last public listener auto-closes even though the + # internal listener remains registered. + remove_public() + await asyncio.sleep(0) + results.append( + check( + "removing the last public listener auto-closes despite the internal listener", + not stream.active, + ) + ) + + # A later public listener, added while only the internal one remains + # registered, must still restart the owned task - this is the bug: the + # registry was non-empty (internal listener) so the old whole-registry + # emptiness check never saw a zero-to-one transition. + remove_public2 = stream.async_add_listener(lambda event: None) + await asyncio.sleep(0) + await asyncio.sleep(0) + results.append( + check( + "re-adding a public listener afterwards reconnects", + session.calls == 2, + f"got {session.calls}", + ) + ) + + remove_public2() + remove_internal() + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -250,6 +311,7 @@ async def main() -> None: await test_cancel_while_blocked_reading(results) await test_close_during_connect(results) await test_close_prevents_reconnect_after_backoff(results) + await test_restart_after_public_readd_with_internal_listener_present(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") From 790d9be8a8144ec06ec324386be181a1e458796c Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:19:01 +1000 Subject: [PATCH 4/9] Refresh config record from REST before the no-op check when disconnected The config-sync listener can only observe server-side changes while connected, so a record that went stale during a disconnect (e.g. auto-closed after the last public listener was removed) could pass add_field/prefer_typed's no-op check and wrongly skip a change the server still needs. Both now call get_config() first whenever stream.connected is false, reusing the existing REST fetch rather than adding new state tracking. --- AGENTS.md | 5 +- teslemetry_stream/vehicle.py | 14 ++++ tests/test_batch_retry_storm.py | 3 + tests/test_config_events.py | 3 + tests/test_config_listener_lifecycle.py | 97 ++++++++++++++++++++++++- tests/test_config_update.py | 3 + tests/test_field_type_coercion.py | 3 + 7 files changed, 126 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 02c66da..bd0b2ce 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,10 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs..environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery. - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. -- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history: `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field; a missing piece is left untouched, and a malformed piece (wrong type) is logged and skipped without touching the other piece or the pending `_config`. Registration is deferred to the first `add_field`/`prefer_typed`/`update_config` call (not `__init__`) because `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop - eager construction-time registration broke the previously-synchronous `TeslemetryStream(vin=...)`/vehicle construction. `async_add_listener(..., internal=True)` marks it so it's excluded from the "last listener removed" auto-close check in `TeslemetryStream` - otherwise a permanently-registered internal listener would keep `_listeners` non-empty forever and the SSE connection would never close once every real listener unsubscribed. Because of this, any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)`. `tests/test_config_events.py` covers the record merge; `tests/test_config_listener_lifecycle.py` covers the lazy registration and the auto-close exclusion. +- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, on the first `add_field`/`prefer_typed`/`update_config` call rather than in `__init__` - `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop, and eager construction-time registration broke synchronous `TeslemetryStream(vin=...)`/vehicle construction. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. +- `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Both the "first listener starts the task" and "last listener removed auto-closes" checks count only non-internal (public) listeners - 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)`. +- Because the config-sync listener can only observe server-side changes while connected, `add_field`/`prefer_typed` call `_refresh_if_disconnected()` first, which re-fetches via REST `get_config()` whenever `stream.connected` is false - otherwise a record that went stale during a disconnect (e.g. auto-closed after the last public listener was removed) could pass the no-op check and wrongly skip a change the server still needs. Test doubles need a `connected` attribute (`True` skips the refresh) for this reason. +- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers lazy registration, the auto-close exclusion, and the disconnected-refresh case; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index c8af59e..d35f44d 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -298,6 +298,7 @@ async def post_config(self, config: dict[str, Any]) -> dict[str, Any]: async def add_field(self, field: Signal | str, interval: int | None = None) -> None: """Handle vehicle data from the stream.""" self._ensure_config_listener() + await self._refresh_if_disconnected() if isinstance(field, Signal): field = field.value @@ -317,10 +318,23 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N async def prefer_typed(self, prefer_typed: bool) -> None: """Set prefer typed.""" self._ensure_config_listener() + await self._refresh_if_disconnected() if self.preferTyped == prefer_typed: return await self.update_config({"prefer_typed": prefer_typed}) + async def _refresh_if_disconnected(self) -> None: + """Refresh the record from the REST API before trusting it for a no-op check. + + The config-sync listener can only observe server-side changes while + connected; while disconnected (e.g. every public listener removed + and the stream auto-closed) it may be stale, and add_field/ + prefer_typed's no-op check would wrongly skip a change the server + actually needs. + """ + if not self.stream.connected: + await self.get_config() + def _enable_field(self, field: Signal) -> None: """Enable a field for streaming from a listener.""" asyncio.create_task(self.add_field(field)) diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index 2035544..a6035d3 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -30,6 +30,9 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True + # Skips add_field/prefer_typed's disconnected-refresh path - these tests + # exercise the write path itself, not the reconnect-refresh behavior. + connected = True def async_add_listener( self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 1b3fb87..81331b9 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -26,6 +26,9 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that captures the config listener.""" manual = True + # Skips add_field/prefer_typed's disconnected-refresh path - these tests + # exercise the event-driven merge itself, not the reconnect-refresh behavior. + connected = True def __init__(self) -> None: self.config_listener: Callable[[dict[str, Any]], None] | None = None diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index dee007d..9ac46c2 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -15,6 +15,11 @@ trigger once every real/public listener was gone - the SSE connection leaked open. ``async_add_listener(..., internal=True)`` now excludes it from that check. +- While disconnected (e.g. right after that auto-close), the config-sync + listener can't observe a server-side change, so add_field/prefer_typed's + no-op check could trust a stale record and wrongly skip a change the + server actually needs. Both now refresh via the REST ``get_config()`` + before that check whenever ``stream.connected`` is false. """ from __future__ import annotations @@ -27,10 +32,29 @@ VIN = "TESTVIN0000000001" +class FakeConfigResponse: + """Stand-in for the aiohttp response get_config() awaits .json() on.""" + + def __init__(self, body: dict[str, Any]) -> None: + self.status = 200 + self._body = body + + async def json(self) -> dict[str, Any]: + return self._body + + class FakeSession: - """A session whose get() is never expected to be called in these tests.""" + """Serves the config REST endpoint; any other GET (e.g. a real SSE + connect) is unexpected in these tests.""" + + def __init__(self) -> None: + self.config_response: dict[str, Any] = {"fields": {}, "prefer_typed": False} + self.get_calls = 0 async def get(self, url: str, **kwargs: Any) -> Any: + if "/api/config/" in url: + self.get_calls += 1 + return FakeConfigResponse(self.config_response) raise AssertionError(f"unexpected session.get({url!r}) - these tests must not connect") @@ -138,10 +162,81 @@ async def patch_config(config: dict[str, Any]) -> dict[str, Any]: ) +async def test_stale_record_refreshed_before_noop_check_when_disconnected( + results: list[bool], +) -> None: + """Close via last-public-listener removal, mutate config externally + (simulated via the REST fixture), then re-add - add_field must not + wrongly no-op against the now-stale locally cached record.""" + stream = make_stream() + vehicle = TeslemetryStreamVehicle(stream, VIN) + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + + # Establish a locally cached record while "connected". + stream._response = object() # type: ignore[assignment] + stream._session.config_response = { # type: ignore[attr-defined] + "fields": {"BatteryLevel": {"interval_seconds": 60}}, + "prefer_typed": False, + } + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "the record is established while connected", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, + f"fields {vehicle.fields}", + ) + ) + + # Last public listener removed -> auto-close -> disconnected. The config + # response fixture is mutated externally while unreachable. + remove_public = stream.async_add_listener(lambda event: None) + stream._response = None + remove_public() + results.append(check("the stream is disconnected", not stream.connected)) + + stream._session.config_response = { # type: ignore[attr-defined] + "fields": {"BatteryLevel": {"interval_seconds": 30}}, + "prefer_typed": False, + } + + sent: list[dict[str, Any]] = [] + + async def patch_config_after_reconnect(config: dict[str, Any]) -> dict[str, Any]: + sent.append(dict(config)) + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config_after_reconnect # type: ignore[assignment,method-assign] + + # Re-add, still disconnected: the stale record (BatteryLevel@60) matches + # what's being requested, so an un-fixed no-op check would wrongly skip. + get_calls_before = stream._session.get_calls # type: ignore[attr-defined] + await vehicle.add_field("BatteryLevel", 60) + + results.append( + check( + "add_field refreshes from the REST API before the no-op check", + stream._session.get_calls == get_calls_before + 1, # type: ignore[attr-defined] + f"get_calls {stream._session.get_calls}", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "add_field does not wrongly no-op against the stale record", + len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + f"sent {sent}", + ) + ) + + async def main(pre_loop_results: list[bool]) -> None: results: list[bool] = list(pre_loop_results) await test_lazy_registration_happens_on_first_use(results) await test_auto_close_after_last_public_listener_removed(results) + await test_stale_record_refreshed_before_noop_check_when_disconnected(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 9acf03b..83e6645 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -29,6 +29,9 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True + # Skips add_field/prefer_typed's disconnected-refresh path - these tests + # exercise the no-op check itself, not the reconnect-refresh behavior. + connected = True def async_add_listener( self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py index 84a33b0..d786022 100644 --- a/tests/test_field_type_coercion.py +++ b/tests/test_field_type_coercion.py @@ -21,6 +21,9 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that just captures listeners.""" manual = True + # Skips add_field's disconnected-refresh path - this test pre-populates + # fields directly and asserts no HTTP happens, unrelated to reconnect-refresh. + connected = True def __init__(self) -> None: # maps Signal value -> wrapped listener callback From 9bcee2282d7a14ec735fc23bdfd119112444f484 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:28:55 +1000 Subject: [PATCH 5/9] Revert refresh-GET, gate the no-op skip on record liveness instead The mandatory get_config() refresh added a failure path and could storm the API with one GET per caller in a batch. The no-op skip is purely an optimization - the server handles a redundant PATCH fine - so add_field/prefer_typed now gate the skip on _record_is_live() (stream connected AND the config topic not filtered out via TeslemetryStream(topics=...)) instead of trying to force the record fresh. When not live, they send unconditionally: one redundant request, same as the pre-feature status quo. --- AGENTS.md | 4 +- teslemetry_stream/vehicle.py | 31 +++-- tests/test_batch_retry_storm.py | 5 +- tests/test_config_events.py | 5 +- tests/test_config_listener_lifecycle.py | 156 ++++++++++++------------ tests/test_config_update.py | 5 +- tests/test_field_type_coercion.py | 5 +- 7 files changed, 107 insertions(+), 104 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index bd0b2ce..fed3a23 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,8 +18,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. - `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, on the first `add_field`/`prefer_typed`/`update_config` call rather than in `__init__` - `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop, and eager construction-time registration broke synchronous `TeslemetryStream(vin=...)`/vehicle construction. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Both the "first listener starts the task" and "last listener removed auto-closes" checks count only non-internal (public) listeners - 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)`. -- Because the config-sync listener can only observe server-side changes while connected, `add_field`/`prefer_typed` call `_refresh_if_disconnected()` first, which re-fetches via REST `get_config()` whenever `stream.connected` is false - otherwise a record that went stale during a disconnect (e.g. auto-closed after the last public listener was removed) could pass the no-op check and wrongly skip a change the server still needs. Test doubles need a `connected` attribute (`True` skips the refresh) for this reason. -- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers lazy registration, the auto-close exclusion, and the disconnected-refresh case; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. +- The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. +- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers lazy registration, the auto-close exclusion, and all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index d35f44d..058df57 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -46,6 +46,7 @@ ShiftState, Signal, SpeedAssistLevel, + SseTopic, State, Status, SunroofInstalledState, @@ -298,12 +299,13 @@ async def post_config(self, config: dict[str, Any]) -> dict[str, Any]: async def add_field(self, field: Signal | str, interval: int | None = None) -> None: """Handle vehicle data from the stream.""" self._ensure_config_listener() - await self._refresh_if_disconnected() if isinstance(field, Signal): field = field.value - if field in self.fields and ( - interval is None or self.fields[field].get("interval_seconds") == interval + if ( + self._record_is_live() + and field in self.fields + and (interval is None or self.fields[field].get("interval_seconds") == interval) ): LOGGER.debug( "Streaming field %s already enabled @ %ss", @@ -318,22 +320,25 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N async def prefer_typed(self, prefer_typed: bool) -> None: """Set prefer typed.""" self._ensure_config_listener() - await self._refresh_if_disconnected() - if self.preferTyped == prefer_typed: + if self._record_is_live() and self.preferTyped == prefer_typed: return await self.update_config({"prefer_typed": prefer_typed}) - async def _refresh_if_disconnected(self) -> None: - """Refresh the record from the REST API before trusting it for a no-op check. + def _record_is_live(self) -> bool: + """Whether the record is being kept current and can gate the no-op skip. - The config-sync listener can only observe server-side changes while - connected; while disconnected (e.g. every public listener removed - and the stream auto-closed) it may be stale, and add_field/ - prefer_typed's no-op check would wrongly skip a change the server - actually needs. + The skip is purely an optimization - the server handles a redundant + PATCH fine - so this only needs to answer "is the config-sync + listener actually able to observe a server-side change right now", + not force the record fresh. That requires both a live connection and + the `config` topic not being filtered out via `TeslemetryStream + (topics=...)`; if either is false, add_field/prefer_typed skip the + no-op check and always send, same as the pre-feature status quo. """ if not self.stream.connected: - await self.get_config() + return False + topics = self.stream.topics + return topics is None or SseTopic.CONFIG in topics def _enable_field(self, field: Signal) -> None: """Enable a field for streaming from a listener.""" diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index a6035d3..5655646 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -30,9 +30,10 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True - # Skips add_field/prefer_typed's disconnected-refresh path - these tests - # exercise the write path itself, not the reconnect-refresh behavior. + # Keeps the record "live" (see _record_is_live) so add_field/prefer_typed's + # no-op check runs - these tests exercise the write path itself. connected = True + topics = None def async_add_listener( self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 81331b9..23913fc 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -26,9 +26,10 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that captures the config listener.""" manual = True - # Skips add_field/prefer_typed's disconnected-refresh path - these tests - # exercise the event-driven merge itself, not the reconnect-refresh behavior. + # Keeps the record "live" (see _record_is_live) so add_field/prefer_typed's + # no-op check runs - these tests exercise the event-driven merge itself. connected = True + topics = None def __init__(self) -> None: self.config_listener: Callable[[dict[str, Any]], None] | None = None diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index 9ac46c2..3b6cb02 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -1,6 +1,6 @@ """Regression tests for the vehicle config-sync listener's lifecycle. -Two defects flagged in review of the config-consume feature: +Defects flagged across review of the config-consume feature: - Eagerly registering the internal config listener in ``TeslemetryStreamVehicle.__init__`` called ``TeslemetryStream. @@ -15,11 +15,16 @@ trigger once every real/public listener was gone - the SSE connection leaked open. ``async_add_listener(..., internal=True)`` now excludes it from that check. -- While disconnected (e.g. right after that auto-close), the config-sync - listener can't observe a server-side change, so add_field/prefer_typed's - no-op check could trust a stale record and wrongly skip a change the - server actually needs. Both now refresh via the REST ``get_config()`` - before that check whenever ``stream.connected`` is false. +- The config-sync listener can only observe a server-side change while + connected AND the ``config`` topic isn't filtered out via + ``TeslemetryStream(topics=...)``. A first attempt at handling this forced + a REST refresh before the no-op check whenever disconnected - but that + added a failure path and could storm the API with GETs for a batch of + callers. Reverted: the no-op *skip* is purely an optimization (a + redundant PATCH is harmless), so it's now gated on the record actually + being live-maintained (``_record_is_live()``) rather than force-freshened; + otherwise add_field/prefer_typed just send unconditionally, exactly the + pre-feature status quo. """ from __future__ import annotations @@ -32,29 +37,10 @@ VIN = "TESTVIN0000000001" -class FakeConfigResponse: - """Stand-in for the aiohttp response get_config() awaits .json() on.""" - - def __init__(self, body: dict[str, Any]) -> None: - self.status = 200 - self._body = body - - async def json(self) -> dict[str, Any]: - return self._body - - class FakeSession: - """Serves the config REST endpoint; any other GET (e.g. a real SSE - connect) is unexpected in these tests.""" - - def __init__(self) -> None: - self.config_response: dict[str, Any] = {"fields": {}, "prefer_typed": False} - self.get_calls = 0 + """A session whose get() is never expected to be called in these tests.""" async def get(self, url: str, **kwargs: Any) -> Any: - if "/api/config/" in url: - self.get_calls += 1 - return FakeConfigResponse(self.config_response) raise AssertionError(f"unexpected session.get({url!r}) - these tests must not connect") @@ -76,6 +62,20 @@ def make_stream(**kwargs: Any) -> TeslemetryStream: ) +def make_vehicle_with_capture( + stream: TeslemetryStream, +) -> tuple[TeslemetryStreamVehicle, list[dict[str, Any]]]: + vehicle = TeslemetryStreamVehicle(stream, VIN) + sent: list[dict[str, Any]] = [] + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + sent.append(dict(config)) + return {"updated_vehicles": 1} + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + return vehicle, sent + + def test_sync_construction_without_a_loop() -> bool: """Must run before any event loop exists - constructing a stream+vehicle synchronously (the library's documented pre-async-context usage) must not @@ -91,7 +91,7 @@ def test_sync_construction_without_a_loop() -> bool: async def test_lazy_registration_happens_on_first_use(results: list[bool]) -> None: stream = make_stream() - vehicle = TeslemetryStreamVehicle(stream, VIN) + vehicle, _sent = make_vehicle_with_capture(stream) results.append( check( @@ -101,10 +101,6 @@ async def test_lazy_registration_happens_on_first_use(results: list[bool]) -> No ) ) - async def patch_config(config: dict[str, Any]) -> dict[str, Any]: - return {"updated_vehicles": 1} - - vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] await vehicle.add_field("BatteryLevel") results.append( @@ -124,12 +120,8 @@ async def patch_config(config: dict[str, Any]) -> dict[str, Any]: async def test_auto_close_after_last_public_listener_removed(results: list[bool]) -> None: stream = make_stream() - vehicle = TeslemetryStreamVehicle(stream, VIN) - - async def patch_config(config: dict[str, Any]) -> dict[str, Any]: - return {"updated_vehicles": 1} + vehicle, _sent = make_vehicle_with_capture(stream) - vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] # Registers the internal (excluded) listener. await vehicle.add_field("BatteryLevel") @@ -162,71 +154,71 @@ async def patch_config(config: dict[str, Any]) -> dict[str, Any]: ) -async def test_stale_record_refreshed_before_noop_check_when_disconnected( - results: list[bool], -) -> None: - """Close via last-public-listener removal, mutate config externally - (simulated via the REST fixture), then re-add - add_field must not - wrongly no-op against the now-stale locally cached record.""" +async def test_cold_stream_add_field_sends_patch_unconditionally(results: list[bool]) -> None: + """A never-connected (or disconnected) stream can't have observed a + server-side change, so the no-op skip must not apply - send the PATCH + unconditionally rather than trying to force the record fresh.""" stream = make_stream() - vehicle = TeslemetryStreamVehicle(stream, VIN) + vehicle, sent = make_vehicle_with_capture(stream) + vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} # matches the request below - async def patch_config(config: dict[str, Any]) -> dict[str, Any]: - return {"updated_vehicles": 1} - - vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + results.append(check("the stream starts disconnected", not stream.connected)) - # Establish a locally cached record while "connected". - stream._response = object() # type: ignore[assignment] - stream._session.config_response = { # type: ignore[attr-defined] - "fields": {"BatteryLevel": {"interval_seconds": 60}}, - "prefer_typed": False, - } await vehicle.add_field("BatteryLevel", 60) + results.append( check( - "the record is established while connected", - vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, - f"fields {vehicle.fields}", + "add_field sends the PATCH even though the record already matches", + len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + f"sent {sent}", ) ) - # Last public listener removed -> auto-close -> disconnected. The config - # response fixture is mutated externally while unreachable. - remove_public = stream.async_add_listener(lambda event: None) - stream._response = None - remove_public() - results.append(check("the stream is disconnected", not stream.connected)) - stream._session.config_response = { # type: ignore[attr-defined] - "fields": {"BatteryLevel": {"interval_seconds": 30}}, - "prefer_typed": False, - } +async def test_filtered_config_topic_sends_patch_unconditionally(results: list[bool]) -> None: + """Even while connected, if `topics=` filters out the config topic the + config-sync listener never receives anything - the record can't be + trusted, so the no-op skip must not apply.""" + stream = make_stream(topics=["state"]) + vehicle, sent = make_vehicle_with_capture(stream) + vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} - sent: list[dict[str, Any]] = [] + stream._response = object() # type: ignore[assignment] # simulate a live connection + results.append(check("the stream is connected", stream.connected)) - async def patch_config_after_reconnect(config: dict[str, Any]) -> dict[str, Any]: - sent.append(dict(config)) - return {"updated_vehicles": 1} + await vehicle.add_field("BatteryLevel", 60) - vehicle.patch_config = patch_config_after_reconnect # type: ignore[assignment,method-assign] + results.append( + check( + "add_field sends the PATCH when the config topic is filtered out", + len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + f"sent {sent}", + ) + ) - # Re-add, still disconnected: the stale record (BatteryLevel@60) matches - # what's being requested, so an un-fixed no-op check would wrongly skip. - get_calls_before = stream._session.get_calls # type: ignore[attr-defined] - await vehicle.add_field("BatteryLevel", 60) +async def test_connected_and_subscribed_record_match_skips(results: list[bool]) -> None: + """The no-op skip only applies once both conditions hold: connected, and + the config topic isn't filtered out (default `topics=None` subscribes + to everything).""" + stream = make_stream() + vehicle, sent = make_vehicle_with_capture(stream) + vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} + + stream._response = object() # type: ignore[assignment] # simulate a live connection results.append( check( - "add_field refreshes from the REST API before the no-op check", - stream._session.get_calls == get_calls_before + 1, # type: ignore[attr-defined] - f"get_calls {stream._session.get_calls}", # type: ignore[attr-defined] + "the stream is connected and subscribed to every topic", + stream.connected and stream.topics is None, ) ) + + await vehicle.add_field("BatteryLevel", 60) + results.append( check( - "add_field does not wrongly no-op against the stale record", - len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + "add_field skips the PATCH when the record is live-maintained and matches", + sent == [], f"sent {sent}", ) ) @@ -236,7 +228,9 @@ async def main(pre_loop_results: list[bool]) -> None: results: list[bool] = list(pre_loop_results) await test_lazy_registration_happens_on_first_use(results) await test_auto_close_after_last_public_listener_removed(results) - await test_stale_record_refreshed_before_noop_check_when_disconnected(results) + await test_cold_stream_add_field_sends_patch_unconditionally(results) + await test_filtered_config_topic_sends_patch_unconditionally(results) + await test_connected_and_subscribed_record_match_skips(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 83e6645..e06c73a 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -29,9 +29,10 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True - # Skips add_field/prefer_typed's disconnected-refresh path - these tests - # exercise the no-op check itself, not the reconnect-refresh behavior. + # Keeps the record "live" (see _record_is_live) so add_field/prefer_typed's + # no-op check runs - these tests exercise that check itself. connected = True + topics = None def async_add_listener( self, callback: Any, filters: dict[str, Any] | None = None, internal: bool = False diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py index d786022..0c1f8e2 100644 --- a/tests/test_field_type_coercion.py +++ b/tests/test_field_type_coercion.py @@ -21,9 +21,10 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that just captures listeners.""" manual = True - # Skips add_field's disconnected-refresh path - this test pre-populates - # fields directly and asserts no HTTP happens, unrelated to reconnect-refresh. + # Keeps the record "live" (see _record_is_live) so add_field's no-op + # check short-circuits, matching this test's pre-populated fields. connected = True + topics = None def __init__(self) -> None: # maps Signal value -> wrapped listener callback From b431af981c9f23ea30f37d8006f3ef8fe1819cf0 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:37:57 +1000 Subject: [PATCH 6/9] Register the config-sync listener at construction, not lazily Lazy registration left a gap: a stream already connected before the listener existed could have dispatched a config event that was simply never seen. The loop hazard that motivated deferring it was never about timing - async_add_listener's schedule_refresh is unconditionally false for internal=True, so it never reaches the asyncio.create_task() call that needs a running loop, regardless of when it's registered. Register it unconditionally in __init__ instead. No connection can now predate the listener, and construction stays loop-free (still asserted by a dedicated regression test). Removes _ensure_config_listener() and its call sites; _record_is_live() is unaffected. --- AGENTS.md | 6 +- teslemetry_stream/vehicle.py | 36 +++------- tests/test_config_events.py | 20 ++---- tests/test_config_listener_lifecycle.py | 88 +++++++++++++------------ 4 files changed, 66 insertions(+), 84 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index fed3a23..a69cd4c 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,10 +16,10 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs..environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery. - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. -- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `_ensure_config_listener()` lazily registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, on the first `add_field`/`prefer_typed`/`update_config` call rather than in `__init__` - `TeslemetryStream.async_add_listener()` calls `asyncio.create_task()` for a stream's first-ever listener, which needs a running loop, and eager construction-time registration broke synchronous `TeslemetryStream(vin=...)`/vehicle construction. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. -- `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Both the "first listener starts the task" and "last listener removed auto-closes" checks count only non-internal (public) listeners - 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)`. +- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `__init__` registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, unconditionally at construction - not lazily - so no connection can ever predate it and miss an event. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. +- `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)`. - The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. -- `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers lazy registration, the auto-close exclusion, and all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. +- `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 all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. - `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. ## Maintaining this file diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 058df57..e09fb57 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -87,11 +87,16 @@ def __init__(self, stream: TeslemetryStream, vin: str): # Callers that arrive while it is running merge into `_config` and # await it instead of starting their own PATCH. self._flight = None - # Registration is deferred to first use (see _ensure_config_listener) - - # eagerly calling async_add_listener here would create the stream's - # owned task via asyncio.create_task before a loop necessarily exists, - # breaking a synchronous `TeslemetryStream(vin=...)`/vehicle construction. - self._config_listener_registered = False + # Registered from birth, not lazily, so no connection can ever + # predate this listener and miss a config event. Safe outside a + # running loop: `internal=True` makes async_add_listener's + # schedule_refresh unconditionally False, so it never reaches the + # asyncio.create_task() call that requires one. + self.stream.async_add_listener( + self._on_config_event, + {Key.VIN: self.vin, Key.CONFIG: None}, + internal=True, + ) @property def config(self) -> dict[str, Any]: @@ -121,24 +126,6 @@ async def get_config(self) -> None: req.raise_for_status() - def _ensure_config_listener(self) -> None: - """Register the internal config-sync listener once, lazily. - - Keeps `fields`/`preferTyped` current when the server applies a - config change outside this client (another client, the console, - a Teslemetry-side migration), so add_field/prefer_typed's no-op - checks don't act on stale history. Marked `internal` so it doesn't - pin the stream open once every public listener is removed. - """ - if self._config_listener_registered: - return - self._config_listener_registered = True - self.stream.async_add_listener( - self._on_config_event, - {Key.VIN: self.vin, Key.CONFIG: None}, - internal=True, - ) - def _on_config_event(self, event: dict[str, Any]) -> None: """Sync the record from a server-pushed config event. @@ -188,7 +175,6 @@ async def update_config(self, config: dict[str, Any]) -> None: flush for this vehicle rather than starting a new one, so that a batch of listeners scheduled at the same time produces one PATCH. """ - self._ensure_config_listener() async with self.lock: self._config = merge(config, self._config) @@ -298,7 +284,6 @@ async def post_config(self, config: dict[str, Any]) -> dict[str, Any]: async def add_field(self, field: Signal | str, interval: int | None = None) -> None: """Handle vehicle data from the stream.""" - self._ensure_config_listener() if isinstance(field, Signal): field = field.value @@ -319,7 +304,6 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N async def prefer_typed(self, prefer_typed: bool) -> None: """Set prefer typed.""" - self._ensure_config_listener() if self._record_is_live() and self.preferTyped == prefer_typed: return await self.update_config({"prefer_typed": prefer_typed}) diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 23913fc..4348b8a 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -3,12 +3,12 @@ The server pushes a ``config`` event (``Key.CONFIG`` / ``SseTopic.CONFIG``) shaped like ``{"vin": ..., "config": {"fields": {...}, "prefer_typed": bool}}``, mirroring the REST ``get_config`` response body. ``TeslemetryStreamVehicle`` -lazily registers an internal listener for it (on first ``add_field``/ -``prefer_typed``/``update_config`` call, not at construction - see -``test_config_listener_lifecycle.py`` for why) so ``fields``/``preferTyped`` -- and therefore the ``add_field``/``prefer_typed`` no-op checks - reflect -current server truth rather than only what this client has itself -requested or observed at connect. +registers an internal listener for it at construction (see +``test_config_listener_lifecycle.py`` for why that's safe even outside a +running loop) so ``fields``/``preferTyped`` - and therefore the +``add_field``/``prefer_typed`` no-op checks - reflect current server truth +rather than only what this client has itself requested or observed at +connect. """ from __future__ import annotations @@ -63,16 +63,10 @@ def check(label: str, ok: bool, detail: str = "") -> bool: def make_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]: - """Build a vehicle that records PATCH payloads instead of sending them. - - Registration is lazy (see ``test_config_listener_lifecycle.py``), so - force it here the same way a real first ``add_field``/``prefer_typed``/ - ``update_config`` call would, to exercise the merge logic in isolation. - """ + """Build a vehicle that records PATCH payloads instead of sending them.""" stream = FakeStream() vehicle = TeslemetryStreamVehicle(stream, VIN) # type: ignore[arg-type] vehicle.sent = [] # type: ignore[attr-defined] - vehicle._ensure_config_listener() async def patch_config(config: dict[str, Any]) -> dict[str, Any]: vehicle.sent.append(dict(config)) # type: ignore[attr-defined] diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index 3b6cb02..f814754 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -2,29 +2,36 @@ Defects flagged across review of the config-consume feature: -- Eagerly registering the internal config listener in - ``TeslemetryStreamVehicle.__init__`` called ``TeslemetryStream. - async_add_listener()`` -> ``asyncio.create_task()`` for the very first - listener, which requires a running event loop. That broke the - previously-synchronous ``TeslemetryStream(vin=...)``/vehicle - construction. Registration is now deferred to first use (inside - ``add_field``/``prefer_typed``/``update_config``, all async). -- The internal listener, once registered, stayed in - ``TeslemetryStream._listeners`` forever, so the "last listener removed" - auto-close check (which fires on an empty ``_listeners``) could never - trigger once every real/public listener was gone - the SSE connection - leaked open. ``async_add_listener(..., internal=True)`` now excludes it - from that check. +- A first attempt eagerly registered the internal config listener in + ``TeslemetryStreamVehicle.__init__``, which risked ``TeslemetryStream. + async_add_listener()`` -> ``asyncio.create_task()`` needing a running + event loop. That was worked around by deferring registration to first + use (inside ``add_field``/``prefer_typed``/``update_config``) - but + lazy registration left a gap: a stream that was already connected + before the listener existed could have dispatched a config event that + was simply never seen. +- The actual fix for the loop hazard was structural, not timing: + ``async_add_listener(..., internal=True)`` makes its + ``schedule_refresh`` (the gate on the ``asyncio.create_task()`` call) + unconditionally false for an internal-only registration - so + registering the config listener eagerly in ``__init__`` is safe outside + a running loop, and the lazy-registration gap is gone: the listener + exists from construction, so no connection can ever predate it. +- The same ``internal=True`` flag also excludes it from the "last + listener removed" auto-close check (and its counterpart "first listener + starts the task" check) - otherwise a permanently-registered internal + listener would keep ``_listeners`` non-empty forever, and a later public + listener's own zero-to-one transition couldn't restart a closed stream. - The config-sync listener can only observe a server-side change while connected AND the ``config`` topic isn't filtered out via - ``TeslemetryStream(topics=...)``. A first attempt at handling this forced - a REST refresh before the no-op check whenever disconnected - but that - added a failure path and could storm the API with GETs for a batch of - callers. Reverted: the no-op *skip* is purely an optimization (a - redundant PATCH is harmless), so it's now gated on the record actually - being live-maintained (``_record_is_live()``) rather than force-freshened; - otherwise add_field/prefer_typed just send unconditionally, exactly the - pre-feature status quo. + ``TeslemetryStream(topics=...)``. A separate attempt at handling *that* + forced a REST refresh before the no-op check whenever disconnected - + reverted, since it added a failure path and could storm the API with + GETs for a batch of callers. The no-op *skip* is purely an optimization + (a redundant PATCH is harmless), so it's gated on the record actually + being live-maintained (``_record_is_live()``) rather than + force-freshened; otherwise add_field/prefer_typed just send + unconditionally, exactly the pre-feature status quo. """ from __future__ import annotations @@ -79,33 +86,32 @@ async def patch_config(config: dict[str, Any]) -> dict[str, Any]: def test_sync_construction_without_a_loop() -> bool: """Must run before any event loop exists - constructing a stream+vehicle synchronously (the library's documented pre-async-context usage) must not - require or start one.""" + require or start one, and the config listener must already be registered + by the time construction returns (internal=True never reaches the + asyncio.create_task() call that would need a loop).""" label = "TeslemetryStream(vin=...) construction outside a running loop does not raise" try: + # TeslemetryStream(vin=...) constructs its own TeslemetryStreamVehicle + # internally (get_vehicle), which is exactly the construction path + # that must stay loop-free. stream = make_stream(vin=VIN) - TeslemetryStreamVehicle(stream, VIN) - return check(label, True) except RuntimeError as error: return check(label, False, f"raised {error!r}") + ok = check(label, True) + return check( + "the config listener is registered by the time construction returns", + len(stream._listeners) == 1, + f"listeners {len(stream._listeners)}", + ) and ok -async def test_lazy_registration_happens_on_first_use(results: list[bool]) -> None: +async def test_registration_happens_at_construction(results: list[bool]) -> None: stream = make_stream() - vehicle, _sent = make_vehicle_with_capture(stream) + _vehicle, _sent = make_vehicle_with_capture(stream) results.append( check( - "no listener is registered at construction", - len(stream._listeners) == 0, - f"listeners {len(stream._listeners)}", - ) - ) - - await vehicle.add_field("BatteryLevel") - - results.append( - check( - "the internal config listener is registered by the first add_field call", + "the internal config listener is registered by construction, before any call", len(stream._listeners) == 1, f"listeners {len(stream._listeners)}", ) @@ -120,10 +126,8 @@ async def test_lazy_registration_happens_on_first_use(results: list[bool]) -> No async def test_auto_close_after_last_public_listener_removed(results: list[bool]) -> None: stream = make_stream() - vehicle, _sent = make_vehicle_with_capture(stream) - - # Registers the internal (excluded) listener. - await vehicle.add_field("BatteryLevel") + # The internal listener registers at construction; no call needed to set it up. + _vehicle, _sent = make_vehicle_with_capture(stream) # A real/public listener on top of the internal one. remove_public = stream.async_add_listener(lambda event: None) @@ -226,7 +230,7 @@ async def test_connected_and_subscribed_record_match_skips(results: list[bool]) async def main(pre_loop_results: list[bool]) -> None: results: list[bool] = list(pre_loop_results) - await test_lazy_registration_happens_on_first_use(results) + await test_registration_happens_at_construction(results) await test_auto_close_after_last_public_listener_removed(results) await test_cold_stream_add_field_sends_patch_unconditionally(results) await test_filtered_config_topic_sends_patch_unconditionally(results) From 744fd2b2ba5a6d8b823afa190a68dd20c1f19ab1 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:44:07 +1000 Subject: [PATCH 7/9] Copy fields from a config event instead of aliasing it _on_config_event stored the same dict object the event exposed to public listeners, so a consumer mutating event["config"]["fields"] in place after delivery silently corrupted the last-known-good record. Store a copy of the fields mapping and each nested field config. --- AGENTS.md | 2 +- teslemetry_stream/vehicle.py | 5 ++++- tests/test_config_events.py | 24 ++++++++++++++++++++++++ 3 files changed, 29 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index a69cd4c..45948ff 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `site_info` events do not carry `tariff_content`/`tariff_content_v2`; the V2 tariff is its own `tariff_content_v2` event/listener (`listen_TariffContentV2`), same envelope shape as `site_info`, with a `None` body meaning an explicit server-side removal rather than "not received yet". Both share the same silence-means-no-change contract - freshness lives in REST, never in event cadence. There is deliberately no library helper recombining `site_info` and `tariff_content_v2` into one document - that would only ever cover the V2 tariff (legacy V1 `tariff_content` has no SSE topic and stays REST-only by design); a consumer wanting both tariffs together should use the REST site_info endpoint. - Releases (tag `v*.*.*`) go through `.github/workflows/release.yml` directly - it's the sole top-level workflow, triggered on the tag push: `lint` + the full `test` python-version matrix (mirrors `ci.yml`) must pass on the exact release SHA before `build` (single Python, build+twine) runs, and only then do the `pypi` environment's protection rules (and its trusted-publishing OIDC) allow `publish-to-pypi`. It must stay a top-level workflow, not a `workflow_call` reusable one - PyPI's trusted publisher is configured for the `release.yml` + `pypi` environment identity, and a reusable-workflow caller signs PEP 740 attestations under the caller's identity instead, which that publisher check rejects. The `pypi` GitHub environment itself (required reviewers, deployment branches) is admin-configured outside this repo's files. `jobs..environment.name`/`.url` cannot reference the `env` context (only `github`, `inputs`, `vars`, `needs`, `secrets`, `strategy`, `matrix` resolve there) - referencing `env.*` there is a workflow-file parse error that fails the whole file at startup, before trigger filtering, so it fails every push (not just tags) with zero jobs and no logs. Use literal values or `vars.*` instead. `release.yml` also carries `workflow_dispatch` so a tag whose run failed before publish can be re-run manually without tag surgery. - `TeslemetryStream(topics=...)` is an optional exact SSE wire-event allowlist sent as the connection's `topics` query param; `SseTopic` in `const.py` is the closed set the server recognizes, and `SSE_VEHICLE_TOPICS`/`SSE_ENERGY_TOPICS`/`SSE_ALL_TOPICS` are client-side presets - flat per-product-kind lists of exact wire names, deliberately not further split by whether a topic happens to have a connect-time snapshot server-side; that's upstream server behavior, not something this library encodes. Omitting `topics` (`None`) is legacy-all forever - every applicable event delivered unfiltered. An explicitly empty iterable is rejected with `ValueError` at construction time rather than silently falling back to legacy-all - "no topics" must not mean "all topics", mirroring the server's own 400 on an empty `topics` value. A bare `str`/`SseTopic` is accepted as a single topic rather than iterated character-by-character - `topics` type-checks `str | Iterable[str] | None` precisely because a lone string also satisfies `Iterable[str]`, the classic footgun. `tests/test_sse_topics.py` covers the tariff listener, its null-removal signal, the `topics` param's URL construction, the empty-iterable rejection, and the bare-string/bare-`SseTopic` case. -- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `__init__` registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, unconditionally at construction - not lazily - so no connection can ever predate it and miss an event. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. +- `TeslemetryStreamVehicle` keeps `fields`/`preferTyped` fresh against server-side changes (another client, the console, a Teslemetry migration), not just this client's own history. `__init__` registers an internal listener (`_on_config_event`, filtered on `{Key.VIN, Key.CONFIG: None}`) on the `config` SSE topic, shaped `{vin, config: {fields, prefer_typed}}` like the REST `get_config` body, unconditionally at construction - not lazily - so no connection can ever predate it and miss an event. A well-typed `fields`/`prefer_typed` piece replaces the corresponding record field (every nested `fields` entry must itself be a dict - one bad entry, e.g. a null, rejects the whole `fields` piece rather than leaving something `add_field` would later crash on); a missing piece is left untouched; a malformed piece is logged and skipped without touching the other piece or the pending `_config`. The stored `fields` dict (and each nested per-field dict) is copied, never the same object handed to public listeners for that same event - a consumer mutating its event in place must not corrupt the record. - `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)`. - The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. - `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 all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index e09fb57..e28c5f6 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -149,7 +149,10 @@ def _on_config_event(self, event: dict[str, Any]) -> None: if isinstance(fields, dict) and all( isinstance(value, dict) for value in fields.values() ): - self.fields = fields + # Copied, not aliased - the event dict is also handed to + # public listeners, and a consumer mutating it in place + # must not corrupt this record. + self.fields = {field: dict(value) for field, value in fields.items()} else: LOGGER.warning( "Ignoring malformed fields in config event for %s: %r", diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 4348b8a..1737550 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -100,6 +100,30 @@ async def main() -> None: ) ) + # Mutating the event dict after delivery must not corrupt the stored + # record - the config-sync listener also hands this same dict to public + # listeners, so storing an alias to it would let a consumer's in-place + # edit silently corrupt the last-known-good record. + aliasing_vehicle, aliasing_stream = make_vehicle() + assert aliasing_stream.config_listener is not None + delivered_event = { + "vin": VIN, + "config": { + "fields": {"CarType": {"interval_seconds": 60}}, + "prefer_typed": False, + }, + } + aliasing_stream.config_listener(delivered_event) + delivered_event["config"]["fields"]["CarType"]["interval_seconds"] = 999 + delivered_event["config"]["fields"]["Injected"] = {"interval_seconds": 1} + results.append( + check( + "mutating the event dict after delivery does not corrupt the record", + aliasing_vehicle.fields == {"CarType": {"interval_seconds": 60}}, + f"fields {aliasing_vehicle.fields}", + ) + ) + # A request that now matches the updated record is skipped - no PATCH sent. await vehicle.add_field("BatteryLevel", 60) results.append( From 28e9c61bde09c816092d876b2532f2d73a9e14a3 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:51:57 +1000 Subject: [PATCH 8/9] Dispatch over a snapshot so mid-dispatch listener adds can't kill the loop Eager config-listener registration means a callback that calls get_vehicle() for an uncached VIN mid-dispatch now inserts into _listeners while listen() is iterating it live, raising "dictionary changed size during iteration" and killing the listen task outside the per-callback error handler. listen() now iterates list(self._listeners.values()), a snapshot, so a callback can safely create vehicles or add listeners mid-dispatch. --- AGENTS.md | 2 +- teslemetry_stream/stream.py | 7 ++- tests/test_stream_lifecycle.py | 88 ++++++++++++++++++++++++++++++++-- 3 files changed, 92 insertions(+), 5 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 45948ff..9113c58 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ 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)`. - The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. - `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 all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. -- `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. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, and close-during-backoff. +- `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 `list(self._listeners.values())`, a snapshot, not the live dict - a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN, which registers that vehicle's internal config listener) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop. `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, and a listener mutating `_listeners` mid-dispatch. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 931d5d4..259dfe2 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -436,7 +436,12 @@ async def listen(self) -> None: try: async for event in self: if event: - for listener, filters, _internal in self._listeners.values(): + # A snapshot, not a live view - a callback that creates a + # vehicle (get_vehicle) or otherwise adds a listener + # mid-dispatch must not mutate _listeners while this is + # iterating it, which would raise RuntimeError and kill + # the loop. + for listener, filters, _internal in list(self._listeners.values()): if recursive_match(filters, event): try: listener(event) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 84ca7d5..ebb0dad 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -12,6 +12,7 @@ import asyncio import contextlib +from collections.abc import Callable from typing import Any import aiohttp @@ -37,13 +38,35 @@ def fail(self, exc: BaseException) -> None: self._blocker.set_exception(exc) +class FakeEventContent: + """Async-iterable response body yielding canned SSE `data:` lines, then + blocking until failed or cancelled (like `FakeContent`).""" + + def __init__(self, lines: list[bytes]) -> None: + self._lines = list(lines) + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeEventContent: + return self + + async def __anext__(self) -> bytes: + if self._lines: + return self._lines.pop(0) + 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: + def __init__(self, content: Any = None) -> None: self.url = "https://fake.teslemetry.com/sse" self.status = 200 - self.content = FakeContent() + self.content = content if content is not None else FakeContent() self.closed = False def close(self) -> None: @@ -57,12 +80,16 @@ def __init__(self) -> None: self.calls = 0 self.responses: list[FakeResponse] = [] self.gate: asyncio.Event | None = None + # Overridable factory for the response's `content` - defaults to the + # blocks-forever FakeContent when unset. + self.content_factory: Callable[[], Any] | None = None async def get(self, url: str, **kwargs: Any) -> FakeResponse: self.calls += 1 if self.gate is not None: await self.gate.wait() - response = FakeResponse() + content = self.content_factory() if self.content_factory else None + response = FakeResponse(content) self.responses.append(response) return response @@ -304,6 +331,60 @@ async def test_restart_after_public_readd_with_internal_listener_present( await asyncio.sleep(0) +async def test_dispatch_survives_listener_creating_vehicle_mid_iteration( + results: list[bool], +) -> None: + """A callback that calls get_vehicle() for an uncached VIN - or otherwise + adds a listener - mid-dispatch inserts into `_listeners` while `listen()` + is iterating it. Dispatching over a snapshot means that must not raise + and kill the loop; both queued events should still be delivered.""" + session = FakeSession() + session.content_factory = lambda: FakeEventContent( + [ + b'data: {"vin": "A", "state": "online"}\n', + b'data: {"vin": "A", "state": "online"}\n', + ] + ) + stream = make_stream(session) + + delivered: list[dict[str, Any]] = [] + + def mutate_during_dispatch(event: dict[str, Any]) -> None: + delivered.append(event) + # Registers a new internal listener - a mid-dispatch mutation of + # the exact dict listen() is iterating. + stream.get_vehicle(f"NEWVIN{len(delivered)}") + + stream.async_add_listener(mutate_during_dispatch, {"vin": None}) + + for _ in range(5): + await asyncio.sleep(0) + + results.append( + check( + "the listen task survives a listener mutating _listeners mid-dispatch", + stream._listen_task is not None and not stream._listen_task.done(), + ) + ) + results.append( + check( + "both queued events are delivered - dispatch continues past the mutation", + len(delivered) == 2, + f"delivered {len(delivered)}", + ) + ) + results.append( + check( + "each callback-created vehicle registered its own internal listener", + len(stream.vehicles) == 2, + f"vehicles {list(stream.vehicles)}", + ) + ) + + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -312,6 +393,7 @@ async def main() -> None: await test_close_during_connect(results) await test_close_prevents_reconnect_after_backoff(results) await test_restart_after_public_readd_with_internal_listener_present(results) + await test_dispatch_survives_listener_creating_vehicle_mid_iteration(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") From b9fd8034ecba8997caaa5a30747b35f898f659f3 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Tue, 11 Aug 2026 22:58:10 +1000 Subject: [PATCH 9/9] Dispatch internal listeners before public ones A public callback could run before the internal config-sync listener purely by registration order, mutate event["config"] in place, and have the internal listener cache the already-mutated value - the defensive copy in _on_config_event happened too late to help. listen() now dispatches over a snapshot sorted internal-first, so bookkeeping listeners always see the pristine event regardless of registration order. --- AGENTS.md | 2 +- teslemetry_stream/stream.py | 7 +++-- tests/test_stream_lifecycle.py | 50 ++++++++++++++++++++++++++++++++++ 3 files changed, 56 insertions(+), 3 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index 9113c58..850cc50 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -20,7 +20,7 @@ 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)`. - The config-sync listener can only observe a server-side change while connected AND while the `config` topic isn't filtered out via `TeslemetryStream(topics=...)`. `add_field`/`prefer_typed`'s no-op skip is gated on `_record_is_live()` (both conditions true) rather than on `fields`/`preferTyped` alone - when not live, they send unconditionally instead of trying to force the record fresh, matching the pre-feature status quo (worst case one redundant PATCH, which the server handles fine). An earlier attempt forced a REST `get_config()` refresh before the check whenever disconnected; that was reverted - it added a failure path and could storm the API with one GET per caller in a batch. Test doubles need `connected` and `topics` attributes (`True`/`None` keep the record "live", matching pre-existing test expectations) for this reason. - `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 all three `_record_is_live()` states (disconnected, topic-filtered, live); `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case. -- `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 `list(self._listeners.values())`, a snapshot, not the live dict - a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN, which registers that vehicle's internal config listener) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop. `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, and a listener mutating `_listeners` mid-dispatch. +- `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. `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, and internal-before-public dispatch order. ## Maintaining this file diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index 259dfe2..2c560b4 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -440,8 +440,11 @@ async def listen(self) -> None: # vehicle (get_vehicle) or otherwise adds a listener # mid-dispatch must not mutate _listeners while this is # iterating it, which would raise RuntimeError and kill - # the loop. - for listener, filters, _internal in list(self._listeners.values()): + # the loop. Internal (bookkeeping) listeners go first, so + # one can cache from the pristine event before any public + # callback gets a chance to mutate it in place. + ordered = sorted(self._listeners.values(), key=lambda item: not item[2]) + for listener, filters, _internal in ordered: if recursive_match(filters, event): try: listener(event) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index ebb0dad..c99875d 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -18,6 +18,9 @@ import aiohttp from teslemetry_stream.stream import TeslemetryStream +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" class FakeContent: @@ -385,6 +388,52 @@ def mutate_during_dispatch(event: dict[str, Any]) -> None: await asyncio.sleep(0) +async def test_internal_listener_sees_event_before_public_mutator(results: list[bool]) -> None: + """A public listener registered BEFORE the internal one - and running + first in registration order - must not get a chance to mutate the event + in place before the internal (bookkeeping) listener has cached from it. + Dispatch order must be internal-first, not registration-order.""" + session = FakeSession() + session.content_factory = lambda: FakeEventContent( + [ + ( + b'data: {"vin": "' + + VIN.encode() + + b'", "config": {"fields": ' + + b'{"BatteryLevel": {"interval_seconds": 60}}, ' + + b'"prefer_typed": true}}\n' + ) + ] + ) + stream = make_stream(session) + + def public_mutator(event: dict[str, Any]) -> None: + # A badly-behaved public consumer mutating its event argument. + event["config"]["fields"]["BatteryLevel"]["interval_seconds"] = 999 + event["config"]["prefer_typed"] = False + + # Registered first (and would run first under registration order) but + # is not internal - the internal config listener, registered second + # (via vehicle construction below), must still see the event first. + stream.async_add_listener(public_mutator) + vehicle = TeslemetryStreamVehicle(stream, VIN) + + for _ in range(5): + await asyncio.sleep(0) + + results.append( + check( + "the internal listener captured the pristine value, not the public mutation", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60} + and vehicle.preferTyped is True, + f"fields {vehicle.fields}, prefer_typed {vehicle.preferTyped}", + ) + ) + + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -394,6 +443,7 @@ async def main() -> None: await test_close_prevents_reconnect_after_backoff(results) await test_restart_after_public_readd_with_internal_listener_present(results) await test_dispatch_survives_listener_creating_vehicle_mid_iteration(results) + await test_internal_listener_sees_event_before_public_mutator(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT")