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

Feature 16: NestQuest API Service (#165)

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 #165.
This commit is contained in:
dev_agent
2026-09-23 04:37:26 -04:00
committed by approver_agent
parent b5c2b55d9b
commit d78ef196d0
43 changed files with 12248 additions and 369 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_missed_sweep.py
+671
View File
@@ -0,0 +1,671 @@
"""Admin-plane missed-sweep trigger tests (task 058c7b69).
These tests exercise ``POST /api/v1/admin/missed-sweep`` — the admin
plane's trigger for the nightly missed-quest sweep — against a REAL
server (uvicorn on an ephemeral port, the SSE tests' documented
approach: httpx's ASGI transport buffers whole bodies, so a live
stream needs a live server), plus the API service's daily scheduler:
- The trigger requires the router's ONE ``require_admin`` check: a
non-admin JWT is 403 and an absent credential is 401, and neither
refusal runs the sweep.
- Seeded past-due open instances: ONE ``nestquest_quest_missed``
transition per instance arrives on the LIVE SSE stream with the
documented payload fields (child_id, child_name, instance_id,
quest_title, window, due_date, due_time, occurred_at — §3), while
the seeded TODAY-due instance publishes nothing (the sweep's rule
is ``due_date`` strictly before ``today``).
- A same-day rerun publishes nothing (the core watermark).
- A run after downtime (``today`` advanced two days) publishes the
accumulated window once — the instance due on day one fires on the
catch-up run — and a further rerun is silent again.
- Publishing is the api/scheduler.py shape — the sweep-BUILT
``(event_type, payload)`` pairs go straight to the publisher, no
per-event re-fetch: all events of one run share the run's ONE
``occurred_at`` stamp, and a deterministic concurrent-deletion case
(the sweep is wrapped to delete one swept instance after it built
its events and committed its watermark) still answers 200 and
streams every built event — a re-fetch there would 500 and lose the
missed event forever.
- The daily scheduler (:class:`api.scheduler.MissedSweepScheduler`)
computes the delay to the configured ``day_rollover_time``, runs the
sweep when the (injectable) clock reaches it, publishes the built
events on its publisher, advances to the NEXT day's rollover, and
cancels cleanly.
The household is seeded through the SAME core copy the app uses (the
``nestquest_core.*`` package registered by ``api/nestquest_core.py``;
see the two-copy caveat there) and the sweep's watermark is read back
through the same copy's :class:`~nestquest_core.dao_meta.MetaStateDao`
— on the SERVER's event loop for the uvicorn-hosted app (the database
belongs to that thread's loop, so assertions schedule there), on the
test's own loop for the scheduler test's standalone database.
"""
from __future__ import annotations
import asyncio
import datetime
import importlib
import threading
import time
from pathlib import Path
from types import SimpleNamespace
import httpx
import pytest
import uvicorn
from api import routes_admin
from api.app import create_app
from api.auth import JwksCache
from api.config import ApiConfig
from api.database import _executor
from api.nestquest_core import core_db as _core_db
from api.nestquest_core import core_migrations as _core_migrations
from api.scheduler import MissedSweepScheduler
from tests.admin_jwt_harness import (
ADMIN_KEY,
AUDIENCE,
ISSUER,
JWKS_URL,
KID,
StubbedFetch,
jwks_for,
make_token,
)
from tests.test_api_panel import PANEL_TOKEN
from tests.test_api_sse import EVENT_QUEST_MISSED, READ_TIMEOUT, _free_port, _stream
# 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_instances = importlib.import_module("nestquest_core.dao_instances")
_dao_meta = importlib.import_module("nestquest_core.dao_meta")
_settings_store = importlib.import_module("nestquest_core.settings_store")
_core_sweep = importlib.import_module("nestquest_core.sweep")
#: The well-known ``nestquest_meta_state`` key holding the sweep
#: watermark — asserted literally so the contract is pinned here.
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
async def _seed_missed_household(
db_path: str, today: datetime.date
) -> SimpleNamespace:
"""Seed the missed-sweep household on ``db_path`` and return records.
Two children; two quests whose ONLY instances are due YESTERDAY
(created via ``materialize`` with ``today`` pinned to yesterday —
the no-past guard accepts the rows, and they land past-due with no
completion events: the sweep's targets), and one quest due TODAY
(not past-due: the sweep must NOT announce it). The yesterday
rules carry ``end_date`` so the later today-walk cannot give them
a second instance. 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)
yesterday_rule = _recurrence.ScheduleRule.from_dict(
{
"rule_type": "daily",
"start_date": yesterday_iso,
"end_date": yesterday_iso,
}
)
await _quest_definitions.create_quest_definition(
database,
"Stale chore",
yesterday_rule,
[ada.id],
[("morning", "09:00")],
)
await _quest_definitions.create_quest_definition(
database,
"Old chore",
yesterday_rule,
[bo.id],
[("evening", "18:30")],
)
# Past-due open instances for the two yesterday quests (the
# walk pins ``today`` to yesterday so the rows are accepted).
await _materialize.materialize(
database, yesterday_iso, yesterday_iso, today=yesterday
)
# Today's quest: created after the yesterday walk so it has no
# past-due instance, then materialized for today only.
await _quest_definitions.create_quest_definition(
database,
"Recoverable chore",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": today_iso}
),
[ada.id],
[("afternoon", "17:00")],
)
await _materialize.materialize(
database, today_iso, today_iso, today=today
)
instances = _dao_instances.QuestInstancesDao(database)
ada_yesterday = await instances.list_by_child_and_date(
ada.id, yesterday_iso
)
assert len(ada_yesterday) == 1
stale = ada_yesterday[0]
bo_yesterday = await instances.list_by_child_and_date(
bo.id, yesterday_iso
)
assert len(bo_yesterday) == 1
old = bo_yesterday[0]
ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
assert len(ada_today) == 1
recoverable = ada_today[0]
return SimpleNamespace(
ada=ada,
bo=bo,
stale_instance=stale,
old_instance=old,
recoverable_instance=recoverable,
yesterday_iso=yesterday_iso,
today_iso=today_iso,
)
finally:
await database.close()
class _AdminServer:
"""Run the app with uvicorn on an ephemeral port, in a thread.
The SSE tests' documented approach (see tests/test_api_sse.py):
uvicorn's factory mode runs ``create_app`` INSIDE the server
thread, so the lifespan — database, publisher, the daily sweep
scheduler — belongs to that thread's event loop. The app's JWKS
cache is swapped for one built around the stubbed fetcher BEFORE
the server starts, so admin JWTs genuinely verify with no network.
"""
def __init__(self, db_path: str) -> None:
self.port = _free_port()
self._server_loop: asyncio.AbstractEventLoop | None = None
self._server = uvicorn.Server(
uvicorn.Config(
self._make_app_factory(db_path),
host="127.0.0.1",
port=self.port,
log_level="warning",
)
)
self._thread = threading.Thread(target=self._server.run, daemon=True)
def _make_app_factory(self, db_path: str):
"""The uvicorn app factory, closing over ``db_path``."""
def factory():
app = create_app(
ApiConfig(
db_path=db_path,
panel_token=PANEL_TOKEN,
oidc_issuer=ISSUER,
oidc_audience=AUDIENCE,
oidc_jwks_url=JWKS_URL,
)
)
app.state.jwks_cache = JwksCache(
JWKS_URL, fetcher=StubbedFetch(jwks_for(ADMIN_KEY, KID))
)
self._server_loop = asyncio.get_running_loop()
self.app = app
return app
return factory
def __enter__(self) -> "_AdminServer":
self._thread.start()
deadline = time.monotonic() + READ_TIMEOUT
while not self._server.started:
if time.monotonic() > deadline or not self._thread.is_alive():
raise RuntimeError("uvicorn did not start")
time.sleep(0.01)
return self
def __exit__(self, *exc) -> None:
self._server.should_exit = True
self._thread.join(timeout=READ_TIMEOUT)
def url(self, path: str) -> str:
"""The full URL for ``path`` on this server."""
return f"http://127.0.0.1:{self.port}{path}"
def run_on_loop(self, coro) -> object:
"""Run ``coro`` on the SERVER's event loop and wait for it.
The app (and its database) lives on the uvicorn thread's loop,
so database assertions schedule there. Bounded by
READ_TIMEOUT so a wedged schedule fails instead of hanging.
"""
return asyncio.run_coroutine_threadsafe(
coro, self._server_loop
).result(timeout=READ_TIMEOUT)
def _watermark(server: _AdminServer) -> str | None:
"""The stored sweep watermark, read on the server's own loop."""
database = server.app.state.db.database
return server.run_on_loop(
_dao_meta.MetaStateDao(database).get(SWEEP_WATERMARK_KEY)
)
@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 missed_app(temp_db_path: str) -> SimpleNamespace:
"""The seeded household served by a live uvicorn app, clock pinned.
``routes_admin._local_now`` is pinned to today 12:00 (the same
pinning discipline the panel/SSE tests use) BEFORE the server
starts, so the trigger route's ``today`` matches the seed; the
downtime test moves the pin forward mid-test. Restored in
``finally`` even on failure.
"""
now_local = datetime.datetime.now().astimezone()
pinned_now = now_local.replace(hour=12, minute=0, second=0, microsecond=0)
today = pinned_now.date()
seed = await _seed_missed_household(temp_db_path, today)
original = routes_admin._local_now
routes_admin._local_now = lambda: pinned_now
try:
with _AdminServer(temp_db_path) as server:
client = httpx.AsyncClient(base_url=server.url(""))
async with client:
yield SimpleNamespace(
client=client,
seed=seed,
server=server,
today=today,
pinned_now=pinned_now,
)
finally:
routes_admin._local_now = original
# --- the trigger: one missed transition per past-due open instance -----
async def test_trigger_publishes_one_missed_event_per_instance(
missed_app,
) -> None:
"""The trigger streams one documented missed transition per instance.
The stream is subscribed FIRST, the sweep is triggered through the
route as an admin, and each emitted frame's type and every
documented payload field are asserted; the TODAY-due instance must
NOT appear (the sweep's rule is strictly past-due).
"""
ns = missed_app
seed = ns.seed
stream = _stream(ns.server.url("/api/v1/panel/events"))
async with stream:
response = await ns.client.post(
"/api/v1/admin/missed-sweep", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {"fired": 2}
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert event_type == "nestquest_quest_missed"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == seed.stale_instance.id
assert payload["quest_title"] == "Stale chore"
assert payload["window"] == "morning"
assert payload["due_date"] == seed.yesterday_iso
assert payload["due_time"] == "09:00"
assert payload["occurred_at"] # strict UTC ISO-8601 stamp
run_stamp = payload["occurred_at"]
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert payload["child_id"] == seed.bo.id
assert payload["child_name"] == "Bo"
assert payload["instance_id"] == seed.old_instance.id
assert payload["quest_title"] == "Old chore"
assert payload["window"] == "evening"
assert payload["due_date"] == seed.yesterday_iso
assert payload["due_time"] == "18:30"
# One run, ONE stamp: the route publishes the sweep-built
# payloads directly, so every event of this run carries the
# same ``occurred_at`` the sweep computed for the whole run.
assert payload["occurred_at"] == run_stamp
# Exactly one transition per instance: today's not-past-due
# quest never arrives.
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
# The run recorded the watermark: the household-local date the
# sweep ran for, under the well-known meta_state key.
assert _watermark(ns.server) == seed.today_iso
async def test_trigger_publishes_sweep_built_payloads_without_refetch(
missed_app,
) -> None:
"""Publishing never re-fetches: a concurrently deleted instance
cannot fail the trigger or lose its already-built missed event.
The core sweep commits the idempotency watermark BEFORE returning,
so a publish loop that re-fetches each instance opens a loss
window: an instance deleted inside that window (e.g. a concurrent
regenerate) would raise ``ValueError`` uncaught — a bare 500 — and
the missed event would be gone forever (the watermark already
advanced, so no future sweep re-examines it). The route publishes
the sweep-built ``(event_type, payload)`` pairs DIRECTLY (the
api/scheduler.py shape), which this test exercises deterministically:
the sweep is wrapped to delete one swept instance AFTER the sweep
built its events and committed its watermark but BEFORE the route
publishes, and the trigger must still answer 200 with ``fired: 2``
and stream BOTH sweep-built payloads — the deleted instance's
included, stamped with the run's ONE shared ``occurred_at``.
"""
ns = missed_app
seed = ns.seed
database = ns.server.app.state.db.database
original_sweep = _core_sweep.run_missed_sweep
async def sweep_then_delete_stale(sweep_database, *, today):
events = await original_sweep(sweep_database, today=today)
# The concurrent deletion lands after the sweep built its
# events and committed its watermark, but before the route
# publishes: the row is gone while the built event survives.
await sweep_database.execute(
"DELETE FROM quest_instances WHERE id = ?",
(seed.stale_instance.id,),
)
return events
_core_sweep.run_missed_sweep = sweep_then_delete_stale
try:
stream = _stream(ns.server.url("/api/v1/panel/events"))
async with stream:
response = await ns.client.post(
"/api/v1/admin/missed-sweep", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {"fired": 2}
# The FIRST event is the deleted instance's — published
# from the payload built before the deletion, never
# re-fetched after it (a re-fetch would 500 here).
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert payload["instance_id"] == seed.stale_instance.id
assert payload["child_name"] == "Ada"
assert payload["quest_title"] == "Stale chore"
assert payload["window"] == "morning"
assert payload["due_date"] == seed.yesterday_iso
assert payload["occurred_at"]
run_stamp = payload["occurred_at"]
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert payload["instance_id"] == seed.old_instance.id
assert payload["occurred_at"] == run_stamp
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
finally:
_core_sweep.run_missed_sweep = original_sweep
# The watermark advanced even though one instance was deleted
# mid-flight — nothing was lost to the publish-time failure window.
assert _watermark(ns.server) == seed.today_iso
async def test_same_day_rerun_publishes_nothing(missed_app) -> None:
"""A second trigger the same day publishes nothing (the watermark)."""
ns = missed_app
headers = _admin_headers()
first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
assert first.status_code == 200
assert first.json() == {"fired": 2}
stream = _stream(ns.server.url("/api/v1/panel/events"))
async with stream:
second = await ns.client.post(
"/api/v1/admin/missed-sweep", headers=headers
)
assert second.status_code == 200
assert second.json() == {"fired": 0}
with pytest.raises(asyncio.TimeoutError):
# The watermark makes the rerun an empty no-op: the stream
# stays silent.
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
async def test_run_after_downtime_publishes_accumulated_window_once(
missed_app,
) -> None:
"""After days of downtime the catch-up run sweeps the window once.
The first trigger sweeps yesterday's instances (watermark := today).
``today`` then jumps two days: the accumulated window
``[watermark, today)`` holds exactly the seeded TODAY-due instance
(no completion events), which fires once — and a same-day rerun is
silent again.
"""
ns = missed_app
headers = _admin_headers()
first = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
assert first.status_code == 200
assert first.json() == {"fired": 2}
# Advance the pinned clock two days: the app is "down" over the
# rollovers in between.
routes_admin._local_now = lambda: ns.pinned_now + datetime.timedelta(
days=2
)
later_today = (ns.today + datetime.timedelta(days=2)).isoformat()
stream = _stream(ns.server.url("/api/v1/panel/events"))
async with stream:
catchup = await ns.client.post(
"/api/v1/admin/missed-sweep", headers=headers
)
assert catchup.status_code == 200
assert catchup.json() == {"fired": 1}
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED
assert payload["instance_id"] == ns.seed.recoverable_instance.id
assert payload["child_id"] == ns.seed.ada.id
assert payload["quest_title"] == "Recoverable chore"
assert payload["window"] == "afternoon"
assert payload["due_date"] == ns.seed.today_iso
assert payload["due_time"] == "17:00"
assert payload["occurred_at"]
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
# The watermark moved to the new today...
assert _watermark(ns.server) == later_today
# ...so a further rerun publishes nothing.
again = await ns.client.post("/api/v1/admin/missed-sweep", headers=headers)
assert again.status_code == 200
assert again.json() == {"fired": 0}
# --- authentication: the router's ONE require_admin dependency ----------
async def test_trigger_refuses_non_admin_and_absent_credentials(
missed_app,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401; none sweep."""
ns = missed_app
non_admin = make_token(groups=("some-other-group",))
forbidden = await ns.client.post(
"/api/v1/admin/missed-sweep",
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden.status_code == 403
assert forbidden.json()["detail"] == (
"Admin access requires membership in the nestquest-admins group"
)
absent = await ns.client.post("/api/v1/admin/missed-sweep")
assert absent.status_code == 401
# Neither refusal ran the sweep: the watermark is still unset, and
# the next admin trigger fires the full seeded set.
assert _watermark(ns.server) is None
after = await ns.client.post(
"/api/v1/admin/missed-sweep", headers=_admin_headers()
)
assert after.status_code == 200
assert after.json() == {"fired": 2}
# --- the daily scheduler -------------------------------------------------
class _RecordingPublisher:
"""A transition-publisher stub recording every published event."""
def __init__(self) -> None:
self.published: list[tuple[str, dict]] = []
def publish(self, event_type: str, payload: dict) -> None:
self.published.append((event_type, payload))
async def test_scheduler_runs_sweep_at_rollover_and_cancels_cleanly(
temp_db_path: str,
) -> None:
"""The scheduler sweeps at the configured rollover and stops clean.
The injectable clock starts at 10:00 and the settings pin the
rollover to 10:05, so the first computed delay is five minutes;
the injectable sleep fast-forwards the clock, the sweep runs for
the (now advanced) clock's date, and the seeded past-due open
instance is published on the recording publisher exactly once.
The next cycle schedules for the NEXT day's rollover (23h55m).
``stop`` cancels the loop task and the scheduler reports stopped.
"""
database = _core_db.NestQuestDatabase(_executor)
await database.open(temp_db_path)
try:
await _core_migrations.apply_migrations(database)
now_local = datetime.datetime.now().astimezone()
start = now_local.replace(
hour=10, minute=0, second=0, microsecond=0
)
ada = await _children.create_child(database, "Ada", sort_order=0)
yesterday = (start.date() - datetime.timedelta(days=1)).isoformat()
await _quest_definitions.create_quest_definition(
database,
"Stale chore",
_recurrence.ScheduleRule.from_dict(
{
"rule_type": "daily",
"start_date": yesterday,
"end_date": yesterday,
}
),
[ada.id],
[("morning", "09:00")],
)
await _materialize.materialize(
database, yesterday, yesterday, today=start.date() - datetime.timedelta(days=1)
)
await _settings_store.update_settings(
database, {"day_rollover_time": "10:05"}
)
clock_state = {"now": start}
delays: list[float] = []
def clock() -> datetime.datetime:
return clock_state["now"]
async def sleep_stub(seconds: float) -> None:
delays.append(seconds)
if len(delays) == 1:
# Fast-forward the clock across the rollover wait ONCE,
# so the first sweep runs at the rollover instant; the
# clock then stays frozen at that instant, so the
# follow-up cycles sweep the same (already-watermarked)
# date and the assertions read a stable state.
clock_state["now"] = clock_state["now"] + datetime.timedelta(
seconds=seconds
)
await asyncio.sleep(0)
publisher = _RecordingPublisher()
scheduler = MissedSweepScheduler(
database, publisher=publisher, clock=clock, sleep=sleep_stub
)
scheduler.start()
try:
deadline = time.monotonic() + READ_TIMEOUT
while not publisher.published:
if time.monotonic() > deadline:
pytest.fail("scheduler never published a missed event")
await asyncio.sleep(0.01)
# The second cycle's scheduling decision may lag the first
# publication by a few awaits; wait for it the same way.
while len(delays) < 2:
if time.monotonic() > deadline:
pytest.fail("scheduler never scheduled a second cycle")
await asyncio.sleep(0.01)
event_type, payload = publisher.published[0]
assert event_type == "nestquest_quest_missed"
assert payload["child_id"] == ada.id
assert payload["quest_title"] == "Stale chore"
assert payload["window"] == "morning"
assert payload["due_date"] == yesterday
assert payload["due_time"] == "09:00"
assert payload["occurred_at"]
# The rollover delay was honored exactly...
assert delays[0] == pytest.approx(300.0)
# ...and the next cycle targets the NEXT day's rollover
# (the just-passed instant schedules for tomorrow).
assert delays[1] == pytest.approx(86400.0)
# The watermark pins the run to the sweep-day's date.
watermark = await _dao_meta.MetaStateDao(database).get(
SWEEP_WATERMARK_KEY
)
assert watermark == start.date().isoformat()
finally:
await scheduler.stop()
assert not scheduler.running
# Exactly one announcement: the watermark makes every further
# cycle an empty no-op.
assert len(publisher.published) == 1
finally:
await database.close()