Maniacs_Mansion/NestQuest
11
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
dev
NestQuest/tests/test_api_panel.py
T

717 lines
25 KiB
Python
Raw Permalink Blame History

"""Panel-plane route tests: service token + snapshot (task ae7e90ae).
These tests exercise the NestQuest API's panel plane against the ASGI
app with httpx:
- ``GET /api/v1/panel/snapshot`` requires the panel service token
(D-012): a request with NO ``Authorization`` header, a MALFORMED
header, and a WRONG token all return 401; the correct Bearer token
returns 200.
- The 200 body is the documented snapshot shape: ``today_iso``,
``cycle_day``, and ``children`` in the household's sort order, each
carrying ``child_id``/``child_name``/``present``/``next_present``,
the per-day rollup counts, and its instances in the documented panel
payload shape (ENTITIES-AND-SERVICES.md §1) with missed instances
OMITTED (D-009).
- ``ApiConfig.from_env`` requires ``NESTQUEST_PANEL_TOKEN`` with the
same fail-fast discipline as ``NESTQUEST_DB_PATH``.
The household is seeded through the SAME core copy the app uses — the
``nestquest_core.*`` package registered by ``api/nestquest_core.py`` —
NOT the conftest-stubbed ``custom_components.nestquest.core`` copy (see
the two-copy caveat in api/nestquest_core.py's docstring: never compare
class or exception identity across the two copies). The seed opens and
migrates its own connection on the app's SQLite path and closes BEFORE
the app's lifespan opens the file, so the two connections never
overlap.
The route's single clock read (:func:`api.routes_panel._local_now`) is
pinned with monkeypatch the same way the coordinator's ``_local_now``
is pinned in test_core_snapshot.py, so the seeded rows and the
snapshot's ``today`` can never straddle midnight.
"""
from __future__ import annotations
import datetime
import importlib
from pathlib import Path
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import httpx
import pytest
from api import routes_panel
from api.app import create_app
from api.config import ApiConfig, ConfigError
from api.database import _executor
from api.nestquest_core import (
core_db,
core_migrations,
core_snapshot,
)
#: The panel service token the test apps are configured with.
PANEL_TOKEN = "panel-service-token-test"
# The bundled core submodules, through the SAME copy the app uses
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
_children = importlib.import_module("nestquest_core.children")
_quest_definitions = importlib.import_module(
"nestquest_core.quest_definitions"
)
_materialize = importlib.import_module("nestquest_core.materialize")
_recurrence = importlib.import_module("nestquest_core.recurrence")
_dao_presence = importlib.import_module("nestquest_core.dao_presence")
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
_completion = importlib.import_module("nestquest_core.completion")
async def _events_for_instance(database: object, instance_id: int) -> list:
"""The instance's completion events, on the app's core copy.
Uses the SAME ``nestquest_core.*`` copy the app uses (the two-copy
caveat: never mix copies) against the app's live connection.
"""
dao = _dao_instances.CompletionEventsDao(database)
return await dao.list_by_instance(instance_id)
async def _derived_state(
database: object, instance: object, today: datetime.date
) -> str:
"""The instance's derived state on the app's core copy.
``derive_state(instance, latest, today)`` from the app's core copy;
the latest event is fetched through the same copy's DAO against the
app's live connection.
"""
dao = _dao_instances.CompletionEventsDao(database)
latest = await dao.get_latest_for_instance(instance.id)
return _completion.derive_state(instance, latest, today)
async def _seed_household(
db_path: str,
today: datetime.date,
pinned_now: datetime.datetime,
) -> SimpleNamespace:
"""Seed a two-child household on ``db_path`` and return its records.
Mirrors the test_core_snapshot.py household: an AWAY child (2-week
custody cycle anchored yesterday, empty week 0, an override returns
her today+2) with an open quest due 10:00 — overdue at the pinned
12:00 — and a present child whose 17:00 quest is completed at
12:00 (on time). A third quest's instance is materialized due
YESTERDAY (via ``materialize`` with ``today`` pinned to yesterday,
since the no-past guard refuses past-dated writes): a past-due OPEN
instance, i.e. the "missed" view the panel payload must omit.
Opens, migrates, seeds, and CLOSES its own connection so the app's
lifespan connection never overlaps it.
"""
database = core_db.NestQuestDatabase(_executor)
await database.open(db_path)
try:
await core_migrations.apply_migrations(database)
today_iso = today.isoformat()
yesterday = today - datetime.timedelta(days=1)
yesterday_iso = yesterday.isoformat()
ada = await _children.create_child(database, "Ada", sort_order=0)
bo = await _children.create_child(database, "Bo", sort_order=1)
# Every rule starts YESTERDAY so the yesterday materialization
# below (which pins ``today`` to yesterday) fires them.
await _quest_definitions.create_quest_definition(
database,
"Pack bag",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": yesterday_iso}
),
[ada.id],
[("morning", "10:00")],
)
await _quest_definitions.create_quest_definition(
database,
"Brush teeth",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": yesterday_iso}
),
[bo.id],
[("morning", "17:00")],
)
# Today's instances first, while Ada has no presence pattern
# (she is present every day, so her instance IS generated).
await _materialize.materialize(
database,
today_iso,
(today + datetime.timedelta(days=3)).isoformat(),
today=today,
)
# The stale quest's rule ends YESTERDAY, so create's own
# regeneration writes nothing and its only instance is the
# yesterday one written below.
await _quest_definitions.create_quest_definition(
database,
"Stale chore",
_recurrence.ScheduleRule.from_dict(
{
"rule_type": "daily",
"start_date": yesterday_iso,
"end_date": yesterday_iso,
}
),
[ada.id],
[("morning", "09:00")],
)
# Materialize yesterday with ``today`` pinned to yesterday: the
# no-past guard accepts the rows, and they land past-due (no
# completion events) — the missed view.
await _materialize.materialize(
database, yesterday_iso, yesterday_iso, today=yesterday
)
instances = _dao_instances.QuestInstancesDao(database)
ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
assert len(ada_today) == 1
pack_bag_instance = ada_today[0]
bo_today = await instances.list_by_child_and_date(bo.id, today_iso)
assert len(bo_today) == 1
brush_teeth_instance = bo_today[0]
ada_yesterday = await instances.list_by_child_and_date(
ada.id, yesterday_iso
)
stale_instance = next(
record
for record in ada_yesterday
if record.due_time == "09:00"
)
# Ada's custody pattern (one home pattern), added AFTER materialization (the
# same order test_core_snapshot.py seeds): 2-week cycle
# anchored yesterday, empty week 0 covers today => absent
# today; an override returns her on today+2.
await _dao_presence.PresencePatternsDao(database).create(
ada.id,
name="Custody",
kind="home",
cycle_length_weeks=2,
anchor_date=yesterday_iso,
pattern="|0,1,2,3,4,5,6",
)
returns_date = today + datetime.timedelta(days=2)
await _dao_presence.PresenceOverridesDao(database).create(
ada.id,
returns_date.isoformat(),
returns_date.isoformat(),
is_present=True,
note="visit",
)
# Complete Bo's today instance at the pinned 12:00 (17:00 due
# => on time).
await _completion.complete_instance(
database,
brush_teeth_instance.id,
actor_source="panel",
actor_child_id=bo.id,
now=pinned_now,
today=today,
)
return SimpleNamespace(
ada=ada,
bo=bo,
pack_bag_instance=pack_bag_instance,
brush_teeth_instance=brush_teeth_instance,
stale_instance=stale_instance,
returns_date=returns_date,
)
finally:
await database.close()
class _PanelRunner:
"""Run the app's lifespan around an httpx AsyncClient.
Same pattern as test_api_health.py's ``_AppRunner``: httpx's ASGI
transport does not run the lifespan, so it is driven explicitly and
shut down in ``__aexit__``'s finally block even on failure.
"""
def __init__(
self, db_path: str, panel_token: str = PANEL_TOKEN
) -> None:
self.app = create_app(
ApiConfig(
db_path=db_path,
panel_token=panel_token,
oidc_issuer="https://authentik.example.com/application/o/nestquest/",
oidc_audience="nestquest-api",
oidc_jwks_url=(
"https://authentik.example.com/application/o/nestquest/jwks/"
),
)
)
async def __aenter__(self) -> tuple["_PanelRunner", httpx.AsyncClient]:
self._lifespan = self.app.router.lifespan_context(self.app)
await self._lifespan.__aenter__()
self._client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=self.app),
base_url="http://testserver",
)
await self._client.__aenter__()
return self, self._client
async def __aexit__(self, *exc) -> None:
try:
await self._client.__aexit__(*exc)
finally:
await self._lifespan.__aexit__(*exc)
@pytest.fixture
def temp_db_path(tmp_path: Path) -> str:
"""A fresh per-test SQLite path under pytest's tmp_path."""
return str(tmp_path / "nestquest.db")
@pytest.fixture
async def panel_client(temp_db_path: str) -> httpx.AsyncClient:
"""An app client with an empty (migrated) household database."""
async with _PanelRunner(temp_db_path) as runner_and_client:
_, client = runner_and_client
yield client
@pytest.fixture
async def seeded_panel(temp_db_path: str, monkeypatch) -> SimpleNamespace:
"""A seeded household served by the app, with the clock pinned.
One host clock read anchors ``today``; the pinned instant is
12:00 UTC that day (past Ada's 10:00 due, before Bo's 17:00). The
route's ``_local_now`` is patched to return the pinned instant.
Yields the runner too so tests reach the app's live database (the
SAME ``nestquest_core`` connection) for assertions.
"""
today = datetime.datetime.now().astimezone().date()
time_zone = ZoneInfo("UTC")
pinned_now = datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=time_zone
)
seed = await _seed_household(temp_db_path, today, pinned_now)
monkeypatch.setattr(routes_panel, "_local_now", lambda: pinned_now)
async with _PanelRunner(temp_db_path) as runner_and_client:
runner, client = runner_and_client
yield SimpleNamespace(
client=client,
database=runner.app.state.db.database,
seed=seed,
today=today,
pinned_now=pinned_now,
pinned_utc_iso=pinned_now.astimezone(
datetime.timezone.utc
).isoformat(),
)
# --- ApiConfig.from_env: the panel token --------------------------------
def test_from_env_reads_panel_token() -> None:
"""A set, non-empty NESTQUEST_PANEL_TOKEN is returned verbatim."""
cfg = ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_PANEL_TOKEN": "panel-token",
"NESTQUEST_OIDC_ISSUER": "https://idp.example.com/issuer/",
"NESTQUEST_OIDC_AUDIENCE": "nestquest-api",
"NESTQUEST_OIDC_JWKS_URL": "https://idp.example.com/jwks/",
}
)
assert cfg.panel_token == "panel-token"
def test_from_env_requires_panel_token_when_missing() -> None:
"""A missing panel token raises ConfigError (no silent default)."""
with pytest.raises(ConfigError):
ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_OIDC_ISSUER": "https://idp.example.com/issuer/",
"NESTQUEST_OIDC_AUDIENCE": "nestquest-api",
"NESTQUEST_OIDC_JWKS_URL": "https://idp.example.com/jwks/",
}
)
def test_from_env_requires_panel_token_when_empty() -> None:
"""An empty/whitespace panel token raises ConfigError."""
with pytest.raises(ConfigError):
ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_PANEL_TOKEN": " ",
"NESTQUEST_OIDC_ISSUER": "https://idp.example.com/issuer/",
"NESTQUEST_OIDC_AUDIENCE": "nestquest-api",
"NESTQUEST_OIDC_JWKS_URL": "https://idp.example.com/jwks/",
}
)
# --- service-token auth --------------------------------------------------
async def test_missing_token_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A request with no Authorization header returns 401."""
response = await panel_client.get("/api/v1/panel/snapshot")
assert response.status_code == 401
assert "www-authenticate" in {
key.lower() for key in response.headers
}
async def test_wrong_token_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A wrong Bearer token returns 401."""
response = await panel_client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": "Bearer not-the-panel-token"},
)
assert response.status_code == 401
async def test_malformed_authorization_header_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A malformed Authorization header returns 401.
Wrong scheme, empty credential, and an empty header all fail the
same way — the response never reveals which part failed.
"""
for header in (
"Basic dXNlcjpwYXNz",
"Bearer",
"Bearer ",
"Token not-the-panel-token",
"",
):
response = await panel_client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": header},
)
assert response.status_code == 401, f"header {header!r} passed"
# --- snapshot payload ----------------------------------------------------
async def test_valid_token_returns_full_snapshot(
seeded_panel: SimpleNamespace,
) -> None:
"""The correct token returns 200 with the documented snapshot shape.
Children in sort order (Ada sort_order 0, Bo 1); per-child counts,
presence, next_present, and cycle day all asserted as explicit
literals; each child's instances are in the documented panel
payload shape (done -> completed spelling, on_time passthrough).
"""
seed = seeded_panel.seed
response = await seeded_panel.client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
)
assert response.status_code == 200
body = response.json()
assert body["today_iso"] == seeded_panel.today.isoformat()
# Ada's 2-week cycle anchored yesterday => today is cycle day 2.
assert body["cycle_day"] == 2
children = body["children"]
assert [child["child_id"] for child in children] == [
seed.ada.id,
seed.bo.id,
], "children appear in sort_order"
ada, bo = children
# Ada: away today, returns on the override date; her 10:00 quest
# is open and overdue at the pinned 12:00.
assert ada["child_name"] == "Ada"
assert ada["present"] is False
assert ada["next_present"] == seed.returns_date.isoformat()
assert ada["due_today"] == 1
assert ada["completed_today"] == 0
assert ada["remaining_today"] == 1
assert ada["completion_pct"] == 0
assert ada["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,
}
]
# Bo: present, completed his 17:00 quest at the pinned 12:00
# (on time), so the panel spelling is "completed".
assert bo["child_name"] == "Bo"
assert bo["present"] is True
assert bo["next_present"] is None
assert bo["due_today"] == 1
assert bo["completed_today"] == 1
assert bo["remaining_today"] == 0
assert bo["completion_pct"] == 100
assert bo["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": seeded_panel.pinned_utc_iso,
"on_time": True,
}
]
async def test_missed_instances_are_omitted_from_panel_payload(
seeded_panel: SimpleNamespace,
) -> None:
"""A missed instance never reaches the panel payload (D-009).
The household carries a past-due OPEN instance due yesterday (the
state the panel reads as "missed"); it must not appear anywhere in
the response, and Ada's payload carries exactly her ONE today
instance — the past-due rows (the stale chore AND her pack bag due
yesterday) are both omitted rather than rendered.
"""
seed = seeded_panel.seed
response = await seeded_panel.client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
)
assert response.status_code == 200
body = response.json()
payload_instances = [
instance
for child in body["children"]
for instance in child["instances"]
]
assert "missed" not in [
instance["state"] for instance in payload_instances
]
past_due_ids = {seed.stale_instance.id}
assert past_due_ids.isdisjoint(
instance["id"] for instance in payload_instances
), "the past-due open instance leaked into the panel payload"
ada = next(
child
for child in body["children"]
if child["child_id"] == seed.ada.id
)
assert [instance["id"] for instance in ada["instances"]] == [
seed.pack_bag_instance.id
]
def test_panel_shaper_drops_missed_views() -> None:
"""The route's shaper omits a missed view (D-009), on the app's copy.
``instance_payload(include_missed=False)`` is what the snapshot
route shapes every child's instances with; exercised here on the
SAME core copy the app uses (the two-copy caveat makes this a
distinct object from the conftest-stubbed copy's identical source).
"""
missed_view = core_snapshot.QuestInstanceView(
instance_id=999,
definition_id=1,
child_id=1,
title="Stale chore",
icon=None,
window="morning",
due_date="2026-09-21",
due_time="09:00",
state="missed",
overdue=False,
completed_at=None,
was_on_time=None,
)
assert core_snapshot.instance_payload(
[missed_view], include_missed=False
) == []
# --- route documentation -------------------------------------------------
async def test_snapshot_route_is_documented(
panel_client: httpx.AsyncClient,
) -> None:
"""The snapshot route and its response shape are in the OpenAPI doc."""
response = await panel_client.get("/openapi.json")
assert response.status_code == 200
schema = response.json()
assert "/api/v1/panel/snapshot" in schema["paths"]
assert "get" in schema["paths"]["/api/v1/panel/snapshot"]
documented = {
model["title"]
for model in schema["components"]["schemas"].values()
}
assert "PanelSnapshotResponse" in documented
assert "PanelChildPayload" in documented
assert "PanelInstancePayload" in documented
# --- complete route -------------------------------------------------------
async def test_complete_writes_panel_completion(
seeded_panel: SimpleNamespace,
) -> None:
"""The first POST completes the instance as the tapped profile.
Returns success; the completion event recorded is
``actor_source='panel'`` with the tapped ``actor_child_id`` and NO
``actor_user_id`` (the panel actor contract), and the instance's
derived state becomes ``done``.
"""
seed = seeded_panel.seed
response = await seeded_panel.client.post(
f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": seed.ada.id},
)
assert response.status_code == 200
assert response.json() == {"status": "done"}
# Assert the recorded event and derived state through the SAME core
# copy the app uses (two-copy caveat: never mix copies), against the
# app's live connection.
events = await _events_for_instance(
seeded_panel.database, seed.pack_bag_instance.id
)
assert len(events) == 1
event = events[0]
assert event.event_type == "completed"
assert event.actor_source == "panel"
assert event.actor_child_id == seed.ada.id
assert event.actor_user_id is None
instance = await _dao_instances.QuestInstancesDao(
seeded_panel.database
).get_by_id(seed.pack_bag_instance.id)
assert await _derived_state(
seeded_panel.database, instance, seeded_panel.today
) == "done"
async def test_complete_is_idempotent(
seeded_panel: SimpleNamespace,
) -> None:
"""A second call is a no-op that still returns success.
The core completion layer's ``appended`` flag decides: the repeat
call appends nothing, so exactly ONE completion event row exists
after two calls.
"""
seed = seeded_panel.seed
headers = {"Authorization": f"Bearer {PANEL_TOKEN}"}
url = f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete"
body = {"actor_child_id": seed.ada.id}
first = await seeded_panel.client.post(url, headers=headers, json=body)
assert first.status_code == 200
second = await seeded_panel.client.post(url, headers=headers, json=body)
assert second.status_code == 200
assert second.json() == {"status": "done"}
events = await _events_for_instance(
seeded_panel.database, seed.pack_bag_instance.id
)
assert len(events) == 1, (
"the idempotent call must not append a second event"
)
async def test_complete_unknown_instance_is_not_found(
panel_client: httpx.AsyncClient,
) -> None:
"""Completing a non-existent instance returns 404."""
response = await panel_client.post(
"/api/v1/panel/instances/999999/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": 1},
)
assert response.status_code == 404
async def test_complete_without_token_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A complete request with no Authorization header returns 401."""
response = await panel_client.post(
"/api/v1/panel/instances/1/complete",
json={"actor_child_id": 1},
)
assert response.status_code == 401
async def test_complete_wrong_child_is_not_found(
seeded_panel: SimpleNamespace,
) -> None:
"""Tapping the wrong child's profile answers 404, not success.
``complete_instance`` does not itself check that the tapped
``actor_child_id`` is the child the instance belongs to; the route
guards it so a wrong-profile tap can never complete another
child's quest.
"""
seed = seeded_panel.seed
response = await seeded_panel.client.post(
f"/api/v1/panel/instances/{seed.pack_bag_instance.id}/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": seed.bo.id},
)
assert response.status_code == 404
# And nothing was completed.
events = await _events_for_instance(
seeded_panel.database, seed.pack_bag_instance.id
)
assert events == []
async def test_complete_route_is_documented(
panel_client: httpx.AsyncClient,
) -> None:
"""The complete route and its request shape are in the OpenAPI doc."""
response = await panel_client.get("/openapi.json")
assert response.status_code == 200
schema = response.json()
path = "/api/v1/panel/instances/{instance_id}/complete"
assert path in schema["paths"]
assert "post" in schema["paths"][path]
documented = {
model["title"]
for model in schema["components"]["schemas"].values()
}
assert "PanelCompleteRequest" in documented
Reference in New Issue View Git Blame Copy Permalink