Add admin-plane transition SSE stream
Extract the panel SSE framing/teardown into the shared api.sse.transition_event_stream helper and serve it on the admin plane as GET /api/v1/admin/events, guarded by the admin router's ONE require_admin and fed by the same app.state.publisher. The panel route now uses the helper with its auth and exposure unchanged. Task 92f75680-6d28-4060-bbfc-be808d5c6361 Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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"]
|
||||
|
||||
+7
-30
@@ -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: <type>\\ndata: <json>\\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"]
|
||||
|
||||
+50
@@ -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: <type>\\ndata: <json>\\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"},
|
||||
)
|
||||
@@ -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: <type>`` / ``data: <json>`` 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
|
||||
+10
-5
@@ -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 ----------------------------------------------
|
||||
|
||||
Reference in New Issue
Block a user