dev_agent
0e726ac64f
Co-authored-by: dev_agent <dev_agent@cubecraftlabs.com> Co-committed-by: dev_agent <dev_agent@cubecraftlabs.com>
334 lines
12 KiB
Python
334 lines
12 KiB
Python
"""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 host_zone, 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)
|
|
# The API host runs UTC, the zone of the pinned clock.
|
|
monkeypatch.setattr(host_zone, "host_timezone", lambda: "UTC")
|
|
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 — the
|
|
panel body additionally carries the effective ``timezone`` its cards
|
|
format the visible date and clock in (none stored here: the UTC
|
|
host's own zone).
|
|
"""
|
|
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, "timezone": "UTC"}
|
|
|
|
|
|
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"
|