Compare commits

...
Author SHA1 Message Date
overseer 933434320a Publish the sweep-built missed events directly in the admin trigger (P1)
The missed-sweep route re-fetched every instance through
publish_quest_missed AFTER core_sweep.run_missed_sweep had already
committed its idempotency watermark.  A concurrent regenerate deleting
an instance — or a transient database error — made that re-fetch raise
ValueError uncaught (a bare 500) and dropped the missed event forever:
the watermark had advanced, so no future sweep re-examines the
instance.  The re-fetch also discarded the ONE shared occurred_at
stamp the sweep computed for the whole run, re-stamping each event
individually.

Publish the already-built (event_type, payload) pairs straight to the
app's ONE publisher instead — the same shape api/scheduler.py uses —
leaving no failure window between the watermark and the announcement.
The deterministic concurrent-deletion test wraps the sweep to delete a
swept instance after it built its events: the trigger must still
answer 200 and stream both sweep-built payloads (it 500'd before the
fix), and all events of one run share the run's occurred_at.

Also add the missing trailing newlines to core/sweep.py and
api/scheduler.py (P3).
2026-09-23 04:17:42 -04:00
overseer dfe8587c7a Run the missed sweep in the API and publish quest-missed on SSE
WHY: The nightly missed-quest sweep only existed on the Home
Assistant plane (the rollover listener fired hass.bus), so the API
service - now the writer of record with its own settings store and
SSE stream - had no way to announce missed quests, and an admin
could not trigger a catch-up run after downtime.

- Extract the sweep policy (watermark read, list_missed_unswept
  query, documented payload build, watermark write, per-database
  lock) into the HA-free core (core/sweep.py), building each event
  through the shared events payload builder so the shape cannot
  drift; the integration's sweep.py becomes a thin shim that fires
  the returned events on hass.bus, keeping its signature.
- Admin trigger POST /api/v1/admin/missed-sweep: ONE clock read,
  ONE core call, each returned event published on the SSE stream
  through the existing publish_quest_missed helper; returns the
  fired count. Watermark makes a same-day rerun silent and a
  post-downtime run sweep the accumulated window once.
- Daily scheduler (api/scheduler.py) wired in the lifespan: reads
  day_rollover_time fresh each cycle, runs the sweep at the
  rollover, publishes on the app publisher, backs off and logs on
  failure, cancels cleanly on shutdown; injectable clock/sleep for
  tests.
- Tests: route trigger over a live uvicorn server with a live SSE
  stream (one transition per instance with documented fields,
  same-day rerun silent, post-downtime catch-up once, non-admin
  403 / absent 401) plus the scheduler rollover/cancellation test.
2026-09-23 04:07:49 -04:00
dev_agent fa4415c0d5 Admin routes: settings read and update (#163) 2026-09-23 03:46:46 -04:00
dev_agent 0255017cdc Admin routes: history query and CSV export (#162) 2026-09-23 03:13:39 -04:00
dev_agent f31f6dc80e Admin routes: uncomplete and regenerate (#161) 2026-09-23 02:43:45 -04:00
12 changed files with 4129 additions and 100 deletions
+12 -1
View File
@@ -48,6 +48,7 @@ from api.database import DatabaseState, make_database
from api.events import TransitionPublisher
from api.routes_admin import router as admin_router
from api.routes_panel import router as panel_router
from api.scheduler import MissedSweepScheduler
def _build_lifespan(db_path: str):
@@ -57,7 +58,11 @@ def _build_lifespan(db_path: str):
path in without threading it through FastAPI's lifespan signature.
The lifespan also installs the app's ONE in-process transition
publisher on ``app.state.publisher`` (see :mod:`api.events`) so
every route and SSE subscriber shares that single instance.
every route and SSE subscriber shares that single instance, and
starts the daily missed-sweep scheduler
(:class:`api.scheduler.MissedSweepScheduler`) against the same
database and publisher — cancelled before the connection closes on
shutdown.
"""
@asynccontextmanager
@@ -65,10 +70,16 @@ def _build_lifespan(db_path: str):
state = make_database(db_path)
app.state.db = state
app.state.publisher = TransitionPublisher()
scheduler = MissedSweepScheduler(
state.database, publisher=app.state.publisher
)
app.state.sweep_scheduler = scheduler
try:
await state.open_and_migrate()
scheduler.start()
yield
finally:
await scheduler.stop()
await state.close()
return lifespan
+50
View File
@@ -116,6 +116,11 @@ core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
#: ``uncomplete_instance``) the panel's write routes call.
core_completion = importlib.import_module(f"{_CORE_MODULE_NAME}.completion")
#: The read-only history layer (``core.history.query_history``,
#: ``export_history_csv``, ``rows_to_csv``) the admin plane's history
#: query and CSV export routes call.
core_history = importlib.import_module(f"{_CORE_MODULE_NAME}.history")
#: The transition-event builders (``core.events.build_quest_completed_event``
#: et al.) the panel's write routes build their SSE events with.
core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
@@ -124,6 +129,17 @@ core_events = importlib.import_module(f"{_CORE_MODULE_NAME}.events")
#: the snapshot builder's ``settings`` parameter takes.
core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
#: The durable settings store (``core.settings_store.load_settings`` /
#: ``update_settings``) the admin plane's settings routes read and
#: write — the API's OWN settings source: a validated JSON document in
#: the existing ``nestquest_meta_state`` table, resolved through the
#: same core settings validators (the integration's HA config-entry
#: options remain a temporary separate source until Feature 18 wires
#: the client to the API).
core_settings_store = importlib.import_module(
f"{_CORE_MODULE_NAME}.settings_store"
)
#: The children business layer (``core.children.create_child`` et al.)
#: the admin plane's children routes call.
core_children = importlib.import_module(f"{_CORE_MODULE_NAME}.children")
@@ -161,13 +177,45 @@ core_dao_presence = importlib.import_module(
f"{_CORE_MODULE_NAME}.dao_presence"
)
#: The materialization walk (``core.materialize.materialize``,
#: ``regenerate_for_definition``, ``regenerate_for_child``) the admin
#: plane's regenerate route calls — the same walk the integration's
#: config/presence changes trigger.
core_materialize = importlib.import_module(f"{_CORE_MODULE_NAME}.materialize")
#: The core constants module (``core.const.DEFAULT_HORIZON_DAYS``) the
#: regenerate route sizes the household rolling horizon with.
core_const = importlib.import_module(f"{_CORE_MODULE_NAME}.const")
#: The typed child DAO (``core.dao_children.ChildrenDao.get``) the
#: regenerate route reads a child's existence with before delegating.
core_dao_children = importlib.import_module(
f"{_CORE_MODULE_NAME}.dao_children"
)
#: The typed definition DAO (``core.dao_rules.QuestDefinitionsDao.get``)
#: the regenerate route reads a definition's existence with before
#: delegating.
core_dao_rules = importlib.import_module(f"{_CORE_MODULE_NAME}.dao_rules")
#: The HA-free missed sweep (``core.sweep.run_missed_sweep`` +
#: ``SWEEP_WATERMARK_KEY``) the admin plane's trigger route calls and
#: the daily scheduler runs — the ONE sweep policy the integration's
#: bus shim shares.
core_sweep = importlib.import_module(f"{_CORE_MODULE_NAME}.sweep")
__all__ = [
"core",
"core_children",
"core_completion",
"core_const",
"core_dao_children",
"core_dao_presence",
"core_dao_rules",
"core_db",
"core_events",
"core_history",
"core_materialize",
"core_migrations",
"core_presence",
"core_presence_management",
@@ -175,4 +223,6 @@ __all__ = [
"core_recurrence",
"core_snapshot",
"core_settings",
"core_settings_store",
"core_sweep",
]
+790 -3
View File
@@ -59,6 +59,103 @@ layer the HA services share, so the two planes cannot drift:
design) and regenerates that override's child. It answers
``{"status": "ok"}``.
The uncomplete and regenerate routes (task da0226b3) follow the same
pattern over :mod:`nestquest_core.completion` and
:mod:`nestquest_core.materialize`:
- ``POST /instances/{instance_id}/uncomplete`` reverses ONE completion:
the handler takes the actor from the VERIFIED JWT (``actor_source``
``'user'`` with ``actor_user_id`` = the token's ``sub`` — the D-008
user-actor shape, so the reversal is attributed to the admin who
called it) and calls :func:`nestquest_core.completion.uncomplete_instance`.
The append-only discipline lives in the core: a reversal APPENDS an
``uncompleted`` event after the original ``completed`` row — the
original is never edited or deleted — and the instance's derived
state (recomputed from the latest event) returns to ``open``. On an
ACTUAL reversal (``CompletionResult.appended`` True) the route
publishes the ``nestquest_quest_uncompleted`` transition through
:func:`api.transitions.publish_quest_uncompleted` (the payload built
by the core builder); an already-open no-op publishes nothing. The
response carries the derived ``state`` and whether THIS call
``appended``.
- ``POST /regenerate`` re-materializes the rolling horizon through the
same core path the integration's config/presence changes use. The
body names ONE scope: ``household`` (the default —
:func:`nestquest_core.materialize.materialize` over
``[today, today + const.DEFAULT_HORIZON_DAYS]``, the core's default
horizon; the API config carries no horizon knob of its own),
``definition`` (:func:`nestquest_core.materialize.regenerate_for_definition`
— deletes the definition's open future instances and re-runs the
walk), or ``child``
(:func:`nestquest_core.materialize.regenerate_for_child`).
The upsert is idempotent, so re-running never duplicates an
instance. The response carries the scope, the scoped id (if any)
and the core's upsert ``count``.
The history routes (task 2d2dda53) follow the same pattern over
:mod:`nestquest_core.history` — the ONE read-only history layer, which
owns every rule so the two routes cannot drift:
- ``GET /history?filter=all|reversals|missed&start=YYYY-MM-DD&end=YYYY-MM-DD``
returns the history rows for ONE filter over the CLOSED due-date
range ``[start, end]``: ``all`` (every completion event — completed
AND uncompleted — whose instance is due in the range), ``reversals``
(the uncompleted events only) and ``missed`` (the DERIVED past-due
open instances, computed by the core's own derive-state rule —
``missed`` is never a stored event_type and nothing is appended).
The range scopes by the instance's ``due_date`` (the Feature 13
rule: history asks what happened to the tasks due that day), not the
event's wall-clock stamp. The filter and date-range policies live
in the core (strict ``YYYY-MM-DD``, ``end >= start``, a
``ValueError`` naming the offending field); the handler is a thin
adapter that threads ONE clock read (:func:`_local_now`, the missed
derivation's ``today``) and serializes the core's rows.
- ``GET /history.csv?...`` exports the SAME filtered range as CSV:
ONE :func:`nestquest_core.history.export_history_csv` call — the
same query path the JSON route uses — answered as ``text/csv`` with
a ``Content-Disposition`` attachment filename and the documented,
stable header row (Admin spec §5 event-row fields).
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
document in the existing ``nestquest_meta_state`` key/value table); the
integration's Home-Assistant config-entry options remain a temporary,
SEPARATE source until Feature 18 wires the client to the API — the two
are deliberately not unified here:
- ``GET /settings`` returns the current effective settings: all ten
documented fields (the rolling horizon, the day rollover time, the
notify target, the three notification times and the four enable
toggles) — the defaults on a fresh database.
- ``PATCH /settings`` updates ONLY the supplied fields: the body is a
JSON object of settings field names to new values, passed STRAIGHT
to :func:`nestquest_core.settings_store.update_settings` — the
unknown-field rejection, the per-field validation (the core settings
validators: a positive int, strict ``HH:MM`` strings, real booleans)
and the validate-everything-before-write atomicity all live in the
core, so a rejected update never changes the stored settings. A
JSON ``null`` value resets its field to the default (the settings
module's uniform None-as-absent rule). The response is the updated
effective settings, the same shape ``GET`` returns.
The missed-sweep trigger (task 058c7b69) follows the same pattern over
:mod:`nestquest_core.sweep` — the HA-free sweep policy the
integration's rollover listener shares:
- ``POST /missed-sweep`` runs the sweep for the API host's local
``today`` (ONE :func:`_local_now` clock read threaded into the core
call) and publishes each returned ``(event_type, payload)`` pair
DIRECTLY on the app's ONE transition publisher — the sweep-built
payload, never re-fetched (the core commits its watermark before
returning, so a failed re-fetch after it would lose the missed event
forever), the same publish-what-the-core-built shape
api/scheduler.py uses. The watermark rule
(a same-day rerun announces nothing, a post-downtime run sweeps the
accumulated window once), the no-completion-event rule and the
read-only-over-the-domain-tables guarantee all live in the core; the
response reports how many transitions were published.
Error mapping (every children, quest-definition and presence route,
deliberately narrow):
@@ -94,6 +191,36 @@ deliberately narrow):
:func:`_raise_presence_override_error`, keyed on the
``override_id:`` prefix the core's unknown-id error carries (the
same convention as the quest-definition 404).
- The UNCOMPLETE route maps the core completion layer's unknown-
``instance_id`` error (prefixed ``instance_id:``, the same
convention the panel complete route keys on) to 404; every other
core ``ValueError`` — a rejected actor shape, which a verified token
without a usable ``sub`` claim produces — is 422: the credential
authenticated but the reversal it asks for names no subject to
attribute. Anything ELSE re-raises: no unexpected failure is
swallowed into a 4xx.
- The REGENERATE route's only 404s are the unknown scope ids, decided
by ONE existence read (the definitions/children DAO ``get``) BEFORE
any core call — the core's regenerate path itself takes no existence
stance. Every core ``ValueError`` (a rejected horizon, a malformed
date — unreachable from this body model, but mapped anyway) is 422;
anything else re-raises.
- The HISTORY routes have no path ids, so they have no 404 case: a
missing query parameter (the request's shape) is FastAPI's 422
before any handler code runs, and every core ``ValueError`` — an
unknown filter, a non-strict or inverted date range, each naming the
offending field (``filter`` / ``start`` / ``end``) — is 422
(:func:`_raise_history_error`): well-formed but semantically
invalid, the same failure class Pydantic reports. Only
``ValueError`` is caught, so an unexpected failure re-raises rather
than becoming a 4xx.
- The SETTINGS routes have no path ids either, so they have no 404
case: a body that is not a JSON object is FastAPI's 422 before any
handler code runs, and every core ``ValueError`` — an unknown field
or an invalid value, each naming the offending field, raised BEFORE
any write so the stored settings are never half-changed — is 422
(:func:`_raise_settings_error`). Only ``ValueError`` is caught, so
an unexpected failure re-raises rather than becoming a 4xx.
PATCH edit semantics: the body model's fields are optional and only
SUPPLIED fields are forwarded (``model_dump(exclude_unset=True)``), so
@@ -104,19 +231,30 @@ silently ignored.
"""
from __future__ import annotations
from typing import Annotated, NoReturn
import datetime
from typing import Annotated, Literal, NoReturn
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
from fastapi import APIRouter, Depends, HTTPException, Request, Response
from pydantic import BaseModel, field_validator, model_validator
from api import transitions as api_transitions
from api.auth import require_admin
from api.database import DatabaseState
from api.nestquest_core import (
core_children,
core_completion,
core_const,
core_dao_children,
core_dao_rules,
core_history,
core_materialize,
core_presence,
core_presence_management,
core_quest_definitions,
core_recurrence,
core_settings,
core_settings_store,
core_sweep,
)
#: All admin-plane routes share this router; the ONE admin dependency
@@ -395,6 +533,156 @@ class AdminPresenceOverrideResponse(BaseModel):
note: str | None
class AdminUncompleteResponse(BaseModel):
"""The result of ``POST /api/v1/admin/instances/{id}/uncomplete``.
``state`` is the instance's derived state AFTER the reversal —
``open``, or ``missed`` when it is past due — and ``appended``
reports whether THIS call appended the reversal event (``False``
for an already-open no-op).
"""
instance_id: int
state: str
appended: bool
class AdminRegenerateRequest(BaseModel):
"""The body of ``POST /api/v1/admin/regenerate``.
ONE scope per request: ``household`` (the default) re-materializes
every active definition over the rolling horizon; ``definition``
and ``child`` scope the rebuild to one id. The model enforces the
scope/id pairing so the handler never has to guess: an id without
its scope, a scope without its id, or a zero/negative id is a
validation error (422) before any handler code runs.
"""
scope: Literal["household", "definition", "child"] = "household"
definition_id: int | None = None
child_id: int | None = None
@field_validator("definition_id", "child_id")
@classmethod
def _positive(cls, value: int | None) -> int | None:
"""Reject zero and negative ids with a validation error."""
if value is not None and value < 1:
raise ValueError(
"definition_id and child_id must be positive integers"
)
return value
@model_validator(mode="after")
def _scope_matches_ids(self) -> "AdminRegenerateRequest":
"""Reject an id that does not belong to the requested scope."""
if self.scope == "household":
if self.definition_id is not None or self.child_id is not None:
raise ValueError(
"scope 'household' takes no definition_id or child_id"
)
elif self.scope == "definition":
if self.definition_id is None:
raise ValueError("scope 'definition' requires definition_id")
if self.child_id is not None:
raise ValueError("scope 'definition' must not carry child_id")
else:
if self.child_id is None:
raise ValueError("scope 'child' requires child_id")
if self.definition_id is not None:
raise ValueError(
"scope 'child' must not carry definition_id"
)
return self
class AdminRegenerateResponse(BaseModel):
"""What a regenerate request rebuilt.
``scope`` echoes the request; ``definition_id``/``child_id`` carry
the scoped id (both ``None`` for the household); ``count`` is the
number of instances the core walk upserted — the core's idempotent
count, so a re-run refreshes rows in place and never duplicates.
"""
scope: str
definition_id: int | None
child_id: int | None
count: int
class AdminHistoryRow(BaseModel):
"""One history row as the admin plane serializes it.
Mirrors :class:`nestquest_core.history.HistoryRow` — the Admin spec
§5 event-row fields, the same set the CSV export's header carries.
A stored event carries its UTC ``occurred_at`` stamp and the
``completed``/``uncompleted`` ``event_type``; a derived missed row
(``missed`` is NEVER a stored event_type) carries ``event_type``
``'missed'``, ``occurred_at`` ``None`` (there is no event) and the
``nightly sweep`` actor.
"""
occurred_at: str | None
event_type: str
child_name: str
child_id: int
quest_title: str
instance_id: int
window: str
due_date: str
due_time: str | None
actor: str
was_on_time: bool | None
class AdminHistoryResponse(BaseModel):
"""The history query's answer.
``filter``/``start``/``end`` echo the request, ``count`` is the row
count, and ``rows`` are the core's rows in its deterministic order
(``due_date``, then instance, then event id — the per-instance
append order, so a completion and its reversal keep their order).
"""
filter: str
start: str
end: str
count: int
rows: list[AdminHistoryRow]
class AdminSettingsResponse(BaseModel):
"""The effective settings as the admin plane serializes them.
Mirrors the core :class:`~nestquest_core.settings.NestQuestSettings`
— all TEN documented fields, resolved (every absent stored field
already fell back to its default in the core). The same shape both
the GET and the PATCH route answer with.
"""
horizon_days: int
day_rollover_time: str
notify_target: str
morning_summary_time: str
afternoon_reminder_time: str
end_of_day_report_time: str
morning_summary_enabled: bool
afternoon_reminder_enabled: bool
end_of_day_report_enabled: bool
celebration_enabled: bool
class AdminMissedSweepResponse(BaseModel):
"""The result of ``POST /api/v1/admin/missed-sweep``.
``fired`` is the number of ``nestquest_quest_missed`` transitions
the run built AND this route published on the SSE stream — zero
when the sweep's watermark already covers today (a same-day rerun).
"""
fired: int
#: 404 detail for a child id the core layer reports as non-existent.
_CHILD_NOT_FOUND_DETAIL = "Child not found"
@@ -478,6 +766,98 @@ def _raise_presence_override_error(error: ValueError) -> NoReturn:
raise HTTPException(status_code=422, detail=str(error)) from error
#: 404 detail for an instance id the core layer reports as non-existent.
_QUEST_INSTANCE_NOT_FOUND_DETAIL = "Quest instance not found"
def _raise_instance_error(error: ValueError) -> NoReturn:
"""Map a core completion ``ValueError`` onto 404 or 422 and raise it.
The uncomplete route's mapping, keyed the same way the panel
complete route keys its 404: the core names a missing instance by
prefixing its error with ``instance_id:``
(:func:`nestquest_core.completion.uncomplete_instance` on an
unknown id), and that one case is this route's 404. Every other
``ValueError`` — a rejected actor shape, which a verified token
without a usable ``sub`` claim produces — is 422 with the core's
message as the detail. The route catches ONLY ``ValueError``, so
an unexpected failure propagates untouched.
"""
if str(error).startswith("instance_id:"):
raise HTTPException(
status_code=404, detail=_QUEST_INSTANCE_NOT_FOUND_DETAIL
) from error
raise HTTPException(status_code=422, detail=str(error)) from error
def _raise_regenerate_error(error: ValueError) -> NoReturn:
"""Map a core materialization ``ValueError`` onto 422 and raise it.
The regenerate route's 404s are decided BEFORE the core call (the
handler's own existence reads of the scoped id), so every
``ValueError`` that survives to here is a rejected argument — a
malformed horizon or date the core refuses before any write — and
is 422 with the core's message as the detail. Nothing else is
caught: an unexpected failure re-raises rather than becoming a 4xx.
"""
raise HTTPException(status_code=422, detail=str(error)) from error
def _raise_history_error(error: ValueError) -> NoReturn:
"""Map a core history ``ValueError`` onto 422 and raise it.
The history routes have no path ids, so they have no 404 case:
every ``ValueError`` that survives to here is a rejected argument —
an unknown filter, a non-strict date, or an inverted range, each
raised by the core naming the offending field (``filter`` /
``start`` / ``end``) BEFORE any read — and is 422 with the core's
message as the detail. Only ``ValueError`` is caught by the
handlers, so an unexpected failure re-raises rather than becoming a
4xx.
"""
raise HTTPException(status_code=422, detail=str(error)) from error
def _raise_settings_error(error: ValueError) -> NoReturn:
"""Map a core settings-store ``ValueError`` onto 422 and raise it.
The settings routes have no path ids either, so they have no 404
case: every ``ValueError`` that survives to here is a rejected
change — an unknown field or an invalid value, each raised by the
core naming the offending field BEFORE any write, so the stored
settings are never half-changed — and is 422 with the core's
message as the detail. Only ``ValueError`` is caught, so an
unexpected failure re-raises rather than becoming a 4xx.
"""
raise HTTPException(status_code=422, detail=str(error)) from error
def _raise_sweep_error(error: ValueError) -> NoReturn:
"""Map a core missed-sweep ``ValueError`` onto 422 and raise it.
The sweep route has no path ids, so it has no 404 case: a core
``ValueError`` — a rejected date the sweep's query refuses before
any read, unreachable from this handler because ``today`` comes
from the route's own clock read — is 422 with the core's message
as the detail. Only ``ValueError`` is caught, so an unexpected
failure re-raises rather than becoming a 4xx.
"""
raise HTTPException(status_code=422, detail=str(error)) from error
def _local_now() -> datetime.datetime:
"""Return the API host's local time as ONE timezone-aware clock read.
Same discipline as the panel route's clock read: a single
``now().astimezone()`` read anchors the household-local date the
regenerate horizon and the uncomplete stamp derive from, so the
calendar date can never disagree inside one request. Kept as a
module function so tests pin the instant the same way the panel
route's ``_local_now`` is pinned.
"""
return datetime.datetime.now().astimezone()
def _rule_from_request(rule: AdminRuleRequest) -> core_recurrence.ScheduleRule:
"""Build the core ScheduleRule from the request's ``rule`` object.
@@ -942,4 +1322,411 @@ async def admin_delete_presence_override(
return {"status": "ok"}
@router.post(
"/instances/{instance_id}/uncomplete",
summary="Reverse one completion (append a reversal event)",
response_model=AdminUncompleteResponse,
)
async def admin_uncomplete_instance(
instance_id: int,
request: Request,
claims: Annotated[dict[str, object], Depends(require_admin)],
) -> AdminUncompleteResponse:
"""Reverse one completion; the original event is never touched.
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). ``claims`` is the SAME
verified-claims dict that dependency produced (FastAPI's
per-request dependency cache), and the reversal is attributed to
it: ``actor_source="user"`` with ``actor_user_id`` = the token's
``sub`` — the D-008 user-actor shape, ``actor_child_id`` never
sent — so the completion log records WHICH admin reversed it.
Thin adapter: ONE timezone-aware clock read (:func:`_local_now`)
and ONE :func:`nestquest_core.completion.uncomplete_instance` call
— the append-only discipline (a reversal APPENDS an ``uncompleted``
event after the original ``completed`` row, which is never edited
or deleted), the already-open no-op and the derived-state
recomputation all live in the core layer.
On an ACTUAL reversal (``CompletionResult.appended`` True) the
route publishes the ``nestquest_quest_uncompleted`` transition on
the SSE stream through
:func:`api.transitions.publish_quest_uncompleted` (the payload is
built by the core builder, never hand-built here; fan-out is
fire-and-forget, see api/events.py). An already-open no-op
(``appended`` False) publishes nothing.
Response: the derived ``state`` after the reversal — ``open``, or
``missed`` when the instance is past due — and whether THIS call
``appended``. Error mapping (:func:`_raise_instance_error`): an
unknown instance is 404; every other core ``ValueError`` is 422.
"""
state: DatabaseState = request.app.state.db
database = state.database
now = _local_now()
sub = claims.get("sub")
try:
result = await core_completion.uncomplete_instance(
database,
instance_id,
actor_source="user",
actor_user_id=sub if isinstance(sub, str) else None,
now=now,
)
except ValueError as error:
_raise_instance_error(error)
if result.appended:
# An ACTUAL reversal: announce the transition once. A no-op
# (already open) publishes nothing.
await api_transitions.publish_quest_uncompleted(
database, instance_id, request.app.state.publisher
)
return AdminUncompleteResponse(
instance_id=instance_id,
state=str(result),
appended=result.appended,
)
@router.post(
"/regenerate",
summary="Re-materialize the rolling horizon for one scope",
response_model=AdminRegenerateResponse,
)
async def admin_regenerate(
body: AdminRegenerateRequest, request: Request
) -> AdminRegenerateResponse:
"""Re-materialize the rolling horizon for the requested scope.
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 over the SAME
materialization path the integration's config/presence changes
trigger, one scope per request:
- ``household`` (the default): ONE
:func:`nestquest_core.materialize.materialize` call over
``[today, today + const.DEFAULT_HORIZON_DAYS]`` — the core's
default horizon, which sizes the window here because the API
config carries no horizon knob of its own. The walk is an
idempotent upsert: re-running refreshes rows in place and never
duplicates an instance.
- ``definition``: ONE existence read
(:class:`~nestquest_core.dao_rules.QuestDefinitionsDao.get` — an
unknown id is 404 BEFORE any core call, which the core's
regenerate path itself takes no stance on), then
:func:`nestquest_core.materialize.regenerate_for_definition`.
- ``child``: the same shape through
:class:`~nestquest_core.dao_children.ChildrenDao.get` and
:func:`nestquest_core.materialize.regenerate_for_child`.
The ONE clock read (:func:`_local_now`) threads the household-local
``today`` into every core call so the walk's no-past guard and the
delete cutoff share one anchor. Every core call returns the
upsert ``count``, which the response passes through. Error
mapping (:func:`_raise_regenerate_error`): every core
``ValueError`` is 422; the scope-id 404s are decided above.
"""
state: DatabaseState = request.app.state.db
database = state.database
today = _local_now().date()
if body.scope == "household":
end_date = today + datetime.timedelta(
days=core_const.DEFAULT_HORIZON_DAYS
)
try:
count = await core_materialize.materialize(
database,
today.isoformat(),
end_date.isoformat(),
today=today,
)
except ValueError as error:
_raise_regenerate_error(error)
return AdminRegenerateResponse(
scope="household", definition_id=None, child_id=None,
count=count,
)
if body.scope == "definition":
definition_id = body.definition_id
# The request model's scope/id consistency validator already
# guarantees the id for this scope; this only narrows the type.
assert definition_id is not None
definition = await core_dao_rules.QuestDefinitionsDao(
database
).get(definition_id)
if definition is None:
raise HTTPException(
status_code=404, detail=_QUEST_DEFINITION_NOT_FOUND_DETAIL
)
try:
count = await core_materialize.regenerate_for_definition(
database, definition_id, today=today
)
except ValueError as error:
_raise_regenerate_error(error)
return AdminRegenerateResponse(
scope="definition",
definition_id=definition_id,
child_id=None,
count=count,
)
child_id = body.child_id
# Same model guarantee as the definition scope above.
assert child_id is not None
child = await core_dao_children.ChildrenDao(database).get(child_id)
if child is None:
raise HTTPException(status_code=404, detail=_CHILD_NOT_FOUND_DETAIL)
try:
count = await core_materialize.regenerate_for_child(
database, child_id, today=today
)
except ValueError as error:
_raise_regenerate_error(error)
return AdminRegenerateResponse(
scope="child", definition_id=None, child_id=child_id, count=count
)
def _history_row_response(row: core_history.HistoryRow) -> AdminHistoryRow:
"""Serialize one core HistoryRow into the documented payload."""
return AdminHistoryRow(
occurred_at=row.occurred_at,
event_type=row.event_type,
child_name=row.child_name,
child_id=row.child_id,
quest_title=row.quest_title,
instance_id=row.instance_id,
window=row.window,
due_date=row.due_date,
due_time=row.due_time,
actor=row.actor,
was_on_time=row.was_on_time,
)
def _settings_response(
settings: core_settings.NestQuestSettings,
) -> AdminSettingsResponse:
"""Serialize one core NestQuestSettings into the documented payload.
Explicit field-by-field (not ``dataclasses.asdict``) so the payload
stays pinned to the ten documented fields even if the core
dataclass later grows another one.
"""
return AdminSettingsResponse(
horizon_days=settings.horizon_days,
day_rollover_time=settings.day_rollover_time,
notify_target=settings.notify_target,
morning_summary_time=settings.morning_summary_time,
afternoon_reminder_time=settings.afternoon_reminder_time,
end_of_day_report_time=settings.end_of_day_report_time,
morning_summary_enabled=settings.morning_summary_enabled,
afternoon_reminder_enabled=settings.afternoon_reminder_enabled,
end_of_day_report_enabled=settings.end_of_day_report_enabled,
celebration_enabled=settings.celebration_enabled,
)
@router.get(
"/history",
summary="Query the completion history for one filter and range",
response_model=AdminHistoryResponse,
)
async def admin_history_query(
request: Request,
start: str,
end: str,
filter: str = "all",
) -> AdminHistoryResponse:
"""Return the history rows for one filter over the closed range.
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: ``filter``
(default ``all``), ``start`` and ``end`` go straight to
:func:`nestquest_core.history.query_history` as STRINGS — the
filter policy (``all`` / ``reversals`` / ``missed``) and the
date-range policy (strict ``YYYY-MM-DD``, ``end >= start``) live in
the core, which raises a ``ValueError`` naming the offending field;
the handler only threads ONE clock read (:func:`_local_now`) as the
missed derivation's ``today`` and serializes the returned rows.
``start``/``end`` are required query parameters (missing is
FastAPI's 422 before any handler code runs); every core
``ValueError`` is 422 (:func:`_raise_history_error`).
"""
state: DatabaseState = request.app.state.db
today = _local_now().date()
try:
rows = await core_history.query_history(
state.database, filter, start, end, today=today
)
except ValueError as error:
_raise_history_error(error)
return AdminHistoryResponse(
filter=filter,
start=start,
end=end,
count=len(rows),
rows=[_history_row_response(row) for row in rows],
)
@router.get(
"/history.csv",
summary="Export the completion history as CSV",
response_class=Response,
)
async def admin_history_csv(
request: Request,
start: str,
end: str,
filter: str = "all",
) -> Response:
"""Export the SAME filtered history range as a CSV attachment.
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 over ONE
:func:`nestquest_core.history.export_history_csv` call — the same
query path the JSON route uses, so the CSV can never drift from the
query's filtered range — with the core's validation policy and the
:mod:`csv`-based serialization (stable documented header row, RFC
4180 quoting) living in the core. The answer is ``text/csv`` with
a ``Content-Disposition`` attachment filename naming the range;
the range is core-validated strict ``YYYY-MM-DD`` by the time the
filename is built, so it is injection-safe. Error mapping is the
JSON route's (:func:`_raise_history_error`).
"""
state: DatabaseState = request.app.state.db
today = _local_now().date()
try:
csv_text = await core_history.export_history_csv(
state.database, filter, start, end, today=today
)
except ValueError as error:
_raise_history_error(error)
filename = f"nestquest-history-{start}-to-{end}.csv"
return Response(
content=csv_text,
media_type="text/csv",
headers={
"Content-Disposition": f'attachment; filename="{filename}"'
},
)
@router.get(
"/settings",
summary="Read the effective settings",
response_model=AdminSettingsResponse,
)
async def admin_get_settings(request: Request) -> AdminSettingsResponse:
"""Return the current effective settings (all ten documented fields).
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
:func:`nestquest_core.settings_store.load_settings` call — the API
owns settings in the database (the integration's HA options remain
a temporary separate source until Feature 18 wires the client to
the API), and the stored-document decode, the per-field validation
and the defaults-for-absent-fields fallback all live in the core.
A corrupt stored document never fails the read: the core falls
back to the defaults (per field where possible).
"""
state: DatabaseState = request.app.state.db
settings = await core_settings_store.load_settings(state.database)
return _settings_response(settings)
@router.patch(
"/settings",
summary="Update the supplied settings fields",
response_model=AdminSettingsResponse,
)
async def admin_update_settings(
changes: dict[str, object], request: Request
) -> AdminSettingsResponse:
"""Update the supplied settings fields and return the new state.
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: the body — a
JSON object of settings field names to new values — goes STRAIGHT
to :func:`nestquest_core.settings_store.update_settings`, which
rejects an unknown field, resolves each supplied field through the
core settings validators (a JSON ``null`` resets its field to the
default) and merges onto the current effective settings only after
every value has validated — so a rejected update never changes the
stored settings. Every core ``ValueError`` is 422
(:func:`_raise_settings_error`); anything else re-raises.
"""
state: DatabaseState = request.app.state.db
try:
settings = await core_settings_store.update_settings(
state.database, changes
)
except ValueError as error:
_raise_settings_error(error)
return _settings_response(settings)
@router.post(
"/missed-sweep",
summary="Run the nightly missed-quest sweep now",
response_model=AdminMissedSweepResponse,
)
async def admin_run_missed_sweep(
request: Request,
) -> AdminMissedSweepResponse:
"""Run the missed-quest sweep and publish what it built.
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 over the SAME
HA-free sweep policy the integration's rollover listener runs
(:mod:`nestquest_core.sweep`): ONE timezone-aware clock read
(:func:`_local_now`) pins the API host's local ``today``, and ONE
:func:`nestquest_core.sweep.run_missed_sweep` call does everything
else — the past-due open-instance query, the documented payload
build and the watermark update (a same-day rerun is an empty
no-op, and a run after downtime sweeps the accumulated window
once). The route is read-only over the domain tables: it never
mutates an instance and never writes a completion event.
Each returned ``(event_type, payload)`` pair is published DIRECTLY
on the app's ONE transition publisher (``request.app.state.publisher``,
fan-out fire-and-forget, see api/events.py) — the exact payload the
sweep built, never re-fetched and never hand-built here, the same
publish-what-the-core-built shape api/scheduler.py uses. This is a
correctness rule, not a style preference: the core commits the
idempotency watermark BEFORE returning, so a re-fetch-based publish
loop opens a loss window — a concurrent regenerate deleting the
instance, or a transient database error, would raise uncaught (a
bare 500) AFTER the watermark advanced, and no future sweep would
ever re-examine that instance. Publishing the built payload leaves
no such window, and every event of one run keeps the ONE shared
``occurred_at`` stamp the sweep computed for it. The response
reports ``fired`` — the number of transitions published. Error
mapping (:func:`_raise_sweep_error`): every core ``ValueError`` is
422; anything else re-raises.
"""
state: DatabaseState = request.app.state.db
database = state.database
today = _local_now().date()
try:
events = await core_sweep.run_missed_sweep(database, today=today)
except ValueError as error:
_raise_sweep_error(error)
for event_type, payload in events:
request.app.state.publisher.publish(event_type, payload)
return AdminMissedSweepResponse(fired=len(events))
__all__ = ["router"]
+173
View File
@@ -0,0 +1,173 @@
"""Daily missed-sweep background scheduler for the API service.
The API service has no Home Assistant, so nobody re-fires the
integration's day-rollover listener for it: this module owns the API
plane's own background job. One :class:`MissedSweepScheduler` per app
(installed on ``app.state`` by the lifespan) sleeps until the
configured ``day_rollover_time`` — read fresh from the settings store
at the top of EVERY cycle, so an admin's settings change applies on
the next scheduling decision without a restart — then runs the
HA-free core's :func:`nestquest_core.sweep.run_missed_sweep` for the
API host's local date and publishes each returned
``nestquest_quest_missed`` event on the app's ONE transition publisher
(the same SSE stream every route publishes through).
All sweep policy lives in the core (the watermark, the
``[watermark, today)`` window, the no-completion-event rule): this
scheduler only decides WHEN. A same-night rerun — e.g. two workers,
or a settings change moving the rollover past itself — fires nothing:
the core's watermark makes the run an empty no-op.
Testability: ``clock`` (an aware-now callable) and ``sleep`` (an
awaitable delay) are injectable, so tests pin the time, fast-forward
through the wait, and observe the computed delays without waiting a
real day. Cancellation: :meth:`stop` cancels the loop task and waits
for it, so the lifespan's shutdown is clean and a mid-run sweep is
interrupted before the connection closes. A failed iteration (a
database hiccup, a rejected setting) is logged and retried after a
fixed backoff rather than silently killing the job.
"""
from __future__ import annotations
import asyncio
import datetime
import logging
from api.nestquest_core import core_settings_store, core_sweep
LOGGER = logging.getLogger(__name__)
#: Backoff after a failed scheduling cycle (settings read or sweep
#: error), so a persistent failure retries every minute instead of
#: spinning or dying.
_RETRY_BACKOFF_SECONDS = 60.0
def _delay_seconds(target: datetime.time, now: datetime.datetime) -> float:
"""Seconds from ``now`` until the next occurrence of ``target``.
``target`` is a plain ``HH:MM`` time of day: the next wall-clock
occurrence strictly AFTER ``now`` (an equal time schedules for
tomorrow, so the just-run sweep is never instantly re-run).
"""
scheduled = now.replace(
hour=target.hour, minute=target.minute, second=0, microsecond=0
)
if scheduled <= now:
scheduled += datetime.timedelta(days=1)
return (scheduled - now).total_seconds()
class MissedSweepScheduler:
"""Run the missed sweep once per day at ``day_rollover_time``."""
def __init__(
self,
database: object,
*,
publisher: object | None = None,
clock: object | None = None,
sleep: object | None = None,
) -> None:
"""Create the scheduler (not yet running).
``database`` is the app's live :class:`~api.database.NestQuestDatabase`.
``publisher`` is the app's ONE
:class:`~api.events.TransitionPublisher` (publishing is skipped
when ``None``). ``clock`` defaults to the API host's local
aware now (``datetime.datetime.now().astimezone()``) — the same
single-clock discipline the routes' ``_local_now`` uses;
``sleep`` defaults to :func:`asyncio.sleep`.
"""
self._database = database
self._publisher = publisher
self._clock = clock if clock is not None else (
lambda: datetime.datetime.now().astimezone()
)
self._sleep = sleep if sleep is not None else asyncio.sleep
self._task: asyncio.Task | None = None
def start(self) -> None:
"""Start the background loop (idempotent: a running one stays)."""
if self._task is None or self._task.done():
self._task = asyncio.create_task(
self._run_forever(), name="nestquest-missed-sweep"
)
async def stop(self) -> None:
"""Cancel the loop and wait for it to finish (safe if never started)."""
if self._task is not None:
self._task.cancel()
try:
await self._task
except asyncio.CancelledError:
pass
self._task = None
@property
def running(self) -> bool:
"""Whether the background loop task is alive."""
return self._task is not None and not self._task.done()
async def _run_forever(self) -> None:
"""Schedule → sweep → repeat, forever (until cancelled)."""
while True:
try:
settings = await core_settings_store.load_settings(
self._database
)
rollover = datetime.time.fromisoformat(
settings.day_rollover_time
)
except Exception:
# Not silent: the failure is logged and retried after
# the backoff, so one bad settings document or a
# transient database error never kills the scheduler.
LOGGER.exception(
"NestQuest missed-sweep scheduler failed to resolve "
"the day rollover time; retrying in %gs",
_RETRY_BACKOFF_SECONDS,
)
await self._sleep(_RETRY_BACKOFF_SECONDS)
continue
delay = _delay_seconds(rollover, self._clock())
LOGGER.info(
"NestQuest missed sweep scheduled for %s (%.0fs from now)",
settings.day_rollover_time,
delay,
)
await self._sleep(delay)
try:
await self._run_once(self._clock().date())
except Exception:
# Not silent either: the sweep error is logged and the
# loop backs off before the next cycle, whose watermark
# re-read makes an already-swept day a no-op.
LOGGER.exception(
"NestQuest scheduled missed sweep failed; retrying in "
"%gs",
_RETRY_BACKOFF_SECONDS,
)
await self._sleep(_RETRY_BACKOFF_SECONDS)
async def _run_once(self, today: datetime.date) -> None:
"""Run ONE sweep for ``today`` and publish what it built.
A failure propagates to :meth:`_run_forever`, which logs it and
backs off before the next cycle; the watermark makes a re-run
of an already-swept day announce nothing twice.
"""
events = await core_sweep.run_missed_sweep(
self._database, today=today
)
for event_type, payload in events:
if self._publisher is not None:
self._publisher.publish(event_type, payload)
LOGGER.info(
"NestQuest scheduled missed sweep for %s published %d event(s)",
today.isoformat(),
len(events),
)
__all__ = ["MissedSweepScheduler"]
+405
View File
@@ -0,0 +1,405 @@
"""Read-only history layer for the admin plane (Admin spec §5, task 2d2dda53).
Assembles the History screen's event rows over the append-only
completion log: a query over a CLOSED date range ``[start_date,
end_date]`` for one of three filters —
- ``all`` — every completion event (completed AND uncompleted) whose
instance is due in the range;
- ``reversals`` — the uncompleted events only;
- ``missed`` — the DERIVED missed instances in the range.
Range semantics (Feature 13's rule, kept): the range scopes by the
INSTANCE's ``due_date`` — the day the task was owed — not the event's
wall-clock ``occurred_at``, because history reporting asks "what
happened to the tasks due that day". ``missed`` is derived, never
stored: rows come from :func:`~.completion.list_missed_for_child`
(past-due open instances, the same ``_derived_state`` rule the
coordinator uses), and nothing is ever appended for them.
Row shape: the Admin spec §5 event-row fields, fixed and documented in
:data:`CSV_COLUMNS` — ``occurred_at``, ``event_type``,
``child_name``, ``child_id``, ``quest_title``, ``instance_id``,
``window``, ``due_date``, ``due_time``, ``actor``, ``was_on_time``.
Derived missed rows carry ``event_type`` ``'missed'`` (never a stored
event_type), no ``occurred_at`` (there is no event), ``was_on_time``
``None`` and the spec's ``nightly sweep`` actor. A stored event's
``actor`` follows the spec's meta shape: the ``actor_user_id`` string
for a user event (``joshua``), ``panel (Child)`` naming the tapped
profile for a panel event.
Validation policy (module-level, shared by both the JSON query and the
CSV export): the filter must be one of :data:`FILTERS` and the range
must be two strict ``YYYY-MM-DD`` dates with ``end >= start``;
violations raise ``ValueError`` naming the offending field
(``filter`` / ``start`` / ``end``) BEFORE any read. The whole
household read runs under the task-reentrant connection lock so every
row of one answer is one snapshot — the same discipline
:mod:`.completion` gives its read-decide-write paths.
:func:`rows_to_csv` serializes rows with the :mod:`csv` module (proper
quoting, one stable header row) and :func:`export_history_csv` runs
query + serialization as ONE call, so the CSV export can never drift
from the JSON query's filtered range. No permission check lives here
(that is the admin router's single ``require_admin`` dependency).
"""
from __future__ import annotations
import csv
import datetime
import io
from dataclasses import dataclass
from .completion import _reentrant_connection_lock, list_missed_for_child
from .dao_children import ChildRecord, ChildrenDao
from .dao_instances import (
EVENT_UNCOMPLETED,
CompletionEventRecord,
CompletionEventsDao,
QuestInstanceRecord,
QuestInstancesDao,
)
from .dao_rules import QuestDefinitionsDao, _validate_date
from .db import NestQuestDatabase
#: The ``all`` filter: every completion event in the range.
FILTER_ALL = "all"
#: The ``reversals`` filter: uncompleted events only.
FILTER_REVERSALS = "reversals"
#: The ``missed`` filter: derived missed instances in the range.
FILTER_MISSED = "missed"
#: The filters the query accepts, in the Admin spec's chip order.
FILTERS = (FILTER_ALL, FILTER_REVERSALS, FILTER_MISSED)
#: The ``event_type`` a derived missed row reports. ``missed`` is a
#: DERIVED state, never a stored event_type (Feature 08).
EVENT_MISSED = "missed"
#: The actor a derived missed row reports (Admin spec §5 meta:
#: ``missed · nightly sweep``).
ACTOR_NIGHTLY_SWEEP = "nightly sweep"
#: The CSV column set, in header order — the Admin spec §5 event-row
#: fields. The header row is exactly this tuple, always, so a consumer
#: can key on column names.
CSV_COLUMNS = (
"occurred_at",
"event_type",
"child_name",
"child_id",
"quest_title",
"instance_id",
"window",
"due_date",
"due_time",
"actor",
"was_on_time",
)
@dataclass(frozen=True)
class HistoryRow:
"""One history row: a stored event or a derived missed instance.
The fields are :data:`CSV_COLUMNS` in order. For a derived missed
row ``occurred_at`` is ``None`` (no event exists) and ``was_on_time``
is ``None``; for a stored event ``event_type`` is
``'completed'``/``'uncompleted'`` and the fields come from the event
joined with its instance, child and definition.
"""
occurred_at: str | None
event_type: str
child_name: str
child_id: int
quest_title: str
instance_id: int
window: str
due_date: str
due_time: str | None
actor: str
was_on_time: bool | None
def _validate_filter(filter_name: str) -> str:
"""Return the filter, or raise ValueError naming the field."""
if filter_name not in FILTERS:
raise ValueError(
f"filter must be one of {', '.join(repr(f) for f in FILTERS)}, "
f"got {filter_name!r}"
)
return filter_name
def _validate_range(start_date: str, end_date: str) -> tuple[str, str]:
"""Validate the closed range and return it, else raise naming the field.
Both bounds must be strict ``YYYY-MM-DD`` dates (the shared
:func:`~.dao_rules._validate_date`, so non-padded or re-ordered
shapes are rejected and a non-string raises naming the field) and
``end`` must be on or after ``start``.
"""
_validate_date(start_date, "start")
_validate_date(end_date, "end")
if end_date < start_date:
raise ValueError(
f"end must be on or after start, got {end_date!r} < "
f"{start_date!r}"
)
return start_date, end_date
def _resolve_today(today: datetime.date | None) -> datetime.date:
"""Return the threaded HA-local date, or the host calendar date."""
if today is None:
return datetime.date.today()
return today
def _event_actor(
event: CompletionEventRecord, actor_child: ChildRecord | None
) -> str:
"""Return the spec-meta actor string for one stored event.
A user event reports the ``actor_user_id`` string (the spec's
``uncompleted · joshua (admin)`` shape); a panel event reports
``panel (Child)`` naming the tapped profile. The tapped child is
never hard-deleted, so the lookup below is a belt-and-braces
fallback to the stored id, never a silent rewrite of policy.
"""
if event.actor_source == "user":
return event.actor_user_id or ""
if actor_child is not None:
return f"panel ({actor_child.display_name})"
return f"panel ({event.actor_child_id})"
def _event_row(
event: CompletionEventRecord,
instance: QuestInstanceRecord,
child: ChildRecord,
quest_title: str,
actor_child: ChildRecord | None,
) -> HistoryRow:
"""Join one stored event with its instance, child and definition."""
return HistoryRow(
occurred_at=event.occurred_at,
event_type=event.event_type,
child_name=child.display_name,
child_id=event.child_id,
quest_title=quest_title,
instance_id=event.instance_id,
window=instance.window,
due_date=instance.due_date,
due_time=instance.due_time,
actor=_event_actor(event, actor_child),
was_on_time=event.was_on_time,
)
def _missed_row(
instance: QuestInstanceRecord,
child: ChildRecord,
quest_title: str,
) -> HistoryRow:
"""Build the derived row for one past-due open instance."""
return HistoryRow(
occurred_at=None,
event_type=EVENT_MISSED,
child_name=child.display_name,
child_id=instance.child_id,
quest_title=quest_title,
instance_id=instance.id,
window=instance.window,
due_date=instance.due_date,
due_time=instance.due_time,
actor=ACTOR_NIGHTLY_SWEEP,
was_on_time=None,
)
def _csv_values(row: HistoryRow) -> list[str]:
"""Return one CSV row's cell values in :data:`CSV_COLUMNS` order.
``None`` cells serialize as empty (a missed row has no
``occurred_at``); ``was_on_time`` serializes as ``true``/``false``
when the event carries the flag and empty when it does not.
"""
return [
row.occurred_at or "",
row.event_type,
row.child_name,
str(row.child_id),
row.quest_title,
str(row.instance_id),
row.window,
row.due_date,
row.due_time or "",
row.actor,
"" if row.was_on_time is None else str(row.was_on_time).lower(),
]
def rows_to_csv(rows: list[HistoryRow]) -> str:
"""Serialize history rows to CSV text with the documented header.
The FIRST row is always :data:`CSV_COLUMNS`, so the header is
stable and deterministic; the :mod:`csv` module handles quoting
(titles or names containing commas or quotes are quoted per RFC
4180) and CRLF line endings.
"""
buffer = io.StringIO()
writer = csv.writer(buffer)
writer.writerow(CSV_COLUMNS)
for row in rows:
writer.writerow(_csv_values(row))
return buffer.getvalue()
async def query_history(
database: NestQuestDatabase,
filter_name: str,
start_date: str,
end_date: str,
*,
today: datetime.date | None = None,
) -> list[HistoryRow]:
"""Return the history rows for ``filter_name`` over the closed range.
Validates the filter and the strict ``YYYY-MM-DD`` range (raising
``ValueError`` naming ``filter`` / ``start`` / ``end``) BEFORE any
read, then builds the rows:
- ``all`` / ``reversals``: every child's events whose instances are
due in the range (:meth:`~.dao_instances.CompletionEventsDao.list_by_child_and_date_range`,
the Feature 13 due-date scope), ``reversals`` keeping only the
``uncompleted`` rows.
- ``missed``: :func:`~.completion.list_missed_for_child` per child —
the SAME derived-state rule the coordinator uses, never a new one,
and never a stored event.
Rows are ordered deterministically by ``due_date``, then instance,
then event id (the per-instance append order, so a completion and
its reversal keep their order). ``today`` pins the HA-local date
the missed derivation compares due dates against (the host calendar
date when omitted). The whole read runs under the task-reentrant
connection lock — :func:`list_missed_for_child` re-enters it — so
one answer is one snapshot.
"""
filter_name = _validate_filter(filter_name)
start_date, end_date = _validate_range(start_date, end_date)
today_date = _resolve_today(today)
keyed: list[tuple[tuple[str, int, int], HistoryRow]] = []
async with _reentrant_connection_lock(database):
children = await ChildrenDao(database).list_all()
children_by_id = {child.id: child for child in children}
instances = QuestInstancesDao(database)
definitions = QuestDefinitionsDao(database)
events = CompletionEventsDao(database)
titles: dict[int, str] = {}
if filter_name == FILTER_MISSED:
for child in children:
for instance in await list_missed_for_child(
database, child.id, start_date, end_date, today=today_date
):
if instance.definition_id not in titles:
definition = await definitions.get(
instance.definition_id
)
assert definition is not None, (
"a materialized instance's definition exists"
)
titles[instance.definition_id] = definition.title
keyed.append(
(
(instance.due_date, instance.id, 0),
_missed_row(
instance,
child,
titles[instance.definition_id],
),
)
)
keyed.sort(key=lambda pair: pair[0])
return [row for _, row in keyed]
for child in children:
for event in await events.list_by_child_and_date_range(
child.id, start_date, end_date
):
if filter_name == FILTER_REVERSALS and (
event.event_type != EVENT_UNCOMPLETED
):
continue
instance = await instances.get_by_id(event.instance_id)
assert instance is not None, (
"an event's instance is never deleted (append-only "
"history keeps its references)"
)
actor_child = (
children_by_id.get(event.actor_child_id)
if event.actor_child_id is not None
else None
)
if instance.definition_id not in titles:
definition = await definitions.get(instance.definition_id)
assert definition is not None, (
"a materialized instance's definition exists"
)
titles[instance.definition_id] = definition.title
keyed.append(
(
(instance.due_date, instance.id, event.id),
_event_row(
event,
instance,
child,
titles[instance.definition_id],
actor_child,
),
)
)
keyed.sort(key=lambda pair: pair[0])
return [row for _, row in keyed]
async def export_history_csv(
database: NestQuestDatabase,
filter_name: str,
start_date: str,
end_date: str,
*,
today: datetime.date | None = None,
) -> str:
"""Return the filtered history as CSV text, header row included.
ONE query + serialization call so the CSV export carries exactly
the JSON query's filtered range and order — the same
:func:`query_history` validation, rows and :func:`rows_to_csv`
serializer, never a second implementation.
"""
rows = await query_history(
database, filter_name, start_date, end_date, today=today
)
return rows_to_csv(rows)
__all__ = [
"ACTOR_NIGHTLY_SWEEP",
"CSV_COLUMNS",
"EVENT_MISSED",
"EVENT_UNCOMPLETED",
"FILTERS",
"FILTER_ALL",
"FILTER_MISSED",
"FILTER_REVERSALS",
"HistoryRow",
"export_history_csv",
"query_history",
"rows_to_csv",
]
@@ -0,0 +1,210 @@
"""Durable settings store for the NestQuest API service.
The API service has NO Home Assistant config entry, so it cannot build
:class:`~.settings.NestQuestSettings` from ``entry.options`` the way the
integration does. This module gives the API its own durable settings
source: a JSON document kept in the EXISTING ``nestquest_meta_state``
key/value table (:class:`~.dao_meta.MetaStateDao`) under the well-known
key :data:`SETTINGS_META_KEY` — no schema change, no new table —
validated through the SAME field table the settings object is built
from (:data:`~.settings._FIELDS`, the table driving
``from_options``/``from_options_resilient``). No validation rule is
re-implemented here.
OWNERSHIP: the API now owns settings in the database. The
integration's Home-Assistant config-entry options remain a temporary,
SEPARATE source until Feature 18 converts the integration into a thin
API client; the two are deliberately NOT unified here, and neither
side reads the other's values.
STORED SERIALIZATION — the document at ``SETTINGS_META_KEY`` is a JSON
object holding exactly the TEN settings fields, keyed by field name
(which equals the ``CONF_*`` option key for every field), complete
after the first update:
- ``horizon_days`` — int >= 1,
- ``day_rollover_time``, ``morning_summary_time``,
``afternoon_reminder_time``, ``end_of_day_report_time`` — strict
24-hour ``HH:MM`` strings,
- ``notify_target`` — a string (a padded value is stripped by the
core's normalizer),
- ``morning_summary_enabled``, ``afternoon_reminder_enabled``,
``end_of_day_report_enabled``, ``celebration_enabled`` — real
booleans.
Only values whose core validators ACCEPTED are ever written, so a
stored document always round-trips through the core's option builders.
An absent key (a fresh database) resolves to the all-defaults
settings, and a corrupt document degrades to defaults or per-field
defaults rather than crashing the service (see :func:`load_settings`).
"""
from __future__ import annotations
import asyncio
import json
from collections.abc import Mapping
from typing import Any
from .const import LOGGER
from .dao_meta import MetaStateDao
from .db import NestQuestDatabase
from .settings import NestQuestSettings, _FIELDS
#: The well-known ``nestquest_meta_state`` key holding the settings
#: document. One key, one JSON object — the same table the missed
#: sweep's watermark already uses, so no migration is needed.
SETTINGS_META_KEY = "settings"
#: The per-field resolution table keyed by field name: each row is the
#: ``(default, validator)`` pair the settings module's ``_FIELDS`` table
#: carries for that field. ``update_settings`` resolves each PROVIDED
#: field through its row's validator — the SAME core validators
#: ``from_options`` applies — and never re-implements one here.
#: (Field names equal the ``CONF_*`` option keys, so one dict serves
#: both the API's field names and the stored document's keys.)
_RESOLUTIONS: dict[str, tuple[Any, Any]] = {
key: (default, validator) for _field, key, default, validator in _FIELDS
}
#: The known field names, in the settings module's field order — the
#: allowed keys of an update body and the stored document's key set.
_FIELD_NAMES: tuple[str, ...] = tuple(
key for _field, key, _default, _validator in _FIELDS
)
#: One asyncio.Lock per (connection wrapper, running loop), the missed
#: sweep's pattern: the whole stored-document read → merge → write span
#: is one critical section, so two concurrent updates queue instead of
#: one losing the other's change — the loser re-reads the document
#: INSIDE the lock and merges onto the winner's settings.
_SETTINGS_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
def _settings_lock(database: NestQuestDatabase) -> asyncio.Lock:
"""Return the settings lock bound to ``database``'s running loop."""
loop = asyncio.get_running_loop()
key = (id(database), id(loop))
lock = _SETTINGS_LOCKS.get(key)
if lock is None:
lock = asyncio.Lock()
_SETTINGS_LOCKS[key] = lock
return lock
async def _load_document(database: NestQuestDatabase) -> dict[str, Any]:
"""Read and decode the stored settings document (empty when unusable).
An absent key — a fresh database that never updated settings — and
an unparseable value (invalid JSON, or JSON that is not an object)
return an empty mapping, which the option builders resolve to the
all-defaults settings; the corrupt cases log a warning first. A
parseable object is returned as stored (its per-field validity is
the builder's concern, not this reader's).
"""
raw = await MetaStateDao(database).get(SETTINGS_META_KEY)
if raw is None:
return {}
try:
document = json.loads(raw)
except ValueError:
LOGGER.warning(
"NestQuest stored settings (meta_state key %r) are not valid "
"JSON; falling back to the default settings",
SETTINGS_META_KEY,
)
return {}
if not isinstance(document, dict):
LOGGER.warning(
"NestQuest stored settings (meta_state key %r) are not a JSON "
"object; falling back to the default settings",
SETTINGS_META_KEY,
)
return {}
return document
def _named(field: str, error: ValueError) -> str:
"""Return a field-named message for a rejected validator value.
The horizon, time and notify-target validators already name their
field in the message; the boolean validator's message is generic
(``expected a boolean, got ...``), so it gets the field prefixed.
Every rejected update's error therefore names its field either way.
"""
message = str(error)
if message.startswith(f"{field} "):
return message
return f"{field}: {message}"
async def load_settings(database: NestQuestDatabase) -> NestQuestSettings:
"""Return the current effective settings stored for ``database``.
Reads :data:`SETTINGS_META_KEY` and resolves the stored document
through the core settings validators, with each absent field
falling back to its default. An absent key and an unparseable
document fall back to the all-defaults settings (via
:func:`_load_document`), and a parseable document with a single
invalid FIELD falls back PER FIELD — through
:meth:`NestQuestSettings.from_options_resilient`, the core's own
resilient builder — so a corrupt sibling never resets the stored
values around it. This never raises: the service must keep
serving reads even over a corrupt store.
"""
document = await _load_document(database)
return NestQuestSettings.from_options_resilient(document)
async def update_settings(
database: NestQuestDatabase, changes: Mapping[str, Any]
) -> NestQuestSettings:
"""Validate ``changes`` and persist them as the effective settings.
``changes`` maps settings field names to new values. EVERY name is
checked against the known field table BEFORE anything is validated
or written: an unknown name raises a ``ValueError`` naming it (and
the known names), leaving the stored settings untouched. Each
PROVIDED field is then resolved from_options-style — its
:data:`~.settings._FIELDS` validator applied, with a JSON ``null``
mapping to the field's default (the settings module's uniform
None-as-absent rule) — and an invalid value raises a ``ValueError``
naming the field, also BEFORE anything is written, so a rejected
update can never leave a half-applied document behind.
The validated changes are merged onto the current effective
settings and the merged (complete, ten-field) document is persisted
in one atomic upsert; the updated settings are returned. The whole
read → merge → write span runs inside the per-database settings
lock, so concurrent updates queue instead of interleaving.
"""
if not isinstance(changes, Mapping):
raise ValueError(
"settings changes must be a JSON object mapping field names "
"to values"
)
unknown = sorted(
str(name) for name in changes if name not in _RESOLUTIONS
)
if unknown:
listed = ", ".join(repr(name) for name in unknown)
known = ", ".join(_FIELD_NAMES)
raise ValueError(
f"unknown settings field(s): {listed}; known fields: {known}"
)
async with _settings_lock(database):
current = await load_settings(database)
document: dict[str, Any] = {
field: getattr(current, field) for field in _FIELD_NAMES
}
for name, raw_value in changes.items():
default, validator = _RESOLUTIONS[name]
try:
document[name] = validator(
default if raw_value is None else raw_value
)
except ValueError as error:
raise ValueError(_named(name, error)) from error
await MetaStateDao(database).set(
SETTINGS_META_KEY, json.dumps(document, sort_keys=True)
)
return NestQuestSettings(**document)
+143
View File
@@ -0,0 +1,143 @@
"""The nightly missed-quest sweep, Home-Assistant-free (Feature 11).
This is the HA-free heart of the missed sweep, shared by the two
planes: the Home Assistant integration's :mod:`~..sweep` shim (which
fires the returned events on ``hass.bus``) and the API service (which
publishes them on the SSE stream and can also trigger the run on
demand). It owns the ONE sweep policy:
At the configured day-rollover time (or on an explicit trigger) this
job finds every instance whose ``due_date`` is before the
household-local date, that has NO completion event, and that falls
after the persisted sweep watermark — and RETURNS one
``nestquest_quest_missed`` event per instance with the documented
payload. Like every core module, it never touches the Home Assistant
event bus and never mutates an instance: ``missed`` is derived state,
never an event_type (Feature 08's contract), so the sweep is read-only
over the domain tables.
Idempotency is the watermark: after a run, ``nestquest_meta_state``
records the household-local date the sweep ran, and the next run only
sweeps instances due in ``[watermark, today)``. A second run the same
night therefore fires nothing new, and a run after days of downtime
sweeps exactly the accumulated window once — no per-instance journal,
no re-announcing old misses. An instance swept as missed that is
later completed simply gains its completion event; history keeps both
rows (append-only), and the sweep's later runs skip it via the
no-completion-event rule.
The whole watermark read → query → build → write span runs inside the
per-database sweep lock (one lock per connection wrapper and running
loop, the settings store's pattern): an overlapping startup sweep and
a rollover sweep queue instead of interleaving, and the loser re-reads
the watermark INSIDE the lock — sees the winner's date — and announces
nothing twice.
The caller supplies ``today`` as a plain :class:`datetime.date`: the
household-local calendar date, read from ``hass.config.time_zone`` by
the integration or from the API host's local clock by the API. This
module never reads a timezone database or config of its own.
"""
from __future__ import annotations
import asyncio
import datetime
import logging
from .const import EVENT_QUEST_MISSED
from .dao_meta import MetaStateDao
from .dao_instances import QuestInstancesDao
from .db import NestQuestDatabase
# Intentional intra-package reuse of the shared payload builder: the
# missed event's payload is the documented §3 instance payload, and
# building it through the ONE builder keeps this sweep's payloads from
# drifting from the completion/uncompleted builders' shape.
from .events import _instance_payload
LOGGER = logging.getLogger(__name__)
#: meta_state key holding the household-local date the sweep last ran.
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
#: One asyncio.Lock per (connection wrapper, running loop), mirroring
#: the settings store's pattern: the whole watermark read → query →
#: build → write span is one critical section, so an overlapping
#: startup sweep and rollover sweep queue instead of double-announcing.
#: The loser re-reads the watermark INSIDE the lock, sees the winner's
#: date, and fires nothing.
_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
"""Return the sweep lock bound to ``database``'s running loop."""
loop = asyncio.get_running_loop()
key = (id(database), id(loop))
lock = _SWEEP_LOCKS.get(key)
if lock is None:
lock = asyncio.Lock()
_SWEEP_LOCKS[key] = lock
return lock
#: Strict UTC ISO-8601 timestamp (the DAO layers' policy): one shape,
#: explicit offset, second precision.
_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
async def run_missed_sweep(
database: NestQuestDatabase,
*,
today: datetime.date,
) -> list[tuple[str, dict]]:
"""Build one missed event per unswept past-due open instance.
``today`` is the household-local calendar date the sweep runs for
(the caller owns the clock: the integration reads it from
``hass.config.time_zone``, the API from its host's local time).
Returns the ``(event_type, payload)`` tuples to announce, EMPTY
when the watermark already covers ``today`` (a same-night rerun) —
the caller decides how to publish them and announces nothing on
its own.
The whole watermark read → query → build → write span runs inside
the per-database sweep lock: an overlapping startup sweep and a
rollover sweep queue instead of interleaving, and the loser
re-reads the watermark inside the lock — sees the winner's date —
and announces nothing twice. Every event in one run shares one
``occurred_at`` stamp (one run, one announcement moment), and the
payload comes from the shared
:func:`~.events._instance_payload` builder so the shape can never
drift from the documented §3 contract.
"""
today_iso = today.isoformat()
dao = MetaStateDao(database)
async with _sweep_lock(database):
watermark = await dao.get(SWEEP_WATERMARK_KEY)
if watermark == today_iso:
LOGGER.debug(
"NestQuest missed sweep already ran for %s", today_iso
)
return []
instances_dao = QuestInstancesDao(database)
rows = await instances_dao.list_missed_unswept(today_iso, watermark)
now_stamp = datetime.datetime.now(datetime.timezone.utc).strftime(
_UTC_TIMESTAMP_FORMAT
)
events: list[tuple[str, dict]] = []
for instance in rows:
events.append(
(
EVENT_QUEST_MISSED,
await _instance_payload(
database, instance.id, now_stamp
),
)
)
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
LOGGER.info(
"NestQuest missed sweep for %s built %d event(s)",
today_iso,
len(events),
)
return events
+36 -96
View File
@@ -1,70 +1,40 @@
"""The nightly missed-quest sweep (Feature 11).
"""The nightly missed-quest sweep (Feature 11) — the Home Assistant shim.
At the configured day-rollover time (the same HA-local listener that
re-materializes the rolling horizon) this job finds every instance
whose ``due_date`` is before the HA-local date, that has NO completion
event, and that falls after the persisted sweep watermark — and fires
ONE ``nestquest_quest_missed`` event per instance with the documented
payload. The sweep is read-only over the domain tables: it never
mutates an instance and never writes a completion event (``missed``
is derived state, never an event_type — Feature 08's contract).
The sweep's policy is the HA-free core's :mod:`~.core.sweep`: the
watermark read → ``list_missed_unswept`` query → payload build →
watermark span, all inside the per-database sweep lock, shared with
the API plane. This module is the integration's thin adapter over it:
at the configured day-rollover time (the same HA-local listener that
re-materializes the rolling horizon) and once at setup it calls the
core's :func:`~.core.sweep.run_missed_sweep` and fires each returned
event on ``hass.bus`` — ONE ``nestquest_quest_missed`` per past-due
open instance with the documented payload. The core never touches
Home Assistant; the firing lives here.
Idempotency is the watermark: after a run, ``nestquest_meta_state``
records the HA-local date the sweep ran, and the next run only
sweeps instances due in ``[watermark, today)``. A second run the
same night therefore fires nothing new, and a run after days of
downtime sweeps exactly the accumulated window once — no
per-instance journal, no re-announcing old misses. An instance swept
as missed that is later completed simply gains its completion event;
history keeps both rows (append-only), and the sweep's later runs
skip it via the no-completion-event rule.
Idempotency is the core's watermark: after a run, ``nestquest_meta_state``
records the HA-local date the sweep ran, and the next run only sweeps
instances due in ``[watermark, today)`` — a second run the same night
fires nothing new, and a run after days of downtime sweeps exactly the
accumulated window once. An instance swept as missed that is later
completed simply gains its completion event; the sweep is read-only
over the domain tables (``missed`` is derived state, never an
event_type — Feature 08's contract).
"""
from __future__ import annotations
import asyncio
import datetime
import logging
from zoneinfo import ZoneInfo
from .const import EVENT_QUEST_MISSED
from .dao_children import ChildrenDao
from .dao_instances import QuestInstancesDao
from .dao_meta import MetaStateDao
from .dao_rules import QuestDefinitionsDao
from .core.sweep import (
SWEEP_WATERMARK_KEY,
run_missed_sweep as _run_core_missed_sweep,
)
from .db import NestQuestDatabase
LOGGER = logging.getLogger(__name__)
#: meta_state key holding the HA-local date the sweep last ran.
SWEEP_WATERMARK_KEY = "missed_sweep_last_run_date"
#: One asyncio.Lock per (connection wrapper, running loop), mirroring
#: the migration runner's pattern: the whole watermark read → query →
#: fire → write span is one critical section, so an overlapping startup
#: sweep and rollover sweep queue instead of double-announcing. The
#: loser re-reads the watermark INSIDE the lock, sees the winner's
#: date, and fires nothing.
_SWEEP_LOCKS: dict[tuple[int, int], asyncio.Lock] = {}
def _sweep_lock(database: NestQuestDatabase) -> asyncio.Lock:
"""Return the sweep lock bound to ``database``'s running loop."""
loop = asyncio.get_running_loop()
key = (id(database), id(loop))
lock = _SWEEP_LOCKS.get(key)
if lock is None:
lock = asyncio.Lock()
_SWEEP_LOCKS[key] = lock
return lock
#: UTC ISO-8601 timestamp policy shared with the DAO layers.
_UTC_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S+00:00"
def _now_stamp() -> str:
return datetime.datetime.now(datetime.timezone.utc).strftime(
_UTC_TIMESTAMP_FORMAT
)
__all__ = ["SWEEP_WATERMARK_KEY", "run_missed_sweep"]
async def run_missed_sweep(
@@ -79,54 +49,24 @@ async def run_missed_sweep(
materialization and completion layers); when omitted it is read
from ``hass.config.time_zone``. Returns the number of events
fired. Fires nothing and returns 0 when the watermark already
covers ``today`` (a same-night rerun).
The whole watermark read → query → fire → write span runs inside
the per-database sweep lock: an overlapping startup sweep and a
rollover sweep queue instead of interleaving, and the loser
re-reads the watermark inside the lock — sees the winner's date —
and announces nothing twice.
covers ``today`` (a same-night rerun) — the watermark read, the
query, the payload build and the watermark write all live in the
core's locked
:func:`~.core.sweep.run_missed_sweep`; this shim only fires each
returned event on the bus and counts them.
"""
today_date = (
today
if today is not None
else datetime.datetime.now(ZoneInfo(hass.config.time_zone)).date()
)
today_iso = today_date.isoformat()
dao = MetaStateDao(database)
async with _sweep_lock(database):
watermark = await dao.get(SWEEP_WATERMARK_KEY)
if watermark == today_iso:
LOGGER.debug(
"NestQuest missed sweep already ran for %s", today_iso
)
return 0
instances_dao = QuestInstancesDao(database)
rows = await instances_dao.list_missed_unswept(today_iso, watermark)
children_dao = ChildrenDao(database)
definitions_dao = QuestDefinitionsDao(database)
now_stamp = _now_stamp()
fired = 0
for instance in rows:
child = await children_dao.get(instance.child_id)
definition = await definitions_dao.get(instance.definition_id)
hass.bus.async_fire(
EVENT_QUEST_MISSED,
{
"child_id": instance.child_id,
"child_name": child.display_name if child else None,
"instance_id": instance.id,
"quest_title": definition.title if definition else None,
"window": instance.window,
"due_date": instance.due_date,
"due_time": instance.due_time,
"occurred_at": now_stamp,
},
)
fired += 1
await dao.set(SWEEP_WATERMARK_KEY, today_iso)
events = await _run_core_missed_sweep(database, today=today_date)
for event_type, payload in events:
hass.bus.async_fire(event_type, payload)
fired = len(events)
LOGGER.info(
"NestQuest missed sweep for %s fired %d event(s)", today_iso, fired
"NestQuest missed sweep for %s fired %d event(s)",
today_date.isoformat(),
fired,
)
return fired
+547
View File
@@ -0,0 +1,547 @@
"""Admin-plane history route tests (task 2d2dda53).
These tests exercise the admin plane's history query and CSV export
routes 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 — signature, issuer, audience,
expiry and the nestquest-admins group — 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.*``; the
two-copy caveat in api/nestquest_core.py), and the reversal the
history tests read is appended through that same copy's
:func:`nestquest_core.completion.uncomplete_instance` — so the rows
the routes report are the rows the core's own write paths produced.
The seeded household (clock pinned to noon, the route's ``_local_now``
pinned to the same instant so nothing straddles midnight):
- Bo's ``Brush teeth`` TODAY instance: a ``completed`` event (panel
tap, on time) plus the appended ``uncompleted`` reversal — the
completion and the reversal the ``all`` and ``reversals`` filters
report.
- Ada's ``Pack bag`` and ``Stale chore`` YESTERDAY instances and Bo's
``Brush teeth`` YESTERDAY instance: open, past due, no events — the
DERIVED missed rows the ``missed`` filter reports.
Proven per the done-condition, everything read back through the API:
- ``GET /api/v1/admin/history`` returns the expected rows for EACH
filter — ``all`` (the completed event AND the reversal),
``reversals`` (the uncompleted event only) and ``missed`` (the
derived past-due open instances, actor ``nightly sweep``, no
``occurred_at``) — and the range scopes by the INSTANCE's due date
(the Feature 13 rule), so the events sit under their instance's due
day, not the wall-clock day.
- ``GET /api/v1/admin/history.csv`` answers ``text/csv`` with a
``Content-Disposition`` attachment filename, and the parsed body
(csv.reader) carries the documented header columns and exactly the
filtered rows.
- Authentication runs through the router's ONE ``require_admin``
dependency: a non-admin JWT is 403 and an absent credential is 401
on both routes.
- Malformed input — an unknown filter, a non-strict date, an inverted
range, a missing parameter — is 422 on both routes (never a 500).
"""
from __future__ import annotations
import csv
import datetime
import importlib
import io
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import pytest
from api import routes_admin
from api.database import _executor
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
SUBJECT,
AdminRunner,
jwks_for,
make_token,
)
from tests.test_api_panel import _seed_household
# The completion, instance-DAO and db layers through the SAME core copy
# the app uses (nestquest_core.*, NOT the conftest-stubbed
# custom_components copy) — used for the tests' seeding only; the
# routes reach them through api.nestquest_core.
_completion = importlib.import_module("nestquest_core.completion")
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
_core_db = importlib.import_module("nestquest_core.db")
#: The documented CSV columns (Admin spec §5 event-row fields), in
#: header order — asserted literally so the contract is pinned here,
#: independent of the core constant.
DOCUMENTED_CSV_COLUMNS = [
"occurred_at",
"event_type",
"child_name",
"child_id",
"quest_title",
"instance_id",
"window",
"due_date",
"due_time",
"actor",
"was_on_time",
]
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _pinned_noon() -> tuple[datetime.date, datetime.datetime]:
"""One host clock read pinned to 12:00 UTC that day.
The seed's completion, the reversal and the route's clock all
anchor to this instant (the reversal five minutes later), so
nothing can straddle midnight mid-test.
"""
today = datetime.datetime.now().astimezone().date()
return today, datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
)
@pytest.fixture
async def history_household(temp_db_path: str, monkeypatch) -> SimpleNamespace:
"""A seeded household served by the admin-plane app, clock pinned.
Beyond :func:`_seed_household`'s rows (Bo's today completion, the
past-due open yesterday instances), the fixture appends ONE
``uncompleted`` reversal on Bo's today instance through the core
completion layer — attributed to the harness subject, stamped five
minutes after the pinned completion — and exposes the yesterday
instances the missed filter must report.
"""
today, pinned_now = _pinned_noon()
yesterday = today - datetime.timedelta(days=1)
seed = await _seed_household(temp_db_path, today, pinned_now)
database = _core_db.NestQuestDatabase(_executor)
await database.open(temp_db_path)
try:
instances = _dao_instances.QuestInstancesDao(database)
ada_yesterday = await instances.list_by_child_and_date(
seed.ada.id, yesterday.isoformat()
)
pack_bag_yesterday = next(
record for record in ada_yesterday if record.due_time == "10:00"
)
bo_yesterday = await instances.list_by_child_and_date(
seed.bo.id, yesterday.isoformat()
)
brush_teeth_yesterday = next(
record for record in bo_yesterday if record.due_time == "17:00"
)
await _completion.uncomplete_instance(
database,
seed.brush_teeth_instance.id,
actor_source="user",
actor_user_id=SUBJECT,
now=pinned_now + datetime.timedelta(minutes=5),
today=today,
)
finally:
await database.close()
monkeypatch.setattr(routes_admin, "_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,
database=runner.app.state.db.database,
today=today,
yesterday=yesterday,
pack_bag_yesterday=pack_bag_yesterday,
brush_teeth_yesterday=brush_teeth_yesterday,
)
# --- authentication: the ONE require_admin dependency guards both routes ----
async def test_history_routes_refuse_non_admin_and_absent(
history_household: SimpleNamespace,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401, both routes."""
client = history_household.client
query = f"start={history_household.yesterday.isoformat()}"
query += f"&end={history_household.today.isoformat()}"
non_admin = make_token(groups=("some-other-group",))
for path in ("/api/v1/admin/history", "/api/v1/admin/history.csv"):
forbidden = await client.get(
f"{path}?{query}",
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden.status_code == 403, path
assert forbidden.json()["detail"] == (
"Admin access requires membership in the nestquest-admins group"
)
unauthenticated = await client.get(f"{path}?{query}")
assert unauthenticated.status_code == 401, path
# --- the JSON history query: the expected rows per filter --------------------
async def test_history_all_returns_completed_and_reversal(
history_household: SimpleNamespace,
) -> None:
"""``all`` returns the completed event AND the reversal, in order.
Both events belong to Bo's Brush teeth instance due TODAY, so the
due-date-scoped range [yesterday, today] reports them under today;
the rows are ordered by due date, then instance, then event id —
the completed row before its reversal.
"""
client = history_household.client
seed = history_household.seed
brush = seed.brush_teeth_instance
yesterday_iso = history_household.yesterday.isoformat()
today_iso = history_household.today.isoformat()
response = await client.get(
"/api/v1/admin/history",
params={"filter": "all", "start": yesterday_iso, "end": today_iso},
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body["filter"] == "all"
assert body["start"] == yesterday_iso
assert body["end"] == today_iso
assert body["count"] == 2
completed, reversal = body["rows"]
assert completed == {
"occurred_at": f"{today_iso}T12:00:00+00:00",
"event_type": "completed",
"child_name": "Bo",
"child_id": brush.child_id,
"quest_title": "Brush teeth",
"instance_id": brush.id,
"window": "morning",
"due_date": today_iso,
"due_time": "17:00",
"actor": "panel (Bo)",
"was_on_time": True,
}
assert reversal == {
"occurred_at": f"{today_iso}T12:05:00+00:00",
"event_type": "uncompleted",
"child_name": "Bo",
"child_id": brush.child_id,
"quest_title": "Brush teeth",
"instance_id": brush.id,
"window": "morning",
"due_date": today_iso,
"due_time": "17:00",
"actor": SUBJECT,
"was_on_time": None,
}
async def test_history_reversals_returns_only_the_uncompleted_event(
history_household: SimpleNamespace,
) -> None:
"""``reversals`` returns the uncompleted event only, user-attributed."""
client = history_household.client
seed = history_household.seed
brush = seed.brush_teeth_instance
today_iso = history_household.today.isoformat()
response = await client.get(
"/api/v1/admin/history",
params={
"filter": "reversals",
"start": today_iso,
"end": today_iso,
},
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body["filter"] == "reversals"
assert body["count"] == 1
(reversal,) = body["rows"]
assert reversal["event_type"] == "uncompleted"
assert reversal["instance_id"] == brush.id
assert reversal["actor"] == SUBJECT
assert reversal["occurred_at"] == f"{today_iso}T12:05:00+00:00"
assert reversal["was_on_time"] is None
async def test_history_missed_returns_derived_missed_instances(
history_household: SimpleNamespace,
) -> None:
"""``missed`` returns the DERIVED past-due open instances.
Three seeded instances are open and due yesterday: Ada's Pack bag
and Stale chore and Bo's Brush teeth. Each is a DERIVED row —
never a stored event: no ``occurred_at``, ``was_on_time`` None and
the spec's ``nightly sweep`` actor.
"""
client = history_household.client
seed = history_household.seed
yesterday_iso = history_household.yesterday.isoformat()
response = await client.get(
"/api/v1/admin/history",
params={
"filter": "missed",
"start": yesterday_iso,
"end": yesterday_iso,
},
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body["filter"] == "missed"
assert body["count"] == 3
rows = body["rows"]
assert all(row["event_type"] == "missed" for row in rows)
assert all(row["occurred_at"] is None for row in rows)
assert all(row["was_on_time"] is None for row in rows)
assert all(row["actor"] == "nightly sweep" for row in rows)
assert all(row["due_date"] == yesterday_iso for row in rows)
assert {
(row["child_name"], row["quest_title"], row["due_time"])
for row in rows
} == {
("Ada", "Pack bag", "10:00"),
("Bo", "Brush teeth", "17:00"),
("Ada", "Stale chore", "09:00"),
}
assert {
row["instance_id"]
for row in rows
} == {
history_household.pack_bag_yesterday.id,
history_household.brush_teeth_yesterday.id,
seed.stale_instance.id,
}
# Deterministic order: due date, then instance id.
keys = [(row["due_date"], row["instance_id"]) for row in rows]
assert keys == sorted(keys)
async def test_history_range_scopes_by_the_instance_due_date(
history_household: SimpleNamespace,
) -> None:
"""The range scopes by the INSTANCE's due date, not the wall clock.
Both events OCCURRED today but belong to the instance due today, so
a yesterday-only ``all`` range reports nothing; the missed filter
over a today-only range reports nothing either (a due-TODAY open
instance is not past due).
"""
client = history_household.client
yesterday_iso = history_household.yesterday.isoformat()
today_iso = history_household.today.isoformat()
yesterday_all = await client.get(
"/api/v1/admin/history",
params={"filter": "all", "start": yesterday_iso, "end": yesterday_iso},
headers=_admin_headers(),
)
assert yesterday_all.status_code == 200
assert yesterday_all.json()["count"] == 0
today_missed = await client.get(
"/api/v1/admin/history",
params={"filter": "missed", "start": today_iso, "end": today_iso},
headers=_admin_headers(),
)
assert today_missed.status_code == 200
assert today_missed.json()["count"] == 0
# --- the CSV export: documented columns and the filtered rows ----------------
async def test_history_csv_carries_documented_columns_and_all_rows(
history_household: SimpleNamespace,
) -> None:
"""The CSV export carries the documented header and the filtered rows.
The body is parsed with csv.reader: the first row is the documented
column set, the remaining rows are exactly the ``all`` filter's
rows (the completion and its reversal) in the same order the JSON
query reports them.
"""
client = history_household.client
seed = history_household.seed
brush = seed.brush_teeth_instance
yesterday_iso = history_household.yesterday.isoformat()
today_iso = history_household.today.isoformat()
response = await client.get(
"/api/v1/admin/history.csv",
params={"filter": "all", "start": yesterday_iso, "end": today_iso},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/csv")
disposition = response.headers["content-disposition"]
assert disposition.startswith("attachment;")
assert (
f'filename="nestquest-history-{yesterday_iso}-to-{today_iso}.csv"'
in disposition
)
parsed = list(csv.reader(io.StringIO(response.text)))
assert parsed[0] == DOCUMENTED_CSV_COLUMNS
assert len(parsed) == 3 # header + completed + reversal
assert parsed[1] == [
f"{today_iso}T12:00:00+00:00",
"completed",
"Bo",
str(brush.child_id),
"Brush teeth",
str(brush.id),
"morning",
today_iso,
"17:00",
"panel (Bo)",
"true",
]
assert parsed[2] == [
f"{today_iso}T12:05:00+00:00",
"uncompleted",
"Bo",
str(brush.child_id),
"Brush teeth",
str(brush.id),
"morning",
today_iso,
"17:00",
SUBJECT,
"",
]
async def test_history_csv_missed_rows_are_derived_rows(
history_household: SimpleNamespace,
) -> None:
"""The CSV export of the ``missed`` filter carries the derived rows.
Same filtered range as the JSON query's missed answer: three rows
with an empty ``occurred_at`` (no event exists), the ``missed``
event type and the ``nightly sweep`` actor.
"""
client = history_household.client
seed = history_household.seed
yesterday_iso = history_household.yesterday.isoformat()
response = await client.get(
"/api/v1/admin/history.csv",
params={
"filter": "missed",
"start": yesterday_iso,
"end": yesterday_iso,
},
headers=_admin_headers(),
)
assert response.status_code == 200
parsed = list(csv.reader(io.StringIO(response.text)))
assert parsed[0] == DOCUMENTED_CSV_COLUMNS
assert len(parsed) == 4 # header + the three derived missed rows
missed_rows = parsed[1:]
assert all(row[1] == "missed" for row in missed_rows)
assert all(row[0] == "" for row in missed_rows)
assert all(row[9] == "nightly sweep" for row in missed_rows)
assert all(row[10] == "" for row in missed_rows)
assert {
(row[2], row[4], row[8]) for row in missed_rows
} == {
("Ada", "Pack bag", "10:00"),
("Bo", "Brush teeth", "17:00"),
("Ada", "Stale chore", "09:00"),
}
assert {row[5] for row in missed_rows} == {
str(history_household.pack_bag_yesterday.id),
str(history_household.brush_teeth_yesterday.id),
str(seed.stale_instance.id),
}
# --- validation: malformed filter and date range are 422, never 500 ----------
@pytest.mark.parametrize(
"params",
[
{"filter": "everything", "start": "2026-03-01", "end": "2026-03-02"},
{"filter": "all", "start": "2026/03/01", "end": "2026-03-02"},
{"filter": "all", "start": "2026-3-1", "end": "2026-03-02"},
{"filter": "all", "start": "2026-03-02", "end": "2026-03-01"},
{"filter": "all", "start": "not-a-date", "end": "2026-03-02"},
],
)
async def test_history_query_rejects_malformed_params_as_422(
history_household: SimpleNamespace, params: dict[str, str]
) -> None:
"""An unknown filter or a rejected range is 422 naming the field."""
client = history_household.client
response = await client.get(
"/api/v1/admin/history", params=params, headers=_admin_headers()
)
assert response.status_code == 422, params
detail = response.json()["detail"]
assert any(
name in detail for name in ("filter", "start", "end")
), detail
async def test_history_csv_rejects_malformed_params_as_422(
history_household: SimpleNamespace,
) -> None:
"""The CSV route maps the same rejections to 422."""
client = history_household.client
unknown_filter = await client.get(
"/api/v1/admin/history.csv",
params={"filter": "galaxy", "start": "2026-03-01", "end": "2026-03-02"},
headers=_admin_headers(),
)
assert unknown_filter.status_code == 422
assert "filter" in unknown_filter.json()["detail"]
inverted = await client.get(
"/api/v1/admin/history.csv",
params={"filter": "all", "start": "2026-03-02", "end": "2026-03-01"},
headers=_admin_headers(),
)
assert inverted.status_code == 422
assert "end must be on or after start" in inverted.json()["detail"]
async def test_history_query_missing_params_are_422(
history_household: SimpleNamespace,
) -> None:
"""A missing required query parameter is FastAPI's 422 (the shape)."""
client = history_household.client
no_params = await client.get(
"/api/v1/admin/history", headers=_admin_headers()
)
assert no_params.status_code == 422
missing_start = await client.get(
"/api/v1/admin/history",
params={"end": "2026-03-02"},
headers=_admin_headers(),
)
assert missing_start.status_code == 422
missing_end = await client.get(
"/api/v1/admin/history",
params={"start": "2026-03-01"},
headers=_admin_headers(),
)
assert missing_end.status_code == 422
+398
View File
@@ -0,0 +1,398 @@
"""Admin-plane settings route tests (task 2b3de7e5).
These tests exercise the admin plane's settings read and update routes
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 — signature, issuer, audience, expiry and the
nestquest-admins group — with NO network access. The stored settings
document is read back through the SAME core copy the app uses
(``nestquest_core.*``; see the two-copy caveat in api/nestquest_core.py)
via its :class:`nestquest_core.dao_meta.MetaStateDao`, so persistence is
proven against the DATABASE the routes wrote, not a parallel copy.
Proven per the done-condition, everything read back through the API:
- ``GET /api/v1/admin/settings`` returns the current effective settings
— the ten documented fields at their defaults on a fresh database.
- ``PATCH /api/v1/admin/settings`` PERSISTS the supplied fields and the
NEXT read reflects them; a later partial update merges onto the
current effective settings (omitted fields keep their stored value),
and the document is stored under the well-known ``nestquest_meta_state``
key.
- An invalid value (a non-positive or boolean horizon, a non-strict
time, a non-bool toggle, a non-string target) is rejected with 422
naming the field, and the stored settings are UNCHANGED — the next
read still shows the previous value. An unknown field is rejected
the same way, also when mixed with a valid field (validate
everything before writing).
- A corrupt stored document (invalid JSON, a non-object, an invalid
field value) does not crash the service: the read falls back to the
defaults, per field where possible.
- Authentication runs through the router's ONE ``require_admin``
dependency: a non-admin JWT is 403 and an absent credential is 401
on both routes — and no refused patch writes anything.
"""
from __future__ import annotations
import importlib
import json
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
# The meta-state DAO through the SAME core copy the app uses
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy) —
# used for the tests' database assertions only; the routes reach it
# through api.nestquest_core.
_dao_meta = importlib.import_module("nestquest_core.dao_meta")
#: The well-known ``nestquest_meta_state`` key the API stores its
#: settings under — asserted literally so the contract is pinned here,
#: independent of the core constant.
SETTINGS_META_KEY = "settings"
#: The all-defaults effective settings a fresh database reports — the
#: documented ten fields, asserted literally so the payload contract is
#: pinned here, independent of the core constants.
DEFAULT_SETTINGS = {
"horizon_days": 14,
"day_rollover_time": "00:00",
"notify_target": "",
"morning_summary_time": "08:00",
"afternoon_reminder_time": "15:00",
"end_of_day_report_time": "20:00",
"morning_summary_enabled": True,
"afternoon_reminder_enabled": True,
"end_of_day_report_enabled": True,
"celebration_enabled": True,
}
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
@pytest.fixture
async def settings_client(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane app client plus its live database.
The same runner the children tests use (stubbed JWKS, no network);
the fixture also exposes the app's ONE database connection so tests
can read (or corrupt) the stored settings document itself.
"""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(
client=client,
database=runner.app.state.db.database,
)
# --- read: the effective settings, defaults on a fresh database --------
async def test_get_settings_returns_defaults_on_a_fresh_database(
settings_client: SimpleNamespace,
) -> None:
"""GET returns the ten documented fields at their defaults."""
response = await settings_client.client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == DEFAULT_SETTINGS
# --- update: persists, and the NEXT read reflects it --------------------
async def test_update_persists_and_the_next_read_reflects_it(
settings_client: SimpleNamespace,
) -> None:
"""PATCH persists the supplied fields; GET reads them back."""
client = settings_client.client
changes = {
"horizon_days": 21,
"day_rollover_time": "03:30",
"notify_target": "notify.mobile_app_dads_phone",
"celebration_enabled": False,
}
patched = await client.patch(
"/api/v1/admin/settings", json=changes, headers=_admin_headers()
)
assert patched.status_code == 200
expected = {**DEFAULT_SETTINGS, **changes}
assert patched.json() == expected
# The NEXT read reflects the persisted settings...
reread = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert reread.status_code == 200
assert reread.json() == expected
# ...and the DATABASE holds the merged document: the complete ten
# fields, keyed by field name, under the well-known meta_state key.
raw = await _dao_meta.MetaStateDao(settings_client.database).get(
SETTINGS_META_KEY
)
assert raw is not None
assert json.loads(raw) == expected
async def test_partial_update_merges_onto_the_current_settings(
settings_client: SimpleNamespace,
) -> None:
"""A later PATCH changes only its own fields; the rest survive."""
client = settings_client.client
first = await client.patch(
"/api/v1/admin/settings",
json={"horizon_days": 7, "notify_target": " notify.x "},
headers=_admin_headers(),
)
assert first.status_code == 200
# The notify target went through the core's own normalizer: stripped.
assert first.json()["horizon_days"] == 7
assert first.json()["notify_target"] == "notify.x"
second = await client.patch(
"/api/v1/admin/settings",
json={"morning_summary_time": "07:45"},
headers=_admin_headers(),
)
assert second.status_code == 200
assert second.json() == {
**DEFAULT_SETTINGS,
"horizon_days": 7,
"notify_target": "notify.x",
"morning_summary_time": "07:45",
}
reread = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert reread.json() == second.json()
async def test_empty_update_is_a_noop(
settings_client: SimpleNamespace,
) -> None:
"""PATCHing no fields leaves the effective settings as they are."""
client = settings_client.client
response = await client.patch(
"/api/v1/admin/settings", json={}, headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == DEFAULT_SETTINGS
async def test_null_resets_a_field_to_its_default(
settings_client: SimpleNamespace,
) -> None:
"""A JSON ``null`` maps to the field's default (None-as-absent)."""
client = settings_client.client
set_horizon = await client.patch(
"/api/v1/admin/settings",
json={"horizon_days": 7},
headers=_admin_headers(),
)
assert set_horizon.status_code == 200
assert set_horizon.json()["horizon_days"] == 7
reset = await client.patch(
"/api/v1/admin/settings",
json={"horizon_days": None},
headers=_admin_headers(),
)
assert reset.status_code == 200
assert reset.json()["horizon_days"] == 14
reread = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert reread.json()["horizon_days"] == 14
# --- validation: rejected values never change the stored settings -------
@pytest.mark.parametrize(
("changes", "field"),
[
({"horizon_days": 0}, "horizon_days"),
({"horizon_days": -3}, "horizon_days"),
# A bool is rejected — the core's guardrail survives the route.
({"horizon_days": True}, "horizon_days"),
({"horizon_days": "soon"}, "horizon_days"),
({"day_rollover_time": "25:00"}, "day_rollover_time"),
({"day_rollover_time": "7:30"}, "day_rollover_time"),
({"morning_summary_enabled": "yes"}, "morning_summary_enabled"),
({"celebration_enabled": 1}, "celebration_enabled"),
({"notify_target": 5}, "notify_target"),
],
)
async def test_invalid_value_is_rejected_and_stored_settings_unchanged(
settings_client: SimpleNamespace, changes: dict[str, object], field: str
) -> None:
"""A rejected value is 422 naming the field; the next read is same."""
client = settings_client.client
seeded = await client.patch(
"/api/v1/admin/settings",
json={
"horizon_days": 21,
"day_rollover_time": "01:30",
"morning_summary_enabled": False,
},
headers=_admin_headers(),
)
assert seeded.status_code == 200
before = seeded.json()
response = await client.patch(
"/api/v1/admin/settings", json=changes, headers=_admin_headers()
)
assert response.status_code == 422, changes
assert field in str(response.json()["detail"]), changes
after = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert after.status_code == 200
assert after.json() == before
async def test_unknown_field_is_rejected_and_stored_settings_unchanged(
settings_client: SimpleNamespace,
) -> None:
"""An unknown field is 422 naming it — alone or beside a valid one."""
client = settings_client.client
unknown = await client.patch(
"/api/v1/admin/settings",
json={"bogus_setting": 1},
headers=_admin_headers(),
)
assert unknown.status_code == 422
assert "bogus_setting" in str(unknown.json()["detail"])
mixed = await client.patch(
"/api/v1/admin/settings",
json={"horizon_days": 7, "bogus_setting": 1},
headers=_admin_headers(),
)
assert mixed.status_code == 422
assert "bogus_setting" in str(mixed.json()["detail"])
# Neither rejection wrote anything, not even the valid sibling.
after = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert after.json() == DEFAULT_SETTINGS
async def test_non_object_body_is_422(
settings_client: SimpleNamespace,
) -> None:
"""A body that is not a JSON object is FastAPI's 422 (the shape)."""
client = settings_client.client
for body in ([1, 2], "nope", 5):
response = await client.patch(
"/api/v1/admin/settings", json=body, headers=_admin_headers()
)
assert response.status_code == 422, body
# A bare JSON null is likewise not an object.
response = await client.patch(
"/api/v1/admin/settings",
content=b"null",
headers={**_admin_headers(), "content-type": "application/json"},
)
assert response.status_code == 422
after = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert after.json() == DEFAULT_SETTINGS
# --- resilience: a corrupt stored document does not crash the service ---
async def test_corrupt_stored_document_falls_back_to_defaults(
settings_client: SimpleNamespace,
) -> None:
"""GET degrades gracefully over an unusable settings document."""
client = settings_client.client
dao = _dao_meta.MetaStateDao(settings_client.database)
# Not JSON at all: all defaults, no crash.
await dao.set(SETTINGS_META_KEY, "{not-json")
response = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == DEFAULT_SETTINGS
# JSON but not an object: all defaults, no crash.
await dao.set(SETTINGS_META_KEY, json.dumps([1, 2]))
response = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == DEFAULT_SETTINGS
# An object with one invalid field: PER-FIELD fallback — the valid
# sibling survives, the invalid one falls back to its default.
await dao.set(
SETTINGS_META_KEY,
json.dumps({"horizon_days": 7, "day_rollover_time": "soon"}),
)
response = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {**DEFAULT_SETTINGS, "horizon_days": 7}
# --- authentication: the ONE require_admin dependency guards both routes
async def test_settings_routes_refuse_non_admin_and_absent_credentials(
settings_client: SimpleNamespace,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401; none write."""
client = settings_client.client
non_admin = make_token(groups=("some-other-group",))
forbidden_get = await client.get(
"/api/v1/admin/settings",
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden_get.status_code == 403
assert forbidden_get.json()["detail"] == (
"Admin access requires membership in the nestquest-admins group"
)
forbidden_patch = await client.patch(
"/api/v1/admin/settings",
json={"horizon_days": 7},
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden_patch.status_code == 403
absent_get = await client.get("/api/v1/admin/settings")
assert absent_get.status_code == 401
absent_patch = await client.patch(
"/api/v1/admin/settings", json={"horizon_days": 7}
)
assert absent_patch.status_code == 401
# No refused request changed anything: the settings are untouched.
after = await client.get(
"/api/v1/admin/settings", headers=_admin_headers()
)
assert after.json() == DEFAULT_SETTINGS
+694
View File
@@ -0,0 +1,694 @@
"""Admin-plane uncomplete and regenerate route tests (task da0226b3).
These tests exercise the admin plane's completion-reversal and
re-materialization routes 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 — signature, issuer, audience, expiry and the
nestquest-admins group — 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.*``; the two-copy caveat
in api/nestquest_core.py), and the route's ONE clock read
(:func:`api.routes_admin._local_now`) is pinned the same way the panel
route's is, so seeded rows and route stamps cannot straddle midnight.
Every done-condition case is proven against the DATABASE, read through
that same core copy:
- Authentication runs through the router's ONE ``require_admin``
dependency: an admin JWT passes, a JWT without the group is 403, an
absent credential is 401 — and nothing is written.
- ``POST /api/v1/admin/instances/{id}/uncomplete``: the completion
events log for the instance holds the ORIGINAL ``completed`` row
UNTOUCHED plus ONE appended ``uncompleted`` row attributed to the
verified token's ``sub`` (actor_source ``user``), and the instance's
DERIVED state is ``open`` again; a second uncomplete is a no-op that
appends nothing.
- On an ACTUAL reversal the ``nestquest_quest_uncompleted`` transition
arrives on the SSE stream exactly once (subscribed live as
:mod:`tests.test_api_sse` does); a no-op uncomplete publishes
nothing.
- ``POST /api/v1/admin/regenerate`` re-materializes without
duplicating: running it twice (household, definition and child
scopes) leaves the instance count stable, and the scoped variants
rebuild only their own scope's rows.
Error mapping (documented in api/routes_admin.py): an unknown
instance is 404 (the core's ``instance_id:`` prefix, the same key the
panel complete route uses); unknown definition/child scope ids are 404
from the route's own existence reads; malformed input — a bad scope,
a scope/id mismatch, a zero id, a non-integer path id — is 422 before
any handler code runs.
"""
from __future__ import annotations
import asyncio
import datetime
import importlib
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import httpx
import pytest
from api import routes_admin
from api.app import create_app
from api.auth import JwksCache
from api.config import ApiConfig
from tests.admin_jwt_harness import (
ADMIN_KEY,
AUDIENCE,
ISSUER,
JWKS_URL,
KID,
AdminRunner,
StubbedFetch,
SUBJECT,
jwks_for,
make_token,
)
from tests.test_api_panel import PANEL_TOKEN, _seed_household
from tests.test_api_sse import _EventStream, _Server, _stream
# The completion and instance-DAO layers through the SAME core copy the
# app uses (nestquest_core.*, NOT the conftest-stubbed
# custom_components copy) — used for the tests' database assertions
# only; the routes reach them through api.nestquest_core.
_completion = importlib.import_module("nestquest_core.completion")
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
#: The documented quest-uncompleted event name (core const, §3).
EVENT_QUEST_UNCOMPLETED = importlib.import_module(
"nestquest_core.const"
).EVENT_QUEST_UNCOMPLETED
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _pinned_noon() -> tuple[datetime.date, datetime.datetime]:
"""One host clock read pinned to 12:00 UTC that day.
The seed's completion and the route's clock both anchor to this
instant, so nothing can straddle midnight mid-test.
"""
today = datetime.datetime.now().astimezone().date()
return today, datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
)
async def _instance_count(database: object) -> int:
"""The whole ``quest_instances`` row count (the sanctioned probe)."""
row = await database.fetch_one("SELECT COUNT(*) FROM quest_instances")
return row[0]
async def _events_for_instance(
database: object, instance_id: int
) -> list:
"""The instance's completion events, in append order."""
dao = _dao_instances.CompletionEventsDao(database)
return await dao.list_by_instance(instance_id)
async def _derived_state(
database: object, instance_id: int, today: datetime.date
) -> str:
"""The instance's derived state through the app's core copy."""
dao = _dao_instances.QuestInstancesDao(database)
instance = await dao.get_by_id(instance_id)
assert instance is not None
latest = await _dao_instances.CompletionEventsDao(
database
).get_latest_for_instance(instance_id)
return _completion.derive_state(instance, latest, today)
@pytest.fixture
async def admin_api(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane app client plus its live database.
The same runner the auth and children tests use (stubbed JWKS, no
network); the fixture exposes the app's ONE database connection so
tests assert the routes' effects on the stored rows themselves.
"""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(
client=client,
database=runner.app.state.db.database,
)
@pytest.fixture
async def admin_household(temp_db_path: str, monkeypatch) -> SimpleNamespace:
"""A seeded household served by the admin-plane app, clock pinned.
``_seed_household`` completes Bo's today instance (17:00 due, 12:00
pinned — on time), which is the completion the uncomplete tests
reverse. The route's ``_local_now`` is pinned to the same instant.
"""
today, pinned_now = _pinned_noon()
seed = await _seed_household(temp_db_path, today, pinned_now)
monkeypatch.setattr(routes_admin, "_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,
database=runner.app.state.db.database,
today=today,
)
class _AdminServer(_Server):
""":mod:`tests.test_api_sse`'s live-uvicorn server, admin-enabled.
Same ephemeral-port uvicorn thread (httpx's ASGI transport buffers
SSE bodies, so a live server is the only way to stream), but the
app factory swaps the JWKS cache for one built around the harness's
stub fetcher, so a real admin JWT verifies with NO network access.
"""
def _make_app_factory(self, db_path: str):
"""The uvicorn app factory with the stubbed JWKS cache."""
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
@pytest.fixture
async def admin_sse(temp_db_path: str) -> SimpleNamespace:
"""A seeded household served by a live uvicorn app, clock pinned.
Same pinning discipline as the panel tests, applied to
:func:`api.routes_admin._local_now` BEFORE the server starts (the
app is created inside uvicorn's factory thread).
"""
today, pinned_now = _pinned_noon()
seed = await _seed_household(temp_db_path, today, pinned_now)
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,
database=server.app.state.db.database,
today=today,
)
finally:
routes_admin._local_now = original
# --- authentication: the ONE require_admin dependency guards every route ---
async def test_uncomplete_and_regenerate_refuse_non_admin_and_absent(
admin_household: SimpleNamespace,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401: nothing written."""
client = admin_household.client
database = admin_household.database
brush = admin_household.seed.brush_teeth_instance
non_admin = make_token(groups=("some-other-group",))
events_before = await _events_for_instance(database, brush.id)
instances_before = await _instance_count(database)
uncomplete_url = f"/api/v1/admin/instances/{brush.id}/uncomplete"
forbidden = await client.post(
uncomplete_url,
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden.status_code == 403
assert forbidden.json()["detail"] == (
"Admin access requires membership in the nestquest-admins group"
)
unauthenticated = await client.post(uncomplete_url)
assert unauthenticated.status_code == 401
regenerate_bodies = (
{},
{"scope": "household"},
{"scope": "child", "child_id": 1},
)
for body in regenerate_bodies:
forbidden = await client.post(
"/api/v1/admin/regenerate",
json=body,
headers={"Authorization": f"Bearer {non_admin}"},
)
assert forbidden.status_code == 403, body
unauthenticated = await client.post(
"/api/v1/admin/regenerate", json=body
)
assert unauthenticated.status_code == 401, body
# No handler ran: the seeded completion is still the ONLY event for
# the instance and the instance table is untouched.
assert (
await _events_for_instance(database, brush.id) == events_before
)
assert await _instance_count(database) == instances_before
# --- uncomplete: the append-only reversal -----------------------------------
async def test_uncomplete_appends_reversal_and_derived_state_is_open(
admin_household: SimpleNamespace,
) -> None:
"""A reversal APPENDS an uncompleted event; the instance is open again.
The original completed row is proven byte-identical after the call
(frozen-dataclass equality), the new row carries the verified
token's ``sub`` as the user actor, and the DERIVED state —
recomputed from the latest event — is ``open`` again.
"""
client = admin_household.client
database = admin_household.database
brush = admin_household.seed.brush_teeth_instance
today = admin_household.today
before = await _events_for_instance(database, brush.id)
assert [event.event_type for event in before] == ["completed"]
assert await _derived_state(database, brush.id, today) == "done"
response = await client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body == {
"instance_id": brush.id,
"state": "open",
"appended": True,
}
events = await _events_for_instance(database, brush.id)
assert len(events) == 2
# The original completed row is UNTOUCHED — never edited, never
# deleted (append-only completion_events).
assert events[0] == before[0]
assert events[0].event_type == "completed"
assert events[0].actor_source == "panel"
# The reversal is the NEW row, attributed to the verified token.
reversal = events[1]
assert reversal.event_type == "uncompleted"
assert reversal.instance_id == brush.id
assert reversal.child_id == brush.child_id
assert reversal.actor_source == "user"
assert reversal.actor_user_id == SUBJECT
assert reversal.actor_child_id is None
assert reversal.was_on_time is None
# The derived state is recomputed from the latest event: open again.
assert await _derived_state(database, brush.id, today) == "open"
async def test_second_uncomplete_is_a_noop_that_appends_nothing(
admin_household: SimpleNamespace,
) -> None:
"""A repeat uncomplete appends NOTHING and still succeeds."""
client = admin_household.client
database = admin_household.database
brush = admin_household.seed.brush_teeth_instance
today = admin_household.today
first = await client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=_admin_headers(),
)
assert first.status_code == 200
assert first.json()["appended"] is True
second = await client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=_admin_headers(),
)
assert second.status_code == 200
body = second.json()
assert body["appended"] is False
assert body["state"] == "open"
events = await _events_for_instance(database, brush.id)
assert len(events) == 2
assert [event.event_type for event in events] == [
"completed",
"uncompleted",
]
assert await _derived_state(database, brush.id, today) == "open"
async def test_uncomplete_unknown_instance_is_404_and_malformed_is_422(
admin_household: SimpleNamespace,
) -> None:
"""An unknown instance is 404; a non-integer path id is 422."""
client = admin_household.client
brush = admin_household.seed.brush_teeth_instance
before = await _events_for_instance(
admin_household.database, brush.id
)
unknown = await client.post(
"/api/v1/admin/instances/999/uncomplete",
headers=_admin_headers(),
)
assert unknown.status_code == 404
assert unknown.json()["detail"] == "Quest instance not found"
malformed = await client.post(
"/api/v1/admin/instances/not-an-int/uncomplete",
headers=_admin_headers(),
)
assert malformed.status_code == 422
# Neither rejection wrote anything.
assert (
await _events_for_instance(admin_household.database, brush.id)
== before
)
async def test_uncomplete_already_open_past_due_reports_missed(
admin_household: SimpleNamespace,
) -> None:
"""A no-op uncomplete still reports the honest DERIVED state.
The seeded stale instance is open and past due: uncompleting it is
a no-op (nothing appended) and the derived state is ``missed`` —
open-but-past-due, never an event type.
"""
client = admin_household.client
database = admin_household.database
stale = admin_household.seed.stale_instance
today = admin_household.today
assert await _derived_state(database, stale.id, today) == "missed"
response = await client.post(
f"/api/v1/admin/instances/{stale.id}/uncomplete",
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body["appended"] is False
assert body["state"] == "missed"
# The no-op appended nothing: the instance still has NO events.
assert await _events_for_instance(database, stale.id) == []
# --- the SSE transition on an actual reversal --------------------------------
async def test_uncomplete_publishes_exactly_one_uncompleted_transition(
admin_sse: SimpleNamespace,
) -> None:
"""An actual reversal streams ONE quest-uncompleted event; a no-op none.
The stream is subscribed FIRST (test_api_sse's live-server reader),
the completed instance is reversed through the admin route, and the
documented payload fields are asserted; the no-op second call adds
no frame.
"""
seed = admin_sse.seed
brush = seed.brush_teeth_instance
stream = _stream(admin_sse.server.url("/api/v1/panel/events"))
async with stream:
first = await admin_sse.client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=_admin_headers(),
)
assert first.status_code == 200
assert first.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"] # strict UTC ISO-8601 stamp
assert "was_on_time" not in payload
# The no-op second call publishes nothing: the stream stays
# silent past the window.
second = await admin_sse.client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=_admin_headers(),
)
assert second.status_code == 200
assert second.json()["appended"] is False
with pytest.raises(asyncio.TimeoutError):
await asyncio.wait_for(stream.queue.get(), timeout=0.3)
# --- regenerate: re-materialize without duplicating --------------------------
async def _create_child(client: object, name: str) -> int:
"""Create one child through the admin children route."""
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def _create_daily_definition(
client: object, assignee_ids: list[int]
) -> int:
"""Create one daily one-window definition for the given assignees."""
created = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Tidy the den",
"rule": {"rule_type": "daily", "start_date": "2026-03-02"},
"assignee_child_ids": assignee_ids,
"windows": ["morning"],
},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def test_regenerate_household_never_duplicates(
admin_household: SimpleNamespace,
) -> None:
"""Two household regenerations leave the instance count stable.
The first call re-materializes the rolling horizon
``[today, today + DEFAULT_HORIZON_DAYS]``; the second walks the
SAME tuples and upserts them in place — the row count never grows
and the core reports the same upsert count both times.
"""
client = admin_household.client
database = admin_household.database
before = await _instance_count(database)
assert before > 0 # the seed already materialized a few days
first = await client.post(
"/api/v1/admin/regenerate", json={}, headers=_admin_headers()
)
assert first.status_code == 200
body = first.json()
assert body["scope"] == "household"
assert body["definition_id"] is None
assert body["child_id"] is None
assert body["count"] >= before
after_first = await _instance_count(database)
assert after_first >= before
second = await client.post(
"/api/v1/admin/regenerate", json={}, headers=_admin_headers()
)
assert second.status_code == 200
assert second.json()["count"] == body["count"]
# The idempotent walk: the second run's row count is EXACTLY the
# first's — no duplicated instances.
assert await _instance_count(database) == after_first
async def test_regenerate_definition_scope_rematerializes(
admin_api: SimpleNamespace,
) -> None:
"""The definition scope rebuilds that definition's horizon, twice, stably.
Creating a definition through the admin route materializes NOTHING
(the core's create path does not regenerate); the scoped regenerate
call is what materializes its instances, and re-running it leaves
the row count exactly where the first run left it.
"""
client = admin_api.client
database = admin_api.database
ada = await _create_child(client, "Ada")
definition_id = await _create_daily_definition(client, [ada])
assert await _instance_count(database) == 0
first = await client.post(
"/api/v1/admin/regenerate",
json={"scope": "definition", "definition_id": definition_id},
headers=_admin_headers(),
)
assert first.status_code == 200
body = first.json()
assert body["scope"] == "definition"
assert body["definition_id"] == definition_id
assert body["child_id"] is None
assert body["count"] > 0
after_first = await _instance_count(database)
assert after_first == body["count"]
second = await client.post(
"/api/v1/admin/regenerate",
json={"scope": "definition", "definition_id": definition_id},
headers=_admin_headers(),
)
assert second.status_code == 200
assert second.json()["count"] == body["count"]
assert await _instance_count(database) == after_first
async def test_regenerate_child_scope_touches_only_that_child(
admin_api: SimpleNamespace,
) -> None:
"""The child scope rebuilds that child's rows and no others'.
Two children share one daily definition; the household walk
materializes both. A child-scoped regenerate for Ada re-runs her
side of the walk (same count) while Bo's rows keep their identity,
and re-running it never grows the table.
"""
client = admin_api.client
database = admin_api.database
instances = _dao_instances.QuestInstancesDao(database)
ada = await _create_child(client, "Ada")
bo = await _create_child(client, "Bo")
# One daily definition assigned to BOTH children: the household walk
# gives each child its own rows.
await _create_daily_definition(client, [ada, bo])
# Materialize the household first so BOTH children have rows.
seeded = await client.post(
"/api/v1/admin/regenerate", json={}, headers=_admin_headers()
)
assert seeded.status_code == 200
assert seeded.json()["scope"] == "household"
ada_rows = await instances.list_by_date_range(
ada, "2026-03-02", "2100-01-01"
)
bo_rows = await instances.list_by_date_range(
bo, "2026-03-02", "2100-01-01"
)
assert ada_rows and bo_rows
total = await _instance_count(database)
assert total == len(ada_rows) + len(bo_rows)
for expected_count in (len(ada_rows), len(ada_rows)):
response = await client.post(
"/api/v1/admin/regenerate",
json={"scope": "child", "child_id": ada},
headers=_admin_headers(),
)
assert response.status_code == 200
body = response.json()
assert body == {
"scope": "child",
"definition_id": None,
"child_id": ada,
"count": expected_count,
}
# Ada's side re-materialized the same number of rows and Bo's
# rows kept their identity; the table never grew.
assert await instances.list_by_date_range(
ada, "2026-03-02", "2100-01-01"
)
assert await instances.list_by_date_range(
bo, "2026-03-02", "2100-01-01"
) == bo_rows
assert await _instance_count(database) == total
async def test_regenerate_unknown_scope_ids_are_404(
admin_api: SimpleNamespace,
) -> None:
"""An unknown definition or child scope id is 404, nothing written."""
client = admin_api.client
unknown_definition = await client.post(
"/api/v1/admin/regenerate",
json={"scope": "definition", "definition_id": 999},
headers=_admin_headers(),
)
assert unknown_definition.status_code == 404
assert unknown_definition.json()["detail"] == "Quest definition not found"
unknown_child = await client.post(
"/api/v1/admin/regenerate",
json={"scope": "child", "child_id": 999},
headers=_admin_headers(),
)
assert unknown_child.status_code == 404
assert unknown_child.json()["detail"] == "Child not found"
assert await _instance_count(admin_api.database) == 0
@pytest.mark.parametrize(
"body",
[
{"scope": "galaxy"}, # unknown scope
{"scope": "definition"}, # scope without its id
{"scope": "child"}, # scope without its id
{"scope": "household", "child_id": 1}, # id without its scope
{"scope": "household", "definition_id": 1}, # id without its scope
{
"scope": "definition",
"definition_id": 1,
"child_id": 2,
}, # scope mismatch
{"scope": "child", "child_id": 0}, # zero id
{"scope": "child", "child_id": "seven"}, # wrong type
],
)
async def test_regenerate_malformed_bodies_are_422(
admin_api: SimpleNamespace, body: dict[str, object]
) -> None:
"""Malformed scope/id combinations are 422 before any core call."""
response = await admin_api.client.post(
"/api/v1/admin/regenerate", json=body, headers=_admin_headers()
)
assert response.status_code == 422, body
assert await _instance_count(admin_api.database) == 0
+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()