Compare commits

...
Author SHA1 Message Date
overseer 7a26845abe Admin routes: quest definitions
The admin plane gains quest-definition routes on the existing admin
router, mirroring the children routes' thin-adapter pattern: the body
model carries the shape, the handler builds ONE ScheduleRule from the
request's rule object through the recurrence model's own from_dict
(rule validation stays the model's construction-time job), and every
call goes to ONE nestquest_core.quest_definitions business function —
the title/window/assignee policies, the whole-set window replacement
and the never-hard-delete policy all live in the core, never in a
handler.

- POST /quest-definitions: title + rule + assignees + windows (+
  optional description/icon), persisted atomically by the core.
- PATCH /quest-definitions/{id}: only supplied fields are forwarded
  (exclude_unset); an explicit null description/icon clears it, a null
  rule/windows is rejected; assignment stays unsettable here.
- PATCH /quest-definitions/{id}/active: the only removal path — the
  flag flips, the row, rule, assignees and windows survive.

Error mapping stays deliberately narrow: the core's definition_id:-
prefixed not-exist error is 404; every other ValueError (including
RuleValidationError) is 422 with the core's message, so a rejected
rule, window, assignee or title from the BODY never disguises itself
as a missing resource and an unexpected error is never swallowed as a
4xx.

Task: b586a2f1-1532-4194-a19a-8f8386d22590 (Feature 16).
2026-09-23 01:44:50 -04:00
dev_agent 0b7f71831b Admin routes: children (#158) 2026-09-23 01:23:57 -04:00
7 changed files with 1953 additions and 181 deletions
+21
View File
@@ -124,12 +124,33 @@ 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 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")
#: The quest-definition business layer (``core.quest_definitions`` —
#: ``create_quest_definition``, ``edit_quest_definition``,
#: ``set_quest_definition_active``) the admin plane's quest-definition
#: routes call.
core_quest_definitions = importlib.import_module(
f"{_CORE_MODULE_NAME}.quest_definitions"
)
#: The recurrence model (``core.recurrence.ScheduleRule`` and
#: ``RuleValidationError``) the quest-definition routes build request
#: rules with — the model validates at construction, the routes never
#: re-implement rule validation.
core_recurrence = importlib.import_module(f"{_CORE_MODULE_NAME}.recurrence")
__all__ = [
"core",
"core_children",
"core_completion",
"core_db",
"core_events",
"core_migrations",
"core_quest_definitions",
"core_recurrence",
"core_snapshot",
"core_settings",
]
+626 -7
View File
@@ -8,20 +8,79 @@ the router level, so each later admin route added to this router
inherits the check). The panel service token is refused outright on
this plane (403), never treated as a JWT.
This task is the AUTH probe only: the ping route is a thin adapter
that performs NO business logic and NO auth of its own — it reuses the
router's already-verified claims (FastAPI caches a dependency's result
per request, so the check runs exactly once) and returns the status
plus the token's subject.
The children routes are THIN adapters over the bundled core (Feature
15): each handler validates the request's shape with a Pydantic model,
resolves the live database off ``app.state.db``, calls ONE
:mod:`nestquest_core.children` business function — the name policy,
the no-explicit-NULL rule, the real-bool check and the permutation
check all live THERE, never in a handler — and serializes the returned
:class:`~nestquest_core.dao_children.ChildRecord` into the documented
payload (``id``, ``display_name``, ``colour``, ``avatar_ref``,
``sort_order``, ``is_active``).
The quest-definition routes follow the same pattern over
:mod:`nestquest_core.quest_definitions`: the body model carries the
shape, the handler builds ONE
:class:`~nestquest_core.recurrence.ScheduleRule` from the request's
``rule`` object through the model's own ``from_dict`` (rule validation
is the MODEL's construction-time job — a bad combination raises
:class:`~nestquest_core.recurrence.RuleValidationError`, never
hand-rolled here), and the call goes to
``create_quest_definition`` / ``edit_quest_definition`` /
``set_quest_definition_active``. The title policy, the window-name
and due-time policy, the assignee checks, the whole-set window
replacement and the never-hard-delete policy (deactivation is the only
removal path) all live in the core. Assignment is NOT settable
through the API: the core's edit path takes no assignees. The edit
route forwards only SUPPLIED fields (``exclude_unset``), like the
children edit.
Error mapping (every children and quest-definition route, deliberately
narrow):
- Malformed input the body model or the path parser rejects (a missing
or wrong-typed field, a non-integer child or definition id) is
FastAPI's 422 before any handler code runs.
- A core ``ValueError`` reporting a NON-EXISTENT child (edit or
set_active on an unknown id) is mapped to 404.
- A core ``ValueError`` reporting a NON-EXISTENT DEFINITION (edit or
set_active on an unknown id — the core prefixes that error with
``definition_id:``) is mapped to 404. Create's rejected-assignee
error names a child from the BODY, not a path id, so it maps to 422.
- Every OTHER core ``ValueError`` — an empty display name, an explicit
``null`` colour/avatar_ref (clearing is not supported), a reorder
list that is not a complete permutation of the children (missing,
duplicate, or unknown ids); a rejected rule
(:class:`~nestquest_core.recurrence.RuleValidationError`: unknown
``rule_type``, a shape-missing field such as a weekly rule without
weekdays, a malformed date, a forbidden field), a bad window name or
due time, a rejected window or assignee list (empty, duplicate,
wrong-typed) — is mapped to 422: the body was well-formed JSON but
semantically invalid, the same failure class Pydantic reports. The
core rejects these BEFORE any write, so a 422 never leaves a
half-applied change behind.
PATCH edit semantics: the body model's fields are optional and only
SUPPLIED fields are forwarded (``model_dump(exclude_unset=True)``), so
an omitted field is never passed to
:func:`nestquest_core.children.edit_child` and stays unchanged — while
an explicit ``null`` IS passed and is rejected by the core rather than
silently ignored.
"""
from __future__ import annotations
from typing import Annotated
from typing import Annotated, NoReturn
from fastapi import APIRouter, Depends
from fastapi import APIRouter, Depends, HTTPException, Request
from pydantic import BaseModel
from api.auth import require_admin
from api.database import DatabaseState
from api.nestquest_core import (
core_children,
core_quest_definitions,
core_recurrence,
)
#: All admin-plane routes share this router; the ONE admin dependency
#: (service-token refusal + JWT verification + nestquest-admins group
@@ -42,6 +101,303 @@ class AdminPingResponse(BaseModel):
sub: str | None
class AdminChildResponse(BaseModel):
"""One child profile as the admin plane serializes it.
Mirrors :class:`~nestquest_core.dao_children.ChildRecord` minus the
internal ``created_at`` stamp.
"""
id: int
display_name: str
colour: str | None
avatar_ref: str | None
sort_order: int
is_active: bool
class AdminChildListResponse(BaseModel):
"""The children list: every profile in the household's display order."""
children: list[AdminChildResponse]
class AdminRuleResponse(BaseModel):
"""A definition's schedule rule as the admin plane serializes it.
Mirrors :meth:`nestquest_core.recurrence.ScheduleRule.to_dict`: the
model-name ``rule_type`` (e.g. ``monthly_weekday``), the weekday
set as a sorted list, and ``None`` for every field the rule's shape
does not use.
"""
rule_type: str
interval: int
weekday_set: list[int] | None
day_of_month: int | None
nth_weekday: int | None
nth_weekday_weekday: int | None
month: int | None
start_date: str
end_date: str | None
class AdminAssigneeResponse(BaseModel):
"""One assigned child: the id and display name the PWA renders."""
id: int
display_name: str
class AdminDefinitionWindowResponse(BaseModel):
"""One window declaration: its name and optional HH:MM due time."""
window: str
due_time: str | None
class AdminQuestDefinitionResponse(BaseModel):
"""One quest definition as the admin plane serializes it.
Mirrors the core's
:class:`~nestquest_core.quest_definitions.CreatedQuestDefinition`
bundle: the stored row (minus the internal ``schedule_rule_id``/
``due_time``/``created_at`` columns) plus its decoded rule,
assignees and windows.
"""
id: int
title: str
description: str | None
icon: str | None
is_active: bool
rule: AdminRuleResponse
assignees: list[AdminAssigneeResponse]
windows: list[AdminDefinitionWindowResponse]
class AdminChildCreateRequest(BaseModel):
"""The body of ``POST /api/v1/admin/children``.
``display_name`` is required (the core trims it and rejects an
empty name); the other fields are optional and default through the
core layer (``colour``/``avatar_ref`` NULL, ``sort_order`` 0).
"""
display_name: str
colour: str | None = None
avatar_ref: str | None = None
sort_order: int = 0
class AdminChildEditRequest(BaseModel):
"""The body of ``PATCH /api/v1/admin/children/{child_id}``.
Every field is optional; ONLY the fields the request supplies are
forwarded to :func:`nestquest_core.children.edit_child` (the route
dumps the model with ``exclude_unset``), so an omitted field stays
unchanged while an explicit ``null`` — a supplied value — reaches
the core and is rejected there rather than silently ignored.
"""
display_name: str | None = None
colour: str | None = None
avatar_ref: str | None = None
sort_order: int | None = None
class AdminChildActiveRequest(BaseModel):
"""The body of ``PATCH /api/v1/admin/children/{child_id}/active``."""
is_active: bool
class AdminChildReorderRequest(BaseModel):
"""The body of ``POST /api/v1/admin/children/reorder``.
``ordered_ids`` must be a complete permutation of the household's
child ids (the core rejects a partial, duplicate, or unknown list
before any write).
"""
ordered_ids: list[int]
class AdminRuleRequest(BaseModel):
"""The ``rule`` object of a quest-definition create or edit body.
The fields of :class:`~nestquest_core.recurrence.ScheduleRule` —
nothing more: the shape is the model's, the defaults are the
model's (``interval`` 1, ``start_date`` 1970-01-01). Only
``rule_type`` is required. The COMBINATION policy (which fields a
shape requires and forbids, the 1..31/1..12/0..6 ranges, the strict
ISO dates) is enforced by the model's constructor, which the route
builds through ``from_dict`` — this model only guards the SHAPE.
"""
rule_type: str
interval: int = 1
weekday_set: list[int] | None = None
day_of_month: int | None = None
nth_weekday: int | None = None
nth_weekday_weekday: int | None = None
month: int | None = None
start_date: str = "1970-01-01"
end_date: str | None = None
class AdminQuestDefinitionCreateRequest(BaseModel):
"""The body of ``POST /api/v1/admin/quest-definitions``.
``title``, ``rule``, ``assignee_child_ids`` and ``windows`` are
required (the core rejects an empty title, a non-list or empty
assignee/window list, and unknown window names); ``description``
and ``icon`` are optional and default to NULL through the core.
``windows`` entries are a window name (``morning``) or a two-item
``[window, due_time]`` pair whose due time is a strict 24-hour
HH:MM (or ``null`` for none) — both forms the core normalizes.
"""
title: str
rule: AdminRuleRequest
assignee_child_ids: list[int]
windows: list[str | tuple[str, str | None]]
description: str | None = None
icon: str | None = None
class AdminQuestDefinitionEditRequest(BaseModel):
"""The body of ``PATCH /api/v1/admin/quest-definitions/{definition_id}``.
Every field is optional; ONLY the fields the request supplies are
forwarded to :func:`nestquest_core.quest_definitions.edit_quest_definition`
(the route dumps the model with ``exclude_unset``), so an omitted
field stays unchanged. An explicit ``null`` ``description``/
``icon`` IS passed and CLEARS the stored value (the core supports
clearing here, unlike the children edit); an explicit ``null``
``rule`` or ``windows`` is passed and rejected by the core. A
supplied ``windows`` list REPLACES the whole window set, and
``assignees`` are deliberately absent — assignment is not settable
through this route.
"""
title: str | None = None
description: str | None = None
icon: str | None = None
rule: AdminRuleRequest | None = None
windows: list[str | tuple[str, str | None]] | None = None
class AdminQuestDefinitionActiveRequest(BaseModel):
"""The body of ``PATCH /api/v1/admin/quest-definitions/{id}/active``."""
is_active: bool
#: 404 detail for a child id the core layer reports as non-existent.
_CHILD_NOT_FOUND_DETAIL = "Child not found"
def _raise_child_error(error: ValueError) -> NoReturn:
"""Map a core children ``ValueError`` onto 404 or 422 and raise it.
The core layer raises ``ValueError`` for two distinct situations,
and this mapping is the children routes' one piece of adapter
logic: a non-existent child on a single-id route (edit, set_active
— the message reads ``child N does not exist``) becomes 404, the
admin plane's not-found answer; every other core ``ValueError``
becomes 422 with the core's message as the detail — it names the
field and what to fix, and leaks nothing but that validation rule.
"""
if "does not exist" in str(error):
raise HTTPException(
status_code=404, detail=_CHILD_NOT_FOUND_DETAIL
) from error
raise HTTPException(status_code=422, detail=str(error)) from error
def _child_response(record: core_children.ChildRecord) -> AdminChildResponse:
"""Serialize one core ChildRecord into the documented payload."""
return AdminChildResponse(
id=record.id,
display_name=record.display_name,
colour=record.colour,
avatar_ref=record.avatar_ref,
sort_order=record.sort_order,
is_active=record.is_active,
)
#: 404 detail for a definition id the core layer reports as non-existent.
_QUEST_DEFINITION_NOT_FOUND_DETAIL = "Quest definition not found"
def _raise_quest_definition_error(error: ValueError) -> NoReturn:
"""Map a core quest-definitions ``ValueError`` onto 404 or 422.
The mapping mirrors :func:`_raise_child_error` and stays narrow:
the core names a missing definition by prefixing its error with
``definition_id:`` (edit or set_active on an unknown id — its id
validator's message reads ``definition_id must be an integer``
WITHOUT the colon, so a malformed id stays 422), and that one case
is the quest-definition routes' 404. Every other ``ValueError`` —
including :class:`~nestquest_core.recurrence.RuleValidationError`,
which subclasses it — becomes 422 with the core's message as the
detail: create's rejected assignee names a child from the BODY (not
a path id, so not 404), a rejected rule/window/title names the
field and what to fix, and none of it leaks more than the rule it
enforces.
"""
if str(error).startswith("definition_id:"):
raise HTTPException(
status_code=404, detail=_QUEST_DEFINITION_NOT_FOUND_DETAIL
) from error
raise HTTPException(status_code=422, detail=str(error)) from error
def _rule_from_request(rule: AdminRuleRequest) -> core_recurrence.ScheduleRule:
"""Build the core ScheduleRule from the request's ``rule`` object.
The recurrence model validates AT CONSTRUCTION — through its own
``from_dict`` here, which raises
:class:`~nestquest_core.recurrence.RuleValidationError` (a
``ValueError``) for any bad combination — so the route re-implements
no rule policy of its own.
"""
return core_recurrence.ScheduleRule.from_dict(
rule.model_dump(exclude_unset=True)
)
def _rule_response(rule: core_recurrence.ScheduleRule) -> AdminRuleResponse:
"""Serialize one core ScheduleRule via its lossless dict form."""
return AdminRuleResponse(**rule.to_dict())
def _quest_definition_response(
bundle: core_quest_definitions.CreatedQuestDefinition,
) -> AdminQuestDefinitionResponse:
"""Serialize one core CreatedQuestDefinition into the payload."""
return AdminQuestDefinitionResponse(
id=bundle.definition.id,
title=bundle.definition.title,
description=bundle.definition.description,
icon=bundle.definition.icon,
is_active=bundle.definition.is_active,
rule=_rule_response(bundle.rule),
assignees=[
AdminAssigneeResponse(id=child.id, display_name=child.display_name)
for child in bundle.assignees
],
windows=[
AdminDefinitionWindowResponse(
window=window.window, due_time=window.due_time
)
for window in bundle.windows
],
)
@router.get("/ping", summary="Admin plane probe")
async def admin_ping(
claims: Annotated[dict[str, object], Depends(require_admin)],
@@ -62,4 +418,267 @@ async def admin_ping(
)
@router.get(
"/children",
summary="List child profiles",
response_model=AdminChildListResponse,
)
async def admin_list_children(request: Request) -> AdminChildListResponse:
"""Return every child profile in display order.
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.children.list_children` call, active AND
inactive children included (the admin screen manages both), in the
core's display order (``sort_order``, then ``id``).
"""
state: DatabaseState = request.app.state.db
records = await core_children.list_children(state.database)
return AdminChildListResponse(
children=[_child_response(record) for record in records]
)
@router.post(
"/children",
summary="Create a child profile",
status_code=201,
response_model=AdminChildResponse,
)
async def admin_create_child(
body: AdminChildCreateRequest, request: Request
) -> AdminChildResponse:
"""Create a child profile and return it as stored.
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 model
carries the shape, and the name policy (trim, reject empty), the
duplicate-name warning and the created_at stamp all live in
:func:`nestquest_core.children.create_child`. Errors are mapped by
:func:`_raise_child_error` — create names no existing child, so a
rejection here is always 422.
"""
state: DatabaseState = request.app.state.db
try:
record = await core_children.create_child(
state.database,
body.display_name,
colour=body.colour,
avatar_ref=body.avatar_ref,
sort_order=body.sort_order,
)
except ValueError as error:
_raise_child_error(error)
return _child_response(record)
@router.patch(
"/children/{child_id}",
summary="Edit a child profile",
response_model=AdminChildResponse,
)
async def admin_edit_child(
child_id: int, body: AdminChildEditRequest, request: Request
) -> AdminChildResponse:
"""Edit a child profile; only supplied fields change.
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 supplied
fields are forwarded with ``exclude_unset`` semantics so an omitted
field is never passed to :func:`nestquest_core.children.edit_child`
and stays unchanged, while an explicit ``null`` is passed and
rejected by the core (clearing is not supported). An unknown child
id is mapped to 404 and any other core ``ValueError`` to 422
(:func:`_raise_child_error`).
"""
state: DatabaseState = request.app.state.db
try:
record = await core_children.edit_child(
state.database,
child_id,
**body.model_dump(exclude_unset=True),
)
except ValueError as error:
_raise_child_error(error)
return _child_response(record)
@router.patch(
"/children/{child_id}/active",
summary="Set a child's active flag",
response_model=AdminChildResponse,
)
async def admin_set_child_active(
child_id: int, body: AdminChildActiveRequest, request: Request
) -> AdminChildResponse:
"""Deactivate (or reactivate) a child; the profile is NEVER deleted.
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: ``is_active``
is passed straight to
:func:`nestquest_core.children.set_child_active` — deactivation is
the only removal path (the core has no delete, so completion
history keeps its references). An unknown child id is mapped to
404 (:func:`_raise_child_error`); the body model already makes the
core's real-bool check unreachable from this route.
"""
state: DatabaseState = request.app.state.db
try:
record = await core_children.set_child_active(
state.database, child_id, body.is_active
)
except ValueError as error:
_raise_child_error(error)
return _child_response(record)
@router.post(
"/children/reorder",
summary="Reorder the children's panel display order",
)
async def admin_reorder_children(
body: AdminChildReorderRequest, request: Request
) -> dict[str, str]:
"""Rewrite the children's display order in one atomic core call.
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 posted id
order goes straight to
:func:`nestquest_core.children.reorder_children`, which rejects a
list that is not a complete permutation of the household (missing,
duplicate, or unknown ids) BEFORE any write — mapped to 422 by
:func:`_raise_child_error`, leaving the stored order untouched. On
success the core returns ``None``; the route answers
``{"status": "ok"}`` (the fresh order is read back via
``GET /children``).
"""
state: DatabaseState = request.app.state.db
try:
await core_children.reorder_children(
state.database, body.ordered_ids
)
except ValueError as error:
_raise_child_error(error)
return {"status": "ok"}
@router.post(
"/quest-definitions",
summary="Create a quest definition",
status_code=201,
response_model=AdminQuestDefinitionResponse,
)
async def admin_create_quest_definition(
body: AdminQuestDefinitionCreateRequest, request: Request
) -> AdminQuestDefinitionResponse:
"""Create a quest definition (rule + assignees + windows) and return it.
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 model
carries the shape, the request's ``rule`` becomes ONE
:class:`~nestquest_core.recurrence.ScheduleRule` (validated by the
model's constructor, :func:`_rule_from_request`), and the title
policy, the assignee and window policies and the atomic
rule+definition+assignees+windows persistence all live in
:func:`nestquest_core.quest_definitions.create_quest_definition`.
Create names no existing definition, so a missing definition can
never be its error: every rejection here is 422
(:func:`_raise_quest_definition_error`).
"""
state: DatabaseState = request.app.state.db
try:
bundle = await core_quest_definitions.create_quest_definition(
state.database,
body.title,
_rule_from_request(body.rule),
body.assignee_child_ids,
body.windows,
description=body.description,
icon=body.icon,
)
except ValueError as error:
_raise_quest_definition_error(error)
return _quest_definition_response(bundle)
@router.patch(
"/quest-definitions/{definition_id}",
summary="Edit a quest definition",
response_model=AdminQuestDefinitionResponse,
)
async def admin_edit_quest_definition(
definition_id: int,
body: AdminQuestDefinitionEditRequest,
request: Request,
) -> AdminQuestDefinitionResponse:
"""Edit a quest definition; only supplied fields change.
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 supplied
fields are forwarded with ``exclude_unset`` semantics so an omitted
field is never passed to
:func:`nestquest_core.quest_definitions.edit_quest_definition` and
stays unchanged; a supplied ``windows`` list replaces the WHOLE
window set in the core, and a supplied ``rule`` is rebuilt into a
ScheduleRule (:func:`_rule_from_request`, validated by the model).
An explicit ``null`` ``rule``/``windows`` reaches the core as None
and is rejected there (422), while an explicit ``null``
``description``/``icon`` clears the stored value — the core edit
path supports clearing, unlike the children edit. An unknown
definition id is mapped to 404 and any other core ``ValueError``
to 422 (:func:`_raise_quest_definition_error`).
"""
state: DatabaseState = request.app.state.db
supplied = body.model_dump(exclude_unset=True)
try:
if isinstance(body.rule, AdminRuleRequest):
supplied["rule"] = _rule_from_request(body.rule)
bundle = await core_quest_definitions.edit_quest_definition(
state.database, definition_id, **supplied
)
except ValueError as error:
_raise_quest_definition_error(error)
return _quest_definition_response(bundle)
@router.patch(
"/quest-definitions/{definition_id}/active",
summary="Set a quest definition's active flag",
response_model=AdminQuestDefinitionResponse,
)
async def admin_set_quest_definition_active(
definition_id: int,
body: AdminQuestDefinitionActiveRequest,
request: Request,
) -> AdminQuestDefinitionResponse:
"""Deactivate (or reactivate) a definition; it is NEVER deleted.
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: ``is_active``
goes straight to
:func:`nestquest_core.quest_definitions.set_quest_definition_active`
— deactivation is the only removal path (the row, rule, assignees
and windows all survive, so completion history keeps its
references). An unknown definition id is mapped to 404
(:func:`_raise_quest_definition_error`); the body model already
makes the core's real-bool check unreachable from this route.
"""
state: DatabaseState = request.app.state.db
try:
bundle = await core_quest_definitions.set_quest_definition_active(
state.database, definition_id, body.is_active
)
except ValueError as error:
_raise_quest_definition_error(error)
return _quest_definition_response(bundle)
__all__ = ["router"]
+189
View File
@@ -0,0 +1,189 @@
"""Shared admin-plane JWT test harness: local keys, stubbed JWKS.
The admin-plane API test modules share ONE harness instead of
duplicating it: the auth contract tests (:mod:`tests.test_api_admin_auth`)
and the children route tests (:mod:`tests.test_api_admin_children`) both
drive the ASGI app with httpx using a locally generated RSA key pair and
a stubbed JWKS fetcher — NO network access:
- ``ADMIN_KEY`` signs the test tokens; :func:`jwks_for` builds the JWKS
document publishing its public half, so genuinely signed tokens
verify, and tokens signed with the second pair (``OTHER_KEY``) fail
the signature check for real.
- :func:`make_token` mints one RS256 JWT the way Authentik would (its
claims, our key), defaulting ``groups`` to the admin group a real
admin's token carries.
- :class:`StubbedFetch` is the counting JWKS fetcher stub.
- :class:`AdminRunner` runs the app's lifespan around an httpx
AsyncClient (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) with the app's JWKS cache swapped for one built
around the stub fetcher.
The harness carries no tests itself; the ``admin_client`` fixture in
``tests/conftest.py`` hands its runner to the test modules.
"""
from __future__ import annotations
import base64
import time
import httpx
import jwt
from cryptography.hazmat.primitives.asymmetric import rsa
from api.app import create_app
from api.auth import ADMIN_GROUP, JwksCache
from api.config import ApiConfig
#: The configured OIDC settings the test apps run with (values a real
#: deployment would get from the NESTQUEST_OIDC_* environment vars).
ISSUER = "https://authentik.example.com/application/o/nestquest/"
AUDIENCE = "nestquest-api"
JWKS_URL = "https://authentik.example.com/application/o/nestquest/jwks/"
#: The ``kid`` the test tokens carry, matching the JWKS entry.
KID = "admin-test-key-1"
#: The subject a valid token authenticates as.
SUBJECT = "admin-user-1"
def _generate_key() -> rsa.RSAPrivateKey:
"""Generate one local RSA key pair (no network, no fixtures)."""
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
#: The key pair whose PUBLIC half the stubbed JWKS publishes — tokens
#: signed with it verify; the second pair below proves they don't when
#: signed otherwise.
ADMIN_KEY = _generate_key()
OTHER_KEY = _generate_key()
def _b64url_uint(value: int) -> str:
"""Base64url-encode a non-negative integer (no padding), per JWK."""
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def jwks_for(private_key: rsa.RSAPrivateKey, kid: str) -> dict[str, object]:
"""Build a JWKS document publishing ``private_key``'s public half."""
numbers = private_key.public_key().public_numbers()
return {
"keys": [
{
"kty": "RSA",
"use": "sig",
"alg": "RS256",
"kid": kid,
"n": _b64url_uint(numbers.n),
"e": _b64url_uint(numbers.e),
}
]
}
class StubbedFetch:
"""A JWKS fetcher stub: returns a fixed document, counts calls."""
def __init__(self, document: dict[str, object]) -> None:
self.document = document
self.calls = 0
async def __call__(self, url: str) -> dict[str, object]:
self.calls += 1
return self.document
def make_token(
signing_key: rsa.RSAPrivateKey = ADMIN_KEY,
*,
kid: str | None = KID,
iss: str = ISSUER,
aud: object = AUDIENCE,
sub: str = SUBJECT,
expires_in: int | None = 300,
nbf_in: int | None = -1,
groups: object = (ADMIN_GROUP,),
) -> str:
"""Mint one RS256 JWT the way Authentik would (its claims, our key).
``groups`` defaults to the admin group (what a real admin user's
token carries); tests pass a different value — or ``None`` for NO
groups claim at all — to prove the group gate rejects it.
"""
now = int(time.time())
claims: dict[str, object] = {"iss": iss, "aud": aud, "sub": sub, "iat": now}
if groups is not None:
claims["groups"] = (
list(groups) if isinstance(groups, (list, tuple)) else groups
)
if nbf_in is not None:
claims["nbf"] = now + nbf_in
if expires_in is not None:
claims["exp"] = now + expires_in
headers: dict[str, str] = {}
if kid is not None:
headers["kid"] = kid
return jwt.encode(
claims, signing_key, algorithm="RS256", headers=headers
)
class AdminRunner:
"""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.
The app's JWKS cache is replaced (before the lifespan runs) with
one built around a counting stub fetcher, so verification goes
through the REAL cache logic but never the network.
"""
def __init__(
self,
db_path: str,
jwks_document: dict[str, object],
*,
ttl_seconds: float | None = None,
backoff_seconds: float | None = None,
kid_refetch_seconds: float | None = None,
) -> None:
self.app = create_app(
ApiConfig(
db_path=db_path,
panel_token="admin-test-panel-token",
oidc_issuer=ISSUER,
oidc_audience=AUDIENCE,
oidc_jwks_url=JWKS_URL,
)
)
self.fetch = StubbedFetch(jwks_document)
cache_kwargs: dict[str, float] = {}
if ttl_seconds is not None:
cache_kwargs["ttl_seconds"] = ttl_seconds
if backoff_seconds is not None:
cache_kwargs["backoff_seconds"] = backoff_seconds
if kid_refetch_seconds is not None:
cache_kwargs["kid_refetch_seconds"] = kid_refetch_seconds
self.app.state.jwks_cache = JwksCache(
JWKS_URL, fetcher=self.fetch, **cache_kwargs
)
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)
+25 -1
View File
@@ -33,8 +33,11 @@ from pathlib import Path
from types import SimpleNamespace
from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from tests.admin_jwt_harness import ADMIN_KEY, KID, AdminRunner, jwks_for
# ---------------------------------------------------------------------------
# Homeassistant module provisioning (explicitly mock-only).
# ---------------------------------------------------------------------------
@@ -953,4 +956,25 @@ def freeze_nestquest_clock(
date=datetime.date,
)
monkeypatch.setattr(nestquest, "datetime", fake_module)
return frozen, requested
return frozen, requested
# ---------------------------------------------------------------------------
# Admin-plane API fixtures (shared by the admin-plane test modules).
#
# The local-key/stubbed-JWKS harness itself lives in
# :mod:`tests.admin_jwt_harness` (imported by name, the same pattern as
# tests.blueprint_helpers); these fixtures hand its runner to every test
# module in tests/ without re-declaring it.
# ---------------------------------------------------------------------------
@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 admin_client(temp_db_path: str) -> httpx.AsyncClient:
"""An app client whose provider publishes the admin test key."""
async with AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID)) as client:
yield client
+17 -173
View File
@@ -2,7 +2,10 @@
These tests exercise the NestQuest API's admin plane against the ASGI
app with httpx, using a locally generated RSA key pair and a stubbed
JWKS fetcher — NO network access:
JWKS fetcher — NO network access. That harness (the key pairs, the
JWKS builder, the token minting and the app runner) is shared with the
other admin-plane route tests in :mod:`tests.admin_jwt_harness`; this
module imports it and proves the auth CONTRACT itself:
- ``GET /api/v1/admin/ping`` requires a valid Authentik OIDC JWT: a
request with NO ``Authorization`` header and a MALFORMED header
@@ -41,28 +44,24 @@ import asyncio
import base64
import json
import time
from pathlib import Path
import httpx
import jwt
import pytest
from cryptography.hazmat.primitives.asymmetric import rsa
from api.app import create_app
from api.auth import ADMIN_GROUP, JwksCache
from api.auth import JwksCache
from api.config import ApiConfig, ConfigError
#: The configured OIDC settings the test apps run with (values a real
#: deployment would get from the NESTQUEST_OIDC_* environment vars).
ISSUER = "https://authentik.example.com/application/o/nestquest/"
AUDIENCE = "nestquest-api"
JWKS_URL = "https://authentik.example.com/application/o/nestquest/jwks/"
#: The ``kid`` the test tokens carry, matching the JWKS entry.
KID = "admin-test-key-1"
#: The subject a valid token authenticates as.
SUBJECT = "admin-user-1"
from tests.admin_jwt_harness import (
ADMIN_KEY,
AUDIENCE,
ISSUER,
JWKS_URL,
KID,
OTHER_KEY,
SUBJECT,
AdminRunner as _AdminRunner,
jwks_for as _jwks_for,
make_token as _make_token,
)
#: The uniform 401 detail every rejection must carry.
REJECTION_DETAIL = "Missing or invalid admin credentials"
@@ -78,53 +77,6 @@ FORBIDDEN_DETAIL = (
)
def _generate_key() -> rsa.RSAPrivateKey:
"""Generate one local RSA key pair (no network, no fixtures)."""
return rsa.generate_private_key(public_exponent=65537, key_size=2048)
#: The key pair whose PUBLIC half the stubbed JWKS publishes — tokens
#: signed with it verify; the second pair below proves they don't when
#: signed otherwise.
ADMIN_KEY = _generate_key()
OTHER_KEY = _generate_key()
def _b64url_uint(value: int) -> str:
"""Base64url-encode a non-negative integer (no padding), per JWK."""
raw = value.to_bytes((value.bit_length() + 7) // 8, "big")
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def _jwks_for(private_key: rsa.RSAPrivateKey, kid: str) -> dict[str, object]:
"""Build a JWKS document publishing ``private_key``'s public half."""
numbers = private_key.public_key().public_numbers()
return {
"keys": [
{
"kty": "RSA",
"use": "sig",
"alg": "RS256",
"kid": kid,
"n": _b64url_uint(numbers.n),
"e": _b64url_uint(numbers.e),
}
]
}
class _StubbedFetch:
"""A JWKS fetcher stub: returns a fixed document, counts calls."""
def __init__(self, document: dict[str, object]) -> None:
self.document = document
self.calls = 0
async def __call__(self, url: str) -> dict[str, object]:
self.calls += 1
return self.document
class _FailingFetch:
"""A JWKS fetcher stub that always fails, counting the attempts."""
@@ -164,41 +116,6 @@ class _FailThenServe:
return self.document
def _make_token(
signing_key: rsa.RSAPrivateKey = ADMIN_KEY,
*,
kid: str | None = KID,
iss: str = ISSUER,
aud: object = AUDIENCE,
sub: str = SUBJECT,
expires_in: int | None = 300,
nbf_in: int | None = -1,
groups: object = (ADMIN_GROUP,),
) -> str:
"""Mint one RS256 JWT the way Authentik would (its claims, our key).
``groups`` defaults to the admin group (what a real admin user's
token carries); tests pass a different value — or ``None`` for NO
groups claim at all — to prove the group gate rejects it.
"""
now = int(time.time())
claims: dict[str, object] = {"iss": iss, "aud": aud, "sub": sub, "iat": now}
if groups is not None:
claims["groups"] = (
list(groups) if isinstance(groups, (list, tuple)) else groups
)
if nbf_in is not None:
claims["nbf"] = now + nbf_in
if expires_in is not None:
claims["exp"] = now + expires_in
headers: dict[str, str] = {}
if kid is not None:
headers["kid"] = kid
return jwt.encode(
claims, signing_key, algorithm="RS256", headers=headers
)
def _token_with_header(
header: dict[str, object], payload: dict[str, object]
) -> str:
@@ -226,79 +143,6 @@ def _verified_payload() -> dict[str, object]:
}
class _AdminRunner:
"""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.
The app's JWKS cache is replaced (before the lifespan runs) with
one built around a counting stub fetcher, so verification goes
through the REAL cache logic but never the network.
"""
def __init__(
self,
db_path: str,
jwks_document: dict[str, object],
*,
ttl_seconds: float | None = None,
backoff_seconds: float | None = None,
kid_refetch_seconds: float | None = None,
) -> None:
self.app = create_app(
ApiConfig(
db_path=db_path,
panel_token="admin-test-panel-token",
oidc_issuer=ISSUER,
oidc_audience=AUDIENCE,
oidc_jwks_url=JWKS_URL,
)
)
self.fetch = _StubbedFetch(jwks_document)
cache_kwargs: dict[str, float] = {}
if ttl_seconds is not None:
cache_kwargs["ttl_seconds"] = ttl_seconds
if backoff_seconds is not None:
cache_kwargs["backoff_seconds"] = backoff_seconds
if kid_refetch_seconds is not None:
cache_kwargs["kid_refetch_seconds"] = kid_refetch_seconds
self.app.state.jwks_cache = JwksCache(
JWKS_URL, fetcher=self.fetch, **cache_kwargs
)
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 admin_client(
temp_db_path: str,
) -> httpx.AsyncClient:
"""An app client whose provider publishes the admin test key."""
async with _AdminRunner(temp_db_path, _jwks_for(ADMIN_KEY, KID)) as client:
yield client
# --- the happy path -------------------------------------------------------
+464
View File
@@ -0,0 +1,464 @@
"""Admin-plane children route tests (task e4845baa).
These tests exercise the admin plane's children 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.
Every done-condition case is proven against the DATABASE, read through
the SAME core copy the app uses (``nestquest_core.*``; see the two-copy
caveat in api/nestquest_core.py):
- 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 no row is written.
- ``POST /api/v1/admin/children``: the created child EXISTS with the
given display name (and the optional colour/avatar_ref/sort_order).
- ``PATCH /api/v1/admin/children/{id}``: an edit PERSISTS the supplied
fields and leaves omitted fields untouched (exclude_unset semantics —
an omitted field is never passed to the core).
- ``PATCH /api/v1/admin/children/{id}/active`` with ``is_active: false``
sets the active flag FALSE and the row STILL EXISTS (children are
never hard-deleted).
- ``POST /api/v1/admin/children/reorder``: the posted order becomes the
rows' sort_order; a list that is not a complete permutation is 422
and leaves the stored order untouched.
Error mapping (documented in api/routes_admin.py): a core ValueError
for a non-existent child is 404; every other core ValueError (an empty
display name, an explicit null colour, a bad permutation) is 422 — the
same failure class FastAPI reports for malformed bodies.
"""
from __future__ import annotations
import importlib
from types import SimpleNamespace
import httpx
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
# The children business layer 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.
_children = importlib.import_module("nestquest_core.children")
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
async def _children_by_id(database: object) -> dict[int, object]:
"""Every child row through the app's core copy, keyed by id."""
records = await _children.list_children(database)
return {record.id: record for record in records}
@pytest.fixture
async def admin_children(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane app client plus its live database.
The same runner the auth tests use (stubbed JWKS, no network); the
fixture also 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,
)
# --- authentication: the ONE require_admin dependency guards every route ---
async def test_children_routes_refuse_non_admin_and_absent_credentials(
admin_children: SimpleNamespace,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401: nothing written."""
client = admin_children.client
body = {"display_name": "Ada"}
non_admin = make_token(groups=("some-other-group",))
forbidden = await client.post(
"/api/v1/admin/children",
json=body,
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("/api/v1/admin/children", json=body)
assert unauthenticated.status_code == 401
unauthenticated_list = await client.get("/api/v1/admin/children")
assert unauthenticated_list.status_code == 401
# No route ran its handler: the household is still empty.
assert await _children_by_id(admin_children.database) == {}
# --- create -----------------------------------------------------------------
async def test_create_child_persists_the_given_name(
admin_children: SimpleNamespace,
) -> None:
"""POST /children creates the row; the child exists with that name."""
response = await admin_children.client.post(
"/api/v1/admin/children",
json={
"display_name": "Ada",
"colour": "#ff8800",
"avatar_ref": "avatar:bee",
"sort_order": 2,
},
headers=_admin_headers(),
)
assert response.status_code == 201
body = response.json()
assert body["display_name"] == "Ada"
assert body["colour"] == "#ff8800"
assert body["avatar_ref"] == "avatar:bee"
assert body["sort_order"] == 2
assert body["is_active"] is True
stored = (await _children_by_id(admin_children.database))[body["id"]]
assert stored.display_name == "Ada"
assert stored.colour == "#ff8800"
assert stored.avatar_ref == "avatar:bee"
assert stored.sort_order == 2
assert stored.is_active is True
async def test_create_child_with_only_a_name_uses_core_defaults(
admin_children: SimpleNamespace,
) -> None:
"""Omitted optional fields default through the core layer."""
response = await admin_children.client.post(
"/api/v1/admin/children",
json={"display_name": "Bo"},
headers=_admin_headers(),
)
assert response.status_code == 201
stored = (await _children_by_id(admin_children.database))[
response.json()["id"]
]
assert stored.display_name == "Bo"
assert stored.colour is None
assert stored.avatar_ref is None
assert stored.sort_order == 0
assert stored.is_active is True
@pytest.mark.parametrize(
"body",
[
{}, # display_name missing
{"display_name": 5}, # wrong type
{"display_name": ""}, # empty: the core rejects it
{"display_name": " "}, # whitespace-only: empty to a parent
{"display_name": "Ada", "sort_order": "first"}, # wrong type
],
)
async def test_create_child_malformed_input_is_422(
admin_children: SimpleNamespace, body: dict[str, object]
) -> None:
"""Malformed bodies are 422 and never create a row."""
response = await admin_children.client.post(
"/api/v1/admin/children", json=body, headers=_admin_headers()
)
assert response.status_code == 422
assert await _children_by_id(admin_children.database) == {}
# --- edit -------------------------------------------------------------------
async def test_edit_child_persists_supplied_fields_and_skips_omitted_ones(
admin_children: SimpleNamespace,
) -> None:
"""PATCH persists exactly the supplied fields (exclude_unset)."""
client = admin_children.client
created = await client.post(
"/api/v1/admin/children",
json={
"display_name": "Ada",
"colour": "#ff8800",
"avatar_ref": "avatar:bee",
},
headers=_admin_headers(),
)
assert created.status_code == 201
child_id = created.json()["id"]
# Only display_name and sort_order: colour/avatar_ref must survive.
edited = await client.patch(
f"/api/v1/admin/children/{child_id}",
json={"display_name": "Ada Lovelace", "sort_order": 4},
headers=_admin_headers(),
)
assert edited.status_code == 200
assert edited.json()["display_name"] == "Ada Lovelace"
assert edited.json()["sort_order"] == 4
stored = (await _children_by_id(admin_children.database))[child_id]
assert stored.display_name == "Ada Lovelace"
assert stored.sort_order == 4
# The omitted fields were never passed to the core: unchanged.
assert stored.colour == "#ff8800"
assert stored.avatar_ref == "avatar:bee"
# A second edit touching only colour leaves the name alone.
recoloured = await client.patch(
f"/api/v1/admin/children/{child_id}",
json={"colour": "#00aa55"},
headers=_admin_headers(),
)
assert recoloured.status_code == 200
stored = (await _children_by_id(admin_children.database))[child_id]
assert stored.colour == "#00aa55"
assert stored.display_name == "Ada Lovelace"
async def test_edit_unknown_child_is_404(
admin_children: SimpleNamespace,
) -> None:
"""An edit naming a child that does not exist is 404."""
response = await admin_children.client.patch(
"/api/v1/admin/children/999",
json={"display_name": "Nobody"},
headers=_admin_headers(),
)
assert response.status_code == 404
assert response.json()["detail"] == "Child not found"
async def test_edit_explicit_null_colour_is_422_and_changes_nothing(
admin_children: SimpleNamespace,
) -> None:
"""An explicit null (clearing) is passed through and rejected by the core.
Only the fields the request SUPPLIES are forwarded: a null colour is
a supplied None, which the core refuses (the DAO cannot express
NULL) — mapped to 422, with the stored colour left untouched.
"""
created = await admin_children.client.post(
"/api/v1/admin/children",
json={"display_name": "Ada", "colour": "#ff8800"},
headers=_admin_headers(),
)
child_id = created.json()["id"]
response = await admin_children.client.patch(
f"/api/v1/admin/children/{child_id}",
json={"colour": None},
headers=_admin_headers(),
)
assert response.status_code == 422
stored = (await _children_by_id(admin_children.database))[child_id]
assert stored.colour == "#ff8800"
# --- set_active -------------------------------------------------------------
async def test_deactivation_sets_the_flag_false_and_never_deletes(
admin_children: SimpleNamespace,
) -> None:
"""is_active=false deactivates the row; the row STILL exists."""
client = admin_children.client
created = await client.post(
"/api/v1/admin/children",
json={"display_name": "Ada"},
headers=_admin_headers(),
)
child_id = created.json()["id"]
deactivated = await client.patch(
f"/api/v1/admin/children/{child_id}/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert deactivated.status_code == 200
assert deactivated.json()["is_active"] is False
# Database proof: the flag is False and the row was NOT hard-deleted.
rows = await _children_by_id(admin_children.database)
stored = rows[child_id]
assert stored.is_active is False
assert stored.display_name == "Ada"
# Reactivation goes through the same route: the flag is set, not
# one-way.
reactivated = await client.patch(
f"/api/v1/admin/children/{child_id}/active",
json={"is_active": True},
headers=_admin_headers(),
)
assert reactivated.status_code == 200
assert reactivated.json()["is_active"] is True
rows = await _children_by_id(admin_children.database)
assert rows[child_id].is_active is True
async def test_set_active_unknown_child_is_404(
admin_children: SimpleNamespace,
) -> None:
"""A set_active naming a child that does not exist is 404."""
response = await admin_children.client.patch(
"/api/v1/admin/children/999/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert response.status_code == 404
assert response.json()["detail"] == "Child not found"
async def test_set_active_malformed_body_is_422(
admin_children: SimpleNamespace,
) -> None:
"""A non-bool is_active is rejected by the body model (422)."""
response = await admin_children.client.patch(
"/api/v1/admin/children/1/active",
json={"is_active": "maybe"},
headers=_admin_headers(),
)
assert response.status_code == 422
# --- reorder ----------------------------------------------------------------
async def test_reorder_writes_the_posted_order_to_sort_order(
admin_children: SimpleNamespace,
) -> None:
"""The posted id order becomes the rows' sort_order."""
client = admin_children.client
ids = []
for name in ("Ada", "Bo", "Cyd"):
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
ids.append(created.json()["id"])
response = await client.post(
"/api/v1/admin/children/reorder",
json={"ordered_ids": [ids[2], ids[0], ids[1]]},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json() == {"status": "ok"}
# Database proof: each posted position became that child's sort_order.
rows = await _children_by_id(admin_children.database)
assert rows[ids[2]].sort_order == 0
assert rows[ids[0]].sort_order == 1
assert rows[ids[1]].sort_order == 2
async def test_reorder_rejects_non_permutations_and_writes_nothing(
admin_children: SimpleNamespace,
) -> None:
"""A list that is not a complete permutation is 422, order intact.
The core rejects the list BEFORE any write, so the stored sort_order
values survive every rejected call unchanged.
"""
client = admin_children.client
ids = []
for name in ("Ada", "Bo"):
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
ids.append(created.json()["id"])
bad_lists = [
ids[:1], # partial: missing an existing child
[ids[0], ids[0], ids[1]], # duplicate id
[*ids, ids[1] + 999], # unknown id
]
for ordered_ids in bad_lists:
response = await client.post(
"/api/v1/admin/children/reorder",
json={"ordered_ids": ordered_ids},
headers=_admin_headers(),
)
assert response.status_code == 422, ordered_ids
# A non-integer id never reaches the core either (body model).
malformed = await client.post(
"/api/v1/admin/children/reorder",
json={"ordered_ids": ["first", "second"]},
headers=_admin_headers(),
)
assert malformed.status_code == 422
# Database proof: no write happened for any rejected call.
rows = await _children_by_id(admin_children.database)
assert [rows[child_id].sort_order for child_id in ids] == [0, 0]
# --- list (the read model the admin PWA re-renders from) ---------------------
async def test_list_children_returns_all_in_display_order(
admin_children: SimpleNamespace,
) -> None:
"""GET /children returns every child (inactive included) in order."""
client = admin_children.client
ada = await client.post(
"/api/v1/admin/children",
json={"display_name": "Ada", "sort_order": 5},
headers=_admin_headers(),
)
bo = await client.post(
"/api/v1/admin/children",
json={"display_name": "Bo", "sort_order": 1},
headers=_admin_headers(),
)
deactivated = await client.patch(
f"/api/v1/admin/children/{ada.json()['id']}/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert deactivated.status_code == 200
response = await client.get(
"/api/v1/admin/children", headers=_admin_headers()
)
assert response.status_code == 200
children = response.json()["children"]
assert [child["id"] for child in children] == [
bo.json()["id"],
ada.json()["id"],
]
by_id = {child["id"]: child for child in children}
assert by_id[ada.json()["id"]]["is_active"] is False
assert by_id[bo.json()["id"]]["is_active"] is True
# The serialized shape is the documented child payload.
assert set(by_id[bo.json()["id"]]) == {
"id",
"display_name",
"colour",
"avatar_ref",
"sort_order",
"is_active",
}
+611
View File
@@ -0,0 +1,611 @@
"""Admin-plane quest-definition route tests (task b586a2f1).
These tests exercise the admin plane's quest-definition 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.
Every done-condition case is proven against the DATABASE, read through
the SAME core copy the app uses (``nestquest_core.*``; see the two-copy
caveat in api/nestquest_core.py):
- 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 no definition is written.
- ``POST /api/v1/admin/quest-definitions``: the created definition
EXISTS with its title, description, icon, decoded rule, assignees
and windows exactly as posted.
- ``PATCH /api/v1/admin/quest-definitions/{id}``: an edit PERSISTS the
supplied fields and leaves omitted fields untouched (exclude_unset
semantics); a supplied ``windows`` list replaces the whole set;
assignment is not settable through the edit.
- ``PATCH /api/v1/admin/quest-definitions/{id}/active`` with
``is_active: false`` sets the flag FALSE and the definition STILL
EXISTS with its rule, assignees and windows (never hard-deleted);
reactivation restores it.
Error mapping (documented in api/routes_admin.py): a core ValueError
for a non-existent DEFINITION is 404; every other core ValueError — a
rejected rule (RuleValidationError), a bad window name or due time, a
rejected assignee list (an unknown assignee arrives in the BODY, so it
is 422, never 404) — is 422, the same failure class FastAPI reports
for malformed bodies.
"""
from __future__ import annotations
import importlib
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
# The quest-definition business layer and its 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 them through api.nestquest_core.
_quest_definitions = importlib.import_module(
"nestquest_core.quest_definitions"
)
_dao_rules = importlib.import_module("nestquest_core.dao_rules")
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _weekly_rule_body() -> dict[str, object]:
"""A valid weekly rule body: Mondays and Wednesdays from 2026-03-02."""
return {
"rule_type": "weekly",
"interval": 1,
"weekday_set": [0, 2],
"start_date": "2026-03-02",
}
def _weekly_rule_dict() -> dict[str, object]:
"""The lossless dict form _weekly_rule_body() must decode back to."""
return {
"rule_type": "weekly",
"interval": 1,
"weekday_set": [0, 2],
"day_of_month": None,
"nth_weekday": None,
"nth_weekday_weekday": None,
"month": None,
"start_date": "2026-03-02",
"end_date": None,
}
async def _create_children(
client: object, names: tuple[str, ...]
) -> list[int]:
"""Create ACTIVE children through the admin children route."""
ids = []
for name in names:
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
ids.append(created.json()["id"])
return ids
async def _create_definition(
client: object, assignee_ids: list[int]
) -> object:
"""POST a valid two-window definition for the given assignees."""
return await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Tidy the den",
"rule": _weekly_rule_body(),
"assignee_child_ids": assignee_ids,
"windows": ["morning", ["evening", "19:30"]],
"description": "Put everything back where it belongs.",
"icon": "toy-box",
},
headers=_admin_headers(),
)
async def _definitions_by_id(database: object) -> dict[int, object]:
"""Every ACTIVE definition bundle through the app's core copy."""
bundles = await _quest_definitions.list_active_definitions(database)
return {bundle.definition.id: bundle for bundle in bundles}
@pytest.fixture
async def admin_questdefs(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 also 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,
)
# --- authentication: the ONE require_admin dependency guards every route ---
async def test_quest_definition_routes_refuse_non_admin_and_absent(
admin_questdefs: SimpleNamespace,
) -> None:
"""A non-admin JWT is 403, an absent credential is 401: nothing written."""
client = admin_questdefs.client
body = {"title": "Sweep the porch"}
non_admin = make_token(groups=("some-other-group",))
forbidden = await client.post(
"/api/v1/admin/quest-definitions",
json=body,
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_create = await client.post(
"/api/v1/admin/quest-definitions", json=body
)
assert unauthenticated_create.status_code == 401
unauthenticated_edit = await client.patch(
"/api/v1/admin/quest-definitions/1", json={"title": "X"}
)
assert unauthenticated_edit.status_code == 401
unauthenticated_active = await client.patch(
"/api/v1/admin/quest-definitions/1/active", json={"is_active": False}
)
assert unauthenticated_active.status_code == 401
# No route ran its handler: no definition was written.
assert await _definitions_by_id(admin_questdefs.database) == {}
# --- create -----------------------------------------------------------------
async def test_create_persists_definition_with_assignees_and_windows(
admin_questdefs: SimpleNamespace,
) -> None:
"""POST /quest-definitions persists rule, assignees and windows."""
client = admin_questdefs.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
response = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Tidy the den",
"rule": _weekly_rule_body(),
"assignee_child_ids": [ada, bo],
"windows": ["morning", ["evening", "19:30"]],
"description": "Put everything back where it belongs.",
"icon": "toy-box",
},
headers=_admin_headers(),
)
assert response.status_code == 201
payload = response.json()
definition_id = payload["id"]
# The serialized shape is the documented definition payload.
assert set(payload) == {
"id",
"title",
"description",
"icon",
"is_active",
"rule",
"assignees",
"windows",
}
assert payload["title"] == "Tidy the den"
assert payload["description"] == "Put everything back where it belongs."
assert payload["icon"] == "toy-box"
assert payload["is_active"] is True
assert payload["rule"] == _weekly_rule_dict()
assert {a["id"]: a["display_name"] for a in payload["assignees"]} == {
ada: "Ada",
bo: "Bo",
}
assert {w["window"]: w["due_time"] for w in payload["windows"]} == {
"morning": None,
"evening": "19:30",
}
# Database proof through the app's core copy: the definition, its
# decoded rule, its assignees and its windows all exist as posted.
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.title == "Tidy the den"
assert stored.definition.description == (
"Put everything back where it belongs."
)
assert stored.definition.icon == "toy-box"
assert stored.definition.is_active is True
assert stored.rule.to_dict() == _weekly_rule_dict()
assert sorted(child.id for child in stored.assignees) == [ada, bo]
assert {(w.window, w.due_time) for w in stored.windows} == {
("morning", None),
("evening", "19:30"),
}
async def test_create_with_required_fields_only_uses_core_defaults(
admin_questdefs: SimpleNamespace,
) -> None:
"""Omitted optional fields default to NULL through the core layer."""
client = admin_questdefs.client
ada = (await _create_children(client, ("Ada",)))[0]
response = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Water the plants",
"rule": _weekly_rule_body(),
"assignee_child_ids": [ada],
"windows": ["morning"],
},
headers=_admin_headers(),
)
assert response.status_code == 201
payload = response.json()
assert payload["description"] is None
assert payload["icon"] is None
stored = (await _definitions_by_id(admin_questdefs.database))[
payload["id"]
]
assert stored.definition.description is None
assert stored.definition.icon is None
async def test_create_without_required_fields_is_422(
admin_questdefs: SimpleNamespace,
) -> None:
"""A body missing any required field is 422 and creates nothing."""
client = admin_questdefs.client
bodies: list[dict[str, object]] = [
{}, # everything missing
{"title": "Tidy the den"}, # rule, assignees, windows missing
{
"title": "Tidy the den",
"rule": _weekly_rule_body(),
}, # assignees and windows missing
{
"title": "Tidy the den",
"rule": _weekly_rule_body(),
"assignee_child_ids": [1],
}, # windows missing
]
for body in bodies:
response = await client.post(
"/api/v1/admin/quest-definitions",
json=body,
headers=_admin_headers(),
)
assert response.status_code == 422, body
assert await _definitions_by_id(admin_questdefs.database) == {}
@pytest.mark.parametrize(
"override",
[
{"title": ""}, # empty: the core rejects it
{"title": " "}, # whitespace-only: empty to a parent
{"assignee_child_ids": []}, # empty assignee list
{"assignee_child_ids": [999, 1000]}, # unknown children: 422 not 404
{"assignee_child_ids": ["Ada"]}, # wrong-typed ids
{"windows": []}, # empty window set
{"windows": "morning"}, # not a list
{"windows": ["attic"]}, # not a const.QUEST_WINDOWS name
{"windows": [["morning", "25:00"]]}, # not a 24-hour time
{"windows": [["morning", "8:00"]]}, # not a strict HH:MM
{"windows": [["morning", None], "morning"]}, # duplicate window
{"rule": {"rule_type": "weekly"}}, # weekly without weekday_set
{"rule": {"rule_type": "fortnightly"}}, # unknown rule_type
{"rule": {"rule_type": "daily", "interval": 0}}, # interval < 1
{"rule": {"rule_type": "weekly", "weekday_set": [7]}}, # out of range
{"rule": {"rule_type": "monthly_day"}}, # missing day_of_month
{"rule": {"rule_type": "daily", "day_of_month": 5}}, # forbidden field
{"rule": {"rule_type": "daily", "start_date": "2026-13-01"}},
{"rule": None}, # an explicit null rule is a supplied invalid value
],
)
async def test_create_rejections_are_422_and_persist_nothing(
admin_questdefs: SimpleNamespace, override: dict[str, object]
) -> None:
"""Rejected bodies are 422 and never create a definition."""
client = admin_questdefs.client
ids = await _create_children(client, ("Ada", "Bo"))
body: dict[str, object] = {
"title": "Tidy the den",
"rule": _weekly_rule_body(),
"assignee_child_ids": ids,
"windows": ["morning", ["evening", "19:30"]],
}
body.update(override)
response = await client.post(
"/api/v1/admin/quest-definitions",
json=body,
headers=_admin_headers(),
)
assert response.status_code == 422, override
assert await _definitions_by_id(admin_questdefs.database) == {}
# --- edit -------------------------------------------------------------------
async def test_edit_persists_supplied_fields_and_leaves_omitted_untouched(
admin_questdefs: SimpleNamespace,
) -> None:
"""PATCH persists exactly the supplied fields (exclude_unset)."""
client = admin_questdefs.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
created = await _create_definition(client, [ada, bo])
assert created.status_code == 201
definition_id = created.json()["id"]
dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
original_rule_id = (await dao.get(definition_id)).schedule_rule_id
# Only title and windows: description, icon, rule and assignees
# must survive; the supplied windows REPLACE the whole set.
edited = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"title": "Clear the den", "windows": ["afternoon"]},
headers=_admin_headers(),
)
assert edited.status_code == 200
assert edited.json()["title"] == "Clear the den"
assert {w["window"]: w["due_time"] for w in edited.json()["windows"]} == {
"afternoon": None
}
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.title == "Clear the den"
assert {(w.window, w.due_time) for w in stored.windows} == {
("afternoon", None)
}
# Omitted fields were never passed to the core: unchanged.
assert stored.definition.description == (
"Put everything back where it belongs."
)
assert stored.definition.icon == "toy-box"
assert stored.rule.to_dict() == _weekly_rule_dict()
# Assignment is not settable through the edit route: untouched.
assert sorted(child.id for child in stored.assignees) == [ada, bo]
# A second edit touching only the rule leaves the metadata and the
# windows alone, and updates the rule row IN PLACE (same rule id).
rescheduled = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"rule": {"rule_type": "daily", "start_date": "2026-04-01"}},
headers=_admin_headers(),
)
assert rescheduled.status_code == 200
assert rescheduled.json()["rule"]["rule_type"] == "daily"
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.title == "Clear the den"
assert {(w.window, w.due_time) for w in stored.windows} == {
("afternoon", None)
}
assert stored.rule.to_dict() == {
"rule_type": "daily",
"interval": 1,
"weekday_set": None,
"day_of_month": None,
"nth_weekday": None,
"nth_weekday_weekday": None,
"month": None,
"start_date": "2026-04-01",
"end_date": None,
}
record = await dao.get(definition_id)
assert record.schedule_rule_id == original_rule_id
rule_row = await _dao_rules.ScheduleRulesDao(
admin_questdefs.database
).get(record.schedule_rule_id)
assert rule_row.rule_type == "daily"
assert rule_row.start_date == "2026-04-01"
async def test_edit_can_clear_description_and_icon_explicitly(
admin_questdefs: SimpleNamespace,
) -> None:
"""An explicit null description/icon clears them (the core's edit path).
Unlike the children edit, the definition edit supports clearing: a
supplied null IS forwarded and writes NULL; everything omitted
stays untouched.
"""
client = admin_questdefs.client
ada = (await _create_children(client, ("Ada",)))[0]
created = await _create_definition(client, [ada])
definition_id = created.json()["id"]
response = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"description": None, "icon": None},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json()["description"] is None
assert response.json()["icon"] is None
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.description is None
assert stored.definition.icon is None
assert stored.definition.title == "Tidy the den"
assert stored.rule.to_dict() == _weekly_rule_dict()
async def test_edit_unknown_definition_is_404(
admin_questdefs: SimpleNamespace,
) -> None:
"""An edit naming a definition that does not exist is 404."""
response = await admin_questdefs.client.patch(
"/api/v1/admin/quest-definitions/999",
json={"title": "Ghost chore"},
headers=_admin_headers(),
)
assert response.status_code == 404
assert response.json()["detail"] == "Quest definition not found"
async def test_edit_rejections_are_422_and_change_nothing(
admin_questdefs: SimpleNamespace,
) -> None:
"""A rejected rule or window set is 422; the stored state survives.
The core rejects BEFORE any write (and the DAO edit runs in one
transaction), so a 422 never leaves a half-applied change behind.
"""
client = admin_questdefs.client
ada = (await _create_children(client, ("Ada",)))[0]
created = await _create_definition(client, [ada])
definition_id = created.json()["id"]
rejected_bodies: list[dict[str, object]] = [
{"rule": {"rule_type": "weekly"}}, # weekly without weekday_set
{"rule": None}, # an explicit null rule is rejected by the core
{"windows": ["attic"]}, # unknown window name
{"windows": None}, # an explicit null window set is rejected
{"title": ""}, # the core rejects an empty title
]
for body in rejected_bodies:
response = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json=body,
headers=_admin_headers(),
)
assert response.status_code == 422, body
# Database proof: nothing changed for any rejected call.
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.title == "Tidy the den"
assert stored.definition.description == (
"Put everything back where it belongs."
)
assert stored.rule.to_dict() == _weekly_rule_dict()
assert {(w.window, w.due_time) for w in stored.windows} == {
("morning", None),
("evening", "19:30"),
}
# --- set_active -------------------------------------------------------------
async def test_set_active_toggles_the_flag_and_the_definition_survives(
admin_questdefs: SimpleNamespace,
) -> None:
"""is_active=false deactivates; the definition STILL exists intact.
Deactivation is the only removal path: the row, rule, assignees and
windows all survive, and reactivation through the same route
restores the flag without anything having been lost.
"""
client = admin_questdefs.client
ada = (await _create_children(client, ("Ada",)))[0]
created = await _create_definition(client, [ada])
assert created.status_code == 201
definition_id = created.json()["id"]
deactivated = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert deactivated.status_code == 200
payload = deactivated.json()
assert payload["is_active"] is False
# Everything else came back unchanged: deactivation only flips a flag.
assert payload["title"] == "Tidy the den"
assert payload["rule"] == _weekly_rule_dict()
assert [a["id"] for a in payload["assignees"]] == [ada]
assert {w["window"]: w["due_time"] for w in payload["windows"]} == {
"morning": None,
"evening": "19:30",
}
# Database proof: the definition left the ACTIVE list but its row,
# rule, assignees and windows were NOT hard-deleted.
assert definition_id not in await _definitions_by_id(
admin_questdefs.database
)
dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
record = await dao.get(definition_id)
assert record is not None
assert record.is_active is False
assert record.title == "Tidy the den"
assert await dao.list_assignees(definition_id) != []
assert await dao.list_windows(definition_id) != []
# Reactivation goes through the same route: the flag is set, not
# one-way, and the surviving rule is still the one that was stored.
reactivated = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}/active",
json={"is_active": True},
headers=_admin_headers(),
)
assert reactivated.status_code == 200
assert reactivated.json()["is_active"] is True
stored = (await _definitions_by_id(admin_questdefs.database))[
definition_id
]
assert stored.definition.is_active is True
assert stored.rule.to_dict() == _weekly_rule_dict()
assert [child.id for child in stored.assignees] == [ada]
async def test_set_active_unknown_definition_is_404(
admin_questdefs: SimpleNamespace,
) -> None:
"""A set_active naming a definition that does not exist is 404."""
response = await admin_questdefs.client.patch(
"/api/v1/admin/quest-definitions/999/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert response.status_code == 404
assert response.json()["detail"] == "Quest definition not found"
async def test_set_active_malformed_body_is_422(
admin_questdefs: SimpleNamespace,
) -> None:
"""A non-bool is_active is rejected by the body model (422)."""
response = await admin_questdefs.client.patch(
"/api/v1/admin/quest-definitions/1/active",
json={"is_active": "maybe"},
headers=_admin_headers(),
)
assert response.status_code == 422