Admin routes: children CRUD over the core business layer
Task e4845baa (feature 16, NestQuest API service): the admin PWA needs to manage the children registry, but the only removal path the domain allows is deactivation and the name/NULL/bool/permutation policies all live in core.children — so the plane gets thin adapters that forward each request to ONE core call and never re-implement a rule. WHY: create/edit/set_active/reorder (+ a thin GET list) hang off the router's ONE require_admin dependency; edit forwards only the fields the request supplied (exclude_unset) so an omitted field stays unchanged while an explicit null is rejected by the core rather than silently ignored. Core ValueErrors map to 404 (non-existent child) or 422 (invalid input, same class Pydantic reports) and are documented in the module docstring; the core rejects before any write, so neither status can leave a half-applied change. The admin JWT test harness (local RSA keys, stubbed JWKS, token minting, app runner) is factored into tests/admin_jwt_harness.py so the children route tests reuse it instead of duplicating it; the auth tests import the same harness with unchanged test bodies, and the shared temp_db_path/admin_client fixtures now live in conftest.py. TZ=UTC uv run pytest -q: 1669 passed (baseline 1652 + 17 new).
This commit is contained in:
@@ -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
@@ -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
@@ -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 -------------------------------------------------------
|
||||
|
||||
|
||||
|
||||
@@ -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",
|
||||
}
|
||||
Reference in New Issue
Block a user