Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Feature 21: Admin Today Snapshot API (#190) (#191)

Browse Source 
Co-authored-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
Co-committed-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
This commit was merged in pull request #191.
This commit is contained in:
dev_agent
2026-09-23 20:01:25 -04:00
committed by approver_agent
parent 22f1a6861d
commit a808f89993
2 changed files with 445 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/routes_admin.py
+117
View File
@@ -116,6 +116,22 @@ owns every rule so the two routes cannot drift:
a ``Content-Disposition`` attachment filename and the documented,
stable header row (Admin spec §5 event-row fields).
The snapshot route (task ef3f4877) is the admin-plane twin of the
panel's ``GET /api/v1/panel/snapshot`` for the admin PWA's Today tab —
the PWA is internet-facing and never holds the panel service token
(D-012), so it reads the same snapshot here:
- ``GET /snapshot`` threads ONE clock read (:func:`_local_now`) into
:func:`nestquest_core.snapshot.build_snapshot` and shapes each
child's instances through
:func:`nestquest_core.snapshot.instance_payload` with
``include_missed=True`` — admin payloads KEEP missed instances with
their ``missed`` state (D-009), where the panel omits them. The
body is the panel's documented shape (``today_iso``, ``cycle_day``,
``children`` with the per-child presence, rollup counts and
``instances``); the builder, the state derivation and the shaping
all live in the core.
The settings routes (task 2b3de7e5) follow the same pattern over
:mod:`nestquest_core.settings_store` — the API's OWN durable settings
store. The API now OWNS settings in the database (a validated JSON
@@ -254,6 +270,7 @@ from api.nestquest_core import (
core_recurrence,
core_settings,
core_settings_store,
core_snapshot,
core_sweep,
)
@@ -683,6 +700,59 @@ class AdminMissedSweepResponse(BaseModel):
fired: int
class AdminInstancePayload(BaseModel):
"""One instance in the admin snapshot payload.
Mirrors :func:`nestquest_core.snapshot.instance_payload` with
``include_missed=True``: ``state`` is ``open`` | ``completed`` |
``missed`` (admin payloads keep missed instances, D-009),
``on_time`` passes the completion event's flag through verbatim
(``None`` unless completed), and ``completed_at`` is the event's
UTC timestamp when completed.
"""
id: int
definition_id: int
child_id: int
title: str
icon: str | None
window: str
due_time: str | None
state: str
overdue: bool
completed_at: str | None
on_time: bool | None
class AdminChildSnapshotPayload(BaseModel):
"""One child's day: presence, rollup counts, and today's instances."""
child_id: int
child_name: str
present: bool
#: ISO date of the child's next present day while they are away;
#: ``None`` when the child is present today.
next_present: str | None
due_today: int
completed_today: int
remaining_today: int
completion_pct: int
instances: list[AdminInstancePayload]
class AdminSnapshotResponse(BaseModel):
"""Today's whole household snapshot for the admin Today tab.
The panel snapshot's shape (``today_iso``, ``cycle_day``,
``children`` in the household's sort order), with missed instances
kept in each child's ``instances``.
"""
today_iso: str
cycle_day: int
children: list[AdminChildSnapshotPayload]
#: 404 detail for a child id the core layer reports as non-existent.
_CHILD_NOT_FOUND_DETAIL = "Child not found"
@@ -1729,4 +1799,51 @@ async def admin_run_missed_sweep(
return AdminMissedSweepResponse(fired=len(events))
@router.get(
"/snapshot",
summary="Today's household snapshot",
response_model=AdminSnapshotResponse,
)
async def admin_snapshot(request: Request) -> AdminSnapshotResponse:
"""Return today's full household snapshot for the admin Today tab.
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). Thin adapter: ONE clock read
(:func:`_local_now`) goes into
:func:`nestquest_core.snapshot.build_snapshot` against the live
database, and each child's instances are shaped with
``instance_payload(include_missed=True)`` — missed instances are
KEPT in the admin payload (D-009) — into
:class:`AdminSnapshotResponse`.
"""
state: DatabaseState = request.app.state.db
snapshot = await core_snapshot.build_snapshot(
state.database, core_settings.NestQuestSettings(), _local_now()
)
return AdminSnapshotResponse(
today_iso=snapshot.today_iso,
cycle_day=snapshot.cycle_day,
children=[
AdminChildSnapshotPayload(
child_id=child.child_id,
child_name=child.child_name,
present=child.present,
next_present=child.next_present,
due_today=child.due_today,
completed_today=child.completed_today,
remaining_today=child.remaining_today,
completion_pct=child.completion_pct,
instances=[
AdminInstancePayload.model_validate(instance)
for instance in core_snapshot.instance_payload(
child.instances, include_missed=True
)
],
)
for child in snapshot.children
],
)
__all__ = ["router"]
tests/test_api_admin_snapshot.py
+328
View File
@@ -0,0 +1,328 @@
"""Admin-plane snapshot route tests (task ef3f4877).
These tests exercise ``GET /api/v1/admin/snapshot`` against the ASGI
app with httpx, reusing the shared local-RSA-key/stubbed-JWKS harness
(:mod:`tests.admin_jwt_harness`) so an admin JWT genuinely
authenticates with NO network access. The household is seeded with
:func:`tests.test_api_panel._seed_household` through the SAME core
copy the app uses (``nestquest_core.*``), and both routes' single
clock read (``_local_now``) is pinned to noon so nothing straddles
midnight.
Proven per the done-condition:
- An admin JWT returns 200 with the documented shape: ``today_iso``,
``cycle_day`` and, per child, the presence fields, the rollup counts
and ``instances`` in the core ``instance_payload`` shape.
- A MISSED instance is KEPT in the admin payload (D-009) and still
OMITTED from the panel payload. The core builder lists only
instances due on ``now.date()``, and ``derive_state`` calls an
instance missed only when its due date is BEFORE that day — so a
real snapshot never carries a missed view. The test therefore
wraps the real builder to add the seeded past-due instance as a
view whose state the core's own ``derive_state`` computes
(``missed``), and reads it through both routes.
- Auth runs through the router's ONE ``require_admin`` dependency: no
credential is 401, the panel service token is refused outright
(403, the credential-type detail), and a non-admin JWT is 403.
- The handler performs exactly ONE clock read per request.
"""
from __future__ import annotations
import dataclasses
import datetime
import importlib
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import pytest
from api import routes_admin, routes_panel
from api.nestquest_core import core_snapshot
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
from tests.test_api_panel import _seed_household
# The completion layer through the SAME core copy the app uses — the
# missed test derives its view's state with the core's own rule.
_completion = importlib.import_module("nestquest_core.completion")
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _panel_headers() -> dict[str, str]:
"""The Authorization header carrying the harness panel token."""
return {"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"}
@pytest.fixture
async def snapshot_household(
temp_db_path: str, monkeypatch
) -> SimpleNamespace:
"""A seeded household served by the admin-plane app, clock pinned.
``today`` comes from one host clock read; the pinned instant is
12:00 UTC that day (past Ada's 10:00 due, before Bo's 17:00). Both
the admin and the panel route's ``_local_now`` return it, and the
admin one counts its calls.
"""
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)
clock_reads: list[datetime.datetime] = []
def _admin_now() -> datetime.datetime:
clock_reads.append(pinned_now)
return pinned_now
monkeypatch.setattr(routes_admin, "_local_now", _admin_now)
monkeypatch.setattr(routes_panel, "_local_now", lambda: pinned_now)
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(
client=client,
seed=seed,
today=today,
pinned_now=pinned_now,
pinned_utc_iso=pinned_now.astimezone(
datetime.timezone.utc
).isoformat(),
clock_reads=clock_reads,
)
# --- authentication: the ONE require_admin dependency ----------------------
async def test_snapshot_without_credential_is_401(
snapshot_household: SimpleNamespace,
) -> None:
"""No Authorization header is the uniform 401."""
response = await snapshot_household.client.get("/api/v1/admin/snapshot")
assert response.status_code == 401
async def test_snapshot_refuses_panel_service_token_outright(
snapshot_household: SimpleNamespace,
) -> None:
"""The panel service token is refused as a credential type (403).
403 with the credential-type detail — not the uniform 401 and not
the non-admin group detail — proves the token was refused outright,
never tried as a JWT.
"""
response = await snapshot_household.client.get(
"/api/v1/admin/snapshot", headers=_panel_headers()
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_snapshot_refuses_non_admin_jwt(
snapshot_household: SimpleNamespace,
) -> None:
"""A valid JWT without the nestquest-admins group is 403."""
non_admin = make_token(groups=("some-other-group",))
response = await snapshot_household.client.get(
"/api/v1/admin/snapshot",
headers={"Authorization": f"Bearer {non_admin}"},
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
# --- the payload -----------------------------------------------------------
async def test_admin_jwt_returns_documented_snapshot(
snapshot_household: SimpleNamespace,
) -> None:
"""An admin JWT returns 200 with the documented snapshot shape.
Every field is asserted as an explicit literal, with ONE clock
read for the request. With no missed view in the real snapshot,
the admin body equals the panel body for the same household.
"""
client = snapshot_household.client
seed = snapshot_household.seed
response = await client.get(
"/api/v1/admin/snapshot", headers=_admin_headers()
)
assert response.status_code == 200
assert len(snapshot_household.clock_reads) == 1
body = response.json()
assert body == {
"today_iso": snapshot_household.today.isoformat(),
# Ada's 2-week cycle anchored yesterday => today is cycle day 2.
"cycle_day": 2,
"children": [
{
"child_id": seed.ada.id,
"child_name": "Ada",
"present": False,
"next_present": seed.returns_date.isoformat(),
"due_today": 1,
"completed_today": 0,
"remaining_today": 1,
"completion_pct": 0,
"instances": [
{
"id": seed.pack_bag_instance.id,
"definition_id": seed.pack_bag_instance.definition_id,
"child_id": seed.ada.id,
"title": "Pack bag",
"icon": None,
"window": "morning",
"due_time": "10:00",
"state": "open",
"overdue": True,
"completed_at": None,
"on_time": None,
}
],
},
{
"child_id": seed.bo.id,
"child_name": "Bo",
"present": True,
"next_present": None,
"due_today": 1,
"completed_today": 1,
"remaining_today": 0,
"completion_pct": 100,
"instances": [
{
"id": seed.brush_teeth_instance.id,
"definition_id": (
seed.brush_teeth_instance.definition_id
),
"child_id": seed.bo.id,
"title": "Brush teeth",
"icon": None,
"window": "morning",
"due_time": "17:00",
"state": "completed",
"overdue": False,
"completed_at": snapshot_household.pinned_utc_iso,
"on_time": True,
}
],
},
],
}
panel = await client.get(
"/api/v1/panel/snapshot", headers=_panel_headers()
)
assert panel.status_code == 200
assert panel.json() == body
async def test_admin_keeps_missed_instance_panel_omits_it(
snapshot_household: SimpleNamespace, monkeypatch
) -> None:
"""A missed view is KEPT by the admin route, OMITTED by the panel's.
The real builder is wrapped to add the seeded past-due Stale chore
instance to Ada's views, its state derived by the core's own
``derive_state`` (no events, due yesterday => ``missed``). Both
routes read the SAME wrapped snapshot, so the only difference is
each route's ``include_missed`` flag.
"""
client = snapshot_household.client
seed = snapshot_household.seed
stale = seed.stale_instance
missed_state = _completion.derive_state(
stale, None, snapshot_household.today
)
assert missed_state == "missed"
missed_view = core_snapshot.QuestInstanceView(
instance_id=stale.id,
definition_id=stale.definition_id,
child_id=stale.child_id,
title="Stale chore",
icon=None,
window=stale.window,
due_date=stale.due_date,
due_time=stale.due_time,
state=missed_state,
overdue=False,
completed_at=None,
was_on_time=None,
)
real_build_snapshot = core_snapshot.build_snapshot
async def _with_missed_view(database, settings, now):
snapshot = await real_build_snapshot(database, settings, now)
return dataclasses.replace(
snapshot,
children=tuple(
dataclasses.replace(
child, instances=(*child.instances, missed_view)
)
if child.child_id == seed.ada.id
else child
for child in snapshot.children
),
)
monkeypatch.setattr(core_snapshot, "build_snapshot", _with_missed_view)
admin = await client.get(
"/api/v1/admin/snapshot", headers=_admin_headers()
)
assert admin.status_code == 200
admin_ada = admin.json()["children"][0]
assert admin_ada["child_id"] == seed.ada.id
assert [instance["id"] for instance in admin_ada["instances"]] == [
seed.pack_bag_instance.id,
stale.id,
]
assert admin_ada["instances"][1] == {
"id": stale.id,
"definition_id": stale.definition_id,
"child_id": seed.ada.id,
"title": "Stale chore",
"icon": None,
"window": "morning",
"due_time": "09:00",
"state": "missed",
"overdue": False,
"completed_at": None,
"on_time": None,
}
panel = await client.get(
"/api/v1/panel/snapshot", headers=_panel_headers()
)
assert panel.status_code == 200
panel_ada = panel.json()["children"][0]
assert [instance["id"] for instance in panel_ada["instances"]] == [
seed.pack_bag_instance.id
], "the panel payload must still omit missed instances"