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

Panel plane: service-token auth and snapshot route (#152)

Browse Source
This commit was merged in pull request #152.
This commit is contained in:
dev_agent
2026-09-22 22:23:16 -04:00
parent ca477125d6
commit 4da762173c
7 changed files with 880 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/app.py
+11
View File
@@ -8,6 +8,10 @@ exposes the two endpoints every later task depends on:
app has imported core, opened the database, and applied migrations;
a cheap ``SELECT 1`` confirms the connection is live without leaking
the filesystem path on an unauthenticated route.
- ``GET /api/v1/panel/snapshot`` — the panel plane's today snapshot
(service-token protected; see :mod:`api.routes_panel`, whose router
carries the shared :func:`api.dependencies.require_panel_token`
check).
- ``GET /docs`` — FastAPI's auto-served OpenAPI document (Swagger UI);
``/openapi.json`` is the raw schema. No extra wiring is needed;
FastAPI serves both by default.
@@ -35,6 +39,7 @@ from fastapi import FastAPI
from api.config import ApiConfig
from api.database import DatabaseState, make_database
from api.routes_panel import router as panel_router
def _build_lifespan(db_path: str):
@@ -79,6 +84,12 @@ def create_app(config: ApiConfig | None = None) -> FastAPI:
lifespan=_build_lifespan(cfg.db_path),
)
# The resolved config rides on app.state so the reusable route
# dependencies (the panel service-token check) read the configured
# values from the app rather than module state.
app.state.config = cfg
app.include_router(panel_router)
@app.get("/health", summary="Service health")
async def health() -> dict[str, str]:
"""Return service health.
api/config.py
+49 -19
View File
@@ -1,16 +1,22 @@
"""Configuration for the NestQuest API service.
The API reads its single configuration knob — the SQLite database path —
from the ``NESTQUEST_DB_PATH`` environment variable. Unlike a dev
convenience default, a missing or empty value is a configuration ERROR,
not a silent fallback to a predictable world-writable temp path: the API
owns the database (it is the writer of record per Feature 16's
contract), and silently dropping data into ``$TMPDIR/nestquest.db``
would lose it at reboot. ``ApiConfig.from_env`` therefore RAISES when
the variable is unset or empty. This mirrors the integration's
"explicit settings object" discipline (Feature 15's
``core.settings``): the API never reads HA config; callers pass the
resolved path into the database wiring.
The API reads its configuration knobs from environment variables:
- ``NESTQUEST_DB_PATH`` — the SQLite database file. The API opens this
file on startup (creating it if absent) and applies migrations.
- ``NESTQUEST_PANEL_TOKEN`` — the static panel-plane service token
(D-012): the credential ``/api/v1/panel/*`` routes require via
:func:`api.dependencies.require_panel_token`. It is stored outside
the repo and handed to the service through the environment.
Unlike a dev convenience default, a missing or empty value is a
configuration ERROR, not a silent fallback: a default database path
would silently drop data at reboot, and a default (or empty) service
token would either authenticate everyone or lock the panel plane out
for good. ``ApiConfig.from_env`` therefore RAISES when either variable
is unset or empty. This mirrors the integration's "explicit settings
object" discipline (Feature 15's ``core.settings``): the API never
reads HA config; callers pass the resolved values into the wiring.
Serving: the app is built with the factory ``api.app:create_app`` — run
it with ``uvicorn --factory api.app:create_app`` so the factory reads
@@ -27,6 +33,14 @@ from dataclasses import dataclass
#: :meth:`ApiConfig.from_env`.
DB_PATH_ENV = "NESTQUEST_DB_PATH"
#: Environment variable holding the static panel-plane service token
#: (D-012), checked by :func:`api.dependencies.require_panel_token`.
#: REQUIRED — unset or empty raises :class:`ConfigError` from
#: :meth:`ApiConfig.from_env`; there is no silent default, because an
#: empty configured token would either authenticate every caller or
#: lock the panel plane out permanently.
PANEL_TOKEN_ENV = "NESTQUEST_PANEL_TOKEN"
class ConfigError(RuntimeError):
"""Raised when required API configuration is missing or invalid."""
@@ -36,12 +50,18 @@ class ConfigError(RuntimeError):
class ApiConfig:
"""Resolved API configuration.
Currently a single field — the SQLite path — but kept as a dataclass
so later tasks (auth, SSE, settings persistence) can extend it
without touching the call sites that read ``db_path``.
Both fields are required (no defaults): ``from_env`` rejects an
unset or empty value for either, and a caller building the dataclass
directly must pass an explicit token — the token check itself fails
closed on an empty configured token, but expecting a caller to
discover that at request time is a wiring mistake, so the field is
mandatory. Kept as a dataclass so later tasks (admin plane, SSE,
settings persistence) can extend it without touching the call sites
that read ``db_path``.
"""
db_path: str
panel_token: str
@classmethod
def from_env(cls, env: dict[str, str] | None = None) -> "ApiConfig":
@@ -52,17 +72,27 @@ class ApiConfig:
Raises :class:`ConfigError` when ``NESTQUEST_DB_PATH`` is unset
or empty/whitespace-only — there is no silent default, because a
default in the temp directory would silently lose data at reboot.
default in the temp directory would silently lose data at reboot
— and likewise when ``NESTQUEST_PANEL_TOKEN`` is unset or empty
(a default token is never invented: the panel plane is closed
until a real one is configured).
"""
source = env if env is not None else os.environ
raw = source.get(DB_PATH_ENV, "").strip()
if not raw:
raw_db_path = source.get(DB_PATH_ENV, "").strip()
if not raw_db_path:
raise ConfigError(
f"{DB_PATH_ENV} must be set to a writable SQLite file path "
f"(it is unset or empty — the API will not fall back to a "
f"temp directory and silently lose data)"
)
return cls(db_path=raw)
raw_panel_token = source.get(PANEL_TOKEN_ENV, "").strip()
if not raw_panel_token:
raise ConfigError(
f"{PANEL_TOKEN_ENV} must be set to the panel service token "
f"(it is unset or empty — the API will not start without "
f"one, and no default token is invented)"
)
return cls(db_path=raw_db_path, panel_token=raw_panel_token)
__all__ = ["ApiConfig", "ConfigError", "DB_PATH_ENV"]
__all__ = ["ApiConfig", "ConfigError", "DB_PATH_ENV", "PANEL_TOKEN_ENV"]
api/dependencies.py
+81
View File
@@ -0,0 +1,81 @@
"""Reusable FastAPI dependencies for the NestQuest API routes.
The panel plane (``/api/v1/panel/*``) is authenticated by the static
service token configured through ``NESTQUEST_PANEL_TOKEN`` (D-012): the
HA integration presents it as a Bearer credential on every panel
request. :func:`require_panel_token` is the ONE implementation of that
check — every panel route (this task's snapshot route and each later
panel route) declares it as a dependency rather than re-implementing
token handling, so the credential, the failure behaviour, and the
401 contract stay in exactly one place.
The dependency FAILS CLOSED: a missing ``Authorization`` header, a
malformed one (wrong scheme or empty credential), a token that does not
match the configured value, and an empty configured token all yield
HTTP 401. The comparison is constant-time (:func:`secrets.compare_digest`)
so the wall-clock difference between "wrong length" and "wrong value"
cannot be used to probe the token byte by byte, and every rejection
carries the same detail — nothing in the response distinguishes which
part of the check failed.
"""
from __future__ import annotations
import secrets
from typing import Annotated
from fastapi import Header, HTTPException, Request
#: The authorization scheme the panel plane expects. Matched
#: case-insensitively, as RFC 9110 requires for scheme names.
_BEARER_SCHEME = "bearer"
#: The 401 detail shared by every rejection — deliberately uniform so
#: the response never reveals which part of the check failed.
_REJECTION_DETAIL = "Missing or invalid panel service token"
#: Challenge header returned with a 401, per the Bearer auth scheme.
_WWW_AUTHENTICATE = {"WWW-Authenticate": "Bearer"}
def _reject() -> HTTPException:
"""Build the single 401 the dependency raises for every failure."""
return HTTPException(
status_code=401,
detail=_REJECTION_DETAIL,
headers=_WWW_AUTHENTICATE,
)
async def require_panel_token(
request: Request,
authorization: Annotated[str | None, Header()] = None,
) -> None:
"""Reject the request with 401 unless it carries the panel token.
Reads the configured token off ``app.state.config`` (set by
:func:`api.app.create_app`) so the dependency stays reusable across
routes and carries no module state of its own. A missing or
malformed ``Authorization`` header, a non-matching credential, and
an EMPTY configured token (a wiring mistake) all raise the same
``HTTPException(401)`` — the check is closed even when the service
is misconfigured.
Returns ``None`` on success; FastAPI discards a ``None`` return,
the route handler simply proceeds.
"""
config = getattr(request.app.state, "config", None)
expected = config.panel_token if config is not None else ""
if not expected or authorization is None:
raise _reject()
scheme, _, credential = authorization.partition(" ")
if scheme.lower() != _BEARER_SCHEME or not credential.strip():
raise _reject()
# compare_digest on bytes so a non-ASCII token raises nothing and a
# timing side channel on the str path is not available.
if not secrets.compare_digest(
credential.strip().encode("utf-8"), expected.encode("utf-8")
):
raise _reject()
__all__ = ["require_panel_token"]
api/nestquest_core.py
+16 -1
View File
@@ -107,4 +107,19 @@ core: ModuleType = _load_core()
core_db = importlib.import_module(f"{_CORE_MODULE_NAME}.db")
core_migrations = importlib.import_module(f"{_CORE_MODULE_NAME}.migrations")
__all__ = ["core", "core_db", "core_migrations"]
#: The panel snapshot builder and its payload shaper
#: (``core.snapshot.build_snapshot`` / ``instance_payload``) — the pure
#: assembly the panel routes call.
core_snapshot = importlib.import_module(f"{_CORE_MODULE_NAME}.snapshot")
#: The explicit settings object (``core.settings.NestQuestSettings``)
#: the snapshot builder's ``settings`` parameter takes.
core_settings = importlib.import_module(f"{_CORE_MODULE_NAME}.settings")
__all__ = [
"core",
"core_db",
"core_migrations",
"core_snapshot",
"core_settings",
]
api/routes_panel.py
+183
View File
@@ -0,0 +1,183 @@
"""Panel-plane routes for the NestQuest API service.
The panel plane (``/api/v1/panel/*``, D-012) serves the wall panel's
reads. Every route on :data:`router` requires the panel service token
through the ONE reusable dependency :func:`api.dependencies.require_panel_token`
(declared once at the router level, so each later panel route added to
this router inherits the check).
The snapshot route is a thin adapter over the bundled core: it performs
NO business logic. It reads the live database off ``app.state.db``,
resolves ONE timezone-aware ``now`` (the API host's local time — the
API is the household's "local" host now that the domain no longer lives
inside Home Assistant), calls the pure
:func:`nestquest_core.snapshot.build_snapshot`, shapes each child's
instances through :func:`nestquest_core.snapshot.instance_payload` with
``include_missed=False`` (D-009: missed instances are OMITTED from the
panel payload), and returns the documented JSON shape:
.. code-block:: json
{
"today_iso": "2026-09-22",
"cycle_day": 2,
"children": [
{
"child_id": 1,
"child_name": "Ada",
"present": false,
"next_present": "2026-09-24",
"due_today": 1,
"completed_today": 0,
"remaining_today": 1,
"completion_pct": 0,
"instances": [
{"id": 7, "definition_id": 3, "child_id": 1,
"title": "Pack bag", "icon": null, "window": "morning",
"due_time": "10:00", "state": "open", "overdue": true,
"completed_at": null, "on_time": null}
]
}
]
}
``children`` follows the household's sort order (the core builder lists
active children ``ORDER BY sort_order, id``); ``state`` is the
documented panel spelling ``open`` | ``completed`` (§1 of
design/ENTITIES-AND-SERVICES.md); ``cycle_day`` is today's 1-based day
in the first scheduled child's custody cycle (0 when no child has one).
The same :class:`PanelSnapshotResponse` Pydantic model is the route's
``response_model``, so the OpenAPI document carries the shape too.
"""
from __future__ import annotations
import datetime
from fastapi import APIRouter, Depends, Request
from pydantic import BaseModel
from api.database import DatabaseState
from api.dependencies import require_panel_token
from api.nestquest_core import core_snapshot, core_settings
#: All panel-plane routes share this router; the service-token check is
#: attached HERE so every panel route (current and later) requires it.
router = APIRouter(
prefix="/api/v1/panel",
dependencies=[Depends(require_panel_token)],
)
class PanelInstancePayload(BaseModel):
"""One instance in the documented panel payload shape (§1).
Mirrors :func:`nestquest_core.snapshot.instance_payload`:
``state`` is ``open`` | ``completed`` (missed instances never reach
the panel payload), ``on_time`` passes the completion event's flag
through verbatim (``None`` for an open instance), and
``completed_at`` is the event's UTC timestamp when completed.
"""
id: int
definition_id: int
child_id: int
title: str
icon: str | None
window: str
due_time: str | None
state: str
overdue: bool
completed_at: str | None
on_time: bool | None
class PanelChildPayload(BaseModel):
"""One child's day: presence, rollup counts, and today's instances."""
child_id: int
child_name: str
present: bool
#: ISO date of the child's next present day while they are away
#: (the away plate renders ``Returns <weekday>, <Mon D>`` from it);
#: ``None`` when the child is present today.
next_present: str | None
due_today: int
completed_today: int
remaining_today: int
completion_pct: int
instances: list[PanelInstancePayload]
class PanelSnapshotResponse(BaseModel):
"""The whole household snapshot the panel renders from.
``today_iso`` anchors the payload's day; ``cycle_day`` is today's
1-based day in the first scheduled child's custody cycle (0 when no
child has a schedule); ``children`` is in the household's sort
order.
"""
today_iso: str
cycle_day: int
children: list[PanelChildPayload]
def _local_now() -> datetime.datetime:
"""Return the API host's local time as ONE timezone-aware clock read.
``now().astimezone()`` attaches the host's local timezone to a
single clock read, so the calendar date and wall-clock time the
builder derives can never disagree. Kept as a module function (not
inlined) so tests pin the instant the same way the coordinator's
``_local_now`` is pinned.
"""
return datetime.datetime.now().astimezone()
@router.get(
"/snapshot",
summary="Today's household snapshot",
response_model=PanelSnapshotResponse,
)
async def panel_snapshot(request: Request) -> PanelSnapshotResponse:
"""Return today's full household snapshot for the panel.
Requires the panel service token (checked by the router's shared
:func:`~api.dependencies.require_panel_token` dependency). Builds
the snapshot through the pure core builder against the live
database and shapes each child's instances with
``instance_payload(include_missed=False)`` — missed instances are
OMITTED from the panel payload (D-009) — then returns the shape
documented in this module's docstring and
:class:`PanelSnapshotResponse`.
"""
state: DatabaseState = request.app.state.db
snapshot = await core_snapshot.build_snapshot(
state.database, core_settings.NestQuestSettings(), _local_now()
)
return PanelSnapshotResponse(
today_iso=snapshot.today_iso,
cycle_day=snapshot.cycle_day,
children=[
PanelChildPayload(
child_id=child.child_id,
child_name=child.child_name,
present=child.present,
next_present=child.next_present,
due_today=child.due_today,
completed_today=child.completed_today,
remaining_today=child.remaining_today,
completion_pct=child.completion_pct,
instances=[
PanelInstancePayload.model_validate(instance)
for instance in core_snapshot.instance_payload(
child.instances, include_missed=False
)
],
)
for child in snapshot.children
],
)
__all__ = ["router"]
tests/test_api_health.py
+17 -5
View File
@@ -69,7 +69,12 @@ class _AppRunner:
"""
def __init__(self, db_path: str) -> None:
self.app = create_app(ApiConfig(db_path=db_path))
# The token value is irrelevant to these tests (the health
# route is unauthenticated); it just satisfies the required
# field — the panel token's own tests live in test_api_panel.py.
self.app = create_app(
ApiConfig(db_path=db_path, panel_token="health-runner-token")
)
async def __aenter__(self) -> httpx.AsyncClient:
self._lifespan = self.app.router.lifespan_context(self.app)
@@ -97,9 +102,15 @@ async def app_client(temp_db_path: str) -> httpx.AsyncClient:
def test_from_env_returns_set_value() -> None:
"""A set, non-empty NESTQUEST_DB_PATH is returned verbatim."""
cfg = ApiConfig.from_env({"NESTQUEST_DB_PATH": "/data/nq.db"})
"""Set, non-empty env values are returned verbatim."""
cfg = ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_PANEL_TOKEN": "panel-token",
}
)
assert cfg.db_path == "/data/nq.db"
assert cfg.panel_token == "panel-token"
def test_from_env_raises_on_empty_value() -> None:
@@ -283,10 +294,11 @@ _NO_HA_SCRIPT = textwrap.dedent(
# Build the app (the lifespan is NOT run here, so the path is never
# opened — it just needs to be a non-empty string to satisfy the
# fail-fast config check). Use a temp path so nothing is created.
# fail-fast config check). Use a temp path so nothing is created;
# the panel token value is irrelevant to this import-only check.
app = create_app(ApiConfig(db_path=str(
Path(tempfile.gettempdir()) / "nq-noha-buildonly.db"
)))
), panel_token="noha-import-token"))
assert app is not None
print("API NO-HA IMPORT OK")
tests/test_api_panel.py
+523
View File
@@ -0,0 +1,523 @@
"""Panel-plane route tests: service token + snapshot (task ae7e90ae).
These tests exercise the NestQuest API's panel plane against the ASGI
app with httpx:
- ``GET /api/v1/panel/snapshot`` requires the panel service token
(D-012): a request with NO ``Authorization`` header, a MALFORMED
header, and a WRONG token all return 401; the correct Bearer token
returns 200.
- The 200 body is the documented snapshot shape: ``today_iso``,
``cycle_day``, and ``children`` in the household's sort order, each
carrying ``child_id``/``child_name``/``present``/``next_present``,
the per-day rollup counts, and its instances in the documented panel
payload shape (ENTITIES-AND-SERVICES.md §1) with missed instances
OMITTED (D-009).
- ``ApiConfig.from_env`` requires ``NESTQUEST_PANEL_TOKEN`` with the
same fail-fast discipline as ``NESTQUEST_DB_PATH``.
The household is seeded through the SAME core copy the app uses — the
``nestquest_core.*`` package registered by ``api/nestquest_core.py`` —
NOT the conftest-stubbed ``custom_components.nestquest.core`` copy (see
the two-copy caveat in api/nestquest_core.py's docstring: never compare
class or exception identity across the two copies). The seed opens and
migrates its own connection on the app's SQLite path and closes BEFORE
the app's lifespan opens the file, so the two connections never
overlap.
The route's single clock read (:func:`api.routes_panel._local_now`) is
pinned with monkeypatch the same way the coordinator's ``_local_now``
is pinned in test_core_snapshot.py, so the seeded rows and the
snapshot's ``today`` can never straddle midnight.
"""
from __future__ import annotations
import datetime
import importlib
from pathlib import Path
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import httpx
import pytest
from api import routes_panel
from api.app import create_app
from api.config import ApiConfig, ConfigError
from api.database import _executor
from api.nestquest_core import (
core_db,
core_migrations,
core_snapshot,
)
#: The panel service token the test apps are configured with.
PANEL_TOKEN = "panel-service-token-test"
# The bundled core submodules, through the SAME copy the app uses
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
_children = importlib.import_module("nestquest_core.children")
_quest_definitions = importlib.import_module(
"nestquest_core.quest_definitions"
)
_materialize = importlib.import_module("nestquest_core.materialize")
_recurrence = importlib.import_module("nestquest_core.recurrence")
_dao_presence = importlib.import_module("nestquest_core.dao_presence")
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
_completion = importlib.import_module("nestquest_core.completion")
async def _seed_household(
db_path: str,
today: datetime.date,
pinned_now: datetime.datetime,
) -> SimpleNamespace:
"""Seed a two-child household on ``db_path`` and return its records.
Mirrors the test_core_snapshot.py household: an AWAY child (2-week
custody cycle anchored yesterday, empty week 0, an override returns
her today+2) with an open quest due 10:00 — overdue at the pinned
12:00 — and a present child whose 17:00 quest is completed at
12:00 (on time). A third quest's instance is materialized due
YESTERDAY (via ``materialize`` with ``today`` pinned to yesterday,
since the no-past guard refuses past-dated writes): a past-due OPEN
instance, i.e. the "missed" view the panel payload must omit.
Opens, migrates, seeds, and CLOSES its own connection so the app's
lifespan connection never overlaps it.
"""
database = core_db.NestQuestDatabase(_executor)
await database.open(db_path)
try:
await core_migrations.apply_migrations(database)
today_iso = today.isoformat()
yesterday = today - datetime.timedelta(days=1)
yesterday_iso = yesterday.isoformat()
ada = await _children.create_child(database, "Ada", sort_order=0)
bo = await _children.create_child(database, "Bo", sort_order=1)
# Every rule starts YESTERDAY so the yesterday materialization
# below (which pins ``today`` to yesterday) fires them.
await _quest_definitions.create_quest_definition(
database,
"Pack bag",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": yesterday_iso}
),
[ada.id],
[("morning", "10:00")],
)
await _quest_definitions.create_quest_definition(
database,
"Brush teeth",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": yesterday_iso}
),
[bo.id],
[("morning", "17:00")],
)
# Today's instances first, while Ada has no presence schedule
# (she is present every day, so her instance IS generated).
await _materialize.materialize(
database,
today_iso,
(today + datetime.timedelta(days=3)).isoformat(),
today=today,
)
# The stale quest is created AFTER the today walk, so its only
# instance is the yesterday one written below.
await _quest_definitions.create_quest_definition(
database,
"Stale chore",
_recurrence.ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": yesterday_iso}
),
[ada.id],
[("morning", "09:00")],
)
# Materialize yesterday with ``today`` pinned to yesterday: the
# no-past guard accepts the rows, and they land past-due (no
# completion events) — the missed view.
await _materialize.materialize(
database, yesterday_iso, yesterday_iso, today=yesterday
)
instances = _dao_instances.QuestInstancesDao(database)
ada_today = await instances.list_by_child_and_date(ada.id, today_iso)
assert len(ada_today) == 1
pack_bag_instance = ada_today[0]
bo_today = await instances.list_by_child_and_date(bo.id, today_iso)
assert len(bo_today) == 1
brush_teeth_instance = bo_today[0]
ada_yesterday = await instances.list_by_child_and_date(
ada.id, yesterday_iso
)
stale_instance = next(
record
for record in ada_yesterday
if record.due_time == "09:00"
)
# Ada's custody schedule, added AFTER materialization (the
# same order test_core_snapshot.py seeds): 2-week cycle
# anchored yesterday, empty week 0 covers today => absent
# today; an override returns her on today+2.
await _dao_presence.PresenceSchedulesDao(database).upsert_by_child(
ada.id,
cycle_length_weeks=2,
anchor_date=yesterday_iso,
pattern="|0,1,2,3,4,5,6",
)
returns_date = today + datetime.timedelta(days=2)
await _dao_presence.PresenceOverridesDao(database).create(
ada.id,
returns_date.isoformat(),
returns_date.isoformat(),
is_present=True,
note="visit",
)
# Complete Bo's today instance at the pinned 12:00 (17:00 due
# => on time).
await _completion.complete_instance(
database,
brush_teeth_instance.id,
actor_source="panel",
actor_child_id=bo.id,
now=pinned_now,
today=today,
)
return SimpleNamespace(
ada=ada,
bo=bo,
pack_bag_instance=pack_bag_instance,
brush_teeth_instance=brush_teeth_instance,
stale_instance=stale_instance,
returns_date=returns_date,
)
finally:
await database.close()
class _PanelRunner:
"""Run the app's lifespan around an httpx AsyncClient.
Same pattern as test_api_health.py's ``_AppRunner``: httpx's ASGI
transport does not run the lifespan, so it is driven explicitly and
shut down in ``__aexit__``'s finally block even on failure.
"""
def __init__(
self, db_path: str, panel_token: str = PANEL_TOKEN
) -> None:
self.app = create_app(
ApiConfig(db_path=db_path, panel_token=panel_token)
)
async def __aenter__(self) -> httpx.AsyncClient:
self._lifespan = self.app.router.lifespan_context(self.app)
await self._lifespan.__aenter__()
self._client = httpx.AsyncClient(
transport=httpx.ASGITransport(app=self.app),
base_url="http://testserver",
)
return await self._client.__aenter__()
async def __aexit__(self, *exc) -> None:
try:
await self._client.__aexit__(*exc)
finally:
await self._lifespan.__aexit__(*exc)
@pytest.fixture
def temp_db_path(tmp_path: Path) -> str:
"""A fresh per-test SQLite path under pytest's tmp_path."""
return str(tmp_path / "nestquest.db")
@pytest.fixture
async def panel_client(temp_db_path: str) -> httpx.AsyncClient:
"""An app client with an empty (migrated) household database."""
async with _PanelRunner(temp_db_path) as client:
yield client
@pytest.fixture
async def seeded_panel(temp_db_path: str, monkeypatch) -> SimpleNamespace:
"""A seeded household served by the app, with the clock pinned.
One host clock read anchors ``today``; the pinned instant is
12:00 UTC that day (past Ada's 10:00 due, before Bo's 17:00). The
route's ``_local_now`` is patched to return the pinned instant.
"""
today = datetime.datetime.now().astimezone().date()
time_zone = ZoneInfo("UTC")
pinned_now = datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=time_zone
)
seed = await _seed_household(temp_db_path, today, pinned_now)
monkeypatch.setattr(routes_panel, "_local_now", lambda: pinned_now)
async with _PanelRunner(temp_db_path) as client:
yield SimpleNamespace(
client=client,
seed=seed,
today=today,
pinned_now=pinned_now,
pinned_utc_iso=pinned_now.astimezone(
datetime.timezone.utc
).isoformat(),
)
# --- ApiConfig.from_env: the panel token --------------------------------
def test_from_env_reads_panel_token() -> None:
"""A set, non-empty NESTQUEST_PANEL_TOKEN is returned verbatim."""
cfg = ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_PANEL_TOKEN": "panel-token",
}
)
assert cfg.panel_token == "panel-token"
def test_from_env_requires_panel_token_when_missing() -> None:
"""A missing panel token raises ConfigError (no silent default)."""
with pytest.raises(ConfigError):
ApiConfig.from_env({"NESTQUEST_DB_PATH": "/data/nq.db"})
def test_from_env_requires_panel_token_when_empty() -> None:
"""An empty/whitespace panel token raises ConfigError."""
with pytest.raises(ConfigError):
ApiConfig.from_env(
{
"NESTQUEST_DB_PATH": "/data/nq.db",
"NESTQUEST_PANEL_TOKEN": " ",
}
)
# --- service-token auth --------------------------------------------------
async def test_missing_token_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A request with no Authorization header returns 401."""
response = await panel_client.get("/api/v1/panel/snapshot")
assert response.status_code == 401
assert "www-authenticate" in {
key.lower() for key in response.headers
}
async def test_wrong_token_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A wrong Bearer token returns 401."""
response = await panel_client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": "Bearer not-the-panel-token"},
)
assert response.status_code == 401
async def test_malformed_authorization_header_is_unauthorized(
panel_client: httpx.AsyncClient,
) -> None:
"""A malformed Authorization header returns 401.
Wrong scheme, empty credential, and an empty header all fail the
same way — the response never reveals which part failed.
"""
for header in (
"Basic dXNlcjpwYXNz",
"Bearer",
"Bearer ",
"Token not-the-panel-token",
"",
):
response = await panel_client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": header},
)
assert response.status_code == 401, f"header {header!r} passed"
# --- snapshot payload ----------------------------------------------------
async def test_valid_token_returns_full_snapshot(
seeded_panel: SimpleNamespace,
) -> None:
"""The correct token returns 200 with the documented snapshot shape.
Children in sort order (Ada sort_order 0, Bo 1); per-child counts,
presence, next_present, and cycle day all asserted as explicit
literals; each child's instances are in the documented panel
payload shape (done -> completed spelling, on_time passthrough).
"""
seed = seeded_panel.seed
response = await seeded_panel.client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
)
assert response.status_code == 200
body = response.json()
assert body["today_iso"] == seeded_panel.today.isoformat()
# Ada's 2-week cycle anchored yesterday => today is cycle day 2.
assert body["cycle_day"] == 2
children = body["children"]
assert [child["child_id"] for child in children] == [
seed.ada.id,
seed.bo.id,
], "children appear in sort_order"
ada, bo = children
# Ada: away today, returns on the override date; her 10:00 quest
# is open and overdue at the pinned 12:00.
assert ada["child_name"] == "Ada"
assert ada["present"] is False
assert ada["next_present"] == seed.returns_date.isoformat()
assert ada["due_today"] == 1
assert ada["completed_today"] == 0
assert ada["remaining_today"] == 1
assert ada["completion_pct"] == 0
assert ada["instances"] == [
{
"id": seed.pack_bag_instance.id,
"definition_id": seed.pack_bag_instance.definition_id,
"child_id": seed.ada.id,
"title": "Pack bag",
"icon": None,
"window": "morning",
"due_time": "10:00",
"state": "open",
"overdue": True,
"completed_at": None,
"on_time": None,
}
]
# Bo: present, completed his 17:00 quest at the pinned 12:00
# (on time), so the panel spelling is "completed".
assert bo["child_name"] == "Bo"
assert bo["present"] is True
assert bo["next_present"] is None
assert bo["due_today"] == 1
assert bo["completed_today"] == 1
assert bo["remaining_today"] == 0
assert bo["completion_pct"] == 100
assert bo["instances"] == [
{
"id": seed.brush_teeth_instance.id,
"definition_id": seed.brush_teeth_instance.definition_id,
"child_id": seed.bo.id,
"title": "Brush teeth",
"icon": None,
"window": "morning",
"due_time": "17:00",
"state": "completed",
"overdue": False,
"completed_at": seeded_panel.pinned_utc_iso,
"on_time": True,
}
]
async def test_missed_instances_are_omitted_from_panel_payload(
seeded_panel: SimpleNamespace,
) -> None:
"""A missed instance never reaches the panel payload (D-009).
The household carries a past-due OPEN instance due yesterday (the
state the panel reads as "missed"); it must not appear anywhere in
the response, and Ada's payload carries exactly her ONE today
instance — the past-due rows (the stale chore AND her pack bag due
yesterday) are both omitted rather than rendered.
"""
seed = seeded_panel.seed
response = await seeded_panel.client.get(
"/api/v1/panel/snapshot",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
)
assert response.status_code == 200
body = response.json()
payload_instances = [
instance
for child in body["children"]
for instance in child["instances"]
]
assert "missed" not in [
instance["state"] for instance in payload_instances
]
past_due_ids = {seed.stale_instance.id}
assert past_due_ids.isdisjoint(
instance["id"] for instance in payload_instances
), "the past-due open instance leaked into the panel payload"
ada = next(
child
for child in body["children"]
if child["child_id"] == seed.ada.id
)
assert [instance["id"] for instance in ada["instances"]] == [
seed.pack_bag_instance.id
]
def test_panel_shaper_drops_missed_views() -> None:
"""The route's shaper omits a missed view (D-009), on the app's copy.
``instance_payload(include_missed=False)`` is what the snapshot
route shapes every child's instances with; exercised here on the
SAME core copy the app uses (the two-copy caveat makes this a
distinct object from the conftest-stubbed copy's identical source).
"""
missed_view = core_snapshot.QuestInstanceView(
instance_id=999,
definition_id=1,
child_id=1,
title="Stale chore",
icon=None,
window="morning",
due_date="2026-09-21",
due_time="09:00",
state="missed",
overdue=False,
completed_at=None,
was_on_time=None,
)
assert core_snapshot.instance_payload(
[missed_view], include_missed=False
) == []
# --- route documentation -------------------------------------------------
async def test_snapshot_route_is_documented(
panel_client: httpx.AsyncClient,
) -> None:
"""The snapshot route and its response shape are in the OpenAPI doc."""
response = await panel_client.get("/openapi.json")
assert response.status_code == 200
schema = response.json()
assert "/api/v1/panel/snapshot" in schema["paths"]
assert "get" in schema["paths"]["/api/v1/panel/snapshot"]
documented = {
model["title"]
for model in schema["components"]["schemas"].values()
}
assert "PanelSnapshotResponse" in documented
assert "PanelChildPayload" in documented
assert "PanelInstancePayload" in documented