diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index bbcb3ad..96cbb41 100644 --- a/custom_components/nestquest/__init__.py +++ b/custom_components/nestquest/__init__.py @@ -31,6 +31,7 @@ from .db import NestQuestDatabase, make_database from .materialize import materialize as _materialize_run from .migrations import apply_migrations from .services import async_deregister_services, async_register_services +from .sse import NestQuestEventStream from .store import async_get_db_path from .admin_allowlist import seed_setup_admin from .sweep import run_missed_sweep @@ -242,6 +243,7 @@ class NestQuestRuntimeData: remove_update_listener: Callable[[], Any] remove_time_change_listener: Callable[[], Any] coordinator: Any = None + event_stream: NestQuestEventStream | None = None async def _async_owner_user_ids(hass: HomeAssistant) -> list[str]: @@ -439,6 +441,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: remove_update_listener: Callable[[], Any] | None = None remove_time_change_listener: Callable[[], Any] | None = None coordinator: Any = None + event_stream: NestQuestEventStream | None = None try: # Seed the admin allowlist on first setup so the owner is never # locked out: an empty database takes the persisted admin copy @@ -496,14 +499,37 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # (unavailable entities, never a setup crash). Created AFTER # services so a failure below unwinds them through the except # path's listener handling. + # ONE API client for the entry, shared by the coordinator's + # snapshot polling and the SSE subscription below (the client + # is stateless — one session, one connection pool). + api_client = coordinator_client_from_entry(hass, entry) + # The shared Feature 10 coordinator: one refresh cycle every + # entity reads from (CONF_UPDATE_INTERVAL seconds, default + # five minutes). Since Feature 18 it polls the API service's + # panel snapshot instead of reading the local database; an + # entry with no panel token yet gets the typed-error stub + # (unavailable entities, never a setup crash). Created AFTER + # services so a failure below unwinds them through the except + # path's listener handling. coordinator = NestQuestCoordinator( hass, entry_id=entry.entry_id, - api_client=coordinator_client_from_entry(hass, entry), + api_client=api_client, update_interval_seconds=entry.options.get( CONF_UPDATE_INTERVAL, DEFAULT_UPDATE_INTERVAL ), ) + # The panel event subscription (the SSE task): a background + # task consumes the API service's transition-event stream and + # re-fires the four documented HA bus events. It reuses the + # coordinator's client, so it only starts when that client + # actually offers a stream — an unconfigured-token stub (or a + # test snapshot client without ``stream_events``) leaves it + # unstarted, and the typed-error path already covers + # availability for those entries. + if callable(getattr(api_client, "stream_events", None)): + event_stream = NestQuestEventStream(hass, api_client) + event_stream.start() runtime_data = NestQuestRuntimeData( entry_id=entry.entry_id, settings=settings, @@ -511,6 +537,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: remove_update_listener=remove_update_listener, remove_time_change_listener=remove_time_change_listener, coordinator=coordinator, + event_stream=event_stream, ) # The runtime record is registered BEFORE the first refresh: # the refresh path (and anything it calls) resolves the entry's @@ -543,6 +570,14 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: # unwind EVERY remover we acquired before the failure, keeping the # original error as the one raised. Removal is best-effort — a # failing remover must not mask the setup error that triggered it. + # The SSE subscription (started just before the runtime record) + # is stopped first: like the coordinator it must never outlive + # the database being closed below. + if event_stream is not None: + try: + await event_stream.stop() + except BaseException: + pass for remove_listener in ( remove_time_change_listener, remove_update_listener, @@ -611,6 +646,13 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: return False unload_error: BaseException | None = None if runtime_data is not None: + # The SSE subscription is stopped FIRST: its task must be gone + # (no frame can fire) before anything else is torn down. + event_stream = getattr(runtime_data, "event_stream", None) + if event_stream is not None: + stop_stream = getattr(event_stream, "stop", None) + if stop_stream is not None: + await stop_stream() remove_update_listener = getattr(runtime_data, "remove_update_listener", None) remove_time_change_listener = getattr( runtime_data, "remove_time_change_listener", None diff --git a/custom_components/nestquest/api_client.py b/custom_components/nestquest/api_client.py index 1a7ed05..7dbf157 100644 --- a/custom_components/nestquest/api_client.py +++ b/custom_components/nestquest/api_client.py @@ -30,11 +30,22 @@ swallowed. There are NO retries and the client logs nothing: refresh cadence is the coordinator's concern (the next task), and the panel token is a secret that must never reach a log line or an error message — the error texts name the method, path, and status only. + +The panel plane's SSE event stream (``GET /api/v1/panel/events``) is +exposed by :meth:`NestQuestApiClient.stream_events` — an async +generator yielding ``(event_type, payload)`` per Server-Sent Events +frame (``event: `` / ``data: `` lines separated by a blank +line). Opening the stream gets the same one-timeout, typed-error +contract as every other call; reading it has NO body-wide timeout (a +healthy stream is long-lived — the subscription manager owns +reconnection and backoff). """ from __future__ import annotations import asyncio +import inspect import json +from collections.abc import AsyncIterator from typing import TYPE_CHECKING, Any from urllib.parse import urlparse @@ -51,6 +62,7 @@ if TYPE_CHECKING: __all__ = [ "COMPLETE_INSTANCE_PATH", "DEFAULT_TIMEOUT_SECONDS", + "EVENTS_PATH", "NestQuestApiError", "NestQuestApiClient", "SNAPSHOT_PATH", @@ -62,6 +74,10 @@ __all__ = [ #: The panel plane's snapshot route (Feature 16). SNAPSHOT_PATH = "/api/v1/panel/snapshot" +#: The panel plane's SSE event-stream route (Feature 16): each +#: transition arrives as an ``event: `` / ``data: `` frame. +EVENTS_PATH = "/api/v1/panel/events" + #: The panel plane's completion route template (Feature 16); the #: ``{instance_id}`` placeholder is filled per call. COMPLETE_INSTANCE_PATH = "/api/v1/panel/instances/{instance_id}/complete" @@ -230,6 +246,121 @@ class NestQuestApiClient: "POST", path, json_body={"actor_child_id": actor_child_id} ) + async def stream_events( + self, + ) -> AsyncIterator[tuple[str, dict[str, Any]]]: + """Yield ``(event_type, payload)`` for every panel SSE frame. + + Opens ``GET /api/v1/panel/events`` with the Bearer credential + and parses the Server-Sent Events framing the API service + emits: ``event: `` and ``data: `` lines closed by a + blank line (one frame per transition event; a frame's ``data`` + may span several ``data:`` lines, joined with newlines — the + standard SSE rule). Line splits across chunk boundaries are + buffered, so frames are parsed correctly whatever chunking the + transport produces. + + Opening the stream carries the SAME one-timeout, typed-error + contract as every other call: a non-2xx response raises + :class:`NestQuestApiError` with its status and a transport + failure (connection refused, DNS, the open timing out) raises + one with ``status=None`` and the original error chained. The + reading loop itself has NO timeout — a healthy stream is + long-lived — but any failure raised from the transport + mid-stream raises typed as well, so the consumer can reconnect. + A frame whose ``data`` is not valid JSON is a protocol failure + and raises typed, carrying the response status it arrived + with. Cancellation is never converted: closing the generator + (``aclose``, i.e. the consumer's task being cancelled) stops + cleanly and closes the underlying response in every path. + """ + url = f"{self._base_url}{EVENTS_PATH}" + headers = {"Authorization": f"Bearer {self._panel_token}"} + # Opening the response gets ONE request timeout, exactly like + # ``_request_json``; the read loop below runs WITHOUT one — a + # body-wide timeout would kill a healthy long-lived stream. + # The response context manager is entered manually (not via + # ``async with``) because the block must span the yields below; + # the ``finally`` closes it on EVERY exit path. + try: + async with asyncio.timeout(self._timeout_seconds): + stream_cm = self._session.request( + "GET", url, headers=headers, json=None + ) + response = await stream_cm.__aenter__() + except asyncio.CancelledError: + # Cancellation is never converted (see ``_request_json``). + raise + except NestQuestApiError: + raise + except Exception as err: + raise NestQuestApiError( + f"NestQuest API GET {EVENTS_PATH} transport failure: {err}", + status=None, + ) from err + + status = response.status + if not 200 <= status < 300: + await self._close_stream(stream_cm) + raise NestQuestApiError( + f"NestQuest API GET {EVENTS_PATH} failed with HTTP {status}", + status=status, + ) + + try: + event_type: str | None = None + data_lines: list[str] = [] + buffer = "" + async for chunk in response: + if isinstance(chunk, (bytes, bytearray)): + chunk = chunk.decode("utf-8") + buffer += chunk + while "\n" in buffer: + line, buffer = buffer.split("\n", 1) + if line.startswith("event:"): + event_type = line[len("event:"):].strip() + elif line.startswith("data:"): + data_lines.append(line[len("data:"):].strip()) + elif not line and event_type is not None: + # A blank line ends the frame. A trailing + # incomplete frame (stream ends without its + # closing blank line) is discarded, per the + # SSE spec's treatment of incomplete events. + try: + payload = json.loads("\n".join(data_lines)) + except ValueError as err: + raise NestQuestApiError( + f"NestQuest API GET {EVENTS_PATH} " + f"delivered a malformed SSE frame: {err}", + status=status, + ) from err + yield event_type, payload + event_type = None + data_lines = [] + except asyncio.CancelledError: + raise + except NestQuestApiError: + raise + except Exception as err: + raise NestQuestApiError( + f"NestQuest API GET {EVENTS_PATH} transport failure: {err}", + status=None, + ) from err + finally: + await self._close_stream(stream_cm) + + @staticmethod + async def _close_stream(stream_cm: Any) -> None: + """Close an opened streaming response context manager. + + Works for aiohttp's response context manager (whose + ``__aexit__`` is a coroutine) and for test stubs that return + either a coroutine or a plain value. + """ + result = stream_cm.__aexit__(None, None, None) + if inspect.isawaitable(result): + await result + async def _request_json( self, method: str, diff --git a/custom_components/nestquest/sse.py b/custom_components/nestquest/sse.py new file mode 100644 index 0000000..8d1ad87 --- /dev/null +++ b/custom_components/nestquest/sse.py @@ -0,0 +1,172 @@ +"""HA-bus subscription over the API service's panel SSE stream. + +The NestQuest API service (Feature 16) publishes every quest +transition on ``GET /api/v1/panel/events`` as Server-Sent Events +frames. This module is the integration-side subscriber: ONE +background task consumes :meth:`~.api_client.NestQuestApiClient +.stream_events` and re-fires each frame on ``hass.bus`` with the SAME +event type and payload fields the local transition path +(:mod:`~.events`) emits — ``nestquest_quest_completed``, +``nestquest_quest_uncompleted``, ``nestquest_quest_missed`` and +``nestquest_child_day_complete`` — so the four blueprints fire +identically whether a completion happens through the integration's own +service or through the API service's panel route. + +Failure handling: a failed stream (API down, network blip, non-2xx, a +dropped connection) is retried FOREVER with a bounded exponential +backoff — :data:`RECONNECT_MIN_SECONDS` after the first failure, +doubling up to :data:`RECONNECT_MAX_SECONDS` — so a long API outage +neither pins the event loop nor spams the log, and a frame received on +a healthy stream resets the backoff. Nothing escapes the task: every +failure is logged with its reason and the loop reconnects; a +:class:`asyncio.CancelledError` (from :meth:`stop` or integration +unload) is the only thing that ends it, and it stops cleanly — no +frame fires after :meth:`stop` returns and the response is closed. +""" +from __future__ import annotations + +import asyncio +from typing import Any + +from .const import LOGGER + +__all__ = [ + "NestQuestEventStream", + "RECONNECT_MAX_SECONDS", + "RECONNECT_MIN_SECONDS", +] + +#: Delay before the FIRST reconnect attempt after a stream failure. +#: One failed frame must not turn into a hot reconnect loop. +RECONNECT_MIN_SECONDS = 5.0 + +#: Ceiling for the exponential reconnect backoff: a long API outage +#: retries at a calm fixed cadence instead of growing without bound. +RECONNECT_MAX_SECONDS = 60.0 + + +class NestQuestEventStream: + """Consume the panel SSE stream in the background, fire the HA bus. + + ONE instance per config entry, built in integration setup with the + SAME API client the coordinator polls (the client is stateless — + one shared session, one connection pool). ``start()`` schedules + the background task; ``stop()`` cancels and awaits it, so unload + never leaks the task and no frame fires after teardown. The task + itself is failure-proof: any exception from the stream (typed + :class:`~.api_client.NestQuestApiError` or anything else) is + logged and retried after the current backoff, never allowed to + escape the task. + """ + + def __init__( + self, + hass: Any, + client: Any, + *, + reconnect_min_seconds: float = RECONNECT_MIN_SECONDS, + reconnect_max_seconds: float = RECONNECT_MAX_SECONDS, + ) -> None: + """Store the hass bus target, the API client, and the backoff. + + The client is anything exposing ``stream_events()`` (the real + :class:`~.api_client.NestQuestApiClient` in production, a stub + in tests). The backoff bounds must be positive and the minimum + must not exceed the maximum — a misconfiguration should fail + loudly at construction, not quietly reconnect at delay 0. + """ + if ( + not isinstance(reconnect_min_seconds, (int, float)) + or isinstance(reconnect_min_seconds, bool) + or reconnect_min_seconds <= 0 + or reconnect_min_seconds > reconnect_max_seconds + ): + raise ValueError( + "reconnect_min_seconds must be a positive number no " + f"greater than reconnect_max_seconds, got " + f"{reconnect_min_seconds!r}" + ) + if ( + not isinstance(reconnect_max_seconds, (int, float)) + or isinstance(reconnect_max_seconds, bool) + or reconnect_max_seconds <= 0 + ): + raise ValueError( + f"reconnect_max_seconds must be a positive number, got " + f"{reconnect_max_seconds!r}" + ) + self._hass = hass + self._client = client + self._min_backoff = float(reconnect_min_seconds) + self._max_backoff = float(reconnect_max_seconds) + self._backoff = float(reconnect_min_seconds) + self._task: asyncio.Task | None = None + + @property + def task(self) -> asyncio.Task | None: + """The background task, or ``None`` when never started/stopped.""" + return self._task + + def start(self) -> None: + """Schedule the subscription task (idempotent while it runs).""" + if self._task is not None and not self._task.done(): + return + self._backoff = self._min_backoff + self._task = asyncio.create_task( + self._run(), name="nestquest-sse-subscription" + ) + + async def stop(self) -> None: + """Cancel and await the subscription task. + + Returns only after the task has actually finished, so unload + can rely on: no further frame is fired after ``stop()``, and + the underlying stream response has been closed. Safe to call + when the task was never started or already stopped. + """ + task, self._task = self._task, None + if task is None: + return + task.cancel() + try: + await task + except asyncio.CancelledError: + # The task's own cancellation surfaces here — expected and + # swallowed; a cancelling the CALLER would re-raise out of + # the surrounding await on its own. + pass + + async def _run(self) -> None: + """Consume stream frames forever, reconnecting on failure.""" + while True: + try: + async for event_type, payload in self._client.stream_events(): + # A frame arrived on a healthy stream: the next + # failure reconnects quickly again. + self._backoff = self._min_backoff + try: + self._hass.bus.async_fire(event_type, dict(payload)) + except Exception as err: + # Firing one frame must never tear down a + # healthy connection: log and keep consuming. + LOGGER.error( + "NestQuest panel event %s could not be fired " + "on the Home Assistant event bus: %s", + event_type, + err, + ) + except asyncio.CancelledError: + # Unload / shutdown: stop cleanly, never converted. + raise + except Exception as err: + delay = self._backoff + self._backoff = min( + self._backoff * 2, self._max_backoff + ) + LOGGER.warning( + "NestQuest panel event stream failed (%s); " + "reconnecting in %.1f s", + err, + delay, + ) + await asyncio.sleep(delay) diff --git a/tests/test_sse_subscription.py b/tests/test_sse_subscription.py new file mode 100644 index 0000000..0a5c26d --- /dev/null +++ b/tests/test_sse_subscription.py @@ -0,0 +1,619 @@ +"""Tests for the panel SSE subscription (HA-bus re-firing task). + +Three layers are exercised, all STUB-driven — no network, no aiohttp, +no real Home Assistant: + +- The API client's ``stream_events`` (transport stub): what goes on + the wire (GET, the events URL, the Bearer header), SSE frame + parsing (``event:``/``data:`` lines, blank-line frame ends, chunk + boundaries, multi-line data), and every failure mode (non-2xx, + connection failure, mid-stream drop, malformed frame) surfacing as + the typed :class:`NestQuestApiError`; cancellation closes the + stream cleanly. +- The :class:`~custom_components.nestquest.sse.NestQuestEventStream` + subscription manager (duck-typed stream client): the four documented + transition events (design/ENTITIES-AND-SERVICES.md §3, exactly as + ``tests/test_api_sse.py`` observes them on the API side) fire on + ``hass.bus`` with the documented payload fields; ``stop()`` cancels + the task so no frame fires afterwards; a stream failure reconnects + with a bounded, reset-on-success backoff and never crashes the + task. +- The integration wiring: setup starts the stream alongside the + runtime record, unload tears it down (no firing afterwards, task + cancelled), and an entry whose client has no stream surface (the + unconfigured-token stub, the test snapshot client) starts none. +""" +from __future__ import annotations + +import asyncio +import json + +import pytest + +from conftest import set_coordinator_client, wire_entry_to_registry + +import custom_components.nestquest as nq_integration +from custom_components.nestquest import ( + async_setup_entry, + async_unload_entry, +) +from custom_components.nestquest import sse as sse_module +from custom_components.nestquest.api_client import ( + NestQuestApiClient, + NestQuestApiError, +) +from custom_components.nestquest.sse import NestQuestEventStream + +BASE_URL = "http://api.test:8000" +TOKEN = "test-panel-token" + +# Captured BEFORE any test patches asyncio.sleep (the reconnect test +# swaps in a recording fake; this reference keeps helpers sleeping for +# real regardless). +_REAL_SLEEP = asyncio.sleep + +#: The four documented transition events with their documented payload +#: fields — the exact shapes the API side streams (test_api_sse.py). +DOCUMENTED_FRAMES = ( + ( + "nestquest_quest_completed", + { + "child_id": 1, + "child_name": "Ada", + "instance_id": 7, + "quest_title": "Pack bag", + "window": "morning", + "due_date": "2026-09-23", + "due_time": "10:00", + "occurred_at": "2026-09-23T12:00:00+00:00", + "was_on_time": False, + }, + ), + ( + "nestquest_quest_uncompleted", + { + "child_id": 1, + "child_name": "Ada", + "instance_id": 7, + "quest_title": "Pack bag", + "window": "morning", + "due_date": "2026-09-23", + "due_time": "10:00", + "occurred_at": "2026-09-23T12:00:00+00:00", + }, + ), + ( + "nestquest_quest_missed", + { + "child_id": 1, + "child_name": "Ada", + "instance_id": 9, + "quest_title": "Stale chore", + "window": "morning", + "due_date": "2026-09-23", + "due_time": "09:00", + "occurred_at": "2026-09-23T23:30:00+00:00", + }, + ), + ( + "nestquest_child_day_complete", + { + "child_id": 1, + "child_name": "Ada", + "quests_due": 1, + "quests_completed": 1, + "occurred_at": "2026-09-23T12:00:00+00:00", + }, + ), +) + +#: The documented payload field sets per event type. +DOCUMENTED_FIELDS = { + "nestquest_quest_completed": { + "child_id", + "child_name", + "instance_id", + "quest_title", + "window", + "due_date", + "due_time", + "occurred_at", + "was_on_time", + }, + "nestquest_quest_uncompleted": { + "child_id", + "child_name", + "instance_id", + "quest_title", + "window", + "due_date", + "due_time", + "occurred_at", + }, + "nestquest_quest_missed": { + "child_id", + "child_name", + "instance_id", + "quest_title", + "window", + "due_date", + "due_time", + "occurred_at", + }, + "nestquest_child_day_complete": { + "child_id", + "child_name", + "quests_due", + "quests_completed", + "occurred_at", + }, +} + +#: A minimal panel snapshot payload for the full-setup wiring tests +#: (the shape ``snapshot_from_api_payload`` consumes; no children). +MINIMAL_SNAPSHOT = {"today_iso": "2026-09-23", "cycle_day": 1, "children": []} + + +# --------------------------------------------------------------------------- +# Stub transports and stream clients (no network anywhere) +# --------------------------------------------------------------------------- + + +class _StreamingResponse: + """A fake aiohttp streaming response: ``status`` + chunk iteration. + + ``chunks`` are the body pieces the stream yields in order; a chunk + may be an exception, raised when the iteration reaches it (a + mid-stream transport drop). ``hold_open`` keeps the stream open + forever after the chunks run out, for cancellation tests. + """ + + def __init__(self, status: int, chunks, *, hold_open: bool = False) -> None: + self.status = status + self._chunks = list(chunks) + self._hold_open = hold_open + self.closed = False + + def __aiter__(self): + return self._iterate() + + async def _iterate(self): + for chunk in self._chunks: + if isinstance(chunk, BaseException): + raise chunk + yield chunk + if self._hold_open: + await asyncio.Event().wait() # cancelled, never ends + + +class _RequestContext: + """Async context manager mirroring ``session.request(...)``'s return. + + Entering performs the request (a scripted transport error raises + THERE, where aiohttp's connector raises on a refused connection); + exiting marks the response closed, so tests can assert the stream + was released. + """ + + def __init__(self, response, error) -> None: + self._response = response + self._error = error + + async def __aenter__(self): + if self._error is not None: + raise self._error + return self._response + + async def __aexit__(self, exc_type, exc, tb) -> bool: + if self._response is not None: + self._response.closed = True + return False + + +class _StreamTransport: + """A no-network stand-in for aiohttp's ``ClientSession``. + + Records every ``request(...)`` call and answers from a scripted + queue of outcomes (a :class:`_StreamingResponse` or an exception + per attempt, consumed in order). + """ + + def __init__(self, *outcomes) -> None: + self.calls: list[dict] = [] + self._outcomes = list(outcomes) + + def request(self, method: str, url: str, **kwargs): + self.calls.append( + { + "method": method, + "url": url, + "headers": kwargs.get("headers"), + "json": kwargs.get("json"), + } + ) + outcome = self._outcomes.pop(0) + if isinstance(outcome, BaseException): + return _RequestContext(None, outcome) + return _RequestContext(outcome, None) + + +def make_stream_client(transport) -> NestQuestApiClient: + """Build a client over the streaming stub with the test credentials.""" + return NestQuestApiClient(BASE_URL, TOKEN, transport) + + +def _frame_text(event_type: str, payload: dict) -> str: + """One documented SSE frame: event line, data line, blank line.""" + return f"event: {event_type}\ndata: {json.dumps(payload)}\n\n" + + +class _ScriptedStreamClient: + """Duck-typed API client: one scripted stream per connection. + + ``script`` holds generator FACTORIES — ``stream_events()`` pops the + next one and returns its async generator, so a test scripts exactly + what each connection attempt yields (or how it fails). + """ + + def __init__(self, script) -> None: + self._script = list(script) + self.calls = 0 + + def stream_events(self): + self.calls += 1 + return self._script.pop(0)() + + +def _frames(*frames): + """A live stream factory: yields ``frames``, then stays open.""" + + def _factory(): + async def _gen(): + for frame in frames: + yield frame + await asyncio.Event().wait() # cancelled, never ends + + return _gen() + + return _factory + + +def _fail(error: BaseException): + """A stream factory that fails before yielding anything.""" + + def _factory(): + async def _gen(): + raise error + yield # noqa: unreachable — makes _gen an async generator + + return _gen() + + return _factory + + +def _drop_after(error: BaseException, *frames): + """A stream factory that yields ``frames`` then drops mid-stream.""" + + def _factory(): + async def _gen(): + for frame in frames: + yield frame + raise error + + return _gen() + + return _factory + + +async def _wait_until(predicate, *, timeout: float = 2.0) -> None: + """Poll ``predicate`` until true or the deadline passes.""" + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while not predicate(): + if loop.time() > deadline: + raise AssertionError("condition not met within the deadline") + await _REAL_SLEEP(0.01) + + +def make_manager( + hass, client, *, min_backoff: float = 0.01, max_backoff: float = 0.05 +) -> NestQuestEventStream: + """A manager with fast (test-sized) backoff bounds.""" + return NestQuestEventStream( + hass, + client, + reconnect_min_seconds=min_backoff, + reconnect_max_seconds=max_backoff, + ) + + +# --------------------------------------------------------------------------- +# The client's stream_events: parsing and typed failures +# --------------------------------------------------------------------------- + + +async def test_stream_events_yields_documented_frames_and_sends_bearer() -> None: + """All four documented frames parse, with the Bearer GET on the wire.""" + body = "".join( + _frame_text(event_type, payload) + for event_type, payload in DOCUMENTED_FRAMES + ) + transport = _StreamTransport(_StreamingResponse(200, [body])) + client = make_stream_client(transport) + + frames = [frame async for frame in client.stream_events()] + + assert frames == list(DOCUMENTED_FRAMES) + call = transport.calls[0] + assert call["method"] == "GET" + assert call["url"] == f"{BASE_URL}/api/v1/panel/events" + assert call["headers"] == {"Authorization": f"Bearer {TOKEN}"} + assert call["json"] is None + + +async def test_stream_events_buffers_lines_across_chunks_and_joins_data() -> None: + """Chunk-split lines parse, and multi-line data joins per SSE rules.""" + first_type, first_payload = DOCUMENTED_FRAMES[0] + # The first frame is cut MID-LINE across two chunks; the second + # frame spreads its JSON over two ``data:`` lines. + chunks = [ + f"event: {first_type}\nda", + f"ta: {json.dumps(first_payload)}\n\nevent: " + f"{DOCUMENTED_FRAMES[1][0]}\n", + f'data: {{"extra":\ndata: true}}\n\n', + ] + transport = _StreamTransport(_StreamingResponse(200, chunks)) + client = make_stream_client(transport) + + frames = [frame async for frame in client.stream_events()] + + assert len(frames) == 2 + assert frames[0] == DOCUMENTED_FRAMES[0] + # The split-out frame's data spread over two ``data:`` lines joins + # with a newline (the SSE standard) and parses as one JSON body. + assert frames[1] == (DOCUMENTED_FRAMES[1][0], {"extra": True}) + + +async def test_stream_events_non_2xx_raises_typed_and_closes() -> None: + """A wrong token (401) is typed with its status; no stream is read.""" + response = _StreamingResponse(401, ["event: x\ndata: {}\n\n"]) + transport = _StreamTransport(response) + client = make_stream_client(transport) + + with pytest.raises(NestQuestApiError) as excinfo: + async for _ in client.stream_events(): + pass + + assert excinfo.value.status == 401 + assert not excinfo.value.is_transport_error + assert TOKEN not in str(excinfo.value) + assert response.closed + + +async def test_stream_events_connection_failure_raises_typed_transport_error() -> None: + """A failure before any response is a typed transport error.""" + transport = _StreamTransport(OSError("connection refused")) + client = make_stream_client(transport) + + with pytest.raises(NestQuestApiError) as excinfo: + async for _ in client.stream_events(): + pass + + assert excinfo.value.status is None + assert excinfo.value.is_transport_error + assert isinstance(excinfo.value.__cause__, OSError) + + +async def test_stream_events_midstream_drop_raises_typed() -> None: + """A transport error mid-stream is typed, so the consumer reconnects.""" + response = _StreamingResponse( + 200, + [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1]), OSError("reset")], + ) + transport = _StreamTransport(response) + client = make_stream_client(transport) + + with pytest.raises(NestQuestApiError) as excinfo: + async for _ in client.stream_events(): + pass + + assert excinfo.value.status is None + assert excinfo.value.is_transport_error + assert isinstance(excinfo.value.__cause__, OSError) + assert response.closed + + +async def test_stream_events_stops_cleanly_on_cancellation() -> None: + """Closing the generator (the consumer task's cancellation) stops + cleanly: no error, the underlying response is closed, and nothing + further is yielded.""" + response = _StreamingResponse( + 200, + [_frame_text(DOCUMENTED_FRAMES[0][0], DOCUMENTED_FRAMES[0][1])], + hold_open=True, + ) + transport = _StreamTransport(response) + client = make_stream_client(transport) + + generator = client.stream_events() + first = await asyncio.wait_for(generator.__anext__(), timeout=2.0) + assert first == DOCUMENTED_FRAMES[0] + + await asyncio.wait_for(generator.aclose(), timeout=2.0) + assert response.closed + + +async def test_stream_events_malformed_frame_raises_typed() -> None: + """A frame whose data is not JSON is a typed protocol failure.""" + response = _StreamingResponse( + 200, ["event: nestquest_quest_completed\ndata: not-json\n\n"] + ) + transport = _StreamTransport(response) + client = make_stream_client(transport) + + with pytest.raises(NestQuestApiError) as excinfo: + async for _ in client.stream_events(): + pass + + assert excinfo.value.status == 200 + assert response.closed + + +# --------------------------------------------------------------------------- +# The subscription manager: firing, teardown, reconnect +# --------------------------------------------------------------------------- + + +async def test_manager_fires_documented_frames_on_ha_bus(hass) -> None: + """Every documented transition fires on hass.bus with its payload.""" + client = _ScriptedStreamClient([_frames(*DOCUMENTED_FRAMES)]) + manager = make_manager(hass, client) + manager.start() + try: + await _wait_until(lambda: len(hass.bus.events) >= 4) + assert hass.bus.events == [tuple(f) for f in DOCUMENTED_FRAMES] + for event_type, payload in DOCUMENTED_FRAMES: + fired = hass.bus.fired(event_type) + assert fired, f"{event_type} never fired" + assert set(fired[-1]) == DOCUMENTED_FIELDS[event_type] + finally: + await manager.stop() + + +async def test_manager_stop_cancels_task_and_blocks_later_frames(hass) -> None: + """stop() cancels and awaits the task: no frame fires afterwards, + even when the (dead) stream is later offered more data.""" + stream_release = asyncio.Event() + first_type, first_payload = DOCUMENTED_FRAMES[0] + later_type, later_payload = DOCUMENTED_FRAMES[1] + + def _factory(): + async def _gen(): + yield (first_type, dict(first_payload)) + await stream_release.wait() + yield (later_type, dict(later_payload)) + + return _gen() + + client = _ScriptedStreamClient([_factory]) + manager = make_manager(hass, client) + manager.start() + await _wait_until(lambda: len(hass.bus.events) == 1) + + task = manager.task + await manager.stop() + assert task.cancelled() + assert manager.task is None + + # Offer the stream more data after teardown: nothing fires. + stream_release.set() + await _REAL_SLEEP(0.05) + assert hass.bus.events == [(first_type, dict(first_payload))] + + +async def test_manager_reconnects_with_bounded_backoff(hass, monkeypatch) -> None: + """A failed stream reconnects with a bounded, reset-on-success + backoff and never crashes the task.""" + delays: list[float] = [] + + async def fake_sleep(delay, *args, **kwargs): + delays.append(delay) + await _REAL_SLEEP(0) + + monkeypatch.setattr(sse_module.asyncio, "sleep", fake_sleep) + + stream_error = ConnectionError("stream reset") + client = _ScriptedStreamClient( + [ + _fail(RuntimeError("api down")), + _fail(RuntimeError("api down")), + _fail(RuntimeError("api down")), + _fail(RuntimeError("api down")), + _drop_after(stream_error, DOCUMENTED_FRAMES[0]), + _frames(*DOCUMENTED_FRAMES), + ] + ) + manager = make_manager(hass, client, min_backoff=0.05, max_backoff=0.2) + manager.start() + try: + await _wait_until(lambda: len(hass.bus.events) >= 5) + # 5s → 10s → 20s (cap) → 20s (cap), then a frame on a healthy + # stream reset it to the 5s minimum for the next failure. + assert delays == [0.05, 0.1, 0.2, 0.2, 0.05] + assert hass.bus.events == [ + tuple(DOCUMENTED_FRAMES[0]), + *[tuple(f) for f in DOCUMENTED_FRAMES], + ] + # The task is still alive (subscribed and idle), not crashed. + assert manager.task is not None and not manager.task.done() + finally: + await manager.stop() + + +def test_manager_rejects_invalid_backoff_bounds(hass) -> None: + """Non-positive or inverted backoff bounds fail at construction.""" + with pytest.raises(ValueError): + NestQuestEventStream(hass, object(), reconnect_min_seconds=0) + with pytest.raises(ValueError): + NestQuestEventStream(hass, object(), reconnect_min_seconds=-1) + with pytest.raises(ValueError): + NestQuestEventStream(hass, object(), reconnect_max_seconds=0) + with pytest.raises(ValueError): + NestQuestEventStream(hass, object(), reconnect_min_seconds=10.0, reconnect_max_seconds=5.0) + + +# --------------------------------------------------------------------------- +# Integration wiring: setup starts the stream, unload stops it +# --------------------------------------------------------------------------- + + +class _SetupClient: + """Client for full-setup wiring tests: scripted snapshot + stream.""" + + def __init__(self, script) -> None: + self._script = list(script) + self.stream_calls = 0 + + async def get_snapshot(self) -> dict: + return dict(MINIMAL_SNAPSHOT) + + def stream_events(self): + self.stream_calls += 1 + return self._script.pop(0)() + + +async def test_setup_starts_stream_and_unload_stops_it(hass, make_entry) -> None: + """Setup starts the subscription alongside the runtime record; unload + tears the task down so no frame fires afterwards.""" + client = _SetupClient([_frames(*DOCUMENTED_FRAMES)]) + set_coordinator_client(make_entry(), client) + entry = wire_entry_to_registry(make_entry(), hass.registry) + assert await nq_integration.async_setup_entry(hass, entry) is True + + runtime = entry.runtime_data + assert runtime.event_stream is not None + await _wait_until(lambda: len(hass.bus.events) >= 4) + assert [event for event, _ in hass.bus.events] == [ + frame[0] for frame in DOCUMENTED_FRAMES + ] + + task = runtime.event_stream.task + assert await nq_integration.async_unload_entry(hass, entry) is True + assert task.cancelled() + assert runtime.event_stream.task is None + + events_after_unload = len(hass.bus.events) + await _REAL_SLEEP(0.05) + assert len(hass.bus.events) == events_after_unload + + +async def test_setup_without_stream_surface_starts_no_subscription( + hass, make_entry +) -> None: + """An entry whose API client has no stream surface (the local-snapshot + stub the harness installs, like the unconfigured-token stub in + production) starts NO subscription — and setup still succeeds.""" + entry = wire_entry_to_registry(make_entry(), hass.registry) + assert await nq_integration.async_setup_entry(hass, entry) is True + assert entry.runtime_data.event_stream is None + assert await nq_integration.async_unload_entry(hass, entry) is True