diff --git a/api/routes_admin.py b/api/routes_admin.py index 420e8c5..2ad7874 100644 --- a/api/routes_admin.py +++ b/api/routes_admin.py @@ -290,6 +290,16 @@ an omitted field is never passed to :func:`nestquest_core.children.edit_child` and stays unchanged — while an explicit ``null`` IS passed and is rejected by the core rather than silently ignored. + +The EVENTS route (``GET /events``) is the admin plane's live +transition stream for the Admin PWA: the SAME four transitions +(quest-completed, quest-uncompleted, quest-missed, child-day-complete) +off the SAME in-process publisher as the panel plane, framed by the +shared :func:`api.sse.transition_event_stream` helper. It inherits the +router's ONE :func:`~api.auth.require_admin` check (the Authorization +header — the PWA reads it with a fetch-based reader because +``EventSource`` cannot set headers), so the panel service token is +refused here exactly as on every other admin route. """ from __future__ import annotations @@ -297,6 +307,7 @@ import datetime from typing import Annotated, Literal, NoReturn from fastapi import APIRouter, Depends, HTTPException, Request, Response +from fastapi.responses import StreamingResponse from pydantic import ( BaseModel, StrictBool, @@ -325,6 +336,7 @@ from api.nestquest_core import ( core_snapshot, core_sweep, ) +from api.sse import transition_event_stream #: All admin-plane routes share this router; the ONE admin dependency #: (service-token refusal + JWT verification + nestquest-admins group @@ -2205,4 +2217,25 @@ async def admin_snapshot(request: Request) -> AdminSnapshotResponse: ) + +@router.get( + "/events", + summary="Stream admin transition events (SSE)", +) +async def admin_events(request: Request) -> StreamingResponse: + """Stream the app's transition events to an admin client + (Server-Sent Events). + + Requires a valid admin JWT (the router's shared + :func:`~api.auth.require_admin` dependency — the ONE check; this + handler performs NO auth of its own), so an unauthenticated or + panel-token request is refused before any stream is established. + + Returns the shared :func:`api.sse.transition_event_stream` response + off the app's single in-process publisher (``app.state.publisher``) + — the same publisher, framing and teardown as the panel stream. + """ + return transition_event_stream(request) + + __all__ = ["router"] diff --git a/api/routes_panel.py b/api/routes_panel.py index 462458c..6405384 100644 --- a/api/routes_panel.py +++ b/api/routes_panel.py @@ -53,9 +53,6 @@ from __future__ import annotations import datetime import importlib -import json - -from collections.abc import AsyncIterator from fastapi import APIRouter, Depends, HTTPException, Request from fastapi.responses import StreamingResponse @@ -70,6 +67,7 @@ from api.nestquest_core import ( core_snapshot, ) from api import transitions as api_transitions +from api.sse import transition_event_stream #: The instance DAO, imported on the app's core copy for the #: child-mismatch guard on the complete route. @@ -328,23 +326,6 @@ async def panel_complete_instance( return {"status": str(result)} -async def _sse_stream(request: Request) -> AsyncIterator[str]: - """Yield the SSE frames for every published transition event. - - One subscriber of the app's single publisher (resolved off - ``app.state.publisher``); each event becomes one - ``event: \\ndata: \\n\\n`` frame. Breaking out of the - publisher's subscription (on client disconnect) tears the - subscription down cleanly — no queue is leaked. - """ - publisher = request.app.state.publisher - async for event_type, payload in publisher.subscribe(): - yield ( - f"event: {event_type}\n" - f"data: {json.dumps(payload)}\n\n" - ) - - @router.get( "/events", summary="Stream panel transition events (SSE)", @@ -357,17 +338,13 @@ async def panel_events(request: Request) -> StreamingResponse: :func:`~api.dependencies.require_panel_token` dependency — the ONE token check; this handler performs NO auth of its own). - Returns a ``text/event-stream`` response whose body is produced by - :func:`_sse_stream` off the app's single in-process publisher - (``app.state.publisher``, installed in the lifespan). The stream - ends when the client disconnects, at which point the subscription - is torn down (its queue removed) with nothing leaked. + Returns the shared :func:`api.sse.transition_event_stream` response + off the app's single in-process publisher (``app.state.publisher``, + installed in the lifespan). The stream ends when the client + disconnects, at which point the subscription is torn down (its + queue removed) with nothing leaked. """ - return StreamingResponse( - _sse_stream(request), - media_type="text/event-stream", - headers={"Cache-Control": "no-cache"}, - ) + return transition_event_stream(request) __all__ = ["router"] diff --git a/api/sse.py b/api/sse.py new file mode 100644 index 0000000..495a6c7 --- /dev/null +++ b/api/sse.py @@ -0,0 +1,50 @@ +"""The ONE Server-Sent Events response helper for the API service. + +Both SSE planes — the panel plane (``/api/v1/panel/events``, panel +service token, LAN-only) and the admin plane (``/api/v1/admin/events``, +Authentik admin JWT) — stream the SAME transitions off the SAME +in-process publisher (``app.state.publisher``, see :mod:`api.events`) +with the SAME framing and teardown, so the stream lives here once and +each plane's route only supplies its own router-level auth. +""" +from __future__ import annotations + +import json +from collections.abc import AsyncIterator + +from fastapi import Request +from fastapi.responses import StreamingResponse + +__all__ = ["transition_event_stream"] + + +async def _sse_stream(request: Request) -> AsyncIterator[str]: + """Yield the SSE frames for every published transition event. + + One subscriber of the app's single publisher (resolved off + ``app.state.publisher``); each event becomes one + ``event: \\ndata: \\n\\n`` frame. Breaking out of the + publisher's subscription (on client disconnect) tears the + subscription down cleanly — no queue is leaked. + """ + publisher = request.app.state.publisher + async for event_type, payload in publisher.subscribe(): + yield ( + f"event: {event_type}\n" + f"data: {json.dumps(payload)}\n\n" + ) + + +def transition_event_stream(request: Request) -> StreamingResponse: + """A ``text/event-stream`` response over the app's transition events. + + Performs NO auth of its own: the calling route's router-level + dependency is the ONE check for its plane. The stream ends when + the client disconnects, at which point the subscription is torn + down (its queue removed) with nothing leaked. + """ + return StreamingResponse( + _sse_stream(request), + media_type="text/event-stream", + headers={"Cache-Control": "no-cache"}, + ) diff --git a/tests/test_api_admin_sse.py b/tests/test_api_admin_sse.py new file mode 100644 index 0000000..6156345 --- /dev/null +++ b/tests/test_api_admin_sse.py @@ -0,0 +1,283 @@ +"""Admin-plane SSE tests: the admin transition stream (task 92f75680). + +These tests exercise ``GET /api/v1/admin/events`` — the admin plane's +Server-Sent Events stream for the Admin PWA — against a REAL server: + +- The stream inherits the admin router's ONE ``require_admin`` + dependency: an absent credential is 401, the panel service token and + a non-admin JWT are 403, all before any stream is established. +- An admin JWT subscribes; each of the four documented transitions + (quest-completed, quest-uncompleted, quest-missed, child-day-complete) + arrives as an ``event: `` / ``data: `` frame carrying its + documented payload fields (design/ENTITIES-AND-SERVICES.md §3). +- The admin and panel streams are fed by the SAME in-process publisher: + one transition arrives, identically, on both. + +Streaming approach: :mod:`tests.test_api_sse`'s live-uvicorn reader +(httpx's ASGITransport buffers SSE bodies), on the admin-enabled +:class:`tests.test_api_admin_uncomplete._AdminServer` whose JWKS cache +is stubbed, so a real admin JWT verifies with NO network access. +""" +from __future__ import annotations + +import datetime +from types import SimpleNamespace +from zoneinfo import ZoneInfo + +import httpx +import pytest + +from api import routes_admin, routes_panel +from tests.admin_jwt_harness import make_token +from tests.test_api_admin_uncomplete import _AdminServer +from tests.test_api_panel import PANEL_TOKEN, _seed_household +from tests.test_api_sse import ( + EVENT_CHILD_DAY_COMPLETE, + EVENT_QUEST_COMPLETED, + EVENT_QUEST_MISSED, + EVENT_QUEST_UNCOMPLETED, + READ_TIMEOUT, + _stream, +) + +EVENTS_PATH = "/api/v1/admin/events" + + +@pytest.fixture +def temp_db_path(tmp_path) -> str: + """A fresh per-test SQLite path under pytest's tmp_path.""" + return str(tmp_path / "nestquest.db") + + +@pytest.fixture +async def admin_live(temp_db_path: str) -> SimpleNamespace: + """A seeded household served by a live admin-enabled uvicorn app. + + Both planes' clocks are pinned to 12:00 UTC (past Ada's 10:00 due, + before Bo's 17:00) BEFORE the server starts — the app is created + inside uvicorn's factory thread. + """ + today = datetime.datetime.now().astimezone().date() + pinned_now = datetime.datetime( + today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC") + ) + seed = await _seed_household(temp_db_path, today, pinned_now) + original_panel = routes_panel._local_now + original_admin = routes_admin._local_now + routes_panel._local_now = lambda: pinned_now + routes_admin._local_now = lambda: pinned_now + try: + with _AdminServer(temp_db_path) as server: + client = httpx.AsyncClient( + base_url=server.url(""), timeout=READ_TIMEOUT + ) + async with client: + yield SimpleNamespace(client=client, seed=seed, server=server) + finally: + routes_panel._local_now = original_panel + routes_admin._local_now = original_admin + + +async def _panel_complete(live: SimpleNamespace, instance_id: int, child_id: int): + """Complete ``instance_id`` through the panel route (a real transition).""" + response = await live.client.post( + f"/api/v1/panel/instances/{instance_id}/complete", + headers={"Authorization": f"Bearer {PANEL_TOKEN}"}, + json={"actor_child_id": child_id}, + ) + assert response.status_code == 200 + + +# --- auth: the ONE require_admin dependency, no stream on refusal ---------- + + +async def test_admin_events_refuses_absent_panel_and_non_admin( + admin_live: SimpleNamespace, +) -> None: + """No credential is 401; the panel token and a non-admin JWT are 403. + + Each refusal is a complete (non-streaming) JSON error response, so + no stream is established for any of them. + """ + client = admin_live.client + + absent = await client.get(EVENTS_PATH) + assert absent.status_code == 401 + assert not absent.headers["content-type"].startswith("text/event-stream") + + panel = await client.get( + EVENTS_PATH, headers={"Authorization": f"Bearer {PANEL_TOKEN}"} + ) + assert panel.status_code == 403 + assert panel.json()["detail"] == ( + "This credential type is not accepted for admin access" + ) + + non_admin = await client.get( + EVENTS_PATH, + headers={ + "Authorization": f"Bearer {make_token(groups=('other-group',))}" + }, + ) + assert non_admin.status_code == 403 + assert non_admin.json()["detail"] == ( + "Admin access requires membership in the nestquest-admins group" + ) + + +async def test_admin_jwt_is_not_accepted_on_panel_events( + admin_live: SimpleNamespace, +) -> None: + """The panel plane is unchanged: an admin JWT does not open it.""" + response = await admin_live.client.get( + "/api/v1/panel/events", + headers={"Authorization": f"Bearer {make_token()}"}, + ) + assert response.status_code == 401 + + +async def test_admin_events_sets_no_cache(admin_live: SimpleNamespace) -> None: + """The admin stream is ``text/event-stream`` with ``no-cache``.""" + async with admin_live.client.stream( + "GET", + EVENTS_PATH, + headers={"Authorization": f"Bearer {make_token()}"}, + ) as response: + assert response.status_code == 200 + assert response.headers["content-type"].startswith("text/event-stream") + assert response.headers["cache-control"] == "no-cache" + + +# --- the four documented transitions on the admin stream ------------------- + + +async def test_admin_stream_carries_completed_and_day_complete( + admin_live: SimpleNamespace, +) -> None: + """A panel completion streams quest-completed then child-day-complete. + + Ada's only quest today is Pack bag, so completing it also clears her + day; both frames arrive on the ADMIN stream with their documented + payload fields. + """ + seed = admin_live.seed + pack_bag = seed.pack_bag_instance + stream = _stream(admin_live.server.url(EVENTS_PATH), make_token()) + async with stream: + await _panel_complete(admin_live, pack_bag.id, seed.ada.id) + + event_type, payload = await stream.next_event() + assert event_type == EVENT_QUEST_COMPLETED == "nestquest_quest_completed" + assert payload["child_id"] == seed.ada.id + assert payload["child_name"] == "Ada" + assert payload["instance_id"] == pack_bag.id + assert payload["quest_title"] == "Pack bag" + assert payload["window"] == "morning" + assert payload["due_date"] == pack_bag.due_date + assert payload["due_time"] == "10:00" + assert payload["occurred_at"] + assert payload["was_on_time"] is False # 10:00 due, 12:00 now + completed_at = payload["occurred_at"] + + event_type, payload = await stream.next_event() + assert event_type == EVENT_CHILD_DAY_COMPLETE + assert event_type == "nestquest_child_day_complete" + assert payload["child_id"] == seed.ada.id + assert payload["child_name"] == "Ada" + assert payload["quests_due"] == 1 + assert payload["quests_completed"] == 1 + assert payload["occurred_at"] == completed_at + + +async def test_admin_stream_carries_uncompleted( + admin_live: SimpleNamespace, +) -> None: + """The admin uncomplete route's reversal streams quest-uncompleted.""" + seed = admin_live.seed + brush = seed.brush_teeth_instance + admin_headers = {"Authorization": f"Bearer {make_token()}"} + stream = _stream(admin_live.server.url(EVENTS_PATH), make_token()) + async with stream: + response = await admin_live.client.post( + f"/api/v1/admin/instances/{brush.id}/uncomplete", + headers=admin_headers, + ) + assert response.status_code == 200 + assert response.json()["appended"] is True + + event_type, payload = await stream.next_event() + assert event_type == EVENT_QUEST_UNCOMPLETED + assert event_type == "nestquest_quest_uncompleted" + assert payload["child_id"] == seed.bo.id + assert payload["child_name"] == "Bo" + assert payload["instance_id"] == brush.id + assert payload["quest_title"] == "Brush teeth" + assert payload["window"] == "morning" + assert payload["due_date"] == brush.due_date + assert payload["due_time"] == "17:00" + assert payload["occurred_at"] + assert "was_on_time" not in payload + + +async def test_admin_stream_carries_missed(admin_live: SimpleNamespace) -> None: + """The missed-sweep publisher's transition streams to the admin client. + + Drives the API-side helper the sweep calls, on the SERVER's loop, + against the app's own database and its ONE publisher. + """ + from api import transitions + + seed = admin_live.seed + stale = seed.stale_instance + app = admin_live.server.app + stream = _stream(admin_live.server.url(EVENTS_PATH), make_token()) + async with stream: + admin_live.server.run_on_loop( + transitions.publish_quest_missed( + app.state.db.database, stale.id, app.state.publisher + ) + ) + event_type, payload = await stream.next_event() + assert event_type == EVENT_QUEST_MISSED == "nestquest_quest_missed" + assert payload["child_id"] == seed.ada.id + assert payload["child_name"] == "Ada" + assert payload["instance_id"] == stale.id + assert payload["quest_title"] == "Stale chore" + assert payload["window"] == "morning" + assert payload["due_date"] == stale.due_date + assert payload["due_time"] == "09:00" + assert payload["occurred_at"] + + +# --- one publisher feeds both planes --------------------------------------- + + +async def test_admin_and_panel_streams_share_one_publisher( + admin_live: SimpleNamespace, +) -> None: + """One transition arrives, identically, on the admin AND panel streams. + + The route publishes once to ``app.state.publisher``; both planes' + subscribers receive the same frames (same ``occurred_at``), which a + second, admin-only publisher could never produce. + """ + seed = admin_live.seed + pack_bag = seed.pack_bag_instance + admin_stream = _stream(admin_live.server.url(EVENTS_PATH), make_token()) + panel_stream = _stream(admin_live.server.url("/api/v1/panel/events")) + async with admin_stream, panel_stream: + await _panel_complete(admin_live, pack_bag.id, seed.ada.id) + + admin_events = [ + await admin_stream.next_event(), + await admin_stream.next_event(), + ] + panel_events = [ + await panel_stream.next_event(), + await panel_stream.next_event(), + ] + assert [event_type for event_type, _ in admin_events] == [ + EVENT_QUEST_COMPLETED, + EVENT_CHILD_DAY_COMPLETE, + ] + assert admin_events == panel_events diff --git a/tests/test_api_sse.py b/tests/test_api_sse.py index aea8034..98a6600 100644 --- a/tests/test_api_sse.py +++ b/tests/test_api_sse.py @@ -158,7 +158,8 @@ class _EventStream: parsed frame with a timeout. """ - def __init__(self) -> None: + def __init__(self, token: str = PANEL_TOKEN) -> None: + self.token = token self.queue: asyncio.Queue = asyncio.Queue() self._task: asyncio.Task | None = None self._client: httpx.AsyncClient | None = None @@ -170,7 +171,7 @@ class _EventStream: self._stream_cm = self._client.stream( "GET", self.url, - headers={"Authorization": f"Bearer {PANEL_TOKEN}"}, + headers={"Authorization": f"Bearer {self.token}"}, ) self._response = await self._stream_cm.__aenter__() assert self._response.status_code == 200 @@ -263,9 +264,13 @@ async def seeded_panel(temp_db_path: str) -> SimpleNamespace: routes_panel._local_now = original -def _stream(url: str) -> _EventStream: - """A subscribed SSE reader for ``url``.""" - return _EventStream()._set_url(url) +def _stream(url: str, token: str = PANEL_TOKEN) -> _EventStream: + """A subscribed SSE reader for ``url``, sending ``token`` as Bearer. + + Defaults to the panel service token; the admin-plane stream tests + pass an admin JWT instead. + """ + return _EventStream(token)._set_url(url) # --- auth and content type ----------------------------------------------