From a5fc9a777ff10c6b172e2e614a391a302a5a1062 Mon Sep 17 00:00:00 2001 From: Brett Adams Date: Wed, 12 Aug 2026 16:33:46 +1000 Subject: [PATCH] fix: replace config no-op-skip machinery with a populated flag + lazy fetch add_field/prefer_typed no longer gate their no-op skip on connection/topic state (_record_is_live). Instead, TeslemetryStreamVehicle tracks whether its record has ever been populated (by a push event or a REST fetch); an unpopulated vehicle awaits a single-flight get_config() before deciding, a populated one trusts the optimistically-updated record outright. A disconnect notification unpopulates the record so a call landing in the reconnect window (before the next connection's config snapshot arrives) refetches instead of trusting stale pre-disconnect data. This replaces the mechanisms proposed in #30 (a second dispatch pass for listeners added mid-event) and #31 (per-connection snapshot tracking) with a smaller, self-contained fix scoped to the vehicle itself. --- AGENTS.md | 6 +- teslemetry_stream/vehicle.py | 70 +++++++--- tests/test_batch_retry_storm.py | 11 +- tests/test_config_events.py | 9 +- tests/test_config_listener_lifecycle.py | 166 ++++++++++++++--------- tests/test_config_update.py | 11 +- tests/test_field_type_coercion.py | 13 +- tests/test_reconnect_config_window.py | 173 ++++++++++++++++++++++++ tests/test_stream_lifecycle.py | 79 +++++++++++ 9 files changed, 430 insertions(+), 108 deletions(-) create mode 100644 tests/test_reconnect_config_window.py diff --git a/AGENTS.md b/AGENTS.md index 54f3c39..8930b8a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -17,9 +17,9 @@ This file is the project's committed home for project-intrinsic agent knowledge: - 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. -- `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.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. +- `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. ## Maintaining this file diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index e28c5f6..3bc1bcf 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -46,7 +46,6 @@ ShiftState, Signal, SpeedAssistLevel, - SseTopic, State, Status, SunroofInstalledState, @@ -73,6 +72,8 @@ class TeslemetryStreamVehicle: preferTyped: bool | None _config: dict[str, Any] _flight: asyncio.Task[None] | None + _populated: bool + _populate_flight: asyncio.Task[None] | None def __init__(self, stream: TeslemetryStream, vin: str): # A dictionary of TelemetryField keys and null values @@ -83,10 +84,18 @@ def __init__(self, stream: TeslemetryStream, vin: str): 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. + self._populated = False # The single in-flight (or most recently completed) coalesced flush. # Callers that arrive while it is running merge into `_config` and # await it instead of starting their own PATCH. self._flight = None + # The single in-flight populating get_config() call, so a batch of + # listeners (e.g. HA integration setup) discovering an unpopulated + # vehicle joins one GET instead of each starting its own. + self._populate_flight = None # 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 @@ -97,6 +106,11 @@ 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. + self.stream.async_add_connection_listener(self._on_connection_event) @property def config(self) -> dict[str, Any]: @@ -120,12 +134,37 @@ async def get_config(self) -> None: self.fields = response.get("fields", {}) self.preferTyped = response.get("prefer_typed", False) + self._populated = True return if req.status == 404: + # No config exists for this vehicle yet - an authoritative + # answer (empty), not a missing one. + self._populated = True return req.raise_for_status() + def _on_connection_event(self, connected: bool) -> None: + """Unpopulate on disconnect - see the __init__ registration comment.""" + if not connected: + self._populated = False + + async def _ensure_populated(self) -> None: + """Lazily fetch current config over REST if not yet known. + + Optimistic updates from `_on_config_event` keep the record fresh + once established; this only covers the gap before a connection's + first snapshot (or after a disconnect) arrives. Concurrent callers + join the same fetch rather than each issuing their own GET. + """ + if self._populated: + return + flight = self._populate_flight + if flight is None or flight.done(): + flight = asyncio.ensure_future(self.get_config()) + self._populate_flight = flight + await asyncio.shield(flight) + def _on_config_event(self, event: dict[str, Any]) -> None: """Sync the record from a server-pushed config event. @@ -140,6 +179,8 @@ def _on_config_event(self, event: dict[str, Any]) -> None: ) return + self._populated = True + if "fields" in config: fields = config["fields"] # Every entry must itself be a dict (e.g. {"interval_seconds": 60} @@ -290,10 +331,10 @@ async def add_field(self, field: Signal | str, interval: int | None = None) -> N if isinstance(field, Signal): field = field.value - if ( - self._record_is_live() - and field in self.fields - and (interval is None or self.fields[field].get("interval_seconds") == interval) + await self._ensure_populated() + + if field in self.fields and ( + interval is None or self.fields[field].get("interval_seconds") == interval ): LOGGER.debug( "Streaming field %s already enabled @ %ss", @@ -307,26 +348,11 @@ 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.""" - if self._record_is_live() and self.preferTyped == prefer_typed: + await self._ensure_populated() + if self.preferTyped == prefer_typed: return await self.update_config({"prefer_typed": prefer_typed}) - def _record_is_live(self) -> bool: - """Whether the record is being kept current and can gate the no-op skip. - - 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: - 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.""" asyncio.create_task(self.add_field(field)) diff --git a/tests/test_batch_retry_storm.py b/tests/test_batch_retry_storm.py index 5655646..c0f31f2 100644 --- a/tests/test_batch_retry_storm.py +++ b/tests/test_batch_retry_storm.py @@ -30,20 +30,23 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True - # 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 ) -> Any: return lambda: None + def async_add_connection_listener(self, callback: Any) -> Any: + return lambda: None + def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle: """Build a vehicle that records payloads and replays canned responses.""" vehicle = TeslemetryStreamVehicle(FakeStream(), vin) # type: ignore[arg-type] + # These tests exercise the write path, not the lazy-populate fetch (the + # fake stream has no REST session to serve one) - mark it populated like + # a real connection's config snapshot already would have. + vehicle._populated = True vehicle.sent = [] # type: ignore[attr-defined] async def patch_config(config: dict[str, Any]) -> dict[str, Any]: diff --git a/tests/test_config_events.py b/tests/test_config_events.py index 1737550..69d3c6e 100644 --- a/tests/test_config_events.py +++ b/tests/test_config_events.py @@ -26,10 +26,6 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that captures the config listener.""" manual = True - # 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 @@ -45,6 +41,11 @@ def async_add_listener( self.config_listener = callback return lambda: None + def async_add_connection_listener( + self, callback: Callable[[bool], None] + ) -> Callable[[], None]: + return lambda: None + class CaptureWarnings(logging.Handler): """Collect formatted WARNING records emitted by the library.""" diff --git a/tests/test_config_listener_lifecycle.py b/tests/test_config_listener_lifecycle.py index f814754..21cb43e 100644 --- a/tests/test_config_listener_lifecycle.py +++ b/tests/test_config_listener_lifecycle.py @@ -22,47 +22,67 @@ 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 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. +- ``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. """ from __future__ import annotations import asyncio from typing import Any +from teslemetry_stream.const import Key from teslemetry_stream.stream import TeslemetryStream from teslemetry_stream.vehicle import TeslemetryStreamVehicle VIN = "TESTVIN0000000001" -class FakeSession: +class RefusingSession: """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") +class FakeConfigResponse: + """Minimal stand-in for the aiohttp response get_config() awaits.""" + + def __init__(self, status: int, body: dict[str, Any]) -> None: + self.status = status + self._body = body + + async def json(self) -> dict[str, Any]: + return self._body + + +class FetchingSession: + """A session that serves a canned config GET and counts calls.""" + + def __init__(self, status: int, body: dict[str, Any]) -> None: + self.calls = 0 + self.status = status + self.body = body + + async def get(self, url: str, **kwargs: Any) -> FakeConfigResponse: + self.calls += 1 + return FakeConfigResponse(self.status, self.body) + + 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: +def make_stream(session: Any, **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. + # connect/listen loop. kwargs.setdefault("manual", True) return TeslemetryStream( - session=FakeSession(), # type: ignore[arg-type] + session=session, access_token="test-token", server="api.teslemetry.com", **kwargs, @@ -94,19 +114,30 @@ def test_sync_construction_without_a_loop() -> bool: # 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) + stream = make_stream(RefusingSession(), vin=VIN) 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 + ok = ( + check( + "the config listener is registered by the time construction returns", + len(stream._listeners) == 1, + f"listeners {len(stream._listeners)}", + ) + and ok + ) + return ( + check( + "the connection listener is registered by the time construction returns", + len(stream._connection_listeners) == 1, + f"connection listeners {len(stream._connection_listeners)}", + ) + and ok + ) async def test_registration_happens_at_construction(results: list[bool]) -> None: - stream = make_stream() + stream = make_stream(RefusingSession()) _vehicle, _sent = make_vehicle_with_capture(stream) results.append( @@ -122,10 +153,17 @@ async def test_registration_happens_at_construction(results: list[bool]) -> None all(is_internal for _, _, is_internal in stream._listeners.values()), ) ) + results.append( + check( + "the connection listener is registered by construction", + len(stream._connection_listeners) == 1, + f"connection listeners {len(stream._connection_listeners)}", + ) + ) async def test_auto_close_after_last_public_listener_removed(results: list[bool]) -> None: - stream = make_stream() + stream = make_stream(RefusingSession()) # The internal listener registers at construction; no call needed to set it up. _vehicle, _sent = make_vehicle_with_capture(stream) @@ -158,70 +196,67 @@ async def test_auto_close_after_last_public_listener_removed(results: list[bool] ) -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() +async def test_unpopulated_add_field_fetches_before_deciding(results: list[bool]) -> None: + """A vehicle that has never received a config event or a REST fetch must + 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} + ) + stream = make_stream(session) vehicle, sent = make_vehicle_with_capture(stream) - vehicle.fields = {"BatteryLevel": {"interval_seconds": 60}} # matches the request below - results.append(check("the stream starts disconnected", not stream.connected)) + results.append(check("the vehicle starts unpopulated", not vehicle._populated)) + # Matches what the fetch will reveal - should fetch, then skip. await vehicle.add_field("BatteryLevel", 60) - + results.append(check("get_config was fetched exactly once", session.calls == 1)) results.append( check( - "add_field sends the PATCH even though the record already matches", - len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + "add_field skips the PATCH once the fetched record matches", + sent == [], f"sent {sent}", ) ) + results.append(check("the vehicle is now populated", vehicle._populated)) - -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}} - - stream._response = object() # type: ignore[assignment] # simulate a live connection - results.append(check("the stream is connected", stream.connected)) - - await vehicle.add_field("BatteryLevel", 60) - + # A second call for a field the fetch didn't mention must not re-fetch, + # and must send since it doesn't match. + await vehicle.add_field("ChargeState") + results.append( + check("a later call does not re-fetch once populated", session.calls == 1) + ) 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}, + "a later call sends when the (now-trusted) record doesn't match", + len(sent) == 1 and "ChargeState" in sent[0]["fields"], f"sent {sent}", ) ) -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() +async def test_populated_add_field_skips_without_fetching(results: list[bool]) -> None: + """Once a config event has populated the record, add_field trusts it + outright - no REST fetch, matching the pre-feature status quo for the + push-driven path.""" + stream = make_stream(RefusingSession()) 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( - "the stream is connected and subscribed to every topic", - stream.connected and stream.topics is None, - ) + vehicle._on_config_event( + { + Key.VIN: VIN, + Key.CONFIG: { + "fields": {"BatteryLevel": {"interval_seconds": 60}}, + "prefer_typed": False, + }, + } ) + results.append(check("the config event populated the vehicle", vehicle._populated)) await vehicle.add_field("BatteryLevel", 60) - results.append( check( - "add_field skips the PATCH when the record is live-maintained and matches", + "add_field skips the PATCH without ever calling session.get", sent == [], f"sent {sent}", ) @@ -232,9 +267,8 @@ 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_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) + await test_unpopulated_add_field_fetches_before_deciding(results) + await test_populated_add_field_skips_without_fetching(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 e06c73a..1c3f38f 100644 --- a/tests/test_config_update.py +++ b/tests/test_config_update.py @@ -29,20 +29,23 @@ class FakeStream: """Minimal stand-in for TeslemetryStream.""" manual = True - # 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 ) -> Any: return lambda: None + def async_add_connection_listener(self, callback: Any) -> Any: + return lambda: None + def make_vehicle(vin: str, responses: list[Any]) -> TeslemetryStreamVehicle: """Build a vehicle that records payloads and replays canned responses.""" vehicle = TeslemetryStreamVehicle(FakeStream(), vin) # type: ignore[arg-type] + # These tests exercise the no-op check itself, not the lazy-populate + # fetch (the fake stream has no REST session to serve one) - mark it + # populated like a real connection's config snapshot already would have. + vehicle._populated = True vehicle.sent = [] # type: ignore[attr-defined] async def patch_config(config: dict[str, Any]) -> dict[str, Any]: diff --git a/tests/test_field_type_coercion.py b/tests/test_field_type_coercion.py index 0c1f8e2..c908ad8 100644 --- a/tests/test_field_type_coercion.py +++ b/tests/test_field_type_coercion.py @@ -21,10 +21,6 @@ class FakeStream: """Minimal stand-in for TeslemetryStream that just captures listeners.""" manual = True - # 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 @@ -45,12 +41,19 @@ def async_add_listener( 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 HTTP. + # 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 diff --git a/tests/test_reconnect_config_window.py b/tests/test_reconnect_config_window.py new file mode 100644 index 0000000..7760268 --- /dev/null +++ b/tests/test_reconnect_config_window.py @@ -0,0 +1,173 @@ +"""Regression test for a reconnect-window gap in the config no-op skip. + +``TeslemetryStream.connect()`` sets ``self._response`` and notifies +connection listeners (``stream.connected`` becomes True) before ``listen()`` +has read a single SSE line - in particular, before the server's post-connect +config snapshot has been dispatched to the vehicle's internal config +listener. ``TeslemetryStreamVehicle`` unpopulates itself on disconnect (see +its ``_on_connection_event``), so a field-config call landing in that window +finds itself unpopulated and awaits a fresh REST fetch rather than trusting +the stale pre-disconnect record. +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from teslemetry_stream.const import Key +from teslemetry_stream.stream import TeslemetryStream +from teslemetry_stream.vehicle import TeslemetryStreamVehicle + +VIN = "TESTVIN0000000001" + + +class FakeResponse: + """A response whose content is never actually iterated in this test.""" + + status = 200 + url = "https://api.teslemetry.com/sse" + + def close(self) -> None: + pass + + +class FakeSession: + """A session whose get() hands back a connected-but-empty response, or + (once armed) serves the vehicle's config GET.""" + + def __init__(self) -> None: + self.config_calls = 0 + self.config_body: dict[str, Any] | None = None + + async def get(self, url: str, **kwargs: Any) -> Any: + if "/api/config/" in url: + self.config_calls += 1 + assert self.config_body is not None, "get_config() called before armed" + return FakeConfigResponse(self.config_body) + return FakeResponse() + + +class FakeConfigResponse: + status = 200 + + def __init__(self, body: dict[str, Any]) -> None: + self._body = body + + async def json(self) -> dict[str, Any]: + return self._body + + +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: + kwargs.setdefault("manual", True) + return TeslemetryStream( + session=FakeSession(), # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + **kwargs, + ) + + +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 + + +async def test_reconnect_window_add_field_refetches_instead_of_trusting_stale_record( + results: list[bool], +) -> None: + stream = make_stream() + vehicle, sent = make_vehicle_with_capture(stream) + + # An initial connection, so a later _close_response() has something to + # actually tear down (and thus something to notify listeners about). + await stream.connect() + + # Pre-disconnect state: server had BatteryLevel @ 60s, learned via a + # push event (so the vehicle is populated). + vehicle._on_config_event( + {Key.VIN: VIN, Key.CONFIG: {"fields": {"BatteryLevel": {"interval_seconds": 60}}}} + ) + results.append(check("the vehicle is populated before disconnect", vehicle._populated)) + + # Disconnect: the connection listener unpopulates the record. + stream._close_response() + results.append( + check("disconnect unpopulates the vehicle", not vehicle._populated) + ) + + # Server-side truth changed while disconnected (unbeknownst to this + # 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 + # re-asserting its desired fields - the exact scenario under test. + scheduled: list[asyncio.Task[None]] = [] + + def on_connect(connected: bool) -> None: + if connected: + scheduled.append(asyncio.ensure_future(vehicle.add_field("BatteryLevel", 60))) + + stream.async_add_connection_listener(on_connect) + + # Reconnect: this flips `connected` True and fires `on_connect` + # synchronously, scheduling (not yet running) the add_field task - well + # before the server's config snapshot for this connection arrives. + await stream.connect() + results.append( + check("stream reports connected immediately after connect()", stream.connected) + ) + + await asyncio.gather(*scheduled) + + results.append( + check( + "add_field fetched fresh config instead of trusting the stale pre-disconnect record", + stream._session.config_calls == 1, # type: ignore[attr-defined] + f"config GETs {stream._session.config_calls}", # type: ignore[attr-defined] + ) + ) + results.append( + check( + "add_field sends a PATCH since the fresh 30s answer doesn't match the desired 60s", + len(sent) == 1 and sent[0]["fields"]["BatteryLevel"] == {"interval_seconds": 60}, + f"sent {sent}", + ) + ) + results.append( + check( + "the desired 60s interval is the vehicle's final state", + vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, + f"fields {vehicle.fields}", + ) + ) + + +async def main() -> None: + results: list[bool] = [] + await test_reconnect_window_add_field_refetches_instead_of_trusting_stale_record(results) + + print("-" * 72) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/tests/test_stream_lifecycle.py b/tests/test_stream_lifecycle.py index c99875d..146f898 100644 --- a/tests/test_stream_lifecycle.py +++ b/tests/test_stream_lifecycle.py @@ -434,6 +434,84 @@ def public_mutator(event: dict[str, Any]) -> None: await asyncio.sleep(0) +async def test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use( + results: list[bool], +) -> None: + """A generic public listener that discovers an uncached VIN and calls + get_vehicle() while handling that VIN's own config event registers a new + internal listener that the current dispatch already snapshotted past - + the new vehicle never sees the event that revealed it and stays + unpopulated. That's fine: its first field-config call finds itself + unpopulated and fetches over REST before deciding, so it still ends up + correct rather than seeded from the (already gone) triggering event.""" + new_vin = "TESTVIN0000000002" + session = FakeSession() + session.content_factory = lambda: FakeEventContent( + [ + ( + b'data: {"vin": "' + + new_vin.encode() + + b'", "config": {"fields": ' + + b'{"BatteryLevel": {"interval_seconds": 60}}, ' + + b'"prefer_typed": true}}\n' + ) + ] + ) + stream = make_stream(session) + + discovered: list[TeslemetryStreamVehicle] = [] + + def generic_listener(event: dict[str, Any]) -> None: + vin = event.get("vin") + if vin and vin not in stream.vehicles: + discovered.append(stream.get_vehicle(vin)) + + stream.async_add_listener(generic_listener, {"vin": None}) + + for _ in range(5): + await asyncio.sleep(0) + + results.append(check("the listener discovered the new vehicle", len(discovered) == 1)) + if not discovered: + return + vehicle = discovered[0] + + results.append( + check( + "the newly discovered vehicle missed the triggering event and stays unpopulated", + not vehicle._populated, + f"populated {vehicle._populated}", + ) + ) + + async def patch_config(config: dict[str, Any]) -> dict[str, Any]: + raise AssertionError("should skip: the lazy fetch below reveals a matching record") + + vehicle.patch_config = patch_config # type: ignore[assignment,method-assign] + + 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] + + await vehicle.add_field("BatteryLevel", 60) + results.append( + check( + "the first field-config call fetches and lands on the correct answer", + vehicle._populated + and vehicle.fields.get("BatteryLevel") == {"interval_seconds": 60}, + f"fields {vehicle.fields}", + ) + ) + + stream.close() + await asyncio.sleep(0) + + async def main() -> None: results: list[bool] = [] await test_add_remove_readd_before_loop_runs(results) @@ -444,6 +522,7 @@ async def main() -> None: 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) + await test_vehicle_discovered_mid_dispatch_fetches_correctly_on_first_use(results) print("-" * 72) print("ALL PASS" if all(results) else "FAILURES PRESENT")