diff --git a/AGENTS.md b/AGENTS.md index 06b1e43..2c98f75 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -22,7 +22,8 @@ This file is the project's committed home for project-intrinsic agent knowledge: - `tests/test_config_events.py` covers the record merge and nested-entry validation; `tests/test_config_listener_lifecycle.py` covers construction-time registration (including outside a running loop), the auto-close exclusion, and the populated/unpopulated no-op-skip gating; `tests/test_stream_lifecycle.py` covers the internal-listener-survives-close restart case and a vehicle discovered mid-dispatch of its own config event (misses that event, stays unpopulated, and self-corrects via the lazy fetch on its first field-config call); `tests/test_reconnect_config_window.py` covers the reconnect-window race (a field-config call between `connect()` and that connection's config snapshot re-fetches rather than trusting the stale pre-disconnect record). - `TeslemetryStream` has no `__aenter__`/`__aexit__` - do not reintroduce `async with TeslemetryStream(...)` in the README or examples. Connection lifecycle is entirely listener-driven: `async_add_listener` connects on the first (public) listener and disconnects on the last one removed; `connect()`/`close()`/`listen()` exist for callers who want to manage the connection themselves instead. - `__anext__` treats a `aiohttp.ClientResponseError` with `status` 401 or 403 as terminal, not transient: it sets `active = False` and raises `TeslemetryStreamAuthenticationError` (chaining the original error) rather than retrying, since a rejected token can never succeed on retry. Every other `aiohttp.ClientError` (including other response statuses) keeps the pre-existing backoff-and-reconnect behavior. `tests/test_auth_failure.py` covers both the 401/403 surfacing and that a genuine transient `ClientError` still retries and reconnects. -- `TeslemetryStream` owns exactly one `_listen_task`: `async_add_listener`'s zero-to-one transition only creates it when absent/done, and `listen()` itself checks `asyncio.current_task()` against `self._listen_task` - a second concurrent `listen()` call joins the owner via `await existing_task` instead of racing it for the connection. `connect()` serializes the actual GET behind `self._connect_lock` and re-checks `self.active` after acquiring both the lock and the response, discarding a response that arrived after a stop/supersede rather than publishing it. Internal reconnect paths (EOF, `ClientError`, unexpected exceptions in `__anext__`) call `_close_response()`, which only clears the response and notifies connection listeners - they must not call `close()`, which additionally flips `active=False` and cancels the owned task, i.e. a real stop. `listen()`'s `finally` calls `_close_response()` unconditionally, so task cancellation (however triggered) still releases the connection. `listen()` dispatches over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch. +- `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 through `_dispatch()` (shared with `ingest()`), over a *sorted* snapshot of `_listeners.values()` (never the live dict) with internal listeners ordered first, regardless of registration order: a callback that adds a listener mid-dispatch (e.g. `get_vehicle()` for an uncached VIN) must not raise `RuntimeError: dictionary changed size during iteration` and kill the loop, and a public callback must not get a chance to mutate the event in place before an internal (bookkeeping) listener has cached from it. `_update_connection_listeners()` has the same hazard for `_connection_listeners` (a connection listener calling `get_vehicle()` registers that vehicle's own connection listener mid-dispatch) and is fixed the same way, over a plain `list(...)` snapshot - no ordering requirement there, since connection listeners don't share a mutable event object. `tests/test_stream_lifecycle.py` covers the add/remove/re-add race, duplicate `listen()` calls, cancellation while blocked reading content, close-during-connect, close-during-backoff, a listener mutating `_listeners` mid-dispatch, internal-before-public dispatch order, and a connection listener mutating `_connection_listeners` mid-dispatch. +- `TeslemetryStream.ingest()` (and `TeslemetryStreamVehicle.ingest()`, the same call with the VIN filled in) is the ingestion point for an observation the library did not read off its own SSE connection - a Bluetooth broadcast, today. It builds the native wire event (`vin`/`data`/`createdAt`) plus an open-ended `metadata` dict (`Metadata.SOURCE`/`Metadata.RAW` in `const.py`) and hands it to `_dispatch`, the single fan-out `listen()` also uses - so native events pass through untouched and a consumer's existing `listen_*` callbacks receive both sources with no translation and no second subscription. Dispatch is arrival-ordered and the stream holds no per-field value: nothing is deduplicated, reordered, or dropped, and there is deliberately no source ranking, precedence, or preferred-source field - which report to believe is the consumer's decision, made on `metadata`. Ingesting neither requires nor opens a connection. Neither library depends on the other: the BLE-side shim that shapes a broadcast into this format lives in `tesla-fleet-api` and the consumer wires the two, following the `aiopowerwall`/`EnergySiteRouter` duck-typing precedent. `tests/test_external_ingest.py` covers the native-event regression, the two sources being indistinguishable apart from metadata, and the no-dedup/no-ranking contract. ## Maintaining this file diff --git a/README.md b/README.md index 6c4fe1e..b28cf88 100644 --- a/README.md +++ b/README.md @@ -6,6 +6,7 @@ This is an asynchronous Python 3 library that connects to the Teslemetry Stream - Listen to various telemetry signals from Tesla vehicles - Handle signals using typed listen methods - Write custom listeners for multiple signals +- Ingest observations from other sources into the same listeners ## Installation @@ -208,6 +209,48 @@ stream = TeslemetryStream( `SSE_VEHICLE_TOPICS`, `SSE_ENERGY_TOPICS`, and `SSE_ALL_TOPICS` are convenience presets that expand to those exact names client-side. +## Ingesting Externally Sourced Events + +Some observations arrive from somewhere other than the SSE connection - a +Bluetooth broadcast read locally, for instance. `ingest` accepts one and +delivers it to the listeners you already registered, in the same wire format +the connection itself sends, so no consumer needs a second subscription or a +translation step: + +```python +from teslemetry_stream import Metadata, Signal + +vehicle = stream.get_vehicle("") +vehicle.listen_Locked(print) + +vehicle.ingest( + {Signal.LOCKED: False}, + { + Metadata.SOURCE: "bluetooth", + Metadata.RAW: "VEHICLELOCKSTATE_SELECTIVE_UNLOCKED", + }, +) +``` + +`metadata` records where the observation came from and what its untranslated +wire value was. It is carried on the event and never acted on: `source` keeps +provenance visible for debugging or for a decision the consumer makes, and +`raw` keeps the fidelity a translation discards (any unlocked state reads as +unlocked, but which one is still worth having). It is a plain dict, so a +source can add keys of its own without a format break; `Metadata` names the +two every source is expected to speak. + +Ingesting holds no client and opens no connection, so an observation is +delivered whether or not the stream is connected. + +**Ordering and deduplication.** Every event - native or ingested - is +dispatched in arrival order, exactly as given. The stream keeps no per-field +value and compares nothing against what came before, so if two sources report +the same field, listeners are called twice: once per report, later report +last, neither dropped. There is deliberately no source ranking, precedence, +or preferred source. Which report to believe is the consumer's decision, and +`metadata` is what it decides on. + ## Public Methods in TeslemetryStream Class ### `__init__(session: aiohttp.ClientSession, access_token: str, server: str | None = None, vin: str | None = None, parse_timestamp: bool = False, manual: bool = False, topics: str | Iterable[str] | None = None)` @@ -249,6 +292,9 @@ Add listener for data updates. ### `listen(self)` Listen to the telemetry stream. +### `ingest(data: dict, vin: str | None = None, metadata: dict | None = None, created_at: str | None = None) -> dict` +Deliver an externally sourced observation to this stream's listeners, in the same wire format the connection sends. `vin` defaults to the stream's own. Returns the event as dispatched. See [Ingesting Externally Sourced Events](#ingesting-externally-sourced-events). + ### `listen_Credits(callback: Callable[[CreditsEvent], None]) -> Callable[[], None]` Add listener for credit events. @@ -272,6 +318,9 @@ Replace Fleet Telemetry configuration for the vehicle. ### `config(self) -> dict` Return current configuration for the vehicle. +### `ingest(data: dict, metadata: dict | None = None, created_at: str | None = None) -> dict` +Ingest an externally sourced observation for this vehicle - `TeslemetryStream.ingest` with the VIN filled in. + ### `listen_State(callback: Callable[[bool], None]) -> Callable[[],None]` Listen for vehicle online state polling. The callback receives a boolean value representing whether the vehicle is online. diff --git a/teslemetry_stream/__init__.py b/teslemetry_stream/__init__.py index 0c69208..a2419af 100644 --- a/teslemetry_stream/__init__.py +++ b/teslemetry_stream/__init__.py @@ -3,6 +3,7 @@ SSE_ENERGY_TOPICS, SSE_VEHICLE_TOPICS, Alert, + Metadata, Signal, SseTopic, ) @@ -22,6 +23,7 @@ "SSE_ENERGY_TOPICS", "SSE_VEHICLE_TOPICS", "Alert", + "Metadata", "Signal", "SseTopic", "TeslemetryStream", diff --git a/teslemetry_stream/const.py b/teslemetry_stream/const.py index ea3f1f8..f882afc 100644 --- a/teslemetry_stream/const.py +++ b/teslemetry_stream/const.py @@ -37,6 +37,22 @@ class Key(StrEnum): URL = "url" TOTALS = "totals" TARIFF_CONTENT_V2 = "tariff_content_v2" + METADATA = "metadata" + + +class Metadata(StrEnum): + """Conventional keys inside an ingested event's `metadata` dict. + + The dict is open ended by design, so a new key can be added without a + format break; these are the ones every source is expected to speak. + """ + + #: Where the observation came from, e.g. "bluetooth". Provenance is + #: recorded so it stays visible, never so the library can rank sources. + SOURCE = "source" + #: The wire value before any translation, kept because collapsing it to + #: the streamed type (e.g. any-unlocked-is-unlocked) discards fidelity. + RAW = "raw" class Signal(StrEnum): diff --git a/teslemetry_stream/stream.py b/teslemetry_stream/stream.py index d3d7cdc..65dc6f8 100644 --- a/teslemetry_stream/stream.py +++ b/teslemetry_stream/stream.py @@ -9,7 +9,7 @@ import aiohttp -from .const import CreditsEvent +from .const import CreditsEvent, Key from .energysite import TeslemetryStreamEnergySite from .exception import TeslemetryStreamAuthenticationError, TeslemetryStreamEnded from .vehicle import TeslemetryStreamVehicle @@ -339,12 +339,7 @@ async def __anext__(self) -> dict[str, Any]: if field == "data": data = json.loads(value) if self.parse_timestamp: - main, _, ns = data["createdAt"].partition(".") - data["timestamp"] = int( - datetime.strptime(main, "%Y-%m-%dT%H:%M:%S") - .replace(tzinfo=timezone.utc) - .timestamp() - ) * 1000 + int(ns[:3]) + data["timestamp"] = _parse_created_at(data["createdAt"]) return cast(dict[str, Any], data) raise TeslemetryStreamEnded() except StopAsyncIteration as e: @@ -452,26 +447,99 @@ async def listen(self) -> None: try: async for event in self: if event: - # A snapshot, not a live view - a callback that creates a - # vehicle (get_vehicle) or otherwise adds a listener - # mid-dispatch must not mutate _listeners while this is - # iterating it, which would raise RuntimeError and kill - # the loop. Internal (bookkeeping) listeners go first, so - # one can cache from the pristine event before any public - # callback gets a chance to mutate it in place. - ordered = sorted(self._listeners.values(), key=lambda item: not item[2]) - for listener, filters, _internal in ordered: - if recursive_match(filters, event): - try: - listener(event) - except Exception as error: - LOGGER.error("Uncaught error in listener: %s", error) + self._dispatch(event) finally: self._close_response() if self._listen_task is current_task: self._listen_task = None LOGGER.debug("Listen has finished") + def _dispatch(self, event: dict[str, Any]) -> None: + """ + Fan one event out to every listener whose filters match it. + + Shared by the SSE reader and by `ingest`, so an externally sourced + event reaches consumers by the same path a native one does. + + :param event: Event to dispatch. + """ + # A snapshot, not a live view - a callback that creates a vehicle + # (get_vehicle) or otherwise adds a listener mid-dispatch must not + # mutate _listeners while this is iterating it, which would raise + # RuntimeError and kill the loop. Internal (bookkeeping) listeners go + # first, so one can cache from the pristine event before any public + # callback gets a chance to mutate it in place. + ordered = sorted(self._listeners.values(), key=lambda item: not item[2]) + for listener, filters, _internal in ordered: + if recursive_match(filters, event): + try: + listener(event) + except Exception as error: + LOGGER.error("Uncaught error in listener: %s", error) + + def ingest( + self, + data: dict[str, Any], + vin: str | None = None, + metadata: dict[str, Any] | None = None, + created_at: str | None = None, + ) -> dict[str, Any]: + """ + Feed an externally sourced observation into this stream's listeners. + + The event is built in the same wire format the SSE connection + delivers - `{"vin": ..., "data": {...}, "createdAt": ...}` - and goes + out through the same dispatch, so a consumer's existing `listen_*` + callbacks receive it with no translation and no separate + subscription. Nothing here connects, reads, or holds a client: the + observation is an argument, and ingesting one works whether or not + the SSE connection is up. + + Every ingested event is dispatched, in arrival order, exactly as it + was given. The stream keeps no per-field value and does not compare + an event against what came before, so two sources reporting the same + field produce two dispatches - the later one simply arrives later, + the way repeated native events already do. There is deliberately no + source ranking, precedence, or deduplication: which source to + believe is the consumer's call, and `metadata` is what it decides on. + + :param data: Signal payload keyed by signal name, e.g. + `{"Locked": True}` or `{"DoorState": {"TrunkFront": False}}`. + :param vin: Vehicle Identification Number. Defaults to the stream's + own `vin` for a single-vehicle client. + :param metadata: Provenance carried alongside the event and never + acted on - see `Metadata` for the conventional keys (`source`, + `raw`). A dict so it can grow without a format break. + :param created_at: Observation time in the stream's own + `%Y-%m-%dT%H:%M:%S.%fZ` format. Defaults to now. + :return: The event as dispatched. + :raises ValueError: If no VIN is available. + :raises TypeError: If `data` or `metadata` is not a dict. + """ + vin = vin or self.vin + if not vin: + raise ValueError("ingest requires a vin, either its own or the stream's") + if not isinstance(data, dict): + raise TypeError("data must be a dict keyed by signal name") + if metadata is not None and not isinstance(metadata, dict): + raise TypeError("metadata must be a dict") + + event: dict[str, Any] = { + # Plain string keys, exactly as a decoded SSE event carries them. + Key.VIN.value: vin, + # Copied, not aliased - a caller reusing its own payload dict + # must not retroactively change an event already dispatched. + Key.DATA.value: dict(data), + Key.CREATED_AT.value: created_at or _now(), + Key.METADATA.value: dict(metadata) if metadata else {}, + } + if self.parse_timestamp: + # Before dispatch, so a malformed created_at raises to the caller + # instead of half-delivering an event. + event["timestamp"] = _parse_created_at(event[Key.CREATED_AT]) + self._dispatch(event) + return event + def listen_Credits( self, callback: Callable[[CreditsEvent], None] ) -> Callable[[], None]: @@ -527,3 +595,23 @@ def recursive_match(dict1: dict[str, Any] | None, dict2: dict[str, Any]) -> bool return False # No differences found return True + + +def _parse_created_at(created_at: str) -> int: + """ + Convert a stream `createdAt` string to epoch milliseconds. + + :param created_at: Timestamp as sent on the wire. + :return: Milliseconds since the epoch. + """ + main, _, ns = created_at.partition(".") + return int( + datetime.strptime(main, "%Y-%m-%dT%H:%M:%S") + .replace(tzinfo=timezone.utc) + .timestamp() + ) * 1000 + int(ns[:3]) + + +def _now() -> str: + """Current time in the same format the stream sends `createdAt` in.""" + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%f")[:-3] + "Z" diff --git a/teslemetry_stream/vehicle.py b/teslemetry_stream/vehicle.py index 70c7949..4182358 100644 --- a/teslemetry_stream/vehicle.py +++ b/teslemetry_stream/vehicle.py @@ -341,6 +341,22 @@ 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}}) + def ingest( + self, + data: dict[str, Any], + metadata: dict[str, Any] | None = None, + created_at: str | None = None, + ) -> dict[str, Any]: + """Ingest an externally sourced observation for this vehicle. + + A convenience for `TeslemetryStream.ingest` with this vehicle's VIN + filled in; see it for the wire format, the metadata dict, and the + dispatch rules. + """ + return self.stream.ingest( + data, vin=self.vin, metadata=metadata, created_at=created_at + ) + 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_external_ingest.py b/tests/test_external_ingest.py new file mode 100644 index 0000000..c71cb87 --- /dev/null +++ b/tests/test_external_ingest.py @@ -0,0 +1,501 @@ +"""Tests for `TeslemetryStream.ingest`, the externally sourced event surface. + +An observation the library did not pull off its own SSE connection (a +Bluetooth broadcast, today) is handed in as a stream-shaped dict and reaches +the same listeners a native event does. What is asserted here: + +- native SSE events are delivered exactly as before, with nothing added; +- a BLE-shaped event and a native event for the same field are + indistinguishable to a consumer apart from their metadata - the field + names and value types the two sources emit are the same; +- every event is dispatched in arrival order with no deduplication and no + source ranking, so two sources reporting one field produce two dispatches. +""" +from __future__ import annotations + +import asyncio +import json +from collections.abc import Callable +from typing import Any + +from teslemetry_stream.const import Metadata, Signal +from teslemetry_stream.stream import TeslemetryStream + +VIN = "TESTVIN0000000001" + +# Both sources report these three fields with the same names and the same +# value types - the payload-compatibility claim this file exists to pin. +NATIVE_LINES = [ + b'data: {"vin": "TESTVIN0000000001", "createdAt": "2026-08-26T03:40:26.399Z",' + b' "data": {"Locked": true}}\n', + b'data: {"vin": "TESTVIN0000000001", "createdAt": "2026-08-26T03:40:36.399Z",' + b' "data": {"ChargePortDoorOpen": false}}\n', + b'data: {"vin": "TESTVIN0000000001", "createdAt": "2026-08-26T03:40:46.399Z",' + b' "data": {"DoorState": {"TrunkFront": true}}}\n', +] + +NATIVE_EVENTS: list[dict[str, Any]] = [ + json.loads(line.decode().partition(": ")[2]) for line in NATIVE_LINES +] + + +class FakeEventContent: + """Async-iterable response body yielding canned SSE lines, then blocking.""" + + def __init__(self, lines: list[bytes]) -> None: + self._lines = list(lines) + self._blocker: asyncio.Future[None] = asyncio.get_running_loop().create_future() + + def __aiter__(self) -> FakeEventContent: + return self + + async def __anext__(self) -> bytes: + if self._lines: + return self._lines.pop(0) + await self._blocker + raise AssertionError("unreachable - blocker only resolves via cancellation") + + +class FakeResponse: + """Minimal stand-in for the aiohttp response `connect()` awaits.""" + + def __init__(self, content: Any) -> None: + self.url = "https://fake.teslemetry.com/sse" + self.status = 200 + self.content = content + self.closed = False + + def close(self) -> None: + self.closed = True + + +class FakeSession: + """Counts `get()` calls and serves canned SSE content.""" + + def __init__(self, lines: list[bytes] | None = None) -> None: + self.calls = 0 + self.lines = lines or [] + + async def get(self, url: str, **kwargs: Any) -> FakeResponse: + self.calls += 1 + return FakeResponse(FakeEventContent(self.lines)) + + +def make_stream( + session: FakeSession, + manual: bool = True, + vin: str | None = None, + parse_timestamp: bool = False, +) -> TeslemetryStream: + return TeslemetryStream( + session=session, # type: ignore[arg-type] + access_token="test-token", + server="api.teslemetry.com", + manual=manual, + vin=vin, + parse_timestamp=parse_timestamp, + ) + + +def make_vehicle(stream: TeslemetryStream) -> Any: + """A vehicle whose config is already known, so listen_* issues no REST.""" + vehicle = stream.get_vehicle(VIN) + vehicle.fields = {s.value: {} for s in Signal} + vehicle._populated = True + return vehicle + + +def dispatch_native(stream: TeslemetryStream, event: dict[str, Any]) -> None: + """Deliver an event the way the SSE reader does. + + `test_native_events_are_delivered_unchanged` is what proves `listen()` + still goes through here. + """ + stream._dispatch(event) + + +async def wait_for(predicate: Callable[[], bool], timeout: float = 2.0) -> bool: + """Poll until `predicate` holds or the timeout expires.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if predicate(): + return True + await asyncio.sleep(0) + return predicate() + + +def check(label: str, ok: bool, detail: str = "") -> bool: + print(f"{label:<72} {'PASS' if ok else 'FAIL'}{' ' + detail if detail else ''}") + return ok + + +async def test_native_events_are_delivered_unchanged(results: list[bool]) -> None: + """The regression that would matter most: adding an ingestion point must + not touch what the SSE connection itself delivers.""" + session = FakeSession(NATIVE_LINES) + stream = make_stream(session, manual=False) + delivered: list[dict[str, Any]] = [] + stream.async_add_listener(delivered.append) + await wait_for(lambda: len(delivered) == len(NATIVE_EVENTS)) + stream.close() + + results.append( + check( + "native SSE events arrive exactly as sent", + delivered == NATIVE_EVENTS, + f"got {delivered}", + ) + ) + results.append( + check( + "native SSE events gain no metadata key", + all("metadata" not in event for event in delivered), + f"got {[sorted(e) for e in delivered]}", + ) + ) + + # parse_timestamp still derives its epoch milliseconds from createdAt. + session = FakeSession(NATIVE_LINES[:1]) + stream = make_stream(session, manual=False, parse_timestamp=True) + stamped: list[dict[str, Any]] = [] + stream.async_add_listener(stamped.append) + await wait_for(lambda: len(stamped) == 1) + stream.close() + results.append( + check( + "native parse_timestamp is unchanged", + bool(stamped) and stamped[0]["timestamp"] == 1787715626399, + f"got {stamped}", + ) + ) + + +async def test_ingested_and_native_events_are_indistinguishable( + results: list[bool], +) -> None: + """A BLE-shaped event and a stream-shaped event for the same field must + look the same to a consumer once through the funnel, apart from metadata.""" + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + + locked: list[Any] = [] + charge_port: list[Any] = [] + trunk: list[Any] = [] + vehicle.listen_Locked(locked.append) + vehicle.listen_ChargePortDoorOpen(charge_port.append) + vehicle.listen_TrunkFront(trunk.append) + + raw: list[dict[str, Any]] = [] + stream.async_add_listener(raw.append, {"vin": VIN}) + + for event in NATIVE_EVENTS: + dispatch_native(stream, event) + + ingested = [ + vehicle.ingest( + {Signal.LOCKED: True}, + {Metadata.SOURCE: "bluetooth", Metadata.RAW: "VEHICLELOCKSTATE_LOCKED"}, + ), + vehicle.ingest( + {Signal.CHARGE_PORT_DOOR_OPEN: False}, + {Metadata.SOURCE: "bluetooth", Metadata.RAW: "CLOSURESTATE_CLOSED"}, + ), + vehicle.ingest( + {Signal.DOOR_STATE: {"TrunkFront": True}}, + {Metadata.SOURCE: "bluetooth", Metadata.RAW: "CLOSURESTATE_OPEN"}, + ), + ] + + results.append( + check( + "typed listeners deliver the same values from both sources", + locked == [True, True] + and charge_port == [False, False] + and trunk == [True, True], + f"got {locked} {charge_port} {trunk}", + ) + ) + results.append( + check( + "and the same Python types", + all(type(a) is type(b) for a, b in (locked, charge_port, trunk)), + f"got {[type(v).__name__ for v in locked + charge_port + trunk]}", + ) + ) + + def comparable(event: dict[str, Any]) -> dict[str, Any]: + return {k: v for k, v in event.items() if k not in ("metadata", "createdAt")} + + results.append( + check( + "ingested events match native ones apart from metadata", + [comparable(e) for e in ingested] == [comparable(e) for e in NATIVE_EVENTS], + f"got {[comparable(e) for e in ingested]}", + ) + ) + results.append( + check( + "an ingested event carries the documented wire keys", + all( + sorted(e) == ["createdAt", "data", "metadata", "vin"] for e in ingested + ), + f"got {[sorted(e) for e in ingested]}", + ) + ) + results.append( + check( + "an ingested event's keys are plain strings, as decoded JSON gives", + all(type(key) is str for event in ingested for key in event), + f"got {[[type(k).__name__ for k in e] for e in ingested]}", + ) + ) + results.append( + check( + "a plain listener sees both sources on one subscription", + len(raw) == 6, + f"got {len(raw)}", + ) + ) + + +async def test_both_sources_reporting_one_field(results: list[bool]) -> None: + """No deduplication and no reordering: every event is dispatched as it + arrives, whichever source it came from.""" + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + + locked: list[Any] = [] + vehicle.listen_Locked(locked.append) + sources: list[Any] = [] + stream.async_add_listener( + lambda e: sources.append(e.get("metadata", {}).get("source")), {"vin": VIN} + ) + + # Bluetooth reports the change first, the stream repeats it a beat later, + # then Bluetooth reports the next change. + vehicle.ingest({Signal.LOCKED: False}, {Metadata.SOURCE: "bluetooth"}) + dispatch_native(stream, {"vin": VIN, "data": {Signal.LOCKED: False}}) + vehicle.ingest({Signal.LOCKED: True}, {Metadata.SOURCE: "bluetooth"}) + + results.append( + check( + "both sources' events are delivered, in arrival order", + locked == [False, False, True], + f"got {locked}", + ) + ) + results.append( + check( + "the repeated value is not deduplicated away", + sources == ["bluetooth", None, "bluetooth"], + f"got {sources}", + ) + ) + + # The same sequence with the sources swapped behaves identically - the + # stream holds no preference between them. + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + swapped: list[Any] = [] + vehicle.listen_Locked(swapped.append) + vehicle.ingest({Signal.LOCKED: False}, {Metadata.SOURCE: "stream"}) + vehicle.ingest({Signal.LOCKED: False}, {Metadata.SOURCE: "bluetooth"}) + vehicle.ingest({Signal.LOCKED: True}, {Metadata.SOURCE: "stream"}) + results.append( + check( + "no source outranks another", + swapped == locked, + f"got {swapped}", + ) + ) + + # An older observation arriving late is still delivered: the stream keeps + # no per-field value to compare it against. + late: list[Any] = [] + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + vehicle.listen_Locked(late.append) + vehicle.ingest({Signal.LOCKED: True}, created_at="2026-08-26T03:40:46.000Z") + vehicle.ingest({Signal.LOCKED: False}, created_at="2026-08-26T03:40:26.000Z") + results.append( + check( + "an out-of-order observation is delivered, not dropped", + late == [True, False], + f"got {late}", + ) + ) + + +async def test_metadata_is_carried_and_never_acted_on(results: list[bool]) -> None: + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + seen: list[dict[str, Any]] = [] + stream.async_add_listener(seen.append, {"vin": VIN}) + + payload = {Signal.LOCKED: False} + metadata = { + Metadata.SOURCE: "bluetooth", + Metadata.RAW: "VEHICLELOCKSTATE_SELECTIVE_UNLOCKED", + "rssi": -67, + } + event = vehicle.ingest(payload, metadata) + results.append( + check( + "metadata is carried verbatim, extra keys included", + event["metadata"] == metadata and seen[0]["metadata"] == metadata, + f"got {event['metadata']}", + ) + ) + + # The caller's own dicts are its own; mutating them after the fact must + # not rewrite an event already delivered. + payload[Signal.LOCKED] = True + metadata["rssi"] = -20 + results.append( + check( + "the dispatched event is unaffected by later caller mutation", + event["data"] == {Signal.LOCKED: False} and event["metadata"]["rssi"] == -67, + f"got {event}", + ) + ) + + plain = vehicle.ingest({Signal.LOCKED: True}) + results.append( + check( + "omitted metadata is an empty dict, not a missing key", + plain["metadata"] == {}, + f"got {plain}", + ) + ) + + +async def test_ingest_needs_no_connection(results: list[bool]) -> None: + """Bluetooth keeps reporting while the SSE connection is down, so + ingesting must neither require nor start one.""" + session = FakeSession() + stream = make_stream(session) + vehicle = make_vehicle(stream) + locked: list[Any] = [] + vehicle.listen_Locked(locked.append) + + vehicle.ingest({Signal.LOCKED: True}, {Metadata.SOURCE: "bluetooth"}) + + results.append( + check( + "an ingested event is delivered while disconnected", + locked == [True] and not stream.connected, + f"got {locked}, connected={stream.connected}", + ) + ) + results.append( + check( + "ingesting opens no connection", + session.calls == 0 and not stream.active, + f"got calls={session.calls}, active={stream.active}", + ) + ) + + +async def test_ingest_rejects_malformed_input(results: list[bool]) -> None: + stream = make_stream(FakeSession(), vin=VIN) + + def raises(exc: type[BaseException], call: Callable[[], Any]) -> bool: + try: + call() + except exc: + return True + except BaseException: + return False + return False + + results.append( + check( + "a non-dict payload raises TypeError", + raises(TypeError, lambda: stream.ingest(["Locked"])), # type: ignore[arg-type] + ) + ) + results.append( + check( + "non-dict metadata raises TypeError", + raises( + TypeError, + lambda: stream.ingest({Signal.LOCKED: True}, metadata="bluetooth"), # type: ignore[arg-type] + ), + ) + ) + results.append( + check( + "the stream's own vin is used when none is given", + stream.ingest({Signal.LOCKED: True})["vin"] == VIN, + ) + ) + results.append( + check( + "a missing vin raises ValueError", + raises( + ValueError, lambda: make_stream(FakeSession()).ingest({Signal.LOCKED: True}) + ), + ) + ) + + +async def test_ingest_shares_the_native_dispatch_guarantees( + results: list[bool], +) -> None: + """Same dispatch, same protections: a raising listener is contained, and + internal (bookkeeping) listeners still run before public ones.""" + stream = make_stream(FakeSession()) + vehicle = make_vehicle(stream) + + order: list[str] = [] + stream.async_add_listener(lambda e: order.append("public"), {"vin": VIN}) + stream.async_add_listener( + lambda e: order.append("internal"), {"vin": VIN}, internal=True + ) + + def boom(event: dict[str, Any]) -> None: + raise RuntimeError("listener blew up") + + stream.async_add_listener(boom, {"vin": VIN}) + delivered: list[Any] = [] + vehicle.listen_Locked(delivered.append) + + vehicle.ingest({Signal.LOCKED: True}, {Metadata.SOURCE: "bluetooth"}) + + results.append( + check( + "internal listeners run before public ones", + order == ["internal", "public"], + f"got {order}", + ) + ) + results.append( + check( + "a raising listener does not stop the rest", + delivered == [True], + f"got {delivered}", + ) + ) + + +async def main() -> None: + results: list[bool] = [] + await test_native_events_are_delivered_unchanged(results) + await test_ingested_and_native_events_are_indistinguishable(results) + await test_both_sources_reporting_one_field(results) + await test_metadata_is_carried_and_never_acted_on(results) + await test_ingest_needs_no_connection(results) + await test_ingest_rejects_malformed_input(results) + await test_ingest_shares_the_native_dispatch_guarantees(results) + # Let the fire-and-forget field-config tasks listen_* schedules finish. + await asyncio.sleep(0) + + print("-" * 82) + print("ALL PASS" if all(results) else "FAILURES PRESENT") + if not all(results): + raise SystemExit(1) + + +if __name__ == "__main__": + asyncio.run(main())