From 8cf7c65f6a39c9011379af02beda078115fc6d7d Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 16:53:10 +1000 Subject: [PATCH 1/5] refactor: remove prefer_typed - the server always streams typed values now The Teslemetry API maintainer confirms prefer_typed is always true server-side, so the per-vehicle opt-in and its untyped (string-encoded) wire-format handling no longer have a reason to exist. - Drop TeslemetryStreamVehicle.prefer_typed()/preferTyped, and the prefer_typed key from get_config/_on_config_event/_flush/config. - Drop the isinstance(data, str) coercion in make_int/make_float/ make_bool - listen_* callbacks now receive whatever the server sends straight through. - Delete tests/test_field_type_coercion.py (its subject, string-to- native coercion, no longer exists); adapt the remaining config/field tests to drop prefer_typed fixtures and assertions. --- AGENTS.md | 6 +- teslemetry_stream/vehicle.py | 63 ++------- tests/test_config_events.py | 66 ++++------ tests/test_config_listener_lifecycle.py | 23 ++-- tests/test_config_update.py | 11 -- tests/test_field_type_coercion.py | 167 ------------------------ tests/test_reconnect_config_window.py | 1 - tests/test_stream_lifecycle.py | 13 +- 8 files changed, 51 insertions(+), 299 deletions(-) delete mode 100644 tests/test_field_type_coercion.py diff --git a/AGENTS.md b/AGENTS.md index 8930b8a..cafccd0 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -6,7 +6,7 @@ This file is the project's committed home for project-intrinsic agent knowledge: - CI runs on push/PR: `.github/workflows/ci.yml`. Uses `uv` (see `uv.lock`). Three jobs: `lint` (ruff + mypy), `test` (matrix), `build` (`uv build` + `twine check` + artifact upload). Ruff and twine, like mypy, aren't declared dev dependencies - CI installs them ephemerally via `uv run --with ...`, matching the existing mypy pattern. - `[tool.ruff]` in `pyproject.toml` selects `E, F, W, I, B, UP, ASYNC, SIM, RUF` and ignores `RUF006` - the stream's background `listen`/refresh tasks (`stream.py`, `vehicle.py`) are intentionally untracked fire-and-forget `asyncio.create_task` calls, not a lint oversight. - The `test` job's step fails outright (non-zero exit) if `tests/test_*.py` matches nothing, rather than skipping - do not reintroduce a silent-skip fallback there. -- `tests/` files are plain scripts (`if __name__ == "__main__"`), not pytest-based - pytest would collect zero tests here. Run each directly, e.g. `uv run python tests/test_field_type_coercion.py`. +- `tests/` files are plain scripts (`if __name__ == "__main__"`), not pytest-based - pytest would collect zero tests here. Run each directly, e.g. `uv run python tests/test_config_events.py`. - `pyproject.toml` has a `[tool.mypy]` config but no dev-dependency group declares mypy, so `uv sync` alone won't install it. CI installs it ephemerally via `uv run --with mypy mypy teslemetry_stream`. - `Signal` in `const.py` tracks ; the config route rejects names it does not know with `fst_err_validation`. Fields the API has retired are not rejected - it accepts the request and names them in a top-level `ignoredFields` list - so the library can lag the published list without breaking. - Config responses are shaped inconsistently: success is flat, `{"updated_vehicles": n}` plus `ignoredFields` when some were dropped, while errors are wrapped, `{"response": null, "error": ...}`. Do not look for `updated_vehicles` under `response`; that lookup silently never matches. @@ -16,9 +16,9 @@ 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`. 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. +- `TeslemetryStreamVehicle` keeps `fields` 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}}` 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` piece replaces the record (every nested entry must itself be a dict - one bad entry, e.g. a null, rejects the whole 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 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. The server always sends typed telemetry values now - `preferTyped`/`prefer_typed()` (a per-vehicle opt-in for typed vs. legacy string-encoded values) no longer exists in this library; `listen_*` callbacks receive native `int`/`float`/`bool` values straight off the wire with no string-coercion path. - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`. -- `add_field`/`prefer_typed` gate their no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields`/`preferTyped` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. +- `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). - `TeslemetryStream` 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. diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 3bc1bcf..b2f8368 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -69,7 +69,6 @@ class TeslemetryStreamVehicle: """Handle streaming field updates.""" fields: dict[str, dict[str, int]] - preferTyped: bool | None _config: dict[str, Any] _flight: asyncio.Task[None] | None _populated: bool @@ -82,11 +81,10 @@ def __init__(self, stream: TeslemetryStream, vin: str): self.lock = asyncio.Lock() # Per-instance: class-level dicts would share pending config between vehicles self.fields = {} - self.preferTyped = None self._config = {} - # Whether fields/preferTyped reflect a real server answer (a push - # event or a REST fetch) rather than just their unset defaults - - # gates add_field/prefer_typed's lazy REST fetch below. + # Whether fields reflects a real server answer (a push event or a + # REST fetch) rather than just its unset default - gates add_field's + # lazy REST fetch below. self._populated = False # The single in-flight (or most recently completed) coalesced flush. # Callers that arrive while it is running merge into `_config` and @@ -106,10 +104,10 @@ def __init__(self, stream: TeslemetryStream, vin: str): {Key.VIN: self.vin, Key.CONFIG: None}, internal=True, ) - # A disconnect can leave fields/preferTyped stale until the next - # connection's config snapshot arrives - unpopulate so a - # field-config call landing in that window awaits a fresh fetch - # instead of trusting the pre-disconnect record. + # A disconnect can leave fields stale until the next connection's + # config snapshot arrives - unpopulate so a field-config call + # landing in that window awaits a fresh fetch instead of trusting + # the pre-disconnect record. self.stream.async_add_connection_listener(self._on_connection_event) @property @@ -117,7 +115,6 @@ def config(self) -> dict[str, Any]: """Return current configuration.""" return { "fields": self.fields, - "prefer_typed": self.preferTyped, } async def get_config(self) -> None: @@ -133,7 +130,6 @@ async def get_config(self) -> None: response = await req.json() self.fields = response.get("fields", {}) - self.preferTyped = response.get("prefer_typed", False) self._populated = True return if req.status == 404: @@ -168,9 +164,8 @@ async def _ensure_populated(self) -> None: 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. + A well-typed `fields` piece replaces the record; a malformed one is + logged and skipped so it can't corrupt the last-known-good record. """ config = event.get(Key.CONFIG) if not isinstance(config, dict): @@ -201,17 +196,6 @@ def _on_config_event(self, event: dict[str, Any]) -> None: 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. @@ -275,10 +259,6 @@ async def _flush(self) -> None: } LOGGER.debug("Configured streaming fields %s", ", ".join(applied)) self.fields = {**self.fields, **applied} - prefer_typed = self._config.get("prefer_typed") - if isinstance(prefer_typed, bool): - LOGGER.debug("Configured streaming typed to %s", prefer_typed) - self.preferTyped = prefer_typed self._config.clear() async def _patch_with_bounded_retry( @@ -346,13 +326,6 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N value = {"interval_seconds": interval} if interval else None await self.update_config({"fields": {field: value}}) - async def prefer_typed(self, prefer_typed: bool) -> None: - """Set prefer typed.""" - await self._ensure_populated() - if self.preferTyped == prefer_typed: - return - await self.update_config({"prefer_typed": prefer_typed}) - def _enable_field(self, field: Signal) -> None: """Enable a field for streaming from a listener.""" asyncio.create_task(self.add_field(field)) @@ -2972,11 +2945,7 @@ def make_int( """Listener factory""" def typer(event: dict[str, Any]) -> None: - data = event["data"][signal] - if isinstance(data, str): - # Handle invalid and None? - data = int(data) - callback(data) + callback(event["data"][signal]) return typer @@ -2987,11 +2956,7 @@ def make_float( """Listener factory""" def typer(event: dict[str, Any]) -> None: - data = event["data"][signal] - if isinstance(data, str): - # Handle invalid and None? - data = float(data) - callback(data) + callback(event["data"][signal]) return typer @@ -3002,11 +2967,7 @@ def make_bool( """Listener factory""" def typer(event: dict[str, Any]) -> None: - data = event["data"][signal] - if isinstance(data, str): - # Handle invalid and None? - data = data == "true" - callback(data) + callback(event["data"][signal]) return typer diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 69d3c6e..201074c 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -1,14 +1,13 @@ """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 (see +shaped like ``{"vin": ..., "config": {"fields": {...}}}``, mirroring the REST +``get_config`` response body. ``TeslemetryStreamVehicle`` 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. +running loop) so ``fields`` - and therefore the ``add_field`` no-op check - +reflects current server truth rather than only what this client has itself +requested or observed at connect. """ from __future__ import annotations @@ -86,18 +85,14 @@ async def main() -> None: stream.config_listener( { "vin": VIN, - "config": { - "fields": {"BatteryLevel": {"interval_seconds": 60}}, - "prefer_typed": True, - }, + "config": {"fields": {"BatteryLevel": {"interval_seconds": 60}}}, } ) 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 config event updates fields", + vehicle.fields == {"BatteryLevel": {"interval_seconds": 60}}, + f"fields {vehicle.fields}", ) ) @@ -109,10 +104,7 @@ async def main() -> None: assert aliasing_stream.config_listener is not None delivered_event = { "vin": VIN, - "config": { - "fields": {"CarType": {"interval_seconds": 60}}, - "prefer_typed": False, - }, + "config": {"fields": {"CarType": {"interval_seconds": 60}}}, } aliasing_stream.config_listener(delivered_event) delivered_event["config"]["fields"]["CarType"]["interval_seconds"] = 999 @@ -154,15 +146,9 @@ async def main() -> None: vehicle, stream = make_vehicle() assert stream.config_listener is not None stream.config_listener( - { - "vin": VIN, - "config": { - "fields": {"BatteryLevel": {}}, - "prefer_typed": False, - }, - } + {"vin": VIN, "config": {"fields": {"BatteryLevel": {}}}} ) - good_fields, good_typed = dict(vehicle.fields), vehicle.preferTyped + good_fields = dict(vehicle.fields) # A non-dict "config" body is entirely rejected and logged. stream.config_listener({"vin": VIN, "config": "not-a-dict"}) @@ -170,39 +156,31 @@ async def main() -> None: 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. + # A malformed "fields" piece is rejected and logged, keeping last-good. handler.messages.clear() - stream.config_listener( - {"vin": VIN, "config": {"fields": "not-a-dict", "prefer_typed": True}} - ) + stream.config_listener({"vin": VIN, "config": {"fields": "not-a-dict"}}) results.append( check( - "a partial config event applies the good field, keeps the bad one", + "a malformed fields piece is rejected, keeps last-good", 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}", + f"fields {vehicle.fields}, warnings {handler.messages}", ) ) - # A config event missing a key entirely leaves that piece untouched. + # A config event omitting "fields" entirely leaves it unchanged. handler.messages.clear() - stream.config_listener( - {"vin": VIN, "config": {"fields": {"CarType": {}}}} - ) + stream.config_listener({"vin": VIN, "config": {}}) 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}", + "a config event omitting fields leaves it unchanged", + vehicle.fields == good_fields, + f"fields {vehicle.fields}", ) ) diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index 21cb43e..699abdb 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -6,8 +6,8 @@ ``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 + use (inside ``add_field``/``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: @@ -22,12 +22,12 @@ 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. -- ``add_field``/``prefer_typed`` no longer gate their no-op skip on - connection/topic state (``_record_is_live()``, since removed). Instead - they gate on whether the record has ever been ``_populated`` - by a - push event or by a REST fetch. An unpopulated record is fetched over - REST before the no-op decision is made; a populated one is trusted - without hitting the network at all. +- ``add_field`` no longer gates its no-op skip on connection/topic state + (``_record_is_live()``, since removed). Instead it gates on whether the + record has ever been ``_populated`` - by a push event or by a REST + fetch. An unpopulated record is fetched over REST before the no-op + decision is made; a populated one is trusted without hitting the + network at all. """ from __future__ import annotations @@ -201,7 +201,7 @@ async def test_unpopulated_add_field_fetches_before_deciding(results: list[bool] not trust its unset defaults - add_field awaits get_config() first, then decides the no-op skip against the answer it got back.""" session = FetchingSession( - 200, {"fields": {"BatteryLevel": {"interval_seconds": 60}}, "prefer_typed": False} + 200, {"fields": {"BatteryLevel": {"interval_seconds": 60}}} ) stream = make_stream(session) vehicle, sent = make_vehicle_with_capture(stream) @@ -245,10 +245,7 @@ async def test_populated_add_field_skips_without_fetching(results: list[bool]) - vehicle._on_config_event( { Key.VIN: VIN, - Key.CONFIG: { - "fields": {"BatteryLevel": {"interval_seconds": 60}}, - "prefer_typed": False, - }, + Key.CONFIG: {"fields": {"BatteryLevel": {"interval_seconds": 60}}}, } ) results.append(check("the config event populated the vehicle", vehicle._populated)) diff --git a/tests/test_config_update.py b/tests/test_config_update.py index 1c3f38f..f44d7e0 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -191,17 +191,6 @@ async def main() -> None: ) ) - # prefer_typed=False must be recorded as False, not coerced to True. - vehicle = make_vehicle(VIN_A, [ACCEPTED]) - await vehicle.prefer_typed(False) - results.append( - check( - "prefer_typed(False) recorded as False", - vehicle.preferTyped is False, - f"got {vehicle.preferTyped!r}", - ) - ) - # Pending config must not be shared between vehicles. first = make_vehicle(VIN_A, [FAILED]) second = make_vehicle(VIN_B, [ACCEPTED]) diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py deleted file mode 100644 index c908ad8..0000000 --- a/tests/test_field_type_coercion.py +++ /dev/null @@ -1,167 +0,0 @@ -"""End-to-end typing checks for the field-type fixes in v0.9.1. - -Exercises the real public ``listen_*`` methods on ``TeslemetryStreamVehicle`` -exactly as a library consumer would, then pushes representative *real-world* -streamed values (as sampled from the Teslemetry na_cache NATS KV store, per the -PR description) through the registered listener and asserts the value and the -Python type delivered to the consumer callback. -""" -from __future__ import annotations - -import asyncio -from typing import Any, Callable - -from teslemetry_stream.const import Signal -from teslemetry_stream.vehicle import TeslemetryStreamVehicle - -VIN = "TESTVIN0000000001" - - -class FakeStream: - """Minimal stand-in for TeslemetryStream that just captures listeners.""" - - manual = True - - def __init__(self) -> None: - # maps Signal value -> wrapped listener callback - self.captured: dict[str, Any] = {} - - 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; - # it's not under test here, so just ignore it. - assert filters is not None - if "data" in filters: - signal = next(iter(filters["data"])) - self.captured[signal] = callback - return lambda: None - - def async_add_connection_listener( - self, callback: Callable[[bool], None] - ) -> Callable[[], None]: - return lambda: None - - -def build_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]: - stream = FakeStream() - vehicle = TeslemetryStreamVehicle(stream, VIN) # type: ignore[arg-type] - # Pre-populate config so _enable_field/add_field short-circuits without - # a lazy REST fetch (the fake stream has no session to serve one). - vehicle.fields = {s.value: {} for s in Signal} - vehicle._populated = True - return vehicle, stream - - -def deliver(stream: FakeStream, signal: Signal, raw: Any) -> Any: - """Feed a raw streamed value through the captured listener, return delivered.""" - box: dict[str, Any] = {} - # The captured callback was registered by listen_* with the consumer callback - # already baked in; re-register a fresh consumer to capture the delivered value. - event = {"vin": VIN, "data": {signal.value: raw}} - stream.captured[signal.value](event) - return box # unused; delivered value captured by the consumer closure - - -async def main() -> None: - results: list[tuple[str, Any, str, Any, str, bool]] = [] - - def check( - label: str, - signal: Signal, - register: Callable[ - [TeslemetryStreamVehicle, Callable[[Any], None]], Any - ], - raw: Any, - expected: Any, - expected_type: type, - ) -> None: - vehicle, stream = build_vehicle() - delivered: dict[str, Any] = {} - register(vehicle, lambda v: delivered.__setitem__("v", v)) - stream.captured[signal.value]({"vin": VIN, "data": {signal.value: raw}}) - got = delivered.get("v") - ok = got == expected and type(got) is expected_type - results.append( - (label, raw, type(raw).__name__, got, type(got).__name__, ok) - ) - - # --- BATCH 1: seat cooling, previously passed raw str through (wrong) --- - check( - "ClimateSeatCoolingFrontLeft", - Signal.CLIMATE_SEAT_COOLING_FRONT_LEFT, - lambda v, cb: v.listen_ClimateSeatCoolingFrontLeft(cb), - "3", 3, int, - ) - check( - "ClimateSeatCoolingFrontRight", - Signal.CLIMATE_SEAT_COOLING_FRONT_RIGHT, - lambda v, cb: v.listen_ClimateSeatCoolingFrontRight(cb), - "0", 0, int, - ) - - # --- BATCH 2a: CurrentLimitMph streams fractional values; make_int used to - # truncate 74.4367 -> 74. Now make_float preserves the fraction. --- - check( - "CurrentLimitMph", - Signal.CURRENT_LIMIT_MPH, - lambda v, cb: v.listen_CurrentLimitMph(cb), - "74.4367", 74.4367, float, - ) - check( - "CurrentLimitMph", - Signal.CURRENT_LIMIT_MPH, - lambda v, cb: v.listen_CurrentLimitMph(cb), - "80.6504", 80.6504, float, - ) - - # --- BATCH 2b: SoftwareUpdateScheduledStartTime streams an int epoch as str; - # previously delivered raw str, now coerced to int. --- - check( - "SoftwareUpdateScheduledStartTime", - Signal.SOFTWARE_UPDATE_SCHEDULED_START_TIME, - lambda v, cb: v.listen_SoftwareUpdateScheduledStartTime(cb), - "1783072740", 1783072740, int, - ) - - # --- Deliberate int divergence (kept as int despite fields.json "real") --- - check( - "CruiseSetSpeed (kept int)", - Signal.CRUISE_SET_SPEED, - lambda v, cb: v.listen_CruiseSetSpeed(cb), - "65", 65, int, - ) - check( - "DiTorquemotor (kept int)", - Signal.DI_TORQUEMOTOR, - lambda v, cb: v.listen_DiTorquemotor(cb), - "120", 120, int, - ) - check( - "ExpectedEnergyPercentAtTripArrival (kept int)", - Signal.EXPECTED_ENERGY_PERCENT_AT_TRIP_ARRIVAL, - lambda v, cb: v.listen_ExpectedEnergyPercentAtTripArrival(cb), - "82", 82, int, - ) - - print(f"{'field':<42} {'raw (stream)':<16} {'delivered':<14} {'type':<7} ok") - print("-" * 88) - all_ok = True - for label, raw, raw_t, got, got_t, ok in results: - all_ok = all_ok and ok - print( - f"{label:<42} {repr(raw):<16} {repr(got):<14} {got_t:<7} " - f"{'PASS' if ok else 'FAIL'}" - ) - print("-" * 88) - print("ALL PASS" if all_ok else "FAILURES PRESENT") - if not all_ok: - raise SystemExit(1) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/tests/test_reconnect_config_window.py b/tests/test_reconnect_config_window.py index 7760268..77fe400 100644 --- a/tests/test_reconnect_config_window.py +++ b/tests/test_reconnect_config_window.py @@ -113,7 +113,6 @@ async def test_reconnect_window_add_field_refetches_instead_of_trusting_stale_re # client yet) - the reconnect's config snapshot will reveal 30s. stream._session.config_body = { # type: ignore[attr-defined] "fields": {"BatteryLevel": {"interval_seconds": 30}}, - "prefer_typed": False, } # A connection listener reacting to reconnect - e.g. an HA integration diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index 146f898..e7e7f66 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -400,8 +400,7 @@ async def test_internal_listener_sees_event_before_public_mutator(results: list[ b'data: {"vin": "' + VIN.encode() + b'", "config": {"fields": ' - + b'{"BatteryLevel": {"interval_seconds": 60}}, ' - + b'"prefer_typed": true}}\n' + + b'{"BatteryLevel": {"interval_seconds": 60}}}}\n' ) ] ) @@ -410,7 +409,6 @@ async def test_internal_listener_sees_event_before_public_mutator(results: list[ 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 @@ -424,9 +422,8 @@ def public_mutator(event: dict[str, Any]) -> None: 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}", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, + f"fields {vehicle.fields}", ) ) @@ -452,8 +449,7 @@ async def test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use( b'data: {"vin": "' + new_vin.encode() + b'", "config": {"fields": ' - + b'{"BatteryLevel": {"interval_seconds": 60}}, ' - + b'"prefer_typed": true}}\n' + + b'{"BatteryLevel": {"interval_seconds": 60}}}}\n' ) ] ) @@ -493,7 +489,6 @@ async def get_config() -> None: # Stand in for the REST fetch a real session would serve - reflects # what the missed event actually carried. vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} - vehicle.preferTyped = True vehicle._populated = True vehicle.get_config = get_config # type: ignore[assignment,method-assign] From 673130c0c175663248656bc90bab12c56cf3f4e6 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 17:04:30 +1000 Subject: [PATCH 2/5] fix: clear stale fields on an authoritative 404, not just mark populated get_config()'s 404 branch marked the vehicle populated without clearing self.fields. A vehicle carrying fields from before a disconnect whose config was then deleted server-side would keep those stale fields forever, causing add_field() to permanently skip the PATCH for a field the server no longer has configured. Reported by Codex review on #33. --- AGENTS.md | 2 +- teslemetry_stream/vehicle.py | 6 +++- tests/test_config_listener_lifecycle.py | 38 +++++++++++++++++++++++++ 3 files changed, 44 insertions(+), 2 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index cafccd0..222c516 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -18,7 +18,7 @@ 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` 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}}` 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` piece replaces the record (every nested entry must itself be a dict - one bad entry, e.g. a null, rejects the whole 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 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. The server always sends typed telemetry values now - `preferTyped`/`prefer_typed()` (a per-vehicle opt-in for typed vs. legacy string-encoded values) no longer exists in this library; `listen_*` callbacks receive native `int`/`float`/`bool` values straight off the wire with no string-coercion path. - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`. -- `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. +- `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). - `TeslemetryStream` 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. diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index b2f8368..5f4d78b 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -134,7 +134,11 @@ async def get_config(self) -> None: return if req.status == 404: # No config exists for this vehicle yet - an authoritative - # answer (empty), not a missing one. + # answer (empty), not a missing one. Clear any fields left over + # from before a disconnect: without this, a config deleted + # elsewhere while offline would leave add_field() no-op'ing + # forever against a record the server no longer has. + self.fields = {} self._populated = True return diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index 699abdb..54a086b 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -260,12 +260,50 @@ async def test_populated_add_field_skips_without_fetching(results: list[bool]) - ) +async def test_authoritative_404_clears_stale_fields(results: list[bool]) -> None: + """A vehicle carrying fields from before a disconnect must not trust them + forever if the config was deleted server-side while it was offline - a + 404 on the reconnect-window fetch is authoritative (no config exists), + so it must clear `fields`, not just mark the record populated.""" + session = FetchingSession(404, {}) + stream = make_stream(session) + vehicle, sent = make_vehicle_with_capture(stream) + + # Simulate a stale record surviving a disconnect: previously populated + # with a field the server no longer has, then unpopulated as a real + # disconnect would leave it. + vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} + vehicle._populated = False + + await vehicle.get_config() + results.append( + check( + "an authoritative 404 clears stale fields, not just marks populated", + vehicle.fields == {}, + f"fields {vehicle.fields}", + ) + ) + + # With the stale record cleared, a later add_field for that same field + # must send - not skip on a match that no longer exists - and without + # re-fetching, since the (now correctly empty) record is populated. + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "add_field sends the PATCH instead of no-op'ing against the stale record", + session.calls == 1 and len(sent) == 1 and "BatteryLevel" in sent[0]["fields"], + f"calls {session.calls}, sent {sent}", + ) + ) + + async def main(pre_loop_results: list[bool]) -> None: results: list[bool] = list(pre_loop_results) await test_registration_happens_at_construction(results) await test_auto_close_after_last_public_listener_removed(results) await test_unpopulated_add_field_fetches_before_deciding(results) await test_populated_add_field_skips_without_fetching(results) + await test_authoritative_404_clears_stale_fields(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") From ad7165ba93cbae0bffd5c1881fc4d7f9deee94a5 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 17:15:22 +1000 Subject: [PATCH 3/5] chore: bump version to 0.11.0 for the prefer_typed removal Matches the minor bump this PR's body already describes for the breaking removal of prefer_typed()/preferTyped - pyproject.toml and uv.lock still said 0.10.1. Reported by Codex review on #33. --- pyproject.toml | 2 +- uv.lock | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 099a1c7..a29fcc6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ requires = ["setuptools>=77.0"] [project] name = "teslemetry_stream" -version = "0.10.1" +version = "0.11.0" license = "Apache-2.0" description = "Teslemetry Streaming API library for Python" readme = "README.md" diff --git a/uv.lock b/uv.lock index 1884a2e..1d72232 100644 --- a/uv.lock +++ b/uv.lock @@ -612,7 +612,7 @@ wheels = [ [[package]] name = "teslemetry-stream" -version = "0.10.1" +version = "0.11.0" source = { editable = "." } dependencies = [ { name = "aiohttp" }, From 350b45435d1dab19f582468a043f7c6475e663af Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 17:15:22 +1000 Subject: [PATCH 4/5] fix: don't strand a field-enable request on a failed config fetch _ensure_populated() let a failed populating GET (non-200/404 status, or a transport/timeout error) propagate out of add_field(). Called directly that's a normal exception a caller can handle, but every listen_* method reaches add_field() through _enable_field()'s fire-and-forget asyncio.create_task() - nobody awaits it, so the exception just became an unretrieved-task-exception log line and the requested field was never configured, silently. A failed fetch isn't authoritative like a 404: catch it, log a warning, and stay unpopulated so add_field() proceeds against whatever fields it currently knows and sends the PATCH, rather than abandoning the request. Reported by Codex review on #33. --- teslemetry_stream/vehicle.py | 12 +++- tests/test_config_listener_lifecycle.py | 77 +++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 1 deletion(-) diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 5f4d78b..a6ff841 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -163,7 +163,17 @@ async def _ensure_populated(self) -> None: if flight is None or flight.done(): flight = asyncio.ensure_future(self.get_config()) self._populate_flight = flight - await asyncio.shield(flight) + try: + await asyncio.shield(flight) + except (aiohttp.ClientError, asyncio.TimeoutError) as error: + # Unlike a 404, a failed fetch isn't authoritative - stay + # unpopulated and let the caller proceed against whatever + # `fields` currently holds, rather than raising into a + # fire-and-forget `_enable_field()` task where nobody would + # ever see the exception or retry the stranded field request. + LOGGER.warning( + "Config fetch failed for %s, proceeding without it: %s", self.vin, error + ) def _on_config_event(self, event: dict[str, Any]) -> None: """Sync the record from a server-pushed config event. diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index 54a086b..b2debe7 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -34,6 +34,8 @@ import asyncio from typing import Any +import aiohttp + from teslemetry_stream.const import Key from teslemetry_stream.stream import TeslemetryStream from teslemetry_stream.vehicle import TeslemetryStreamVehicle @@ -58,6 +60,20 @@ def __init__(self, status: int, body: dict[str, Any]) -> None: async def json(self) -> dict[str, Any]: return self._body + def raise_for_status(self) -> None: + if self.status >= 400: + url = "https://fake/api/config" + raise aiohttp.ClientResponseError( + request_info=aiohttp.RequestInfo( + url=url, # type: ignore[arg-type] + method="GET", + headers={}, # type: ignore[arg-type] + real_url=url, # type: ignore[arg-type] + ), + history=(), + status=self.status, + ) + class FetchingSession: """A session that serves a canned config GET and counts calls.""" @@ -72,6 +88,17 @@ async def get(self, url: str, **kwargs: Any) -> FakeConfigResponse: return FakeConfigResponse(self.status, self.body) +class TransportFailingSession: + """A session whose config GET always raises a transport-level error.""" + + def __init__(self) -> None: + self.calls = 0 + + async def get(self, url: str, **kwargs: Any) -> Any: + self.calls += 1 + raise aiohttp.ClientConnectionError("simulated connection failure") + + def check(label: str, ok: bool, detail: str = "") -> bool: print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") return ok @@ -297,6 +324,54 @@ async def test_authoritative_404_clears_stale_fields(results: list[bool]) -> Non ) +async def test_unpopulated_add_field_survives_a_failed_status_fetch( + results: list[bool], +) -> None: + """A non-200/404 status on the populating fetch (e.g. a transient 5xx) + is not authoritative like a 404 - it must not strand the field request + the way it silently would inside a fire-and-forget `_enable_field()` + task, where nobody would ever see the raised exception or retry. + add_field must stay unpopulated and still send the PATCH.""" + session = FetchingSession(500, {}) + stream = make_stream(session) + vehicle, sent = make_vehicle_with_capture(stream) + + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "add_field sends the PATCH despite the failed fetch, not stranding it", + len(sent) == 1 and "BatteryLevel" in sent[0]["fields"], + f"sent {sent}", + ) + ) + results.append( + check( + "the vehicle stays unpopulated after a failed (non-authoritative) fetch", + not vehicle._populated, + ) + ) + + +async def test_unpopulated_add_field_survives_a_transport_failure( + results: list[bool], +) -> None: + """A transport-level failure (ClientError/timeout) on the populating + fetch must be handled the same way as a bad status: proceed to the + PATCH rather than stranding the field request.""" + session = TransportFailingSession() + stream = make_stream(session) + vehicle, sent = make_vehicle_with_capture(stream) + + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "add_field sends the PATCH despite a transport failure", + len(sent) == 1 and "BatteryLevel" in sent[0]["fields"], + f"sent {sent}", + ) + ) + + async def main(pre_loop_results: list[bool]) -> None: results: list[bool] = list(pre_loop_results) await test_registration_happens_at_construction(results) @@ -304,6 +379,8 @@ async def main(pre_loop_results: list[bool]) -> None: await test_unpopulated_add_field_fetches_before_deciding(results) await test_populated_add_field_skips_without_fetching(results) await test_authoritative_404_clears_stale_fields(results) + await test_unpopulated_add_field_survives_a_failed_status_fetch(results) + await test_unpopulated_add_field_survives_a_transport_failure(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT") From 767c019e6a3ada42e56fcb69e7ca4ca656384eb1 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 17:17:12 +1000 Subject: [PATCH 5/5] revert: restore string-to-native coercion in make_int/make_float/make_bool prefer_typed is only a default now, not a guarantee: some vehicles haven't picked it up and still stream string-encoded numeric/boolean values. Dropping the isinstance(data, str) coercion in make_int/make_float/make_bool would silently hand those vehicles' consumers raw strings instead of int/float/bool. Restores tests/test_field_type_coercion.py, which regression-tests this coercion against real observed telemetry. The prefer_typed preference surface itself (prefer_typed()/preferTyped, the config-sync handling) stays removed - only the wire-format coercion is back. Per maintainer review on #33. --- AGENTS.md | 4 +- teslemetry_stream/vehicle.py | 24 ++++- tests/test_field_type_coercion.py | 167 ++++++++++++++++++++++++++++++ 3 files changed, 190 insertions(+), 5 deletions(-) create mode 100644 tests/test_field_type_coercion.py diff --git a/AGENTS.md b/AGENTS.md index 222c516..9bc42d4 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,9 +16,9 @@ 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` 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}}` 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` piece replaces the record (every nested entry must itself be a dict - one bad entry, e.g. a null, rejects the whole 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 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. The server always sends typed telemetry values now - `preferTyped`/`prefer_typed()` (a per-vehicle opt-in for typed vs. legacy string-encoded values) no longer exists in this library; `listen_*` callbacks receive native `int`/`float`/`bool` values straight off the wire with no string-coercion path. +- `TeslemetryStreamVehicle` keeps `fields` 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}}` 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` piece replaces the record (every nested entry must itself be a dict - one bad entry, e.g. a null, rejects the whole 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 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. `preferTyped`/`prefer_typed()` (the per-vehicle opt-in that used to control this) no longer exists in this library - prefer_typed is enabled by default server-side now, so there is nothing left to toggle. It is still only a default, not a guarantee: some vehicles haven't picked it up and still stream string-encoded numeric/boolean values, so `make_int`/`make_float`/`make_bool` (`vehicle.py`) keep coercing a `str` payload to native `int`/`float`/`bool` rather than assuming every vehicle is typed. `tests/test_field_type_coercion.py` covers this coercion against real observed telemetry. - `TeslemetryStream.async_add_listener(..., internal=True)` marks a listener as bookkeeping-only (the vehicle's config-sync listener is the only current user). Its `schedule_refresh` gate on the `asyncio.create_task()` call is unconditionally false for `internal=True`, regardless of loop state - this is what makes eager, construction-time registration of the config listener safe outside a running event loop, not deferred timing. Both the "first listener starts the task" and "last listener removed auto-closes" checks also count only non-internal (public) listeners for the same underlying reason: an internal listener alone must not itself start the task, and one surviving an auto-close must not block a later public listener's own zero-to-one transition from restarting it. Any stream stand-in passed to `TeslemetryStreamVehicle` (test doubles included) must implement `async_add_listener(callback, filters, internal=False)` and `async_add_connection_listener(callback)`. -- `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. +- `add_field` gates its no-op skip on `TeslemetryStreamVehicle._populated`, not on connection/topic state: an unpopulated vehicle awaits `_ensure_populated()` (a single-flight `get_config()` REST fetch - concurrent callers, e.g. a batch of `listen_*` calls at HA integration setup, join one GET instead of each starting their own) before deciding; a populated one trusts `fields` outright with no network call. `_populated` is set by a successful `get_config()` (200 or 404 - both are an authoritative answer; 404 also clears `fields`, since no config existing is itself the authoritative state, not a fetch to ignore) and by every `_on_config_event` push, and cleared by an `_on_connection_event` disconnect notification (registered via `async_add_connection_listener` at construction, alongside the config-sync listener) - a disconnect leaves the record possibly stale until the next connection's config snapshot arrives, so a field-config call landing in that reconnect window re-fetches instead of trusting pre-disconnect data. A failed populating fetch (`aiohttp.ClientError`/timeout, or a non-200/404 status via `raise_for_status()`) is not authoritative like a 404: `_ensure_populated()` catches it, logs, and leaves the vehicle unpopulated rather than letting it propagate - every `listen_*` method reaches `add_field()` through `_enable_field()`'s fire-and-forget `asyncio.create_task()`, where an uncaught exception would just silently abandon the field request instead of sending the PATCH. - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). - `TeslemetryStream` 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. diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index a6ff841..5daf84e 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -2959,7 +2959,13 @@ def make_int( """Listener factory""" def typer(event: dict[str, Any]) -> None: - callback(event["data"][signal]) + data = event["data"][signal] + if isinstance(data, str): + # Some vehicles still stream string-encoded values even with + # prefer_typed enabled by default - keep coercing rather than + # assuming every vehicle is typed. + data = int(data) + callback(data) return typer @@ -2970,7 +2976,13 @@ def make_float( """Listener factory""" def typer(event: dict[str, Any]) -> None: - callback(event["data"][signal]) + data = event["data"][signal] + if isinstance(data, str): + # Some vehicles still stream string-encoded values even with + # prefer_typed enabled by default - keep coercing rather than + # assuming every vehicle is typed. + data = float(data) + callback(data) return typer @@ -2981,7 +2993,13 @@ def make_bool( """Listener factory""" def typer(event: dict[str, Any]) -> None: - callback(event["data"][signal]) + data = event["data"][signal] + if isinstance(data, str): + # Some vehicles still stream string-encoded values even with + # prefer_typed enabled by default - keep coercing rather than + # assuming every vehicle is typed. + data = data == "true" + callback(data) return typer diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py new file mode 100644 index 0000000..c908ad8 --- /dev/null +++ b/tests/test_field_type_coercion.py @@ -0,0 +1,167 @@ +"""End-to-end typing checks for the field-type fixes in v0.9.1. + +Exercises the real public ``listen_*`` methods on ``TeslemetryStreamVehicle`` +exactly as a library consumer would, then pushes representative *real-world* +streamed values (as sampled from the Teslemetry na_cache NATS KV store, per the +PR description) through the registered listener and asserts the value and the +Python type delivered to the consumer callback. +""" +from __future__ import annotations + +import asyncio +from typing import Any, Callable + +from teslemetry_stream.const import Signal +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + + +class FakeStream: + """Minimal stand-in for TeslemetryStream that just captures listeners.""" + + manual = True + + def __init__(self) -> None: + # maps Signal value -> wrapped listener callback + self.captured: dict[str, Any] = {} + + 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; + # it's not under test here, so just ignore it. + assert filters is not None + if "data" in filters: + signal = next(iter(filters["data"])) + self.captured[signal] = callback + return lambda: None + + def async_add_connection_listener( + self, callback: Callable[[bool], None] + ) -> Callable[[], None]: + return lambda: None + + +def build_vehicle() -> tuple[TeslemetryStreamVehicle, FakeStream]: + stream = FakeStream() + vehicle = TeslemetryStreamVehicle(stream, VIN) # type: ignore[arg-type] + # Pre-populate config so _enable_field/add_field short-circuits without + # a lazy REST fetch (the fake stream has no session to serve one). + vehicle.fields = {s.value: {} for s in Signal} + vehicle._populated = True + return vehicle, stream + + +def deliver(stream: FakeStream, signal: Signal, raw: Any) -> Any: + """Feed a raw streamed value through the captured listener, return delivered.""" + box: dict[str, Any] = {} + # The captured callback was registered by listen_* with the consumer callback + # already baked in; re-register a fresh consumer to capture the delivered value. + event = {"vin": VIN, "data": {signal.value: raw}} + stream.captured[signal.value](event) + return box # unused; delivered value captured by the consumer closure + + +async def main() -> None: + results: list[tuple[str, Any, str, Any, str, bool]] = [] + + def check( + label: str, + signal: Signal, + register: Callable[ + [TeslemetryStreamVehicle, Callable[[Any], None]], Any + ], + raw: Any, + expected: Any, + expected_type: type, + ) -> None: + vehicle, stream = build_vehicle() + delivered: dict[str, Any] = {} + register(vehicle, lambda v: delivered.__setitem__("v", v)) + stream.captured[signal.value]({"vin": VIN, "data": {signal.value: raw}}) + got = delivered.get("v") + ok = got == expected and type(got) is expected_type + results.append( + (label, raw, type(raw).__name__, got, type(got).__name__, ok) + ) + + # --- BATCH 1: seat cooling, previously passed raw str through (wrong) --- + check( + "ClimateSeatCoolingFrontLeft", + Signal.CLIMATE_SEAT_COOLING_FRONT_LEFT, + lambda v, cb: v.listen_ClimateSeatCoolingFrontLeft(cb), + "3", 3, int, + ) + check( + "ClimateSeatCoolingFrontRight", + Signal.CLIMATE_SEAT_COOLING_FRONT_RIGHT, + lambda v, cb: v.listen_ClimateSeatCoolingFrontRight(cb), + "0", 0, int, + ) + + # --- BATCH 2a: CurrentLimitMph streams fractional values; make_int used to + # truncate 74.4367 -> 74. Now make_float preserves the fraction. --- + check( + "CurrentLimitMph", + Signal.CURRENT_LIMIT_MPH, + lambda v, cb: v.listen_CurrentLimitMph(cb), + "74.4367", 74.4367, float, + ) + check( + "CurrentLimitMph", + Signal.CURRENT_LIMIT_MPH, + lambda v, cb: v.listen_CurrentLimitMph(cb), + "80.6504", 80.6504, float, + ) + + # --- BATCH 2b: SoftwareUpdateScheduledStartTime streams an int epoch as str; + # previously delivered raw str, now coerced to int. --- + check( + "SoftwareUpdateScheduledStartTime", + Signal.SOFTWARE_UPDATE_SCHEDULED_START_TIME, + lambda v, cb: v.listen_SoftwareUpdateScheduledStartTime(cb), + "1783072740", 1783072740, int, + ) + + # --- Deliberate int divergence (kept as int despite fields.json "real") --- + check( + "CruiseSetSpeed (kept int)", + Signal.CRUISE_SET_SPEED, + lambda v, cb: v.listen_CruiseSetSpeed(cb), + "65", 65, int, + ) + check( + "DiTorquemotor (kept int)", + Signal.DI_TORQUEMOTOR, + lambda v, cb: v.listen_DiTorquemotor(cb), + "120", 120, int, + ) + check( + "ExpectedEnergyPercentAtTripArrival (kept int)", + Signal.EXPECTED_ENERGY_PERCENT_AT_TRIP_ARRIVAL, + lambda v, cb: v.listen_ExpectedEnergyPercentAtTripArrival(cb), + "82", 82, int, + ) + + print(f"{'field':<42} {'raw (stream)':<16} {'delivered':<14} {'type':<7} ok") + print("-" * 88) + all_ok = True + for label, raw, raw_t, got, got_t, ok in results: + all_ok = all_ok and ok + print( + f"{label:<42} {repr(raw):<16} {repr(got):<14} {got_t:<7} " + f"{'PASS' if ok else 'FAIL'}" + ) + print("-" * 88) + print("ALL PASS" if all_ok else "FAILURES PRESENT") + if not all_ok: + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())