diff --git a/AGENTS.md b/AGENTS.md index 345a8fe..3837053 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -69,6 +69,8 @@ Commands (vehicle/commands.py) - protobuf-based signed command implementation (A `Router` (`router/base.py`) is an entity-agnostic composition wrapper (not part of the inheritance chain) that chains an ordered list of two-or-more backends sharing a common method surface and dispatches each method call down the chain with automatic per-command failover: it tries the first backend that has the method and, on any exception except `BluetoothUnconfirmedCommand`, retries the same call on the next backend that has it, returning the first success (raising the last error only if every applicable backend fails, `AttributeError` only if none has the method). Non-callable attributes resolve to the first backend that has them. Constructor: `Router(primary, secondary, *more_backends, health=None)`. The health check (`bool` | sync callable | async callable returning `bool`; omitted = attempt primary, fail over on exception with no probe) gates **only the primary**; the rest of the chain is reached purely through per-command failover — there is deliberately no per-backend health matrix. Double-execution caveat: a non-idempotent command that fails mid-flight can be re-run on the next backend, except for `BluetoothUnconfirmedCommand`, which propagates without replay. +`ObservationFunnel` (`funnel.py`) is the **read** side, a separate mechanism from the command `Router` and not in its inheritance chain. It is a **funnel, not a selector**: every attached publisher feeds the same per-field listeners, so a field bound to one source survives that source dropping. There is deliberately no source health, availability, grace window, failback delay, priority, stickiness or per-field selection anywhere in it — **unavailability is a value a source reports** (a null/SNA reading), never something the funnel infers from a link dropping; inferring it would be the funnel asserting data it does not have. The only arbitration is `publish()` ignoring an observation older than the last one for that field and not re-dispatching an unchanged value; both are hard-coded, not configurable. It is **entirely synchronous and can never originate a request**: no `async def`/`await`, no polling loop, no request callable, no scheduling task — `tests/test_funnel.py::TestFunnelCannotOriginateWork` locks that in against the module's own AST, so keep the module synchronous rather than adding a fetch path. Polling belongs entirely to an external consumer, which may gate its own schedule on `listen_demand(paths, cb)` (a read-only observer over the activation counts) and feed results back through `VehicleDataResultPublisher.publish_result(dict)` — that publisher holds no client, session, or callable able to obtain one. Publishers push into the funnel (which is itself the `ObservationSink`) via `publish(Observation)`; `observed_at` values must come from one monotonic clock shared by every publisher on a funnel. `value(path)` returns the last observed value, its `None` meaning either never observed or reported unavailable. Fields are deliberately three (`Locked`, `ChargePortDoorOpen`, `DoorState.TrunkFront`); translations are positive allowlists, and an unmapped VCSEC enum or absent JSON leaf emits no observation rather than a guess, while an explicit JSON null emits an unavailable value. `BleBroadcastPublisher` reuses the existing `VehicleBluetooth` `listen_vehicle_lock_state`/`listen_charge_port`/`listen_front_trunk` seams and never connects, reads, or commands; because `VEHICLELOCKSTATE_UNLOCKED` is 0 with no proto3 presence, every VCSEC status broadcast reports a lock state and the funnel deduplicates the repeats. VCSEC `INTERNAL_LOCKED`/`SELECTIVE_UNLOCKED` and closure `UNKNOWN`/`FAILED_UNLATCH` are unmapped pending live-frame validation. + `VehicleRouter` and `EnergySiteRouter` (`router/vehicle.py`, `router/energysite.py`) are thin entity-specific `Router` subclasses. `VehicleRouter(bluetooth_primary, teslemetry_secondary)` pairs a `VehicleBluetooth` primary with a cloud (`TeslemetryVehicle`) secondary; `EnergySiteRouter(local_energysite, teslemetry_energysite)` pairs a duck-typed local `EnergySite`-shaped object (e.g. aiopowerwall's `PowerwallEnergySite`, no dependency added) with a cloud `TeslemetryEnergySite` fallback. Both re-export from `router/__init__.py` (`tesla_fleet_api.router.Router` etc.) and from `tesla/__init__.py` (`tesla_fleet_api.tesla.Router`) for backward compatibility. They have no factory on the `Vehicles`/`EnergySites` collections. This repo owns the RSA keypair lifecycle and cloud registration (`Tesla.get_rsa_private_key`, `EnergySite.add_authorized_client`) that aiopowerwall's local signed transport depends on but does not implement itself; see `docs/energy_local_control.md` for the end-to-end pairing + `EnergySiteRouter` composition flow. The cloud-only `set_island_mode`/`go_off_grid`/`reconnect_grid` (`tesla/energysite.py`) can only send an unsigned `grpc_command`, which gateways can acknowledge without actuating the contactor — rather than ship that as a silent no-op, they unconditionally raise `SignedCommandRequired` (`exceptions.py`); only the signed local path via `add_authorized_client` + `EnergySiteRouter` actually actuates, and a success response from that transport still doesn't prove the contactor moved — verify state after the call. ### Vehicle Collections diff --git a/tesla_fleet_api/__init__.py b/tesla_fleet_api/__init__.py index f7207a3..3d5d0df 100644 --- a/tesla_fleet_api/__init__.py +++ b/tesla_fleet_api/__init__.py @@ -4,6 +4,15 @@ __version__ = "1.10.1" from tesla_fleet_api.const import Region, is_valid_region +from tesla_fleet_api.funnel import ( + BleBroadcastPublisher, + FieldPath, + Observation, + ObservationFunnel, + ObservationSink, + Publisher, + VehicleDataResultPublisher, +) from tesla_fleet_api.tariff import ( TariffPeriod, TariffRate, @@ -23,6 +32,12 @@ from tesla_fleet_api.util import firmware_at_least, firmware_compare __all__ = [ + "BleBroadcastPublisher", + "FieldPath", + "Observation", + "ObservationFunnel", + "ObservationSink", + "Publisher", "Region", "TariffPeriod", "TariffRate", @@ -33,6 +48,7 @@ "Teslemetry", "TeslemetryClientRegistration", "Tessie", + "VehicleDataResultPublisher", "firmware_at_least", "firmware_compare", "get_tariff_periods", diff --git a/tesla_fleet_api/funnel.py b/tesla_fleet_api/funnel.py new file mode 100644 index 0000000..f3cbbdc --- /dev/null +++ b/tesla_fleet_api/funnel.py @@ -0,0 +1,455 @@ +"""Observation funnel: many push sources, one listener per field. + +A field bound to a single source disappears whenever that source is +unavailable. Every publisher attached here feeds the same listeners, so a +field survives the loss of any one source. Nothing in this module chooses +between sources: unavailability is a value a source reports, never something +the funnel infers from a link dropping. + +The funnel is a pure consumer of observations pushed to it. It is entirely +synchronous and holds no request callable, polling loop, HTTP/BLE read, or +scheduling task: a source the funnel could drive is a source it could be made +to poll. +""" + +from __future__ import annotations + +import time +from collections.abc import Mapping +from dataclasses import dataclass +from typing import TYPE_CHECKING, Any, Callable, Iterator, Protocol, TypeVar + +from tesla_fleet_api.const import LOGGER, StrEnum +from tesla_protocol.command.vcsec_pb2 import ( + ClosureState_E, + VehicleLockState_E, +) + +if TYPE_CHECKING: + from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth + +Unsubscribe = Callable[[], None] +Clock = Callable[[], float] + +# A reading, or None for a source reporting the field itself as unavailable. +Value = bool | None + +T = TypeVar("T") + +_FATAL_CALLBACK_ERRORS = (KeyboardInterrupt, SystemExit) + + +class FieldPath(StrEnum): + """A canonical routable field, valued to match its public signal name. + + A compound signal is split into one path per facet, because a broadcast and + a ``vehicle_data`` result each expose individual leaves. + """ + + LOCKED = "Locked" + CHARGE_PORT_DOOR_OPEN = "ChargePortDoorOpen" + DOOR_STATE_TRUNK_FRONT = "DoorState.TrunkFront" + + +@dataclass(frozen=True, slots=True) +class Observation: + """One timestamped value for one field.""" + + path: FieldPath + value: Value + # Seconds off one monotonic clock shared by every publisher on a funnel; + # observations from different sources are ordered against each other. + observed_at: float + + +class ObservationSink(Protocol): + """The handle a publisher is given to feed the funnel.""" + + def publish(self, observation: Observation) -> None: ... + + +class Publisher(Protocol): + """A source of observations for one or more canonical fields.""" + + @property + def paths(self) -> frozenset[FieldPath]: ... + + def attach(self, sink: ObservationSink) -> Unsubscribe: ... + + def request(self, paths: frozenset[FieldPath]) -> None: + """Begin supplying ``paths``: subscription accounting, not a data request.""" + + def release(self, paths: frozenset[FieldPath]) -> None: + """Stop supplying ``paths``.""" + + +def _dispatch(callback: Callable[[T], None], value: T) -> None: + try: + callback(value) + except _FATAL_CALLBACK_ERRORS: + raise + except BaseException: + LOGGER.exception("observation funnel listener callback failed") + + +def _noop() -> None: + return None + + +@dataclass(eq=False) +class _Attached: + publisher: Publisher + detach: Unsubscribe + + +@dataclass(eq=False) +class _DemandObserver: + paths: frozenset[FieldPath] + callback: Callable[[bool], None] + active: bool + + +class ObservationFunnel: + """Funnels observations from every attached publisher into one listener set. + + Many to one: a listener registered for a field receives what any publisher + reports for it, in arrival order. There is no source selection, ranking, or + health model here, and detaching or silencing a publisher produces no value + of its own. + """ + + def __init__(self) -> None: + self._publishers: list[_Attached] = [] + self._listeners: dict[FieldPath, list[Callable[[Value], None]]] = {} + self._demand_observers: list[_DemandObserver] = [] + self._observed: dict[FieldPath, Observation] = {} + + # -- Publishers -------------------------------------------------------- + + def attach(self, publisher: Publisher) -> Unsubscribe: + """Attach ``publisher`` and return its detach closure.""" + attached = _Attached(publisher=publisher, detach=_noop) + # Registered before attaching so a publisher that reports an + # observation synchronously is not discarded as unknown. + self._publishers.append(attached) + attached.detach = publisher.attach(self) + active = self._active_paths(publisher) + if active: + publisher.request(active) + + def detach() -> None: + if attached not in self._publishers: + return + self._publishers.remove(attached) + active = self._active_paths(publisher) + if active: + publisher.release(active) + attached.detach() + + return detach + + def _active_paths(self, publisher: Publisher) -> frozenset[FieldPath]: + return frozenset(p for p in publisher.paths if self._listeners.get(p)) + + # -- Sink --------------------------------------------------------------- + + def publish(self, observation: Observation) -> None: + """Accept one observation from any publisher and fan it out.""" + path = observation.path + previous = self._observed.get(path) + if previous is not None and observation.observed_at < previous.observed_at: + # A later reading of this field already stands. + return + self._observed[path] = observation + if previous is not None and previous.value == observation.value: + return + for callback in list(self._listeners.get(path, ())): + # A callback may publish, and that nested update has already given + # every listener the newer value; continuing here would leave the + # rest holding a value the funnel no longer holds. + if self._observed[path] is not observation: + return + _dispatch(callback, observation.value) + + # -- Values and listeners ----------------------------------------------- + + def value(self, path: FieldPath) -> Value: + """The last value observed for ``path``. + + ``None`` means there is no reading: either nothing has ever been + observed, or a source reported the field as unavailable. + """ + observation = self._observed.get(path) + return None if observation is None else observation.value + + def listen(self, path: FieldPath, callback: Callable[[Value], None]) -> Unsubscribe: + """Register a listener for ``path``, called on each changed value. + + The first listener for a path activates it on every capable publisher; + the last one to leave releases it. Registration dispatches nothing; + the value standing at that moment is :meth:`value`. + """ + listeners = self._listeners.setdefault(path, []) + listeners.append(callback) + if len(listeners) == 1: + paths = frozenset({path}) + for attached in list(self._publishers): + if path in attached.publisher.paths: + attached.publisher.request(paths) + self._notify_demand() + + released = False + + def unsubscribe() -> None: + # Guarded per registration: the same callback may be registered + # twice, and a repeated release must not drop the other one. + nonlocal released + if released: + return + released = True + try: + listeners.remove(callback) + except ValueError: + return + if not listeners: + paths = frozenset({path}) + for attached in list(self._publishers): + if path in attached.publisher.paths: + attached.publisher.release(paths) + self._notify_demand() + + return unsubscribe + + # -- Demand ------------------------------------------------------------- + + def listen_demand( + self, paths: frozenset[FieldPath], callback: Callable[[bool], None] + ) -> Unsubscribe: + """Observe whether any of ``paths`` has at least one live listener. + + Derived from the activation counts the funnel already keeps. It + reports; it never starts work of its own. + """ + observer = _DemandObserver( + paths=paths, callback=callback, active=self._demand(paths) + ) + self._demand_observers.append(observer) + _dispatch(callback, observer.active) + + released = False + + def unsubscribe() -> None: + nonlocal released + if released: + return + released = True + try: + self._demand_observers.remove(observer) + except ValueError: + pass + + return unsubscribe + + def _demand(self, paths: frozenset[FieldPath]) -> bool: + return any(self._listeners.get(p) for p in paths) + + def _notify_demand(self) -> None: + for observer in list(self._demand_observers): + active = self._demand(observer.paths) + if active != observer.active: + observer.active = active + _dispatch(observer.callback, active) + + +# -- Bluetooth broadcast publisher ----------------------------------------- + +# UNLOCKED is 0 and the enum has no proto3 presence, so every status broadcast +# reports a lock state and the funnel deduplicates the repeats. +# +# INTERNAL_LOCKED and SELECTIVE_UNLOCKED are deliberately unmapped: reducing +# either to one boolean is unvalidated against live frames, and emitting +# nothing keeps the last confirmed value instead of guessing. +_LOCK_STATES: Mapping[int, bool] = { + VehicleLockState_E.VEHICLELOCKSTATE_LOCKED: True, + VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED: False, +} + +# UNKNOWN and FAILED_UNLATCH are unmapped for the same reason. +_CLOSURE_STATES: Mapping[int, bool] = { + ClosureState_E.CLOSURESTATE_CLOSED: False, + ClosureState_E.CLOSURESTATE_OPEN: True, + ClosureState_E.CLOSURESTATE_AJAR: True, + ClosureState_E.CLOSURESTATE_OPENING: True, + ClosureState_E.CLOSURESTATE_CLOSING: True, +} + +_BROADCAST_MAPS: Mapping[FieldPath, Mapping[int, bool]] = { + FieldPath.LOCKED: _LOCK_STATES, + FieldPath.CHARGE_PORT_DOOR_OPEN: _CLOSURE_STATES, + FieldPath.DOOR_STATE_TRUNK_FRONT: _CLOSURE_STATES, +} + +_BROADCAST_PATHS = frozenset(_BROADCAST_MAPS) + + +class BleBroadcastPublisher: + """Publishes VCSEC status broadcasts from an existing BLE session. + + Registration only: it subscribes to the broadcast listeners a + ``VehicleBluetooth`` already fans out, and never connects, reads, or + commands. A dropped BLE session ends the broadcasts and says nothing about + the fields themselves, so it publishes nothing. + """ + + def __init__( + self, + vehicle: VehicleBluetooth[Any], + *, + clock: Clock = time.monotonic, + ) -> None: + self._vehicle = vehicle + self._clock = clock + self._sink: ObservationSink | None = None + self._subscriptions: dict[FieldPath, Unsubscribe] = {} + + @property + def paths(self) -> frozenset[FieldPath]: + return _BROADCAST_PATHS + + def attach(self, sink: ObservationSink) -> Unsubscribe: + self._sink = sink + + def detach() -> None: + self.release(frozenset(self._subscriptions)) + self._sink = None + + return detach + + def request(self, paths: frozenset[FieldPath]) -> None: + for path in paths: + if path in self._subscriptions or path not in _BROADCAST_MAPS: + continue + self._subscriptions[path] = _LISTENER_FOR[path]( + self._vehicle, self._observer(path) + ) + + def release(self, paths: frozenset[FieldPath]) -> None: + for path in paths: + unsubscribe = self._subscriptions.pop(path, None) + if unsubscribe is not None: + unsubscribe() + + def _observer(self, path: FieldPath) -> Callable[[int], None]: + states = _BROADCAST_MAPS[path] + + def on_broadcast(raw: int) -> None: + sink = self._sink + if sink is None or raw not in states: + return + sink.publish( + Observation(path=path, value=states[raw], observed_at=self._clock()) + ) + + return on_broadcast + + +_LISTENER_FOR: Mapping[ + FieldPath, Callable[["VehicleBluetooth[Any]", Callable[[int], None]], Unsubscribe] +] = { + FieldPath.LOCKED: lambda vehicle, cb: vehicle.listen_vehicle_lock_state(cb), + FieldPath.CHARGE_PORT_DOOR_OPEN: lambda vehicle, cb: vehicle.listen_charge_port(cb), + FieldPath.DOOR_STATE_TRUNK_FRONT: lambda vehicle, cb: vehicle.listen_front_trunk( + cb + ), +} + + +# -- Supplied vehicle_data result publisher --------------------------------- + +_ABSENT = object() + + +def _leaf(section: object, key: str) -> object: + """A present leaf, or ``_ABSENT``: a missing key is not a null reading.""" + if isinstance(section, Mapping) and key in section: + return section[key] # pyright: ignore[reportUnknownVariableType] + return _ABSENT + + +def _int_code(value: object) -> int | None: + """A JSON integer code, rejecting ``bool`` so ``False`` cannot pass as 0.""" + if isinstance(value, bool) or not isinstance(value, int): + return None + return value + + +class VehicleDataResultPublisher: + """Translates a caller-supplied ``vehicle_data`` result into observations. + + The result is an argument. This class holds no client, session, endpoint, + or callable able to obtain one, so nothing here can originate a request or + cause a refresh. + """ + + def __init__(self, *, clock: Clock = time.monotonic) -> None: + self._clock = clock + self._sink: ObservationSink | None = None + + @property + def paths(self) -> frozenset[FieldPath]: + return frozenset(FieldPath) + + def attach(self, sink: ObservationSink) -> Unsubscribe: + self._sink = sink + + def detach() -> None: + self._sink = None + + return detach + + def request(self, paths: frozenset[FieldPath]) -> None: + """Passive source: activation subscribes to nothing.""" + + def release(self, paths: frozenset[FieldPath]) -> None: + """Passive source: activation subscribes to nothing.""" + + def publish_result( + self, result: Mapping[str, Any], *, observed_at: float | None = None + ) -> tuple[Observation, ...]: + """Translate a supplied result and feed the audited leaves to the sink.""" + at = self._clock() if observed_at is None else observed_at + observations = tuple(self._translate(result, at)) + sink = self._sink + if sink is not None: + for observation in observations: + sink.publish(observation) + return observations + + def _translate( + self, result: Mapping[str, Any], observed_at: float + ) -> Iterator[Observation]: + payload: Mapping[str, Any] = result + response = _leaf(payload, "response") + if isinstance(response, Mapping): + payload = response # pyright: ignore[reportUnknownVariableType] + + vehicle_state = _leaf(payload, "vehicle_state") + charge_state = _leaf(payload, "charge_state") + + # An explicit null is the vehicle reporting the field unavailable; an + # absent or unrecognised leaf is no reading at all. + locked = _leaf(vehicle_state, "locked") + if locked is None or isinstance(locked, bool): + yield Observation(FieldPath.LOCKED, locked, observed_at) + + charge_port = _leaf(charge_state, "charge_port_door_open") + if charge_port is None or isinstance(charge_port, bool): + yield Observation(FieldPath.CHARGE_PORT_DOOR_OPEN, charge_port, observed_at) + + front_trunk = _leaf(vehicle_state, "ft") + if front_trunk is None: + yield Observation(FieldPath.DOOR_STATE_TRUNK_FRONT, None, observed_at) + # ``ft`` is an ajar/open code, and only 0 and 1 have a documented + # boolean meaning. + elif (code := _int_code(front_trunk)) in (0, 1): + yield Observation(FieldPath.DOOR_STATE_TRUNK_FRONT, code == 1, observed_at) diff --git a/tesla_fleet_api/router/__init__.py b/tesla_fleet_api/router/__init__.py index 4de5f9f..4913eba 100644 --- a/tesla_fleet_api/router/__init__.py +++ b/tesla_fleet_api/router/__init__.py @@ -1,4 +1,4 @@ -"""Routing wrappers that chain backends with per-command failover.""" +"""Routing wrappers with per-command failover across backends.""" from tesla_fleet_api.router.base import HealthCheck, Router from tesla_fleet_api.router.vehicle import VehicleRouter diff --git a/tests/test_funnel.py b/tests/test_funnel.py new file mode 100644 index 0000000..81f3fb4 --- /dev/null +++ b/tests/test_funnel.py @@ -0,0 +1,554 @@ +"""Unit tests for the ObservationFunnel core: fan-in, dedup, activation, demand. + +Uses plain fake publishers so no BLE hardware, network access, or event loop is +involved; the funnel is synchronous by construction. +""" + +from __future__ import annotations + +import ast +from pathlib import Path +from typing import Any +from unittest import TestCase + +from tesla_fleet_api.funnel import ( + FieldPath, + Observation, + ObservationFunnel, + ObservationSink, + Unsubscribe, + Value, +) + +ALL_PATHS = frozenset(FieldPath) +LOCKED = FieldPath.LOCKED +CHARGE_PORT = FieldPath.CHARGE_PORT_DOOR_OPEN +TRUNK = FieldPath.DOOR_STATE_TRUNK_FRONT + + +class _FakePublisher: + """A publisher a test drives directly, recording activation accounting.""" + + def __init__(self, paths: frozenset[FieldPath] = ALL_PATHS) -> None: + self.requested: list[frozenset[FieldPath]] = [] + self.released: list[frozenset[FieldPath]] = [] + self.detached = 0 + self._paths = paths + self._sink: ObservationSink | None = None + + @property + def paths(self) -> frozenset[FieldPath]: + return self._paths + + def attach(self, sink: ObservationSink) -> Unsubscribe: + self._sink = sink + + def detach() -> None: + self.detached += 1 + self._sink = None + + return detach + + def request(self, paths: frozenset[FieldPath]) -> None: + self.requested.append(paths) + + def release(self, paths: frozenset[FieldPath]) -> None: + self.released.append(paths) + + # -- test driver ------------------------------------------------------- + + def emit(self, path: FieldPath, value: Value, observed_at: float) -> None: + assert self._sink is not None + self._sink.publish(Observation(path=path, value=value, observed_at=observed_at)) + + +class TestFanIn(TestCase): + """Many publishers, one listener: nothing here chooses between sources.""" + + def test_both_sources_reach_the_same_listener_with_no_transient_none(self) -> None: + """The bug this exists to prevent: a field blanking when one source drops. + + Both publishers feed the same listener. The stream carries the field, + then goes away entirely and Bluetooth carries it, then Bluetooth goes + away and the stream carries it again. Nothing the funnel emits across + the whole run is ``None``, because no source ever reported one. + """ + funnel = ObservationFunnel() + stream = _FakePublisher() + ble = _FakePublisher() + detach_stream = funnel.attach(stream) + detach_ble = funnel.attach(ble) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + stream.emit(LOCKED, True, 1.0) + self.assertEqual(seen, [True]) + + # Stream to Bluetooth: the source is gone, not the field. + detach_stream() + self.assertEqual(seen, [True]) + self.assertIs(funnel.value(LOCKED), True) + + ble.emit(LOCKED, False, 2.0) + ble.emit(LOCKED, True, 3.0) + self.assertEqual(seen, [True, False, True]) + + # Bluetooth back to stream, the other direction. + detach_ble() + detach_stream = funnel.attach(stream) + stream.emit(LOCKED, False, 4.0) + + self.assertEqual(seen, [True, False, True, False]) + self.assertNotIn(None, seen) + self.assertIs(funnel.value(LOCKED), False) + + def test_both_sources_stay_live_together_and_interleave(self) -> None: + """Neither source is a standby: both are heard, in arrival order.""" + funnel = ObservationFunnel() + stream = _FakePublisher() + ble = _FakePublisher() + funnel.attach(stream) + funnel.attach(ble) + + seen: list[Value] = [] + funnel.listen(CHARGE_PORT, seen.append) + + stream.emit(CHARGE_PORT, True, 1.0) + ble.emit(CHARGE_PORT, False, 2.0) + stream.emit(CHARGE_PORT, True, 3.0) + + self.assertEqual(seen, [True, False, True]) + + def test_detaching_every_publisher_emits_nothing(self) -> None: + """Transport loss is not a reading, so the funnel asserts nothing.""" + funnel = ObservationFunnel() + stream = _FakePublisher() + ble = _FakePublisher() + detach_stream = funnel.attach(stream) + detach_ble = funnel.attach(ble) + + seen: list[Value] = [] + funnel.listen(TRUNK, seen.append) + stream.emit(TRUNK, True, 1.0) + + detach_stream() + detach_ble() + + self.assertEqual(seen, [True]) + self.assertIs(funnel.value(TRUNK), True) + + def test_a_reported_unavailable_value_is_delivered_as_such(self) -> None: + """Unavailability is a value a source reports, and it is passed through.""" + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + publisher.emit(LOCKED, True, 1.0) + publisher.emit(LOCKED, None, 2.0) + publisher.emit(LOCKED, True, 3.0) + + self.assertEqual(seen, [True, None, True]) + + def test_detaching_twice_is_idempotent(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + detach = funnel.attach(publisher) + funnel.listen(LOCKED, lambda _: None) + + detach() + detach() + + self.assertEqual(publisher.detached, 1) + self.assertEqual(publisher.released, [frozenset({LOCKED})]) + + +class TestSurvivingLogic(TestCase): + """The only arbitration left: drop the out-of-order, drop the unchanged.""" + + def test_an_observation_older_than_the_last_one_is_ignored(self) -> None: + funnel = ObservationFunnel() + stream = _FakePublisher() + ble = _FakePublisher() + funnel.attach(stream) + funnel.attach(ble) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + stream.emit(LOCKED, True, 10.0) + ble.emit(LOCKED, False, 5.0) + + self.assertEqual(seen, [True]) + self.assertIs(funnel.value(LOCKED), True) + + def test_a_stale_frame_cannot_win_by_repeating_after_a_newer_one(self) -> None: + """A skipped repeat still advances the clock it is compared against.""" + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + publisher.emit(LOCKED, True, 10.0) + publisher.emit(LOCKED, True, 20.0) # unchanged, not re-dispatched + publisher.emit(LOCKED, False, 15.0) # older than the 20.0 reading + + self.assertEqual(seen, [True]) + + def test_an_unchanged_value_is_not_re_dispatched(self) -> None: + funnel = ObservationFunnel() + stream = _FakePublisher() + ble = _FakePublisher() + funnel.attach(stream) + funnel.attach(ble) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + stream.emit(LOCKED, True, 1.0) + stream.emit(LOCKED, True, 2.0) + ble.emit(LOCKED, True, 3.0) + ble.emit(LOCKED, False, 4.0) + stream.emit(LOCKED, False, 5.0) + + self.assertEqual(seen, [True, False]) + + def test_a_repeated_unavailable_value_is_also_deduplicated(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + publisher.emit(LOCKED, None, 1.0) + publisher.emit(LOCKED, None, 2.0) + + self.assertEqual(seen, [None]) + + def test_an_equal_timestamp_is_not_treated_as_older(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(LOCKED, seen.append) + + publisher.emit(LOCKED, True, 1.0) + publisher.emit(LOCKED, False, 1.0) + + self.assertEqual(seen, [True, False]) + + def test_fields_are_ordered_independently(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + locked: list[Value] = [] + trunk: list[Value] = [] + funnel.listen(LOCKED, locked.append) + funnel.listen(TRUNK, trunk.append) + + publisher.emit(LOCKED, True, 10.0) + publisher.emit(TRUNK, True, 5.0) + + self.assertEqual(locked, [True]) + self.assertEqual(trunk, [True]) + + +class TestValues(TestCase): + def test_an_unobserved_field_has_no_value(self) -> None: + funnel = ObservationFunnel() + for path in FieldPath: + self.assertIsNone(funnel.value(path)) + + def test_registration_dispatches_nothing(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + funnel.listen(LOCKED, lambda _: None) + publisher.emit(LOCKED, True, 1.0) + + late: list[Value] = [] + funnel.listen(LOCKED, late.append) + + self.assertEqual(late, []) + self.assertIs(funnel.value(LOCKED), True) + + +class TestActivation(TestCase): + def test_first_listener_activates_and_last_releases_exactly_once(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + first = funnel.listen(LOCKED, lambda _: None) + second = funnel.listen(LOCKED, lambda _: None) + self.assertEqual(publisher.requested, [frozenset({LOCKED})]) + + first() + self.assertEqual(publisher.released, []) + second() + self.assertEqual(publisher.released, [frozenset({LOCKED})]) + + # A repeated release must not double-count either. + second() + self.assertEqual(publisher.released, [frozenset({LOCKED})]) + + def test_the_same_callback_twice_needs_two_releases(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + def callback(_: Value) -> None: + return None + + first = funnel.listen(LOCKED, callback) + second = funnel.listen(LOCKED, callback) + first() + self.assertEqual(publisher.released, []) + second() + self.assertEqual(publisher.released, [frozenset({LOCKED})]) + + def test_activation_is_per_path(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + funnel.listen(LOCKED, lambda _: None) + funnel.listen(TRUNK, lambda _: None) + + self.assertEqual(publisher.requested, [frozenset({LOCKED}), frozenset({TRUNK})]) + + def test_a_publisher_is_asked_only_for_paths_it_supplies(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher(paths=frozenset({LOCKED})) + funnel.attach(publisher) + + funnel.listen(TRUNK, lambda _: None) + self.assertEqual(publisher.requested, []) + + funnel.listen(LOCKED, lambda _: None) + self.assertEqual(publisher.requested, [frozenset({LOCKED})]) + + def test_a_publisher_attached_later_is_asked_for_active_paths(self) -> None: + funnel = ObservationFunnel() + funnel.listen(LOCKED, lambda _: None) + + publisher = _FakePublisher() + funnel.attach(publisher) + + self.assertEqual(publisher.requested, [frozenset({LOCKED})]) + + def test_detaching_releases_the_active_paths(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + detach = funnel.attach(publisher) + funnel.listen(LOCKED, lambda _: None) + + detach() + + self.assertEqual(publisher.released, [frozenset({LOCKED})]) + self.assertEqual(publisher.detached, 1) + + +class TestDispatchSafety(TestCase): + def test_a_callback_that_publishes_leaves_no_listener_on_the_old_value( + self, + ) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + first: list[Value] = [] + second: list[Value] = [] + + def republish(value: Value) -> None: + first.append(value) + if value is True: + publisher.emit(LOCKED, False, 2.0) + + funnel.listen(LOCKED, republish) + funnel.listen(LOCKED, second.append) + + publisher.emit(LOCKED, True, 1.0) + + self.assertEqual(first, [True, False]) + # The second listener saw only the value that still stands. + self.assertEqual(second, [False]) + self.assertIs(funnel.value(LOCKED), False) + + def test_a_listener_exception_does_not_stop_later_listeners(self) -> None: + funnel = ObservationFunnel() + publisher = _FakePublisher() + funnel.attach(publisher) + + def explode(_: Value) -> None: + raise RuntimeError("listener failed") + + seen: list[Value] = [] + funnel.listen(LOCKED, explode) + funnel.listen(LOCKED, seen.append) + + with self.assertLogs("tesla_fleet_api", level="ERROR"): + publisher.emit(LOCKED, True, 1.0) + + self.assertEqual(seen, [True]) + + +class TestDemand(TestCase): + def test_demand_reports_initial_state_then_only_aggregate_edges(self) -> None: + funnel = ObservationFunnel() + seen: list[bool] = [] + funnel.listen_demand(ALL_PATHS, seen.append) + self.assertEqual(seen, [False]) + + locked = funnel.listen(LOCKED, lambda _: None) + self.assertEqual(seen, [False, True]) + + # Still demanded: a second path rising is not an aggregate edge. + trunk = funnel.listen(TRUNK, lambda _: None) + self.assertEqual(seen, [False, True]) + + locked() + self.assertEqual(seen, [False, True]) + trunk() + self.assertEqual(seen, [False, True, False]) + + def test_demand_starts_true_when_a_listener_already_exists(self) -> None: + funnel = ObservationFunnel() + funnel.listen(LOCKED, lambda _: None) + + seen: list[bool] = [] + funnel.listen_demand(ALL_PATHS, seen.append) + self.assertEqual(seen, [True]) + + def test_demand_ignores_listeners_outside_its_path_set(self) -> None: + funnel = ObservationFunnel() + seen: list[bool] = [] + funnel.listen_demand(frozenset({LOCKED}), seen.append) + + funnel.listen(TRUNK, lambda _: None) + self.assertEqual(seen, [False]) + + funnel.listen(LOCKED, lambda _: None) + self.assertEqual(seen, [False, True]) + + def test_unsubscribe_removes_only_that_observer(self) -> None: + funnel = ObservationFunnel() + kept: list[bool] = [] + dropped: list[bool] = [] + funnel.listen_demand(ALL_PATHS, kept.append) + release = funnel.listen_demand(ALL_PATHS, dropped.append) + + release() + release() + funnel.listen(LOCKED, lambda _: None) + + self.assertEqual(kept, [False, True]) + self.assertEqual(dropped, [False]) + + def test_releasing_one_of_two_identical_registrations_keeps_the_other( + self, + ) -> None: + funnel = ObservationFunnel() + seen: list[bool] = [] + funnel.listen_demand(ALL_PATHS, seen.append) + release = funnel.listen_demand(ALL_PATHS, seen.append) + + release() + funnel.listen(LOCKED, lambda _: None) + + self.assertEqual(seen, [False, False, True]) + + +class TestFunnelCannotOriginateWork(TestCase): + """The load-bearing invariant: the funnel is structurally unable to poll. + + Asserted against the module's own syntax tree rather than its behaviour, + because a request path that exists but is merely unused would still be a + request path. + """ + + tree: ast.Module + + @classmethod + def setUpClass(cls) -> None: + import tesla_fleet_api.funnel as module + + source = Path(module.__file__).read_text(encoding="utf-8") + cls.tree = ast.parse(source) + + def test_the_module_is_entirely_synchronous(self) -> None: + for node in ast.walk(self.tree): + self.assertNotIsInstance(node, ast.AsyncFunctionDef) + self.assertNotIsInstance(node, ast.Await) + self.assertNotIsInstance(node, ast.AsyncFor) + self.assertNotIsInstance(node, ast.AsyncWith) + + def test_the_module_imports_no_transport_or_scheduling_machinery(self) -> None: + forbidden = { + "asyncio", + "aiohttp", + "aiofiles", + "bleak", + "threading", + "sched", + "requests", + "urllib", + "socket", + } + imported: set[str] = set() + for node in ast.walk(self.tree): + if isinstance(node, ast.Import): + imported.update(alias.name.split(".")[0] for alias in node.names) + elif isinstance(node, ast.ImportFrom) and node.module: + imported.add(node.module.split(".")[0]) + self.assertEqual(imported & forbidden, set()) + + def test_the_module_calls_nothing_that_could_originate_a_request(self) -> None: + forbidden = { + "sleep", + "create_task", + "ensure_future", + "run_coroutine_threadsafe", + "call_later", + "call_soon", + "vehicle_data", + "charge_state", + "vehicle_state", + "connect", + "connect_if_needed", + "wake_up", + "_send", + "_request", + "_getVehicleSecurity", + "_getInfotainment", + "Thread", + "Timer", + } + called: set[str] = set() + for node in ast.walk(self.tree): + if not isinstance(node, ast.Call): + continue + func = node.func + if isinstance(func, ast.Name): + called.add(func.id) + elif isinstance(func, ast.Attribute): + called.add(func.attr) + self.assertEqual(called & forbidden, set()) + + def test_the_funnel_exposes_no_awaitable_member(self) -> None: + import inspect + + funnel: Any = ObservationFunnel() + for name, member in inspect.getmembers(funnel): + self.assertFalse( + inspect.iscoroutinefunction(member), + msg=f"{name} is a coroutine function", + ) diff --git a/tests/test_funnel_bluetooth.py b/tests/test_funnel_bluetooth.py new file mode 100644 index 0000000..c4f0936 --- /dev/null +++ b/tests/test_funnel_bluetooth.py @@ -0,0 +1,332 @@ +"""Tests for BleBroadcastPublisher over a real VehicleBluetooth's listener seams. + +Broadcasts are injected through ``vehicle._on_message``, the same routing path +the vehicle uses in production, so the publisher is exercised against the real +``BroadcastListeners`` registries rather than a stand-in. No BLE connection, +GATT traffic, or event loop is involved. +""" + +from __future__ import annotations + +from typing import Any +from unittest import TestCase +from unittest.mock import AsyncMock, MagicMock + +from cryptography.hazmat.primitives.asymmetric import ec + +from tesla_fleet_api.funnel import ( + BleBroadcastPublisher, + FieldPath, + Observation, + ObservationFunnel, + Value, + VehicleDataResultPublisher, +) +from tesla_fleet_api.tesla.vehicle.bluetooth import VehicleBluetooth +from tesla_protocol.command.universal_message_pb2 import ( + Destination, + Domain, + RoutableMessage, +) +from tesla_protocol.command.vcsec_pb2 import ( + ClosureState_E, + ClosureStatuses, + FromVCSECMessage, + VehicleLockState_E, + VehicleStatus, +) + +VIN = "5YJXCAE43LF123456" +DOMAIN = Domain.DOMAIN_VEHICLE_SECURITY + + +class _Clock: + def __init__(self, now: float = 0.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + +def _make_vehicle(*, connected: bool = True) -> VehicleBluetooth[Any]: + parent = MagicMock() + parent.private_key = ec.generate_private_key(ec.SECP256R1()) + vehicle = VehicleBluetooth(parent, VIN) + vehicle.connect_if_needed = AsyncMock() # type: ignore[method-assign] + vehicle.connect = AsyncMock() # type: ignore[method-assign] + vehicle.client = MagicMock() + vehicle.client.is_connected = connected + vehicle.client.write_gatt_char = AsyncMock() + return vehicle + + +def _broadcast(status: VehicleStatus) -> RoutableMessage: + """An unsolicited (unaddressed) VCSEC status broadcast.""" + return RoutableMessage( + from_destination=Destination(domain=DOMAIN), + protobuf_message_as_bytes=FromVCSECMessage( + vehicleStatus=status + ).SerializeToString(), + ) + + +def _lock(state: VehicleLockState_E) -> RoutableMessage: + return _broadcast(VehicleStatus(vehicleLockState=state)) + + +def _closures(**kwargs: ClosureState_E) -> RoutableMessage: + return _broadcast(VehicleStatus(closureStatuses=ClosureStatuses(**kwargs))) + + +CLOSURE_PATHS = frozenset( + {FieldPath.CHARGE_PORT_DOOR_OPEN, FieldPath.DOOR_STATE_TRUNK_FRONT} +) + + +class _Sink: + """Collects observations straight off the publisher.""" + + def __init__(self) -> None: + self.observations: list[Observation] = [] + + def publish(self, observation: Observation) -> None: + self.observations.append(observation) + + +def _attached( + vehicle: VehicleBluetooth[Any], *, paths: frozenset[FieldPath] | None = None +) -> tuple[BleBroadcastPublisher, _Sink]: + publisher = BleBroadcastPublisher(vehicle, clock=_Clock()) + sink = _Sink() + publisher.attach(sink) + publisher.request(frozenset(FieldPath) if paths is None else paths) + return publisher, sink + + +class TestBroadcastTranslation(TestCase): + def test_ordinary_lock_states_map_to_booleans(self) -> None: + vehicle = _make_vehicle() + _, sink = _attached(vehicle) + + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)) + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED)) + + self.assertEqual( + [(o.path, o.value) for o in sink.observations], + [(FieldPath.LOCKED, True), (FieldPath.LOCKED, False)], + ) + + def test_unvalidated_lock_states_emit_no_observation(self) -> None: + """An unmapped enum keeps the last confirmed value instead of guessing.""" + vehicle = _make_vehicle() + _, sink = _attached(vehicle) + + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_INTERNAL_LOCKED)) + vehicle._on_message( + _lock(VehicleLockState_E.VEHICLELOCKSTATE_SELECTIVE_UNLOCKED) + ) + + self.assertEqual(sink.observations, []) + + def test_closure_states_map_to_booleans(self) -> None: + vehicle = _make_vehicle() + _, sink = _attached(vehicle, paths=CLOSURE_PATHS) + + for state, expected in ( + (ClosureState_E.CLOSURESTATE_CLOSED, False), + (ClosureState_E.CLOSURESTATE_OPEN, True), + (ClosureState_E.CLOSURESTATE_AJAR, True), + (ClosureState_E.CLOSURESTATE_OPENING, True), + (ClosureState_E.CLOSURESTATE_CLOSING, True), + ): + sink.observations.clear() + vehicle._on_message(_closures(chargePort=state, frontTrunk=state)) + self.assertEqual( + {(o.path, o.value) for o in sink.observations}, + { + (FieldPath.CHARGE_PORT_DOOR_OPEN, expected), + (FieldPath.DOOR_STATE_TRUNK_FRONT, expected), + }, + msg=f"closure state {state}", + ) + + def test_ambiguous_closure_states_emit_no_observation(self) -> None: + vehicle = _make_vehicle() + _, sink = _attached(vehicle, paths=CLOSURE_PATHS) + + for state in ( + ClosureState_E.CLOSURESTATE_UNKNOWN, + ClosureState_E.CLOSURESTATE_FAILED_UNLATCH, + ): + vehicle._on_message(_closures(chargePort=state, frontTrunk=state)) + + self.assertEqual(sink.observations, []) + + def test_a_broadcast_without_closures_emits_no_closure_observation(self) -> None: + """Closures have proto3 presence, so an absent submessage says nothing.""" + vehicle = _make_vehicle() + _, sink = _attached(vehicle, paths=CLOSURE_PATHS) + + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)) + + self.assertEqual(sink.observations, []) + + def test_every_status_broadcast_carries_a_lock_state(self) -> None: + """UNLOCKED is 0 with no presence, so the funnel dedupes the repeats.""" + vehicle = _make_vehicle() + funnel = ObservationFunnel() + publisher = BleBroadcastPublisher(vehicle, clock=_Clock(1.0)) + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + + # A closure-only broadcast still reports vehicleLockState = UNLOCKED. + vehicle._on_message(_closures(frontTrunk=ClosureState_E.CLOSURESTATE_CLOSED)) + vehicle._on_message(_closures(frontTrunk=ClosureState_E.CLOSURESTATE_OPEN)) + + self.assertEqual(seen, [False]) + + +class TestBroadcastActivation(TestCase): + def test_request_and_release_bracket_the_ble_listener(self) -> None: + vehicle = _make_vehicle() + publisher, sink = _attached(vehicle, paths=frozenset({FieldPath.LOCKED})) + + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)) + self.assertEqual(len(sink.observations), 1) + + publisher.release(frozenset({FieldPath.LOCKED})) + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED)) + self.assertEqual(len(sink.observations), 1) + + def test_only_requested_paths_are_subscribed(self) -> None: + vehicle = _make_vehicle() + _, sink = _attached(vehicle, paths=frozenset({FieldPath.LOCKED})) + + vehicle._on_message( + _closures( + chargePort=ClosureState_E.CLOSURESTATE_OPEN, + frontTrunk=ClosureState_E.CLOSURESTATE_OPEN, + ) + ) + + self.assertEqual([o.path for o in sink.observations], [FieldPath.LOCKED]) + + def test_detach_drops_every_subscription(self) -> None: + vehicle = _make_vehicle() + publisher = BleBroadcastPublisher(vehicle, clock=_Clock()) + sink = _Sink() + detach = publisher.attach(sink) + publisher.request(frozenset(FieldPath)) + + detach() + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)) + + self.assertEqual(sink.observations, []) + + def test_the_publisher_never_drives_the_transport(self) -> None: + vehicle = _make_vehicle() + publisher, _ = _attached(vehicle) + publisher.release(frozenset(FieldPath)) + + vehicle.connect.assert_not_awaited() # type: ignore[attr-defined] + vehicle.connect_if_needed.assert_not_awaited() # type: ignore[attr-defined] + vehicle.client.write_gatt_char.assert_not_awaited() + + def test_a_lost_session_publishes_nothing(self) -> None: + """A dropped link ends the broadcasts; it is not a reading of the field.""" + vehicle = _make_vehicle() + funnel = ObservationFunnel() + detach = funnel.attach(BleBroadcastPublisher(vehicle, clock=_Clock(1.0))) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_LOCKED)) + + vehicle.client.is_connected = False + detach() + + self.assertEqual(seen, [True]) + self.assertIs(funnel.value(FieldPath.LOCKED), True) + + +class TestRegressionWalkthrough(TestCase): + """The reported failure: lock, charge port and front trunk go unavailable. + + Firmware routed the three fields to Bluetooth broadcasts. With Bluetooth + unreachable they had no other source, so the entities blanked. Here the + same funnel is fed by both a supplied ``vehicle_data`` result and BLE + broadcasts, and each field keeps a value whichever source is producing. + """ + + def _compose( + self, + ) -> tuple[ + ObservationFunnel, + VehicleBluetooth[Any], + VehicleDataResultPublisher, + dict[FieldPath, list[Value]], + ]: + vehicle = _make_vehicle() + funnel = ObservationFunnel() + funnel.attach(BleBroadcastPublisher(vehicle, clock=_Clock(100.0))) + result_publisher = VehicleDataResultPublisher(clock=_Clock(0.0)) + funnel.attach(result_publisher) + + seen: dict[FieldPath, list[Value]] = {path: [] for path in FieldPath} + for path in FieldPath: + funnel.listen(path, seen[path].append) + return funnel, vehicle, result_publisher, seen + + def test_unreachable_bluetooth_does_not_blank_the_three_fields(self) -> None: + funnel, _, result_publisher, seen = self._compose() + + # Bluetooth never connects, so it broadcasts nothing at all. + result_publisher.publish_result( + { + "response": { + "vehicle_state": {"locked": True, "ft": 0}, + "charge_state": {"charge_port_door_open": False}, + } + }, + observed_at=1.0, + ) + + self.assertEqual( + seen, + { + FieldPath.LOCKED: [True], + FieldPath.CHARGE_PORT_DOOR_OPEN: [False], + FieldPath.DOOR_STATE_TRUNK_FRONT: [False], + }, + ) + for path in FieldPath: + self.assertIsNotNone(funnel.value(path)) + + def test_bluetooth_recovering_updates_the_same_listeners(self) -> None: + funnel, vehicle, result_publisher, seen = self._compose() + result_publisher.publish_result( + { + "response": { + "vehicle_state": {"locked": True, "ft": 0}, + "charge_state": {"charge_port_door_open": False}, + } + }, + observed_at=1.0, + ) + + # Bluetooth comes up and the vehicle is opened up. + vehicle._on_message(_lock(VehicleLockState_E.VEHICLELOCKSTATE_UNLOCKED)) + vehicle._on_message( + _closures( + chargePort=ClosureState_E.CLOSURESTATE_OPEN, + frontTrunk=ClosureState_E.CLOSURESTATE_OPEN, + ) + ) + + self.assertEqual(seen[FieldPath.LOCKED], [True, False]) + self.assertEqual(seen[FieldPath.CHARGE_PORT_DOOR_OPEN], [False, True]) + self.assertEqual(seen[FieldPath.DOOR_STATE_TRUNK_FRONT], [False, True]) + for values in seen.values(): + self.assertNotIn(None, values) diff --git a/tests/test_funnel_vehicle_data.py b/tests/test_funnel_vehicle_data.py new file mode 100644 index 0000000..000c4aa --- /dev/null +++ b/tests/test_funnel_vehicle_data.py @@ -0,0 +1,257 @@ +"""Tests for VehicleDataResultPublisher: audited leaves, and no way to fetch. + +Every result here is a literal dictionary written in the test. The publisher is +given no client, session, or callable, so a value it produces can only have come +from that literal - which is what makes the "cannot request" claim checkable +rather than asserted. +""" + +from __future__ import annotations + +import inspect +from typing import Any +from unittest import TestCase + +from tesla_fleet_api.funnel import ( + FieldPath, + ObservationFunnel, + Value, + VehicleDataResultPublisher, +) + +# A trimmed but structurally real vehicle_data response. +RESULT: dict[str, Any] = { + "response": { + "id": 1234567890, + "vin": "5YJXCAE43LF123456", + "state": "online", + "charge_state": { + "battery_level": 72, + "charge_port_door_open": True, + "charge_port_latch": "Engaged", + }, + "climate_state": {"inside_temp": 21.5}, + "vehicle_state": { + "locked": True, + "ft": 0, + "rt": 0, + "df": 0, + "car_version": "2025.14.3", + }, + } +} + + +class _Clock: + def __init__(self, now: float = 0.0) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + +def _translate(result: dict[str, Any]) -> dict[FieldPath, Value]: + publisher = VehicleDataResultPublisher(clock=_Clock()) + return {o.path: o.value for o in publisher.publish_result(result)} + + +class TestAuditedLeaves(TestCase): + def test_maps_exactly_the_three_audited_leaves(self) -> None: + self.assertEqual( + _translate(RESULT), + { + FieldPath.LOCKED: True, + FieldPath.CHARGE_PORT_DOOR_OPEN: True, + FieldPath.DOOR_STATE_TRUNK_FRONT: False, + }, + ) + + def test_a_bare_response_body_is_accepted(self) -> None: + self.assertEqual(_translate(RESULT["response"]), _translate(RESULT)) + + def test_absent_leaves_emit_no_observation(self) -> None: + self.assertEqual( + _translate({"response": {"vehicle_state": {"car_version": "2025.14.3"}}}), + {}, + ) + + def test_a_null_leaf_is_an_explicit_unavailable_reading(self) -> None: + """A null is the vehicle reporting the field unavailable, not silence.""" + self.assertEqual( + _translate( + { + "response": { + "vehicle_state": {"locked": None, "ft": None}, + "charge_state": {"charge_port_door_open": None}, + } + } + ), + { + FieldPath.LOCKED: None, + FieldPath.CHARGE_PORT_DOOR_OPEN: None, + FieldPath.DOOR_STATE_TRUNK_FRONT: None, + }, + ) + + def test_a_non_boolean_locked_emits_no_observation(self) -> None: + self.assertEqual( + _translate({"response": {"vehicle_state": {"locked": "true"}}}), {} + ) + + def test_front_trunk_maps_only_the_documented_codes(self) -> None: + for code, expected in ((0, False), (1, True)): + self.assertEqual( + _translate({"response": {"vehicle_state": {"ft": code}}}), + {FieldPath.DOOR_STATE_TRUNK_FRONT: expected}, + ) + for code in (2, 3, -1): + self.assertEqual( + _translate({"response": {"vehicle_state": {"ft": code}}}), + {}, + msg=f"ft code {code}", + ) + + def test_a_boolean_front_trunk_is_not_read_as_a_code(self) -> None: + self.assertEqual(_translate({"response": {"vehicle_state": {"ft": False}}}), {}) + + def test_unaudited_leaves_are_never_routed(self) -> None: + """No reflective flattening: a sibling closure leaf produces nothing.""" + self.assertEqual( + _translate({"response": {"vehicle_state": {"rt": 1, "df": 1}}}), {} + ) + + def test_a_malformed_result_is_ignored_rather_than_guessed(self) -> None: + for result in ( + {}, + {"response": None}, + {"response": {"vehicle_state": None}}, + {"response": {"vehicle_state": "locked"}}, + ): + self.assertEqual(_translate(result), {}, msg=f"{result}") + + +class TestSuppliedResultFunnelling(TestCase): + def test_a_supplied_result_reaches_listeners(self) -> None: + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=_Clock()) + funnel.attach(publisher) + + seen: dict[FieldPath, list[Value]] = {path: [] for path in FieldPath} + for path in FieldPath: + funnel.listen(path, seen[path].append) + + publisher.publish_result(RESULT, observed_at=1.0) + + self.assertEqual( + seen, + { + FieldPath.LOCKED: [True], + FieldPath.CHARGE_PORT_DOOR_OPEN: [True], + FieldPath.DOOR_STATE_TRUNK_FRONT: [False], + }, + ) + + def test_a_repeated_result_is_not_re_dispatched(self) -> None: + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=_Clock()) + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + + publisher.publish_result(RESULT, observed_at=1.0) + publisher.publish_result(RESULT, observed_at=2.0) + + self.assertEqual(seen, [True]) + + def test_a_supplied_result_never_expires_by_itself(self) -> None: + """Age is not a reading either: nothing here blanks a stale value.""" + clock = _Clock() + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=clock) + funnel.attach(publisher) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + publisher.publish_result(RESULT) + + clock.now = 100_000.0 + self.assertEqual(seen, [True]) + self.assertIs(funnel.value(FieldPath.LOCKED), True) + + def test_activation_subscribes_a_passive_source_to_nothing(self) -> None: + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=_Clock()) + funnel.attach(publisher) + before = dict(vars(publisher)) + + release = funnel.listen(FieldPath.LOCKED, lambda _: None) + release() + + self.assertEqual(vars(publisher), before) + + def test_publishing_before_attach_translates_but_reaches_no_listener(self) -> None: + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=_Clock()) + + seen: list[Value] = [] + funnel.listen(FieldPath.LOCKED, seen.append) + observations = publisher.publish_result(RESULT, observed_at=1.0) + + self.assertEqual(len(observations), 3) + self.assertEqual(seen, []) + + +class TestPublisherCannotRequestData(TestCase): + """The publisher's only data source is the dictionary handed to it.""" + + def test_it_exposes_no_coroutine_and_no_awaitable_member(self) -> None: + publisher = VehicleDataResultPublisher(clock=_Clock()) + for name, member in inspect.getmembers(publisher): + self.assertFalse( + inspect.iscoroutinefunction(member), + msg=f"{name} is a coroutine function", + ) + self.assertFalse(inspect.isawaitable(member), msg=f"{name} is awaitable") + + def test_it_holds_no_client_session_or_fetch_callable(self) -> None: + publisher = VehicleDataResultPublisher(clock=_Clock()) + self.assertEqual(set(vars(publisher)), {"_clock", "_sink"}) + + # The clock is the only callable it keeps, and it takes no arguments. + self.assertEqual( + list(inspect.signature(publisher._clock).parameters), # type: ignore[attr-defined] + [], + ) + + def test_it_yields_nothing_when_no_result_is_supplied(self) -> None: + """With the fake input withheld there is no other source to fall back on.""" + clock = _Clock() + funnel = ObservationFunnel() + publisher = VehicleDataResultPublisher(clock=clock) + funnel.attach(publisher) + + seen: list[Value] = [] + for path in FieldPath: + funnel.listen(path, seen.append) + + clock.now = 10_000.0 + self.assertEqual(seen, []) + for path in FieldPath: + self.assertIsNone(funnel.value(path)) + + def test_an_object_that_could_fetch_is_never_called(self) -> None: + """A result-shaped mapping whose lookups are counted proves the reads.""" + calls: list[str] = [] + + class _Tripwire(dict[str, Any]): + def __getitem__(self, key: str) -> Any: + calls.append(key) + return super().__getitem__(key) + + def fetch(self) -> None: # pragma: no cover - must never be reached + raise AssertionError("the publisher invoked a fetch callable") + + publisher = VehicleDataResultPublisher(clock=_Clock()) + publisher.publish_result(_Tripwire(RESULT)) + self.assertEqual(calls, ["response"])