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

Task 9d121e76: Reverse proxy and TLS for the API (#214)

Browse Source
This commit was merged in pull request #214.
This commit is contained in:
dev_agent
2026-09-24 08:00:04 -04:00
parent 9b068926a9
commit f6d222764c
104 changed files with 19763 additions and 185 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_admin_occurrences.py
+431
View File
@@ -0,0 +1,431 @@
"""Admin-plane occurrence-preview route tests (task 5f843564).
These tests exercise ``POST /api/v1/admin/quest-definitions/occurrences-preview``
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 with NO network access.
- Authentication runs through the router's ONE ``require_admin``
dependency: no credential is 401, the panel service token is 403
(refused as a credential type), a JWT without the admin group is
403, an admin JWT is 200.
- The returned dates are asserted EXACTLY for a weekly rule with an
interval and a start_date, a daily rule with an interval, a rule
with an end_date (the list stops at it), a month-end-clamped monthly
rule, and the default start (the pinned host clock's today); one
case also asserts parity with the core's own ``occurrences_between``
over the documented window.
- A rejected rule (the same core ``from_dict`` validation create/edit
use), a non-strict ``start_date`` and a ``count`` outside 1..50 are
422.
"""
from __future__ import annotations
import datetime
import importlib
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import pytest
from api import routes_admin
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
# The recurrence engine through the SAME core copy the app uses
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
_recurrence = importlib.import_module("nestquest_core.recurrence")
PREVIEW_URL = "/api/v1/admin/quest-definitions/occurrences-preview"
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
#: The pinned host instant: Wednesday 2026-09-23, noon UTC.
PINNED_NOW = datetime.datetime(2026, 9, 23, 12, 0, 0, tzinfo=ZoneInfo("UTC"))
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _weekly_rule() -> dict[str, object]:
"""Every OTHER week's Tuesday and Friday, anchored on Tue 2026-03-03."""
return {
"rule_type": "weekly",
"interval": 2,
"weekday_set": [1, 4],
"start_date": "2026-03-03",
}
@pytest.fixture
async def preview(temp_db_path: str, monkeypatch) -> SimpleNamespace:
"""An admin-plane app client with the host clock pinned (and counted)."""
clock_reads: list[datetime.datetime] = []
def _admin_now() -> datetime.datetime:
clock_reads.append(PINNED_NOW)
return PINNED_NOW
monkeypatch.setattr(routes_admin, "_local_now", _admin_now)
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(client=client, clock_reads=clock_reads)
async def _post(client: object, body: dict[str, object]) -> object:
"""POST a preview body with an admin JWT."""
return await client.post(PREVIEW_URL, json=body, headers=_admin_headers())
# --- authentication: the ONE require_admin dependency ----------------------
async def test_preview_without_credential_is_401(
preview: SimpleNamespace,
) -> None:
"""No Authorization header is the uniform 401."""
response = await preview.client.post(
PREVIEW_URL, json={"rule": _weekly_rule()}
)
assert response.status_code == 401
async def test_preview_refuses_panel_service_token(
preview: SimpleNamespace,
) -> None:
"""The panel service token is refused as a credential type (403)."""
response = await preview.client.post(
PREVIEW_URL,
json={"rule": _weekly_rule()},
headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_preview_refuses_non_admin_jwt(
preview: SimpleNamespace,
) -> None:
"""A valid JWT without the nestquest-admins group is 403."""
non_admin = make_token(groups=("some-other-group",))
response = await preview.client.post(
PREVIEW_URL,
json={"rule": _weekly_rule()},
headers={"Authorization": f"Bearer {non_admin}"},
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
async def test_preview_admin_jwt_is_200(preview: SimpleNamespace) -> None:
"""An admin JWT is 200 with the documented ``dates`` shape."""
response = await _post(
preview.client, {"rule": _weekly_rule(), "start_date": "2026-03-01"}
)
assert response.status_code == 200
assert set(response.json()) == {"dates"}
# --- the returned dates ----------------------------------------------------
async def test_weekly_interval_rule_from_before_its_start(
preview: SimpleNamespace,
) -> None:
"""Every other week's Tue/Fri, starting at the rule's own start_date.
The preview starts on Sun 2026-03-01, before the rule's anchor, so
the first date is the anchor itself; the off weeks (Mar 10-16,
Mar 24-30) are skipped.
"""
response = await _post(
preview.client,
{"rule": _weekly_rule(), "start_date": "2026-03-01", "count": 6},
)
assert response.status_code == 200
assert response.json()["dates"] == [
"2026-03-03",
"2026-03-06",
"2026-03-17",
"2026-03-20",
"2026-03-31",
"2026-04-03",
]
async def test_weekly_interval_rule_from_mid_cadence(
preview: SimpleNamespace,
) -> None:
"""A start inside an on-week keeps the rule's anchor cadence."""
response = await _post(
preview.client,
{"rule": _weekly_rule(), "start_date": "2026-03-18", "count": 4},
)
assert response.status_code == 200
assert response.json()["dates"] == [
"2026-03-20",
"2026-03-31",
"2026-04-03",
"2026-04-14",
]
async def test_daily_interval_rule(preview: SimpleNamespace) -> None:
"""Every third day from 2026-01-01, previewed from 2026-01-05."""
response = await _post(
preview.client,
{
"rule": {
"rule_type": "daily",
"interval": 3,
"start_date": "2026-01-01",
},
"start_date": "2026-01-05",
"count": 4,
},
)
assert response.status_code == 200
assert response.json()["dates"] == [
"2026-01-07",
"2026-01-10",
"2026-01-13",
"2026-01-16",
]
async def test_rule_with_end_date_stops_at_it(
preview: SimpleNamespace,
) -> None:
"""A daily rule ending 2026-05-04 yields only its four dates."""
response = await _post(
preview.client,
{
"rule": {
"rule_type": "daily",
"start_date": "2026-05-01",
"end_date": "2026-05-04",
},
"start_date": "2026-04-28",
"count": 10,
},
)
assert response.status_code == 200
assert response.json()["dates"] == [
"2026-05-01",
"2026-05-02",
"2026-05-03",
"2026-05-04",
]
async def test_monthly_day_rule_clamps_to_month_end(
preview: SimpleNamespace,
) -> None:
"""The core's month-end policy: the 31st fires on Feb 28th."""
response = await _post(
preview.client,
{
"rule": {
"rule_type": "monthly_day",
"day_of_month": 31,
"start_date": "2026-01-31",
},
"start_date": "2026-01-01",
"count": 3,
},
)
assert response.status_code == 200
assert response.json()["dates"] == [
"2026-01-31",
"2026-02-28",
"2026-03-31",
]
async def test_default_start_is_host_today_with_default_count(
preview: SimpleNamespace,
) -> None:
"""No start_date: ONE clock read's date; no count: ten dates."""
response = await _post(preview.client, {"rule": {"rule_type": "daily"}})
assert response.status_code == 200
expected = [
(datetime.date(2026, 9, 23) + datetime.timedelta(days=n)).isoformat()
for n in range(10)
]
assert response.json()["dates"] == expected
assert preview.clock_reads == [PINNED_NOW]
async def test_supplied_start_date_reads_no_clock(
preview: SimpleNamespace,
) -> None:
"""A supplied start_date is used as-is: the host clock is not read."""
response = await _post(
preview.client, {"rule": _weekly_rule(), "start_date": "2026-03-01"}
)
assert response.status_code == 200
assert preview.clock_reads == []
async def test_dates_match_core_occurrences_between_over_window(
preview: SimpleNamespace,
) -> None:
"""The route's dates ARE the core's walk over the 366-day window."""
rule_body = {
"rule_type": "monthly_weekday",
"nth_weekday": -1,
"nth_weekday_weekday": 4,
"start_date": "2026-01-01",
}
response = await _post(
preview.client,
{"rule": rule_body, "start_date": "2026-02-10", "count": 50},
)
assert response.status_code == 200
core_dates = _recurrence.occurrences_between(
_recurrence.ScheduleRule.from_dict(rule_body),
"2026-02-10",
"2027-02-11",
)
assert response.json()["dates"] == core_dates
assert len(core_dates) == 12
async def test_yearly_rule_shows_next_occurrence_within_window(
preview: SimpleNamespace,
) -> None:
"""The 366-day window always reaches a yearly rule's next date."""
response = await _post(
preview.client,
{
"rule": {
"rule_type": "yearly",
"month": 9,
"day_of_month": 22,
"start_date": "2020-09-22",
},
"start_date": "2026-09-23",
},
)
assert response.status_code == 200
assert response.json()["dates"] == ["2027-09-22"]
async def test_start_near_calendar_end_is_422(
preview: SimpleNamespace,
) -> None:
"""A window that would reach 9999-12-31 is rejected, not clamped."""
response = await _post(
preview.client,
{"rule": {"rule_type": "daily"}, "start_date": "9999-12-30"},
)
assert response.status_code == 422
assert "too close to the end of the calendar" in response.json()["detail"]
async def test_latest_accepted_start_date(preview: SimpleNamespace) -> None:
"""The last start whose window ends before 9999-12-31 is accepted."""
response = await _post(
preview.client,
{
"rule": {"rule_type": "daily"},
"start_date": "9998-12-29",
"count": 1,
},
)
assert response.status_code == 200
assert response.json()["dates"] == ["9998-12-29"]
# --- rejections: 422 -------------------------------------------------------
@pytest.mark.parametrize(
"rule",
[
{"rule_type": "weekly"}, # weekly without weekday_set
{"rule_type": "fortnightly"}, # unknown rule_type
{"rule_type": "daily", "interval": 0},
{"rule_type": "daily", "weekday_set": [1]}, # forbidden field
{"rule_type": "daily", "start_date": "2026-3-1"},
{
"rule_type": "daily",
"start_date": "2026-05-04",
"end_date": "2026-05-01",
},
],
)
async def test_malformed_rule_is_422(
preview: SimpleNamespace, rule: dict[str, object]
) -> None:
"""A rule the core's from_dict rejects is 422, as on create/edit."""
response = await _post(
preview.client, {"rule": rule, "start_date": "2026-03-01"}
)
assert response.status_code == 422
async def test_missing_rule_is_422(preview: SimpleNamespace) -> None:
"""The rule is required: the body model's 422."""
response = await _post(preview.client, {"start_date": "2026-03-01"})
assert response.status_code == 422
@pytest.mark.parametrize(
"start_date", ["2026-3-1", "20260301", "tomorrow", "2026-02-30", ""]
)
async def test_non_strict_start_date_is_422(
preview: SimpleNamespace, start_date: str
) -> None:
"""start_date must be a strict YYYY-MM-DD date (the core's check)."""
response = await _post(
preview.client, {"rule": _weekly_rule(), "start_date": start_date}
)
assert response.status_code == 422
@pytest.mark.parametrize("count", [0, -1, 51, 1000, True, "5", 2.5])
async def test_count_out_of_range_is_422(
preview: SimpleNamespace, count: object
) -> None:
"""count must be a strict JSON integer in 1..50."""
response = await _post(
preview.client,
{"rule": _weekly_rule(), "start_date": "2026-03-01", "count": count},
)
assert response.status_code == 422
@pytest.mark.parametrize("count", [1, 50])
async def test_count_bounds_are_inclusive(
preview: SimpleNamespace, count: int
) -> None:
"""count 1 and 50 are accepted and truncate the list exactly."""
response = await _post(
preview.client,
{
"rule": {"rule_type": "daily"},
"start_date": "2026-01-01",
"count": count,
},
)
assert response.status_code == 200
assert len(response.json()["dates"]) == count
tests/test_api_admin_presence_consequence.py
+448
View File
@@ -0,0 +1,448 @@
"""Admin override consequence-count route tests (task 7616e3eb).
``POST /api/v1/admin/presence-overrides/consequence`` answers, BEFORE an
override is saved, how many of the child's upcoming open quest
instances saving it would remove (``{"removed": N}``) — the ADMIN-SPEC
§6 "This removes N upcoming tasks" warning. These tests exercise it
against the ASGI app with the shared stubbed-JWKS harness
(:mod:`tests.admin_jwt_harness`):
- a two-week custody pattern (present the whole first week of the
cycle, absent the whole second) where an absent override over a
known span removes an EXACT number of instances;
- ranges that remove nothing (already-absent days, a present
override) count 0;
- an instance with a completion event inside the range is NOT counted
(D-005 — regeneration never deletes it);
- a ``skip_on_away = false`` definition contributes nothing;
- the count MATCHES a direct materialise comparison: the override is
then actually created (which regenerates the child) and the removed
open future instances are counted from the database;
- the preview writes nothing;
- the four auth outcomes, an unknown child 404 and a malformed range
422.
The cycle is anchored on the REAL host "today" (the create route's
regeneration reads the host clock, so the comparison test cannot pin
it): days today..today+6 are cycle week 0 (present every day), days
today+7..today+13 week 1 (absent every day), today+14 week 0 again.
"""
from __future__ import annotations
import datetime
import importlib
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
# The DAOs and materializer through the SAME core copy the app uses
# (nestquest_core.*, NOT the conftest-stubbed custom_components copy).
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
_dao_presence = importlib.import_module("nestquest_core.dao_presence")
_materialize = importlib.import_module("nestquest_core.materialize")
CONSEQUENCE_URL = "/api/v1/admin/presence-overrides/consequence"
OVERRIDES_URL = "/api/v1/admin/presence-overrides"
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
#: The two-week custody pattern: week 0 every weekday, week 1 none.
CUSTODY_PATTERN = {"0": [0, 1, 2, 3, 4, 5, 6], "1": []}
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _day(offset: int) -> str:
"""The ISO date ``offset`` days after the host's today."""
return (datetime.date.today() + datetime.timedelta(days=offset)).isoformat()
def _body(
child_id: int, start: int, end: int, *, is_present: bool = False
) -> dict[str, object]:
"""A consequence body over ``[today+start, today+end]``."""
return {
"child_id": child_id,
"start_date": _day(start),
"end_date": _day(end),
"is_present": is_present,
}
async def _create_child(client: object, name: str = "Ada") -> int:
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def _create_daily_definition(
client: object,
child_id: int,
title: str,
windows: list[str],
*,
skip_on_away: bool = True,
) -> int:
created = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": title,
"rule": {"rule_type": "daily"},
"assignee_child_ids": [child_id],
"windows": windows,
"skip_on_away": skip_on_away,
},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def _set_custody_schedule(client: object, child_id: int) -> None:
"""Anchor the two-week custody cycle on today (week 0 starts today)."""
response = await client.put(
f"/api/v1/admin/children/{child_id}/presence-schedule",
json={
"cycle_length_weeks": 2,
"anchor_date": _day(0),
"pattern": CUSTODY_PATTERN,
},
headers=_admin_headers(),
)
assert response.status_code == 200
async def _materialize_horizon(database: object) -> None:
"""Materialize the household horizon so the stored state is current."""
today = datetime.date.today()
await _materialize.materialize(
database, _day(0), _day(14), today=today
)
async def _open_future_keys(database: object, child_id: int) -> set[tuple]:
"""The child's stored open (no completion event) future instance keys."""
instances = await _dao_instances.QuestInstancesDao(
database
).list_by_date_range(child_id, _day(0), _day(60))
events = _dao_instances.CompletionEventsDao(database)
keys = set()
for instance in instances:
if not await events.list_by_instance(instance.id):
keys.add(
(instance.definition_id, instance.due_date, instance.window)
)
return keys
async def _complete(
database: object, child_id: int, definition_id: int, due_date: str,
window: str,
) -> None:
"""Append a completion event to the stored instance."""
instance = await _dao_instances.QuestInstancesDao(database).get(
definition_id, child_id, due_date, window
)
assert instance is not None
await _dao_instances.CompletionEventsDao(database).append(
instance.id,
child_id,
"completed",
"user",
datetime.datetime.now(datetime.timezone.utc).isoformat(
timespec="seconds"
),
True,
actor_user_id="admin-user-1",
)
async def _consequence(client: object, body: dict[str, object]) -> int:
response = await client.post(
CONSEQUENCE_URL, json=body, headers=_admin_headers()
)
assert response.status_code == 200, response.text
return response.json()["removed"]
@pytest.fixture
async def custody(temp_db_path: str) -> SimpleNamespace:
"""A child on the two-week custody cycle with a two-window chore.
Also assigns a ``skip_on_away = false`` daily chore (generated
regardless of presence) and materializes the horizon, so the stored
instances are the walk's current output.
"""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
database = runner.app.state.db.database
child_id = await _create_child(client)
chore_id = await _create_daily_definition(
client, child_id, "Brush teeth", ["morning", "evening"]
)
always_id = await _create_daily_definition(
client, child_id, "Feed the fish", ["afternoon"],
skip_on_away=False,
)
await _set_custody_schedule(client, child_id)
await _materialize_horizon(database)
yield SimpleNamespace(
client=client,
database=database,
child_id=child_id,
chore_id=chore_id,
always_id=always_id,
)
# --- the count ---------------------------------------------------------------
async def test_custody_override_removes_a_known_number(
custody: SimpleNamespace,
) -> None:
"""Absent over today+2..today+8 removes days +2..+6 x two windows.
Days +7 and +8 are already absent (cycle week 1), and the
skip_on_away=false chore is generated regardless of presence, so
exactly 5 present days x 2 windows = 10 instances would go.
"""
removed = await _consequence(
custody.client, _body(custody.child_id, 2, 8)
)
assert removed == 10
async def test_already_absent_range_removes_nothing(
custody: SimpleNamespace,
) -> None:
"""An absent override over cycle week 1 changes nothing: 0."""
assert await _consequence(
custody.client, _body(custody.child_id, 7, 13)
) == 0
async def test_present_override_removes_nothing(
custody: SimpleNamespace,
) -> None:
"""A present override only ever adds instances: 0."""
assert await _consequence(
custody.client, _body(custody.child_id, 2, 8, is_present=True)
) == 0
async def test_past_range_removes_nothing(custody: SimpleNamespace) -> None:
"""A range entirely before today touches no upcoming instance: 0."""
assert await _consequence(
custody.client, _body(custody.child_id, -20, -1)
) == 0
async def test_completed_instance_in_range_is_not_counted(
custody: SimpleNamespace,
) -> None:
"""D-005: an instance with a completion event is never removed."""
await _complete(
custody.database, custody.child_id, custody.chore_id, _day(3),
"morning",
)
removed = await _consequence(
custody.client, _body(custody.child_id, 2, 8)
)
assert removed == 9
async def test_skip_on_away_false_definition_contributes_nothing(
temp_db_path: str,
) -> None:
"""A child whose only chore ignores presence: absent removes 0."""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
database = runner.app.state.db.database
child_id = await _create_child(client)
await _create_daily_definition(
client, child_id, "Feed the fish", ["morning"],
skip_on_away=False,
)
await _set_custody_schedule(client, child_id)
await _materialize_horizon(database)
assert await _consequence(client, _body(child_id, 0, 14)) == 0
async def test_count_matches_a_direct_materialise_comparison(
custody: SimpleNamespace,
) -> None:
"""Saving the override removes EXACTLY the previewed instances.
With a completed instance inside the range, the preview is taken,
then the override is really created (its core path regenerates the
child) and the open future instances that disappeared are counted
from the database. The completed instance survives the regeneration.
"""
client, database = custody.client, custody.database
await _complete(
database, custody.child_id, custody.chore_id, _day(4), "evening"
)
body = _body(custody.child_id, 1, 9)
removed = await _consequence(client, body)
before = await _open_future_keys(database, custody.child_id)
created = await client.post(
OVERRIDES_URL, json=body, headers=_admin_headers()
)
assert created.status_code == 201
after = await _open_future_keys(database, custody.child_id)
assert after <= before
assert len(before - after) == removed
# Days +1..+6 present x two windows, minus the completed one.
assert removed == 11
completed = await _dao_instances.QuestInstancesDao(database).get(
custody.chore_id, custody.child_id, _day(4), "evening"
)
assert completed is not None
async def test_preview_writes_nothing(custody: SimpleNamespace) -> None:
"""No override is stored and no instance is removed by the preview."""
client, database = custody.client, custody.database
before = await _open_future_keys(database, custody.child_id)
assert await _consequence(client, _body(custody.child_id, 0, 14)) > 0
assert await _open_future_keys(database, custody.child_id) == before
overrides = await _dao_presence.PresenceOverridesDao(
database
).list_by_child_and_range(custody.child_id, "2000-01-01", "2100-01-01")
assert overrides == []
# --- errors ------------------------------------------------------------------
async def test_unknown_child_is_404(custody: SimpleNamespace) -> None:
response = await custody.client.post(
CONSEQUENCE_URL, json=_body(9999, 1, 2), headers=_admin_headers()
)
assert response.status_code == 404
@pytest.mark.parametrize(
"overrides",
[
{"start_date": "2026-02-30"},
{"end_date": "2026-1-5"},
{"start_date": "2026-03-10", "end_date": "2026-03-01"},
{"is_present": "sometimes"},
{"child_id": "one"},
],
)
async def test_malformed_range_or_status_is_422(
custody: SimpleNamespace, overrides: dict[str, object]
) -> None:
body = {**_body(custody.child_id, 1, 2), **overrides}
response = await custody.client.post(
CONSEQUENCE_URL, json=body, headers=_admin_headers()
)
assert response.status_code == 422
@pytest.mark.parametrize(
"overrides",
[
{"child_id": True},
{"child_id": False},
{"child_id": "1"},
{"child_id": 1.0},
{"is_present": 1},
{"is_present": 0},
{"is_present": "true"},
{"is_present": "false"},
],
)
async def test_coercible_child_id_or_status_is_422(
custody: SimpleNamespace, overrides: dict[str, object]
) -> None:
"""Strict fields: a bool id or a numeric/string status is rejected.
The fixture child's id is 1, so a coerced ``true`` would otherwise
target it and return 200. Nothing is stored and nothing removed.
"""
assert custody.child_id == 1
database = custody.database
before = await _open_future_keys(database, custody.child_id)
body = {**_body(custody.child_id, 0, 14), **overrides}
response = await custody.client.post(
CONSEQUENCE_URL, json=body, headers=_admin_headers()
)
assert response.status_code == 422
assert await _open_future_keys(database, custody.child_id) == before
overrides_stored = await _dao_presence.PresenceOverridesDao(
database
).list_by_child_and_range(custody.child_id, "2000-01-01", "2100-01-01")
assert overrides_stored == []
# --- authentication: the router's ONE require_admin --------------------------
async def test_no_credential_is_401(custody: SimpleNamespace) -> None:
response = await custody.client.post(
CONSEQUENCE_URL, json=_body(custody.child_id, 1, 2)
)
assert response.status_code == 401
async def test_panel_service_token_is_403(custody: SimpleNamespace) -> None:
response = await custody.client.post(
CONSEQUENCE_URL,
json=_body(custody.child_id, 1, 2),
headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_non_admin_jwt_is_403(custody: SimpleNamespace) -> None:
non_admin = make_token(groups=("some-other-group",))
response = await custody.client.post(
CONSEQUENCE_URL,
json=_body(custody.child_id, 1, 2),
headers={"Authorization": f"Bearer {non_admin}"},
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
async def test_admin_jwt_is_200(custody: SimpleNamespace) -> None:
response = await custody.client.post(
CONSEQUENCE_URL,
json=_body(custody.child_id, 1, 2),
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json() == {"removed": 4}
tests/test_api_admin_presence_reads.py
+345
View File
@@ -0,0 +1,345 @@
"""Admin-plane presence READ route tests (task 07fde892).
``GET /api/v1/admin/children/{child_id}/presence-schedule`` and
``GET /api/v1/admin/presence-overrides`` against the ASGI app with
httpx, through the shared stubbed-JWKS harness
(:mod:`tests.admin_jwt_harness`) — no network access.
- The schedule read answers ``{"schedule": {...}}`` in the PUT
response's shape for a seeded schedule, ``{"schedule": null}`` for a
child with none (present every day — NOT a 404), and 404 for an
unknown child.
- The overrides read answers ``{"overrides": [...]}`` in the POST
response's shape, ordered by start date, child id, id; ``child_id``
and the ``start``/``end`` range each narrow the list; a malformed or
inverted range is 422 and an unknown ``child_id`` is 404.
- Both routes inherit the router's ONE ``require_admin``: no
credential 401, the panel service token 403, a non-admin JWT 403, an
admin JWT 200.
Seeding goes through the admin write routes, so the reads are proven
against rows the real write path stored.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import ADMIN_KEY, KID, AdminRunner, jwks_for, make_token
OVERRIDES_URL = "/api/v1/admin/presence-overrides"
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
def _schedule_url(child_id: int) -> str:
return f"/api/v1/admin/children/{child_id}/presence-schedule"
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
async def _create_child(client: object, name: str) -> int:
"""Create one child through the admin children route; return the id."""
created = await client.post(
"/api/v1/admin/children",
json={"display_name": name},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def _create_override(
client: object,
child_id: int,
start_date: str,
end_date: str,
*,
is_present: bool = False,
note: str | None = None,
) -> dict[str, object]:
"""Create one override through the admin write route; return it."""
created = await client.post(
OVERRIDES_URL,
json={
"child_id": child_id,
"start_date": start_date,
"end_date": end_date,
"is_present": is_present,
"note": note,
},
headers=_admin_headers(),
)
assert created.status_code == 201, created.text
return created.json()
@pytest.fixture
async def reads(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane client over a fresh database."""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(client=client)
@pytest.fixture
async def seeded(reads: SimpleNamespace) -> SimpleNamespace:
"""Two children with interleaved overrides (Ada has a schedule)."""
client = reads.client
ada = await _create_child(client, "Ada")
ben = await _create_child(client, "Ben")
put = await client.put(
_schedule_url(ada),
json={
"cycle_length_weeks": 2,
"anchor_date": "2026-01-05",
"pattern": {"0": [4, 0, 2], "1": []},
},
headers=_admin_headers(),
)
assert put.status_code == 200
ada_march = await _create_override(
client, ada, "2026-03-01", "2026-03-07", note="grandparents"
)
ben_january = await _create_override(
client, ben, "2026-01-10", "2026-01-12", is_present=True
)
ben_march = await _create_override(client, ben, "2026-03-01", "2026-03-01")
ada_may = await _create_override(client, ada, "2026-05-20", "2026-05-20")
return SimpleNamespace(
client=client,
ada=ada,
ben=ben,
ada_march=ada_march,
ben_january=ben_january,
ben_march=ben_march,
ada_may=ada_may,
)
# --- presence schedule read ------------------------------------------------
async def test_schedule_read_returns_the_stored_schedule(
seeded: SimpleNamespace,
) -> None:
"""A seeded schedule reads back in the PUT response's shape."""
response = await seeded.client.get(
_schedule_url(seeded.ada), headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {
"schedule": {
"child_id": seeded.ada,
"cycle_length_weeks": 2,
"anchor_date": "2026-01-05",
"pattern": {"0": [0, 2, 4], "1": []},
}
}
async def test_schedule_read_without_schedule_is_null_not_404(
seeded: SimpleNamespace,
) -> None:
"""A child with no schedule (present every day) is schedule: null."""
response = await seeded.client.get(
_schedule_url(seeded.ben), headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {"schedule": None}
async def test_schedule_read_unknown_child_is_404(
reads: SimpleNamespace,
) -> None:
"""An unknown child is 404 with the children routes' detail."""
response = await reads.client.get(
_schedule_url(999), headers=_admin_headers()
)
assert response.status_code == 404
assert response.json() == {"detail": "Child not found"}
# --- presence overrides read -----------------------------------------------
async def test_overrides_read_lists_all_in_stable_order(
seeded: SimpleNamespace,
) -> None:
"""Unfiltered: every override, by start date, then child id, then id."""
response = await seeded.client.get(
OVERRIDES_URL, headers=_admin_headers()
)
assert response.status_code == 200
assert response.json() == {
"overrides": [
seeded.ben_january,
seeded.ada_march,
seeded.ben_march,
seeded.ada_may,
]
}
assert seeded.ada_march == {
"id": seeded.ada_march["id"],
"child_id": seeded.ada,
"start_date": "2026-03-01",
"end_date": "2026-03-07",
"is_present": False,
"note": "grandparents",
}
async def test_overrides_read_empty_household(reads: SimpleNamespace) -> None:
"""No overrides at all is an empty list."""
response = await reads.client.get(OVERRIDES_URL, headers=_admin_headers())
assert response.status_code == 200
assert response.json() == {"overrides": []}
async def test_overrides_read_filters_by_child(
seeded: SimpleNamespace,
) -> None:
"""child_id keeps only that child's overrides."""
response = await seeded.client.get(
OVERRIDES_URL,
params={"child_id": seeded.ada},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json() == {
"overrides": [seeded.ada_march, seeded.ada_may]
}
async def test_overrides_read_filters_by_overlapping_range(
seeded: SimpleNamespace,
) -> None:
"""start/end keep overrides overlapping the closed range (inclusive)."""
# 2026-03-07 is Ada's March override's LAST day: the boundary counts.
response = await seeded.client.get(
OVERRIDES_URL,
params={"start": "2026-03-07", "end": "2026-05-20"},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json() == {
"overrides": [seeded.ada_march, seeded.ada_may]
}
async def test_overrides_read_open_ended_range_filters(
seeded: SimpleNamespace,
) -> None:
"""start alone keeps later-ending, end alone keeps earlier-starting."""
from_april = await seeded.client.get(
OVERRIDES_URL, params={"start": "2026-04-01"}, headers=_admin_headers()
)
assert from_april.json() == {"overrides": [seeded.ada_may]}
until_january = await seeded.client.get(
OVERRIDES_URL, params={"end": "2026-01-31"}, headers=_admin_headers()
)
assert until_january.json() == {"overrides": [seeded.ben_january]}
async def test_overrides_read_combines_child_and_range(
seeded: SimpleNamespace,
) -> None:
"""child_id and the range narrow together."""
response = await seeded.client.get(
OVERRIDES_URL,
params={"child_id": seeded.ben, "start": "2026-02-01", "end": "2026-12-31"},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json() == {"overrides": [seeded.ben_march]}
@pytest.mark.parametrize(
"params",
[
{"start": "2026-03-10", "end": "2026-03-01"},
{"start": "2026-3-1"},
{"end": "not-a-date"},
{"start": "2026-02-30"},
],
)
async def test_overrides_read_invalid_range_is_422(
seeded: SimpleNamespace, params: dict[str, str]
) -> None:
"""A malformed date or an inverted range is the core's 422."""
response = await seeded.client.get(
OVERRIDES_URL, params=params, headers=_admin_headers()
)
assert response.status_code == 422
async def test_overrides_read_unknown_child_is_404(
reads: SimpleNamespace,
) -> None:
"""An unknown child_id filter is 404, never an empty list."""
response = await reads.client.get(
OVERRIDES_URL, params={"child_id": 999}, headers=_admin_headers()
)
assert response.status_code == 404
assert response.json() == {"detail": "Child not found"}
# --- authentication: the router's ONE require_admin --------------------------
@pytest.fixture(params=["schedule", "overrides"])
async def read_url(request: pytest.FixtureRequest, reads: SimpleNamespace) -> str:
"""Each read route's URL (the schedule route for a real child)."""
if request.param == "overrides":
return OVERRIDES_URL
return _schedule_url(await _create_child(reads.client, "Ada"))
async def test_read_without_credential_is_401(
reads: SimpleNamespace, read_url: str
) -> None:
response = await reads.client.get(read_url)
assert response.status_code == 401
async def test_read_refuses_panel_service_token(
reads: SimpleNamespace, read_url: str
) -> None:
response = await reads.client.get(
read_url, headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"}
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_read_refuses_non_admin_jwt(
reads: SimpleNamespace, read_url: str
) -> None:
non_admin = make_token(groups=("some-other-group",))
response = await reads.client.get(
read_url, headers={"Authorization": f"Bearer {non_admin}"}
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
async def test_read_admin_jwt_is_200(
reads: SimpleNamespace, read_url: str
) -> None:
response = await reads.client.get(read_url, headers=_admin_headers())
assert response.status_code == 200
tests/test_api_admin_questdef.py
+127 -3
View File
@@ -18,8 +18,9 @@ caveat in api/nestquest_core.py):
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.
semantics); a supplied ``windows`` list replaces the whole set; an
omitted ``assignee_child_ids`` leaves the assignees untouched (the
supplied-list cases live in test_api_admin_questdef_assignees.py).
- ``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);
@@ -215,6 +216,7 @@ async def test_create_persists_definition_with_assignees_and_windows(
"description",
"icon",
"is_active",
"skip_on_away",
"rule",
"assignees",
"windows",
@@ -223,6 +225,7 @@ async def test_create_persists_definition_with_assignees_and_windows(
assert payload["description"] == "Put everything back where it belongs."
assert payload["icon"] == "toy-box"
assert payload["is_active"] is True
assert payload["skip_on_away"] is True
assert payload["rule"] == _weekly_rule_dict()
assert {a["id"]: a["display_name"] for a in payload["assignees"]} == {
ada: "Ada",
@@ -395,7 +398,7 @@ async def test_edit_persists_supplied_fields_and_leaves_omitted_untouched(
)
assert stored.definition.icon == "toy-box"
assert stored.rule.to_dict() == _weekly_rule_dict()
# Assignment is not settable through the edit route: untouched.
# assignee_child_ids was omitted: the assignees are untouched.
assert sorted(child.id for child in stored.assignees) == [ada, bo]
# A second edit touching only the rule leaves the metadata and the
@@ -521,9 +524,130 @@ async def test_edit_rejections_are_422_and_change_nothing(
}
@pytest.mark.parametrize("coercible", [0, 1, "true", "false"])
async def test_create_rejects_coercible_skip_on_away_and_persists_nothing(
admin_questdefs: SimpleNamespace, coercible: object
) -> None:
"""A numeric or string skip_on_away is 422, not coerced to a bool."""
client = admin_questdefs.client
dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
ada = (await _create_children(client, ("Ada",)))[0]
existing = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Feed the cat",
"rule": _weekly_rule_body(),
"assignee_child_ids": [ada],
"windows": ["morning"],
"skip_on_away": False,
},
headers=_admin_headers(),
)
assert existing.status_code == 201
existing_id = existing.json()["id"]
response = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Tidy the den",
"rule": _weekly_rule_body(),
"assignee_child_ids": [ada],
"windows": ["morning"],
"skip_on_away": coercible,
},
headers=_admin_headers(),
)
assert response.status_code == 422
assert set(await _definitions_by_id(admin_questdefs.database)) == {
existing_id
}
assert (await dao.get(existing_id)).skip_on_away is False
@pytest.mark.parametrize("coercible", [0, 1, "true", "false"])
async def test_edit_rejects_coercible_skip_on_away_and_changes_nothing(
admin_questdefs: SimpleNamespace, coercible: object
) -> None:
"""A numeric or string skip_on_away edit is 422; the flag is kept."""
client = admin_questdefs.client
dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
ada = (await _create_children(client, ("Ada",)))[0]
definition_id = (await _create_definition(client, [ada])).json()["id"]
# Start from each stored value so a coerced write would be visible.
for stored in (True, False):
set_flag = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"skip_on_away": stored},
headers=_admin_headers(),
)
assert set_flag.status_code == 200
response = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"skip_on_away": coercible, "title": "Renamed"},
headers=_admin_headers(),
)
assert response.status_code == 422
stored_definition = await dao.get(definition_id)
assert stored_definition.skip_on_away is stored
assert stored_definition.title == "Tidy the den"
# --- set_active -------------------------------------------------------------
async def test_edit_toggles_skip_on_away_and_omission_keeps_it(
admin_questdefs: SimpleNamespace,
) -> None:
"""A supplied skip_on_away persists; an edit omitting it keeps it."""
client = admin_questdefs.client
dao = _dao_rules.QuestDefinitionsDao(admin_questdefs.database)
ada = (await _create_children(client, ("Ada",)))[0]
created = await _create_definition(client, [ada])
assert created.status_code == 201
definition_id = created.json()["id"]
assert created.json()["skip_on_away"] is True
toggled_off = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"skip_on_away": False},
headers=_admin_headers(),
)
assert toggled_off.status_code == 200
assert toggled_off.json()["skip_on_away"] is False
assert (await dao.get(definition_id)).skip_on_away is False
# An edit that omits the flag leaves the stored False untouched.
renamed = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"title": "Tidy the den properly"},
headers=_admin_headers(),
)
assert renamed.status_code == 200
assert renamed.json()["skip_on_away"] is False
stored = await dao.get(definition_id)
assert stored.title == "Tidy the den properly"
assert stored.skip_on_away is False
toggled_on = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"skip_on_away": True},
headers=_admin_headers(),
)
assert toggled_on.status_code == 200
assert toggled_on.json()["skip_on_away"] is True
assert (await dao.get(definition_id)).skip_on_away is True
# A non-boolean value is a malformed body: 422, nothing changes.
rejected = await client.patch(
f"/api/v1/admin/quest-definitions/{definition_id}",
json={"skip_on_away": "sometimes"},
headers=_admin_headers(),
)
assert rejected.status_code == 422
assert (await dao.get(definition_id)).skip_on_away is True
async def test_set_active_toggles_the_flag_and_the_definition_survives(
admin_questdefs: SimpleNamespace,
) -> None:
tests/test_api_admin_questdef_assignees.py
+431
View File
@@ -0,0 +1,431 @@
"""Admin-plane editable assignees on the definition edit route (task 1c8096f6).
``PATCH /api/v1/admin/quest-definitions/{definition_id}`` accepts an
optional ``assignee_child_ids`` list that REPLACES the definition's
assignee set inside the core edit's ONE transaction
(:func:`nestquest_core.quest_definitions.edit_quest_definition`):
- newly listed children are assigned and unlisted ones unassigned; the
response and the database both carry the resulting set;
- an empty list, a duplicate, an unknown child or an inactive child is
422 and NOTHING is written — the stored assignees (and any other
field supplied in the same body) are unchanged;
- a JSON boolean, float or numeric string id is 422 on create and edit
(never coerced to an int) and nothing is written;
- an omitted ``assignee_child_ids`` leaves the assignees unchanged;
- an assignment change regenerates the definition's future instances
through the existing core path;
- the route stays behind the router's ONE ``require_admin`` dependency
(absent credential 401, panel service token 403, non-admin JWT 403,
admin JWT 200).
Every case is proven against the DATABASE, read through the SAME core
copy the app uses (``nestquest_core.*``; see api/nestquest_core.py).
"""
from __future__ import annotations
import datetime
import importlib
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
_dao_rules = importlib.import_module("nestquest_core.dao_rules")
_dao_instances = importlib.import_module("nestquest_core.dao_instances")
_materialize = importlib.import_module("nestquest_core.materialize")
_quest_definitions = importlib.import_module(
"nestquest_core.quest_definitions"
)
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _edit_path(definition_id: int) -> str:
"""The edit route for one definition."""
return f"/api/v1/admin/quest-definitions/{definition_id}"
@pytest.fixture
async def admin_assignees(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane app client plus its live database."""
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,
)
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]) -> int:
"""POST a daily morning definition for the given assignees."""
created = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Brush teeth",
"rule": {"rule_type": "daily"},
"assignee_child_ids": assignee_ids,
"windows": ["morning"],
},
headers=_admin_headers(),
)
assert created.status_code == 201
return created.json()["id"]
async def _stored_assignee_ids(
database: object, definition_id: int
) -> set[int]:
"""The definition's assignee ids as stored."""
dao = _dao_rules.QuestDefinitionsDao(database)
return {child.id for child in await dao.list_assignees(definition_id)}
def _response_assignee_ids(response: object) -> set[int]:
"""The assignee ids a definition payload reports."""
return {assignee["id"] for assignee in response.json()["assignees"]}
# --- add / remove / replace -------------------------------------------------
@pytest.mark.parametrize(
("initial", "requested"),
[
(("Ada",), ("Ada", "Bo")), # add a child
(("Ada", "Bo"), ("Ada",)), # remove a child
(("Ada", "Bo"), ("Cy",)), # replace the whole set
],
)
async def test_edit_replaces_the_assignee_set(
admin_assignees: SimpleNamespace,
initial: tuple[str, ...],
requested: tuple[str, ...],
) -> None:
"""A supplied list becomes exactly the stored and returned set."""
client = admin_assignees.client
names = ("Ada", "Bo", "Cy")
ids = dict(zip(names, await _create_children(client, names)))
definition_id = await _create_definition(
client, [ids[name] for name in initial]
)
wanted = {ids[name] for name in requested}
response = await client.patch(
_edit_path(definition_id),
json={"assignee_child_ids": [ids[name] for name in requested]},
headers=_admin_headers(),
)
assert response.status_code == 200
assert _response_assignee_ids(response) == wanted
assert {a["display_name"] for a in response.json()["assignees"]} == set(
requested
)
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == wanted
# The rest of the definition is untouched by an assignee-only edit.
assert response.json()["title"] == "Brush teeth"
assert [w["window"] for w in response.json()["windows"]] == ["morning"]
async def test_edit_with_assignees_and_other_fields_applies_both(
admin_assignees: SimpleNamespace,
) -> None:
"""Assignees and metadata supplied together land in one edit."""
client = admin_assignees.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
definition_id = await _create_definition(client, [ada])
response = await client.patch(
_edit_path(definition_id),
json={"title": "Floss", "assignee_child_ids": [bo]},
headers=_admin_headers(),
)
assert response.status_code == 200
assert response.json()["title"] == "Floss"
assert _response_assignee_ids(response) == {bo}
dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
assert (await dao.get(definition_id)).title == "Floss"
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {bo}
# --- rejections leave the stored state unchanged ----------------------------
@pytest.mark.parametrize(
("case", "expected_detail"),
[
("empty", "assignee_child_ids must not be empty"),
("unknown", "assignee_child_ids: child 999 does not exist"),
("inactive", "assignee_child_ids: child {cy} is inactive"),
("duplicate", "assignee_child_ids must not contain duplicate"),
("null", "assignee_child_ids must be a list"),
("mixed_unknown", "assignee_child_ids: child 999 does not exist"),
("wrong_type", None), # the body model's own 422
],
)
async def test_rejected_assignee_list_is_422_and_changes_nothing(
admin_assignees: SimpleNamespace,
case: str,
expected_detail: str | None,
) -> None:
"""Empty/unknown/inactive/malformed lists are 422 with no mutation.
The same body also renames the definition, proving the whole edit
is all-or-nothing: neither the assignees nor the title change.
"""
client = admin_assignees.client
ada, bo, cy, di = await _create_children(
client, ("Ada", "Bo", "Cy", "Di")
)
definition_id = await _create_definition(client, [ada, bo])
deactivated = await client.patch(
f"/api/v1/admin/children/{cy}/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert deactivated.status_code == 200
requested: object = {
"empty": [],
"unknown": [999],
"inactive": [ada, cy],
"duplicate": [ada, ada],
"null": None,
# A valid new child alongside an unknown one: di must NOT be
# assigned (all-or-nothing), and ada/bo must NOT be removed.
"mixed_unknown": [di, 999],
"wrong_type": ["Ada"],
}[case]
response = await client.patch(
_edit_path(definition_id),
json={"title": "Renamed", "assignee_child_ids": requested},
headers=_admin_headers(),
)
assert response.status_code == 422, case
if expected_detail is not None:
assert response.json()["detail"].startswith(
expected_detail.format(cy=cy)
), response.json()
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {ada, bo}
dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
assert (await dao.get(definition_id)).title == "Brush teeth"
#: JSON ids the lax ``int`` parser would coerce (``true`` -> 1).
_COERCIBLE_IDS = [True, 1.0, "1"]
@pytest.mark.parametrize("coercible", _COERCIBLE_IDS)
async def test_edit_rejects_coercible_assignee_ids_and_changes_nothing(
admin_assignees: SimpleNamespace, coercible: object
) -> None:
"""A bool, float or numeric-string id edit is 422; nothing changes."""
client = admin_assignees.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
# Assign bo only, so a coerced id 1 (ada) would be a visible change.
definition_id = await _create_definition(client, [bo])
assert ada == 1
response = await client.patch(
_edit_path(definition_id),
json={"title": "Renamed", "assignee_child_ids": [coercible]},
headers=_admin_headers(),
)
assert response.status_code == 422
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {bo}
dao = _dao_rules.QuestDefinitionsDao(admin_assignees.database)
assert (await dao.get(definition_id)).title == "Brush teeth"
@pytest.mark.parametrize("coercible", _COERCIBLE_IDS)
async def test_create_rejects_coercible_assignee_ids_and_persists_nothing(
admin_assignees: SimpleNamespace, coercible: object
) -> None:
"""A bool, float or numeric-string id create is 422; nothing is written."""
client = admin_assignees.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
assert ada == 1
existing_id = await _create_definition(client, [bo])
response = await client.post(
"/api/v1/admin/quest-definitions",
json={
"title": "Tidy the den",
"rule": {"rule_type": "daily"},
"assignee_child_ids": [coercible],
"windows": ["morning"],
},
headers=_admin_headers(),
)
assert response.status_code == 422
bundles = await _quest_definitions.list_active_definitions(
admin_assignees.database
)
assert [bundle.definition.id for bundle in bundles] == [existing_id]
assert await _stored_assignee_ids(
admin_assignees.database, existing_id
) == {bo}
async def test_rejected_assignees_on_unknown_definition_is_404(
admin_assignees: SimpleNamespace,
) -> None:
"""An unknown definition id is still 404 when assignees are supplied."""
client = admin_assignees.client
(ada,) = await _create_children(client, ("Ada",))
response = await client.patch(
_edit_path(999),
json={"assignee_child_ids": [ada]},
headers=_admin_headers(),
)
assert response.status_code == 404
assert response.json()["detail"] == "Quest definition not found"
# --- omitted keeps ----------------------------------------------------------
async def test_omitted_assignees_are_unchanged(
admin_assignees: SimpleNamespace,
) -> None:
"""An edit without assignee_child_ids never touches the assignees."""
client = admin_assignees.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
definition_id = await _create_definition(client, [ada, bo])
response = await client.patch(
_edit_path(definition_id),
json={"title": "Floss"},
headers=_admin_headers(),
)
assert response.status_code == 200
assert _response_assignee_ids(response) == {ada, bo}
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {ada, bo}
# --- regeneration -----------------------------------------------------------
async def test_assignment_change_regenerates_future_instances(
admin_assignees: SimpleNamespace,
) -> None:
"""Swapping the assignee moves the future open instances with it.
The regeneration lives in the core edit path the route calls: the
definition's open instances at or after today are deleted and the
horizon re-materialized for the NEW assignee set.
"""
client = admin_assignees.client
database = admin_assignees.database
ada, bo = await _create_children(client, ("Ada", "Bo"))
definition_id = await _create_definition(client, [ada])
today = datetime.date.today()
start = today.isoformat()
end = (today + datetime.timedelta(days=7)).isoformat()
await _materialize.materialize(database, start, end, today=today)
instances = _dao_instances.QuestInstancesDao(database)
assert await instances.list_by_date_range(ada, start, end)
assert await instances.list_by_date_range(bo, start, end) == []
response = await client.patch(
_edit_path(definition_id),
json={"assignee_child_ids": [bo]},
headers=_admin_headers(),
)
assert response.status_code == 200
assert await instances.list_by_date_range(ada, start, end) == []
regenerated = await instances.list_by_date_range(bo, start, end)
assert regenerated
assert {i.definition_id for i in regenerated} == {definition_id}
# --- authentication ---------------------------------------------------------
async def test_assignee_edit_auth_outcomes(
admin_assignees: SimpleNamespace,
) -> None:
"""401 absent, 403 panel token, 403 non-admin, 200 admin."""
client = admin_assignees.client
ada, bo = await _create_children(client, ("Ada", "Bo"))
definition_id = await _create_definition(client, [ada])
body = {"assignee_child_ids": [bo]}
absent = await client.patch(_edit_path(definition_id), json=body)
assert absent.status_code == 401
panel = await client.patch(
_edit_path(definition_id),
json=body,
headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
)
assert panel.status_code == 403
non_admin_token = make_token(groups=("some-other-group",))
non_admin = await client.patch(
_edit_path(definition_id),
json=body,
headers={"Authorization": f"Bearer {non_admin_token}"},
)
assert non_admin.status_code == 403
# None of the refused calls reached the handler.
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {ada}
admin = await client.patch(
_edit_path(definition_id), json=body, headers=_admin_headers()
)
assert admin.status_code == 200
assert await _stored_assignee_ids(
admin_assignees.database, definition_id
) == {bo}
tests/test_api_admin_questdef_list.py
+299
View File
@@ -0,0 +1,299 @@
"""Admin-plane quest-definition LIST route tests (task d4667573).
``GET /api/v1/admin/quest-definitions`` 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
with NO network access.
- An admin JWT lists EVERY definition — active AND inactive — in rising
id order, each in the same documented shape as the create/edit
responses (id, title, description, icon, is_active, rule, assignees,
windows).
- An empty household lists nothing: 200 with an empty list.
- Authentication runs through the router's ONE ``require_admin``
dependency: an absent credential is 401, the panel service token is
refused outright (403, the credential-type detail), and a JWT without
the nestquest-admins group is 403.
"""
from __future__ import annotations
from types import SimpleNamespace
import pytest
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
#: The admin list route under test.
LIST_PATH = "/api/v1/admin/quest-definitions"
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
@pytest.fixture
async def admin_client(temp_db_path: str) -> SimpleNamespace:
"""An admin-plane app client over a fresh (empty) household."""
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield client
async def _seed_active_and_inactive(client: object) -> tuple[int, int, int]:
"""Seed one child, an ACTIVE definition and an INACTIVE one.
Returns ``(child_id, active_id, inactive_id)``. Seeded through the
admin routes so the rows are exactly what the admin plane writes.
"""
child = await client.post(
"/api/v1/admin/children",
json={"display_name": "Ada"},
headers=_admin_headers(),
)
assert child.status_code == 201
child_id = child.json()["id"]
active = await client.post(
LIST_PATH,
json={
"title": "Tidy the den",
"rule": {
"rule_type": "weekly",
"interval": 1,
"weekday_set": [0, 2],
"start_date": "2026-03-02",
},
"assignee_child_ids": [child_id],
"windows": ["morning", ["evening", "19:30"]],
"description": "Put everything back where it belongs.",
"icon": "toy-box",
},
headers=_admin_headers(),
)
assert active.status_code == 201
inactive = await client.post(
LIST_PATH,
json={
"title": "Water the plants",
"rule": {
"rule_type": "daily",
"interval": 1,
"start_date": "2026-03-01",
},
"assignee_child_ids": [child_id],
"windows": ["afternoon"],
},
headers=_admin_headers(),
)
assert inactive.status_code == 201
inactive_id = inactive.json()["id"]
deactivated = await client.patch(
f"{LIST_PATH}/{inactive_id}/active",
json={"is_active": False},
headers=_admin_headers(),
)
assert deactivated.status_code == 200
return child_id, active.json()["id"], inactive_id
async def test_admin_lists_active_and_inactive_definitions_with_full_shape(
admin_client: object,
) -> None:
"""Both definitions come back, by rising id, in the documented shape."""
child_id, active_id, inactive_id = await _seed_active_and_inactive(
admin_client
)
response = await admin_client.get(LIST_PATH, headers=_admin_headers())
assert response.status_code == 200
assert response.json() == {
"definitions": [
{
"id": active_id,
"title": "Tidy the den",
"description": "Put everything back where it belongs.",
"icon": "toy-box",
"is_active": True,
"skip_on_away": True,
"rule": {
"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,
},
"assignees": [{"id": child_id, "display_name": "Ada"}],
"windows": [
{"window": "morning", "due_time": None},
{"window": "evening", "due_time": "19:30"},
],
},
{
"id": inactive_id,
"title": "Water the plants",
"description": None,
"icon": None,
"is_active": False,
"skip_on_away": True,
"rule": {
"rule_type": "daily",
"interval": 1,
"weekday_set": None,
"day_of_month": None,
"nth_weekday": None,
"nth_weekday_weekday": None,
"month": None,
"start_date": "2026-03-01",
"end_date": None,
},
"assignees": [{"id": child_id, "display_name": "Ada"}],
"windows": [{"window": "afternoon", "due_time": None}],
},
]
}
async def test_list_items_match_the_create_response_shape(
admin_client: object,
) -> None:
"""Each listed item is exactly what the create route returned."""
child = await admin_client.post(
"/api/v1/admin/children",
json={"display_name": "Ben"},
headers=_admin_headers(),
)
created = await admin_client.post(
LIST_PATH,
json={
"title": "Feed the cat",
"rule": {
"rule_type": "daily",
"interval": 1,
"start_date": "2026-03-01",
},
"assignee_child_ids": [child.json()["id"]],
"windows": ["morning"],
},
headers=_admin_headers(),
)
assert created.status_code == 201
response = await admin_client.get(LIST_PATH, headers=_admin_headers())
assert response.status_code == 200
assert response.json() == {"definitions": [created.json()]}
async def test_skip_on_away_round_trips_true_and_false_through_list(
admin_client: object,
) -> None:
"""skip_on_away posted true, false, or omitted lists back as stored."""
child = await admin_client.post(
"/api/v1/admin/children",
json={"display_name": "Cleo"},
headers=_admin_headers(),
)
child_id = child.json()["id"]
async def create(title: str, extra: dict[str, object]) -> dict:
response = await admin_client.post(
LIST_PATH,
json={
"title": title,
"rule": {
"rule_type": "daily",
"interval": 1,
"start_date": "2026-03-01",
},
"assignee_child_ids": [child_id],
"windows": ["morning"],
**extra,
},
headers=_admin_headers(),
)
assert response.status_code == 201
return response.json()
skipping = await create("Skip when away", {"skip_on_away": True})
keeping = await create("Keep when away", {"skip_on_away": False})
defaulted = await create("Default", {})
assert skipping["skip_on_away"] is True
assert keeping["skip_on_away"] is False
assert defaulted["skip_on_away"] is True
response = await admin_client.get(LIST_PATH, headers=_admin_headers())
assert response.status_code == 200
listed = {item["id"]: item for item in response.json()["definitions"]}
assert listed[skipping["id"]]["skip_on_away"] is True
assert listed[keeping["id"]]["skip_on_away"] is False
assert listed[defaulted["id"]]["skip_on_away"] is True
assert response.json() == {"definitions": [skipping, keeping, defaulted]}
async def test_empty_household_lists_no_definitions(
admin_client: object,
) -> None:
"""With no definitions stored the list is 200 and empty."""
response = await admin_client.get(LIST_PATH, headers=_admin_headers())
assert response.status_code == 200
assert response.json() == {"definitions": []}
async def test_list_without_credentials_is_401(admin_client: object) -> None:
"""An absent credential is the uniform 401."""
response = await admin_client.get(LIST_PATH)
assert response.status_code == 401
async def test_list_with_panel_service_token_is_403(
admin_client: object,
) -> None:
"""The panel service token is refused outright, never a JWT."""
response = await admin_client.get(
LIST_PATH,
headers={"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"},
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_list_with_non_admin_jwt_is_403(admin_client: object) -> None:
"""A valid JWT without the nestquest-admins group is 403."""
non_admin = make_token(groups=("some-other-group",))
response = await admin_client.get(
LIST_PATH, headers={"Authorization": f"Bearer {non_admin}"}
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
tests/test_api_admin_snapshot.py
+328
View File
@@ -0,0 +1,328 @@
"""Admin-plane snapshot route tests (task ef3f4877).
These tests exercise ``GET /api/v1/admin/snapshot`` 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 with NO network access. The household is seeded with
:func:`tests.test_api_panel._seed_household` through the SAME core
copy the app uses (``nestquest_core.*``), and both routes' single
clock read (``_local_now``) is pinned to noon so nothing straddles
midnight.
Proven per the done-condition:
- An admin JWT returns 200 with the documented shape: ``today_iso``,
``cycle_day`` and, per child, the presence fields, the rollup counts
and ``instances`` in the core ``instance_payload`` shape.
- A MISSED instance is KEPT in the admin payload (D-009) and still
OMITTED from the panel payload. The core builder lists only
instances due on ``now.date()``, and ``derive_state`` calls an
instance missed only when its due date is BEFORE that day — so a
real snapshot never carries a missed view. The test therefore
wraps the real builder to add the seeded past-due instance as a
view whose state the core's own ``derive_state`` computes
(``missed``), and reads it through both routes.
- Auth runs through the router's ONE ``require_admin`` dependency: no
credential is 401, the panel service token is refused outright
(403, the credential-type detail), and a non-admin JWT is 403.
- The handler performs exactly ONE clock read per request.
"""
from __future__ import annotations
import dataclasses
import datetime
import importlib
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import pytest
from api import routes_admin, routes_panel
from api.nestquest_core import core_snapshot
from tests.admin_jwt_harness import (
ADMIN_KEY,
KID,
AdminRunner,
jwks_for,
make_token,
)
from tests.test_api_panel import _seed_household
# The completion layer through the SAME core copy the app uses — the
# missed test derives its view's state with the core's own rule.
_completion = importlib.import_module("nestquest_core.completion")
#: The panel service token the harness app is configured with.
HARNESS_PANEL_TOKEN = "admin-test-panel-token"
#: The 403 detail for the panel service token on an admin route.
PANEL_TOKEN_REJECTION_DETAIL = (
"This credential type is not accepted for admin access"
)
#: The 403 detail for a valid JWT without the admin group.
FORBIDDEN_DETAIL = (
"Admin access requires membership in the nestquest-admins group"
)
def _admin_headers() -> dict[str, str]:
"""The Authorization header of a valid nestquest-admins JWT."""
return {"Authorization": f"Bearer {make_token()}"}
def _panel_headers() -> dict[str, str]:
"""The Authorization header carrying the harness panel token."""
return {"Authorization": f"Bearer {HARNESS_PANEL_TOKEN}"}
@pytest.fixture
async def snapshot_household(
temp_db_path: str, monkeypatch
) -> SimpleNamespace:
"""A seeded household served by the admin-plane app, clock pinned.
``today`` comes from one host clock read; the pinned instant is
12:00 UTC that day (past Ada's 10:00 due, before Bo's 17:00). Both
the admin and the panel route's ``_local_now`` return it, and the
admin one counts its calls.
"""
today = datetime.datetime.now().astimezone().date()
pinned_now = datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
)
seed = await _seed_household(temp_db_path, today, pinned_now)
clock_reads: list[datetime.datetime] = []
def _admin_now() -> datetime.datetime:
clock_reads.append(pinned_now)
return pinned_now
monkeypatch.setattr(routes_admin, "_local_now", _admin_now)
monkeypatch.setattr(routes_panel, "_local_now", lambda: pinned_now)
runner = AdminRunner(temp_db_path, jwks_for(ADMIN_KEY, KID))
async with runner as client:
yield SimpleNamespace(
client=client,
seed=seed,
today=today,
pinned_now=pinned_now,
pinned_utc_iso=pinned_now.astimezone(
datetime.timezone.utc
).isoformat(),
clock_reads=clock_reads,
)
# --- authentication: the ONE require_admin dependency ----------------------
async def test_snapshot_without_credential_is_401(
snapshot_household: SimpleNamespace,
) -> None:
"""No Authorization header is the uniform 401."""
response = await snapshot_household.client.get("/api/v1/admin/snapshot")
assert response.status_code == 401
async def test_snapshot_refuses_panel_service_token_outright(
snapshot_household: SimpleNamespace,
) -> None:
"""The panel service token is refused as a credential type (403).
403 with the credential-type detail — not the uniform 401 and not
the non-admin group detail — proves the token was refused outright,
never tried as a JWT.
"""
response = await snapshot_household.client.get(
"/api/v1/admin/snapshot", headers=_panel_headers()
)
assert response.status_code == 403
assert response.json()["detail"] == PANEL_TOKEN_REJECTION_DETAIL
async def test_snapshot_refuses_non_admin_jwt(
snapshot_household: SimpleNamespace,
) -> None:
"""A valid JWT without the nestquest-admins group is 403."""
non_admin = make_token(groups=("some-other-group",))
response = await snapshot_household.client.get(
"/api/v1/admin/snapshot",
headers={"Authorization": f"Bearer {non_admin}"},
)
assert response.status_code == 403
assert response.json()["detail"] == FORBIDDEN_DETAIL
# --- the payload -----------------------------------------------------------
async def test_admin_jwt_returns_documented_snapshot(
snapshot_household: SimpleNamespace,
) -> None:
"""An admin JWT returns 200 with the documented snapshot shape.
Every field is asserted as an explicit literal, with ONE clock
read for the request. With no missed view in the real snapshot,
the admin body equals the panel body for the same household.
"""
client = snapshot_household.client
seed = snapshot_household.seed
response = await client.get(
"/api/v1/admin/snapshot", headers=_admin_headers()
)
assert response.status_code == 200
assert len(snapshot_household.clock_reads) == 1
body = response.json()
assert body == {
"today_iso": snapshot_household.today.isoformat(),
# Ada's 2-week cycle anchored yesterday => today is cycle day 2.
"cycle_day": 2,
"children": [
{
"child_id": seed.ada.id,
"child_name": "Ada",
"present": False,
"next_present": seed.returns_date.isoformat(),
"due_today": 1,
"completed_today": 0,
"remaining_today": 1,
"completion_pct": 0,
"instances": [
{
"id": seed.pack_bag_instance.id,
"definition_id": seed.pack_bag_instance.definition_id,
"child_id": seed.ada.id,
"title": "Pack bag",
"icon": None,
"window": "morning",
"due_time": "10:00",
"state": "open",
"overdue": True,
"completed_at": None,
"on_time": None,
}
],
},
{
"child_id": seed.bo.id,
"child_name": "Bo",
"present": True,
"next_present": None,
"due_today": 1,
"completed_today": 1,
"remaining_today": 0,
"completion_pct": 100,
"instances": [
{
"id": seed.brush_teeth_instance.id,
"definition_id": (
seed.brush_teeth_instance.definition_id
),
"child_id": seed.bo.id,
"title": "Brush teeth",
"icon": None,
"window": "morning",
"due_time": "17:00",
"state": "completed",
"overdue": False,
"completed_at": snapshot_household.pinned_utc_iso,
"on_time": True,
}
],
},
],
}
panel = await client.get(
"/api/v1/panel/snapshot", headers=_panel_headers()
)
assert panel.status_code == 200
assert panel.json() == body
async def test_admin_keeps_missed_instance_panel_omits_it(
snapshot_household: SimpleNamespace, monkeypatch
) -> None:
"""A missed view is KEPT by the admin route, OMITTED by the panel's.
The real builder is wrapped to add the seeded past-due Stale chore
instance to Ada's views, its state derived by the core's own
``derive_state`` (no events, due yesterday => ``missed``). Both
routes read the SAME wrapped snapshot, so the only difference is
each route's ``include_missed`` flag.
"""
client = snapshot_household.client
seed = snapshot_household.seed
stale = seed.stale_instance
missed_state = _completion.derive_state(
stale, None, snapshot_household.today
)
assert missed_state == "missed"
missed_view = core_snapshot.QuestInstanceView(
instance_id=stale.id,
definition_id=stale.definition_id,
child_id=stale.child_id,
title="Stale chore",
icon=None,
window=stale.window,
due_date=stale.due_date,
due_time=stale.due_time,
state=missed_state,
overdue=False,
completed_at=None,
was_on_time=None,
)
real_build_snapshot = core_snapshot.build_snapshot
async def _with_missed_view(database, settings, now):
snapshot = await real_build_snapshot(database, settings, now)
return dataclasses.replace(
snapshot,
children=tuple(
dataclasses.replace(
child, instances=(*child.instances, missed_view)
)
if child.child_id == seed.ada.id
else child
for child in snapshot.children
),
)
monkeypatch.setattr(core_snapshot, "build_snapshot", _with_missed_view)
admin = await client.get(
"/api/v1/admin/snapshot", headers=_admin_headers()
)
assert admin.status_code == 200
admin_ada = admin.json()["children"][0]
assert admin_ada["child_id"] == seed.ada.id
assert [instance["id"] for instance in admin_ada["instances"]] == [
seed.pack_bag_instance.id,
stale.id,
]
assert admin_ada["instances"][1] == {
"id": stale.id,
"definition_id": stale.definition_id,
"child_id": seed.ada.id,
"title": "Stale chore",
"icon": None,
"window": "morning",
"due_time": "09:00",
"state": "missed",
"overdue": False,
"completed_at": None,
"on_time": None,
}
panel = await client.get(
"/api/v1/panel/snapshot", headers=_panel_headers()
)
assert panel.status_code == 200
panel_ada = panel.json()["children"][0]
assert [instance["id"] for instance in panel_ada["instances"]] == [
seed.pack_bag_instance.id
], "the panel payload must still omit missed instances"
tests/test_api_admin_sse.py
+283
View File
@@ -0,0 +1,283 @@
"""Admin-plane SSE tests: the admin transition stream (task 92f75680).
These tests exercise ``GET /api/v1/admin/events`` — the admin plane's
Server-Sent Events stream for the Admin PWA — against a REAL server:
- The stream inherits the admin router's ONE ``require_admin``
dependency: an absent credential is 401, the panel service token and
a non-admin JWT are 403, all before any stream is established.
- An admin JWT subscribes; each of the four documented transitions
(quest-completed, quest-uncompleted, quest-missed, child-day-complete)
arrives as an ``event: <type>`` / ``data: <json>`` frame carrying its
documented payload fields (design/ENTITIES-AND-SERVICES.md §3).
- The admin and panel streams are fed by the SAME in-process publisher:
one transition arrives, identically, on both.
Streaming approach: :mod:`tests.test_api_sse`'s live-uvicorn reader
(httpx's ASGITransport buffers SSE bodies), on the admin-enabled
:class:`tests.test_api_admin_uncomplete._AdminServer` whose JWKS cache
is stubbed, so a real admin JWT verifies with NO network access.
"""
from __future__ import annotations
import datetime
from types import SimpleNamespace
from zoneinfo import ZoneInfo
import httpx
import pytest
from api import routes_admin, routes_panel
from tests.admin_jwt_harness import make_token
from tests.test_api_admin_uncomplete import _AdminServer
from tests.test_api_panel import PANEL_TOKEN, _seed_household
from tests.test_api_sse import (
EVENT_CHILD_DAY_COMPLETE,
EVENT_QUEST_COMPLETED,
EVENT_QUEST_MISSED,
EVENT_QUEST_UNCOMPLETED,
READ_TIMEOUT,
_stream,
)
EVENTS_PATH = "/api/v1/admin/events"
@pytest.fixture
def temp_db_path(tmp_path) -> str:
"""A fresh per-test SQLite path under pytest's tmp_path."""
return str(tmp_path / "nestquest.db")
@pytest.fixture
async def admin_live(temp_db_path: str) -> SimpleNamespace:
"""A seeded household served by a live admin-enabled uvicorn app.
Both planes' clocks are pinned to 12:00 UTC (past Ada's 10:00 due,
before Bo's 17:00) BEFORE the server starts — the app is created
inside uvicorn's factory thread.
"""
today = datetime.datetime.now().astimezone().date()
pinned_now = datetime.datetime(
today.year, today.month, today.day, 12, 0, 0, tzinfo=ZoneInfo("UTC")
)
seed = await _seed_household(temp_db_path, today, pinned_now)
original_panel = routes_panel._local_now
original_admin = routes_admin._local_now
routes_panel._local_now = lambda: pinned_now
routes_admin._local_now = lambda: pinned_now
try:
with _AdminServer(temp_db_path) as server:
client = httpx.AsyncClient(
base_url=server.url(""), timeout=READ_TIMEOUT
)
async with client:
yield SimpleNamespace(client=client, seed=seed, server=server)
finally:
routes_panel._local_now = original_panel
routes_admin._local_now = original_admin
async def _panel_complete(live: SimpleNamespace, instance_id: int, child_id: int):
"""Complete ``instance_id`` through the panel route (a real transition)."""
response = await live.client.post(
f"/api/v1/panel/instances/{instance_id}/complete",
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
json={"actor_child_id": child_id},
)
assert response.status_code == 200
# --- auth: the ONE require_admin dependency, no stream on refusal ----------
async def test_admin_events_refuses_absent_panel_and_non_admin(
admin_live: SimpleNamespace,
) -> None:
"""No credential is 401; the panel token and a non-admin JWT are 403.
Each refusal is a complete (non-streaming) JSON error response, so
no stream is established for any of them.
"""
client = admin_live.client
absent = await client.get(EVENTS_PATH)
assert absent.status_code == 401
assert not absent.headers["content-type"].startswith("text/event-stream")
panel = await client.get(
EVENTS_PATH, headers={"Authorization": f"Bearer {PANEL_TOKEN}"}
)
assert panel.status_code == 403
assert panel.json()["detail"] == (
"This credential type is not accepted for admin access"
)
non_admin = await client.get(
EVENTS_PATH,
headers={
"Authorization": f"Bearer {make_token(groups=('other-group',))}"
},
)
assert non_admin.status_code == 403
assert non_admin.json()["detail"] == (
"Admin access requires membership in the nestquest-admins group"
)
async def test_admin_jwt_is_not_accepted_on_panel_events(
admin_live: SimpleNamespace,
) -> None:
"""The panel plane is unchanged: an admin JWT does not open it."""
response = await admin_live.client.get(
"/api/v1/panel/events",
headers={"Authorization": f"Bearer {make_token()}"},
)
assert response.status_code == 401
async def test_admin_events_sets_no_cache(admin_live: SimpleNamespace) -> None:
"""The admin stream is ``text/event-stream`` with ``no-cache``."""
async with admin_live.client.stream(
"GET",
EVENTS_PATH,
headers={"Authorization": f"Bearer {make_token()}"},
) as response:
assert response.status_code == 200
assert response.headers["content-type"].startswith("text/event-stream")
assert response.headers["cache-control"] == "no-cache"
# --- the four documented transitions on the admin stream -------------------
async def test_admin_stream_carries_completed_and_day_complete(
admin_live: SimpleNamespace,
) -> None:
"""A panel completion streams quest-completed then child-day-complete.
Ada's only quest today is Pack bag, so completing it also clears her
day; both frames arrive on the ADMIN stream with their documented
payload fields.
"""
seed = admin_live.seed
pack_bag = seed.pack_bag_instance
stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
async with stream:
await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_COMPLETED == "nestquest_quest_completed"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == pack_bag.id
assert payload["quest_title"] == "Pack bag"
assert payload["window"] == "morning"
assert payload["due_date"] == pack_bag.due_date
assert payload["due_time"] == "10:00"
assert payload["occurred_at"]
assert payload["was_on_time"] is False # 10:00 due, 12:00 now
completed_at = payload["occurred_at"]
event_type, payload = await stream.next_event()
assert event_type == EVENT_CHILD_DAY_COMPLETE
assert event_type == "nestquest_child_day_complete"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["quests_due"] == 1
assert payload["quests_completed"] == 1
assert payload["occurred_at"] == completed_at
async def test_admin_stream_carries_uncompleted(
admin_live: SimpleNamespace,
) -> None:
"""The admin uncomplete route's reversal streams quest-uncompleted."""
seed = admin_live.seed
brush = seed.brush_teeth_instance
admin_headers = {"Authorization": f"Bearer {make_token()}"}
stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
async with stream:
response = await admin_live.client.post(
f"/api/v1/admin/instances/{brush.id}/uncomplete",
headers=admin_headers,
)
assert response.status_code == 200
assert response.json()["appended"] is True
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_UNCOMPLETED
assert event_type == "nestquest_quest_uncompleted"
assert payload["child_id"] == seed.bo.id
assert payload["child_name"] == "Bo"
assert payload["instance_id"] == brush.id
assert payload["quest_title"] == "Brush teeth"
assert payload["window"] == "morning"
assert payload["due_date"] == brush.due_date
assert payload["due_time"] == "17:00"
assert payload["occurred_at"]
assert "was_on_time" not in payload
async def test_admin_stream_carries_missed(admin_live: SimpleNamespace) -> None:
"""The missed-sweep publisher's transition streams to the admin client.
Drives the API-side helper the sweep calls, on the SERVER's loop,
against the app's own database and its ONE publisher.
"""
from api import transitions
seed = admin_live.seed
stale = seed.stale_instance
app = admin_live.server.app
stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
async with stream:
admin_live.server.run_on_loop(
transitions.publish_quest_missed(
app.state.db.database, stale.id, app.state.publisher
)
)
event_type, payload = await stream.next_event()
assert event_type == EVENT_QUEST_MISSED == "nestquest_quest_missed"
assert payload["child_id"] == seed.ada.id
assert payload["child_name"] == "Ada"
assert payload["instance_id"] == stale.id
assert payload["quest_title"] == "Stale chore"
assert payload["window"] == "morning"
assert payload["due_date"] == stale.due_date
assert payload["due_time"] == "09:00"
assert payload["occurred_at"]
# --- one publisher feeds both planes ---------------------------------------
async def test_admin_and_panel_streams_share_one_publisher(
admin_live: SimpleNamespace,
) -> None:
"""One transition arrives, identically, on the admin AND panel streams.
The route publishes once to ``app.state.publisher``; both planes'
subscribers receive the same frames (same ``occurred_at``), which a
second, admin-only publisher could never produce.
"""
seed = admin_live.seed
pack_bag = seed.pack_bag_instance
admin_stream = _stream(admin_live.server.url(EVENTS_PATH), make_token())
panel_stream = _stream(admin_live.server.url("/api/v1/panel/events"))
async with admin_stream, panel_stream:
await _panel_complete(admin_live, pack_bag.id, seed.ada.id)
admin_events = [
await admin_stream.next_event(),
await admin_stream.next_event(),
]
panel_events = [
await panel_stream.next_event(),
await panel_stream.next_event(),
]
assert [event_type for event_type, _ in admin_events] == [
EVENT_QUEST_COMPLETED,
EVENT_CHILD_DAY_COMPLETE,
]
assert admin_events == panel_events
tests/test_api_sse.py
+10 -5
View File
@@ -158,7 +158,8 @@ class _EventStream:
parsed frame with a timeout.
"""
def __init__(self) -> None:
def __init__(self, token: str = PANEL_TOKEN) -> None:
self.token = token
self.queue: asyncio.Queue = asyncio.Queue()
self._task: asyncio.Task | None = None
self._client: httpx.AsyncClient | None = None
@@ -170,7 +171,7 @@ class _EventStream:
self._stream_cm = self._client.stream(
"GET",
self.url,
headers={"Authorization": f"Bearer {PANEL_TOKEN}"},
headers={"Authorization": f"Bearer {self.token}"},
)
self._response = await self._stream_cm.__aenter__()
assert self._response.status_code == 200
@@ -263,9 +264,13 @@ async def seeded_panel(temp_db_path: str) -> SimpleNamespace:
routes_panel._local_now = original
def _stream(url: str) -> _EventStream:
"""A subscribed SSE reader for ``url``."""
return _EventStream()._set_url(url)
def _stream(url: str, token: str = PANEL_TOKEN) -> _EventStream:
"""A subscribed SSE reader for ``url``, sending ``token`` as Bearer.
Defaults to the panel service token; the admin-plane stream tests
pass an admin JWT instead.
"""
return _EventStream(token)._set_url(url)
# --- auth and content type ----------------------------------------------
tests/test_core_sweep.py
+295
View File
@@ -0,0 +1,295 @@
"""Cross-process safety of the missed-sweep watermark claim.
Feature 11's sweep guards its announcement with the watermark; the
original guard was a read-then-set inside a per-PROCESS asyncio lock,
so two PROCESSES sharing one SQLite file (the HA integration's
rollover sweep and the API service's scheduler) could both read the
old watermark and both announce the same missed batch. These tests
pin the fix:
- :meth:`MetaStateDao.claim` is the atomic, database-level primitive:
one conditional upsert whose ``rowcount`` says whether THIS caller
changed the row. The first claimant of a value wins; a second
claimant of the same value — even on an INDEPENDENT connection to
the same file, the shape two processes produce — loses.
- :func:`run_missed_sweep` gates the announcement on that claim: two
independent ``NestQuestDatabase`` wrappers (two processes' shape)
racing one seeded database yields exactly one announcer.
- The documented sweep semantics are unchanged: a same-night rerun
announces nothing, and a run after days of downtime announces the
accumulated ``[watermark, today)`` window exactly once.
Everything here runs against the bundled HA-free core copy
(``custom_components.nestquest.core``) with a plain
``asyncio.to_thread`` executor — no Home Assistant, no API server.
"""
from __future__ import annotations
import asyncio
import datetime
from types import SimpleNamespace
import pytest
from custom_components.nestquest.core.dao_instances import (
QuestInstancesDao,
)
from custom_components.nestquest.core.dao_meta import MetaStateDao
from custom_components.nestquest.core.db import NestQuestDatabase
from custom_components.nestquest.core.migrations import apply_migrations
from custom_components.nestquest.core.children import create_child
from custom_components.nestquest.core.quest_definitions import (
create_quest_definition,
)
from custom_components.nestquest.core.materialize import materialize
from custom_components.nestquest.core.recurrence import ScheduleRule
from custom_components.nestquest.core.sweep import (
SWEEP_WATERMARK_KEY,
run_missed_sweep,
)
#: One missed event per unswept past-due instance — the sweep's only
#: event type.
EVENT_QUEST_MISSED = "nestquest_quest_missed"
async def _executor(fn, *args, **kwargs):
"""Run a synchronous callable on a worker thread and await it.
The :class:`NestQuestDatabase` executor contract (the shape of
``hass.async_add_executor_job``), so sqlite3 calls never block the
event loop.
"""
return await asyncio.to_thread(fn, *args, **kwargs)
async def _open_database(db_path: str) -> NestQuestDatabase:
"""Open (without migrating) one independent connection wrapper."""
database = NestQuestDatabase(_executor)
await database.open(db_path)
return database
async def _seed_missed_household(
db_path: str, today: datetime.date
) -> SimpleNamespace:
"""Seed a household with two past-due open instances on ``db_path``.
One child, two quests whose ONLY instances are due YESTERDAY
(materialized with ``today`` pinned to yesterday, ``end_date``
closing the rules so the later today-walk cannot duplicate them).
Seeding uses its OWN connection, opened and closed here, so the
connections the tests race with start fresh.
"""
database = await _open_database(db_path)
try:
await apply_migrations(database)
today_iso = today.isoformat()
yesterday = today - datetime.timedelta(days=1)
yesterday_iso = yesterday.isoformat()
rule = ScheduleRule.from_dict(
{
"rule_type": "daily",
"start_date": yesterday_iso,
"end_date": yesterday_iso,
}
)
ada = await create_child(database, "Ada", sort_order=0)
await create_quest_definition(
database, "Stale chore", rule, [ada.id], [("morning", "09:00")]
)
await create_quest_definition(
database, "Old chore", rule, [ada.id], [("evening", "18:30")]
)
await materialize(database, yesterday_iso, yesterday_iso, today=yesterday)
# A quest due TODAY (created after the yesterday walk so it has
# no past-due instance): the post-downtime catch-up run's
# target — inside [watermark=today, later_today) but not
# inside the day-one sweep's window.
await create_quest_definition(
database,
"Recoverable chore",
ScheduleRule.from_dict(
{"rule_type": "daily", "start_date": today_iso}
),
[ada.id],
[("afternoon", "17:00")],
)
await materialize(database, today_iso, today_iso, today=today)
instances = QuestInstancesDao(database)
rows = await instances.list_by_child_and_date(ada.id, yesterday_iso)
assert len(rows) == 2
today_rows = await instances.list_by_child_and_date(
ada.id, today_iso
)
assert len(today_rows) == 1
return SimpleNamespace(
child=ada,
instances=rows,
recoverable_instance=today_rows[0],
yesterday_iso=yesterday_iso,
today_iso=today_iso,
)
finally:
await database.close()
# --- the claim primitive: atomic at the database level ------------------
async def test_claim_first_caller_wins_and_second_same_value_loses(
temp_db_path: str,
) -> None:
"""claim() is a conditional upsert: first True, same-value second False.
A different value always wins (an advance re-arms the check), and
the stored value follows the winner.
"""
database = await _open_database(temp_db_path)
try:
await apply_migrations(database)
dao = MetaStateDao(database)
assert await dao.get(SWEEP_WATERMARK_KEY) is None
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is True
# Same value again: the row is already there — nothing changed.
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-01") is False
assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-01"
# A DIFFERENT value is an advance, not a re-claim: it wins.
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is True
assert await dao.get(SWEEP_WATERMARK_KEY) == "2026-09-02"
assert await dao.claim(SWEEP_WATERMARK_KEY, "2026-09-02") is False
finally:
await database.close()
async def test_claim_is_atomic_across_independent_connections(
temp_db_path: str,
) -> None:
"""Two connections to one file cannot both claim the same value.
This is the two-process shape: independent sqlite3 connections (one
per ``NestQuestDatabase`` wrapper), racing the SAME key on the SAME
event loop with no shared lock. The single conditional upsert is
serialized by SQLite itself, so exactly one caller's rowcount is 1.
"""
first = await _open_database(temp_db_path)
second = await _open_database(temp_db_path)
try:
await apply_migrations(first)
won = await asyncio.gather(
MetaStateDao(first).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
MetaStateDao(second).claim(SWEEP_WATERMARK_KEY, "2026-09-01"),
)
assert sorted(won) == [False, True]
# One stored row, holding the winner's value.
assert await MetaStateDao(first).get(SWEEP_WATERMARK_KEY) == (
"2026-09-01"
)
assert await MetaStateDao(second).get(SWEEP_WATERMARK_KEY) == (
"2026-09-01"
)
finally:
await first.close()
await second.close()
# --- the sweep: the announcement is gated by the atomic claim -----------
async def test_two_process_shape_race_exactly_one_announces(
temp_db_path: str,
) -> None:
"""Two independent connections racing the sweep: ONE announcer.
Each wrapper carries its own connection AND its own per-process
sweep lock, so nothing in-process serialises the two runs — the
only guard left is the database-level claim. Both read the unset
watermark and both build the events, but exactly one claim wins:
one run returns the missed events, the loser returns [].
"""
today = datetime.datetime.now().astimezone().date()
seed = await _seed_missed_household(temp_db_path, today)
first = await _open_database(temp_db_path)
second = await _open_database(temp_db_path)
try:
results = await asyncio.gather(
run_missed_sweep(first, today=today),
run_missed_sweep(second, today=today),
)
announcers = [r for r in results if r]
assert len(announcers) == 1
event_type, payload = announcers[0][0]
assert event_type == EVENT_QUEST_MISSED
assert payload["due_date"] == seed.yesterday_iso
# Both instances of the seeded window, one event each.
assert len(announcers[0]) == 2
losers = [r for r in results if not r]
assert losers == [[]]
# The watermark holds the run's date for whoever looks next.
assert (
await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
== seed.today_iso
)
finally:
await first.close()
await second.close()
async def test_same_night_rerun_announces_nothing(temp_db_path: str) -> None:
"""A second sweep for the same date is an empty no-op (the watermark)."""
today = datetime.datetime.now().astimezone().date()
await _seed_missed_household(temp_db_path, today)
database = await _open_database(temp_db_path)
try:
first = await run_missed_sweep(database, today=today)
assert len(first) == 2
second = await run_missed_sweep(database, today=today)
assert second == []
finally:
await database.close()
async def test_post_downtime_run_announces_accumulated_window_once(
temp_db_path: str,
) -> None:
"""After days of downtime the catch-up run sweeps the window once.
The first run sweeps yesterday's instances (watermark := today);
``today`` then jumps two days and the catch-up run — on a
DIFFERENT connection, the two-process shape — announces the
accumulated ``[watermark, later_today)`` window exactly once, and
a rerun on the original connection is silent again.
"""
today = datetime.datetime.now().astimezone().date()
later_today = today + datetime.timedelta(days=2)
seed = await _seed_missed_household(temp_db_path, today)
first = await _open_database(temp_db_path)
second = await _open_database(temp_db_path)
try:
# Day one: yesterday's two instances fire once.
day_one = await run_missed_sweep(first, today=today)
assert len(day_one) == 2
assert all(
payload["due_date"] == seed.yesterday_iso
for _, payload in day_one
)
# The downtime catch-up: the window is [watermark, later_today).
catchup = await run_missed_sweep(second, today=later_today)
assert len(catchup) == 1
event_type, payload = catchup[0]
assert event_type == EVENT_QUEST_MISSED
assert payload["due_date"] == seed.today_iso
assert payload["instance_id"] == seed.recoverable_instance.id
assert payload["child_id"] == seed.child.id
# A further rerun announces nothing.
assert await run_missed_sweep(first, today=later_today) == []
assert (
await MetaStateDao(first).get(SWEEP_WATERMARK_KEY)
== later_today.isoformat()
)
finally:
await first.close()
await second.close()
tests/test_dao_rules.py
+67
View File
@@ -847,6 +847,7 @@ def test_definition_update_cannot_change_assignee_or_active(
"description",
"icon",
"due_time",
"skip_on_away",
}
assert await definitions.update(
definition.id, title="A2"
@@ -862,6 +863,72 @@ def test_definition_update_cannot_change_assignee_or_active(
_with_db(tmp_path, "definition-update-shape.db")(_body)
def test_definition_update_toggles_skip_on_away_and_omission_keeps_it(
tmp_path,
) -> None:
async def _body(database, rules, definitions, children, child):
rule = await rules.create("daily", "2026-09-14")
definition = await definitions.create("A", rule.id, NOW)
assert definition.skip_on_away is True
assert await definitions.update(
definition.id, skip_on_away=False
) == 1
assert (await definitions.get(definition.id)).skip_on_away is False
# Omitting the flag (None) leaves the stored False untouched.
assert await definitions.update(definition.id, title="A2") == 1
kept = await definitions.get(definition.id)
assert kept.title == "A2"
assert kept.skip_on_away is False
assert await definitions.update(definition.id) == 0
assert await definitions.update(
definition.id, skip_on_away=True
) == 1
assert (await definitions.get(definition.id)).skip_on_away is True
_with_db(tmp_path, "definition-update-skip-on-away.db")(_body)
@pytest.mark.parametrize("bad", [2, -1, 0.0, 1.5, "1", "true", b"\x01"])
def test_skip_on_away_rejected_unless_zero_or_one_at_every_write(
tmp_path, bad
) -> None:
"""Every DAO entry point taking the flag rejects non-0/1 values.
This is the only guard on migrated databases, whose column was
added without the fresh-schema CHECK constraint.
"""
async def _body(database, rules, definitions, children, child):
rule = await rules.create("daily", "2026-09-14")
definition = await definitions.create("A", rule.id, NOW)
with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
await definitions.create("B", rule.id, NOW, skip_on_away=bad)
with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
await definitions.create_with_rule_and_windows(
"C",
schedule_rule_to_storage(
ScheduleRule(
rule_type=RuleType.DAILY, start_date="2026-09-14"
)
),
NOW,
[child.id],
[("morning", None)],
skip_on_away=bad,
)
with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
await definitions.edit_definition(
definition.id, title="A2", skip_on_away=bad
)
with pytest.raises(ValueError, match="skip_on_away must be 0 or 1"):
await definitions.update(
definition.id, title="A2", skip_on_away=bad
)
stored = await definitions.list_all()
assert [(d.title, d.skip_on_away) for d in stored] == [("A", True)]
_with_db(tmp_path, "definition-skip-on-away-invalid.db")(_body)
def test_definition_update_no_fields_returns_zero(tmp_path) -> None:
async def _body(database, rules, definitions, children, child):
rule = await rules.create("daily", "2026-09-14")
tests/test_materialize.py
+127
View File
@@ -657,6 +657,133 @@ def test_materialize_honours_overrides(tmp_path) -> None:
_with_db(tmp_path, "materialize-override.db")(_body)
async def _absent_midweek_setup(database, skip_on_away: bool):
"""One child absent on days 2..4 of a future week, one daily quest.
Returns ``(child_id, definition_id, start, end_iso, absent_dates)``.
"""
from custom_components.nestquest.core.dao_presence import (
PresenceOverridesDao,
)
child = await ChildrenDao(database).create("Ada", NOW)
start = _future_monday()
end_iso = (start + datetime.timedelta(days=6)).isoformat()
absent = {
(start + datetime.timedelta(days=offset)).isoformat()
for offset in (2, 3, 4)
}
await PresenceOverridesDao(database).create(
child.id, min(absent), max(absent), False, note="away"
)
created = await create_quest_definition(
database,
"Daily chore",
ScheduleRule(rule_type=RuleType.DAILY, start_date=start.isoformat()),
[child.id],
["morning"],
skip_on_away=skip_on_away,
)
assert created.definition.skip_on_away is skip_on_away
return child.id, created.definition.id, start, end_iso, absent
def test_materialize_skip_on_away_true_skips_absent_dates(tmp_path) -> None:
async def _body(database):
child_id, _, start, end_iso, absent = await _absent_midweek_setup(
database, skip_on_away=True
)
count = await materialize(database, start.isoformat(), end_iso)
records = await QuestInstancesDao(database).list_by_date_range(
child_id, start.isoformat(), end_iso
)
due_dates = {record.due_date for record in records}
assert count == 4
assert due_dates.isdisjoint(absent)
assert len(due_dates) == 4
_with_db(tmp_path, "skip-on-away-true.db")(_body)
def test_materialize_skip_on_away_false_generates_absent_dates(
tmp_path,
) -> None:
async def _body(database):
child_id, _, start, end_iso, absent = await _absent_midweek_setup(
database, skip_on_away=False
)
count = await materialize(database, start.isoformat(), end_iso)
records = await QuestInstancesDao(database).list_by_date_range(
child_id, start.isoformat(), end_iso
)
due_dates = {record.due_date for record in records}
assert count == 7
assert absent <= due_dates
assert len(due_dates) == 7
_with_db(tmp_path, "skip-on-away-false.db")(_body)
def test_edit_skip_on_away_regenerates_absent_dates(tmp_path) -> None:
"""Toggling the flag through the edit path rebuilds future instances."""
async def _body(database):
child_id, definition_id, start, end_iso, absent = (
await _absent_midweek_setup(database, skip_on_away=True)
)
await materialize(database, start.isoformat(), end_iso)
dao = QuestInstancesDao(database)
async def due_dates() -> set[str]:
records = await dao.list_by_date_range(
child_id, start.isoformat(), end_iso
)
return {record.due_date for record in records}
assert (await due_dates()).isdisjoint(absent)
edited = await edit_quest_definition(
database, definition_id, skip_on_away=False
)
assert edited.definition.skip_on_away is False
assert absent <= await due_dates()
# None leaves the stored flag unchanged.
unchanged = await edit_quest_definition(
database, definition_id, title="Daily chore (renamed)"
)
assert unchanged.definition.skip_on_away is False
edited = await edit_quest_definition(
database, definition_id, skip_on_away=True
)
assert edited.definition.skip_on_away is True
assert (await due_dates()).isdisjoint(absent)
_with_db(tmp_path, "skip-on-away-edit.db")(_body)
def test_skip_on_away_rejects_non_bool(tmp_path) -> None:
async def _body(database):
child = await ChildrenDao(database).create("Ada", NOW)
rule = ScheduleRule(rule_type=RuleType.DAILY, start_date="2026-01-01")
with pytest.raises(ValueError, match="skip_on_away must be a real bool"):
await create_quest_definition(
database, "Chore", rule, [child.id], ["morning"],
skip_on_away=1,
)
created = await create_quest_definition(
database, "Chore", rule, [child.id], ["morning"]
)
with pytest.raises(ValueError, match="skip_on_away must be a real bool"):
await edit_quest_definition(
database, created.definition.id, skip_on_away="false"
)
stored = await QuestDefinitionsDao(database).get(created.definition.id)
assert stored.skip_on_away is True
_with_db(tmp_path, "skip-on-away-validation.db")(_body)
def test_materialize_rejects_malformed_or_inverted_bounds(tmp_path) -> None:
async def _body(database):
start = (_future_monday()).isoformat()
tests/test_migrations.py
+93 -2
View File
@@ -153,7 +153,9 @@ def test_migration_list_shape() -> None:
migration 3 brings pre-D-008 definitions to the multi-assignee
model, migration 4 adds the windows table, migration 5 rebuilds
quest_instances onto the widened (definition, child, date, window)
key, and migration 6 adds actor_child_id to completion_events.
key, migration 6 adds actor_child_id to completion_events,
migration 7 adds the meta_state table and migration 8 adds
skip_on_away to quest_definitions.
"""
assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
assert callable(MIGRATIONS[1])
@@ -162,7 +164,8 @@ def test_migration_list_shape() -> None:
assert callable(MIGRATIONS[4])
assert callable(MIGRATIONS[5])
assert MIGRATIONS[6] == SCHEMA_V7_META_STATE_DDL
assert len(MIGRATIONS) == 7
assert callable(MIGRATIONS[7])
assert len(MIGRATIONS) == 8
# ---------------------------------------------------------------------------
@@ -564,6 +567,13 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
database.fetch_all("PRAGMA table_info(quest_definitions)")
)
assert "child_id" not in {c[1] for c in columns}
# The rebuilt row reads the pre-flag behaviour: skip on away.
skip = _run(
database.fetch_one(
"SELECT skip_on_away FROM quest_definitions WHERE id = 1"
)
)
assert skip == (1,)
# The single assignee moved into the assignees table.
assignees = _run(
database.fetch_all(
@@ -637,6 +647,87 @@ def test_v2_definitions_rebuilt_to_multi_assignee(tmp_path) -> None:
_run(database.close())
# ---------------------------------------------------------------------------
# Migration 8: skip_on_away on quest_definitions (Feature 22)
# ---------------------------------------------------------------------------
def _seed_v7_database(database) -> None:
"""Build a genuine version-7 database holding one definition.
Migrations 1..7 run first; the fresh v1 DDL already carries
``skip_on_away`` (canonical shape), so the column is dropped to
reproduce the exact pre-migration-8 table a shipped v7 file holds,
and a definition row is written through that old column list.
"""
assert _migrate(database, MIGRATIONS[:7]) == 7
_run(
database.execute(
"ALTER TABLE quest_definitions DROP COLUMN skip_on_away"
)
)
columns = _run(database.fetch_all("PRAGMA table_info(quest_definitions)"))
assert "skip_on_away" not in {c[1] for c in columns}
_run(
database.execute(
"INSERT INTO schedule_rules (id, rule_type, interval, "
"start_date) VALUES (1, 'daily', 1, '2026-09-01')"
)
)
_run(
database.execute(
"INSERT INTO quest_definitions (id, title, schedule_rule_id, "
"is_active, created_at) VALUES (1, 'Dishes', 1, 1, "
"'2026-09-01T00:00:00+00:00')"
)
)
def test_v7_definitions_gain_skip_on_away_defaulting_true(tmp_path) -> None:
from custom_components.nestquest.core.dao_rules import (
QuestDefinitionsDao,
)
database = _open_db(tmp_path / "v7-upgrade.db")
try:
_seed_v7_database(database)
before = _version(database)
assert before == 7
final = _migrate(database)
assert final == before + 1
assert _version(database) == before + 1
columns = {
c[1]: c
for c in _run(
database.fetch_all("PRAGMA table_info(quest_definitions)")
)
}
assert "skip_on_away" in columns
# (cid, name, type, notnull, dflt_value, pk)
assert columns["skip_on_away"][2:5] == ("INTEGER", 1, "1")
record = _run(QuestDefinitionsDao(database).get(1))
assert record is not None
assert record.title == "Dishes"
assert record.skip_on_away is True
finally:
_run(database.close())
def test_skip_on_away_migration_is_noop_on_fresh_schema(tmp_path) -> None:
"""A fresh file already has the column; step 8 only stamps."""
database = _open_db(tmp_path / "fresh-v8.db")
try:
assert _migrate(database) == len(MIGRATIONS)
columns = _run(
database.fetch_all("PRAGMA table_info(quest_definitions)")
)
assert [c[1] for c in columns].count("skip_on_away") == 1
finally:
_run(database.close())
# ---------------------------------------------------------------------------
# Idempotence: second start is a no-op
# ---------------------------------------------------------------------------
tests/test_schema.py
+25
View File
@@ -322,6 +322,7 @@ def test_quest_definitions_columns_types_and_constraints(tmp_path) -> None:
("due_time", "TEXT", 0, None, 0),
("is_active", "INTEGER", 1, "1", 0),
("created_at", "TEXT", 1, None, 0),
("skip_on_away", "INTEGER", 1, "1", 0),
]
finally:
_run(database.close())
@@ -633,6 +634,30 @@ def test_schedule_rules_rule_type_check_enforced(tmp_path) -> None:
_run(database.close())
@pytest.mark.parametrize("bad_flag", [2, -1])
def test_quest_definitions_skip_on_away_check_enforced(
tmp_path, bad_flag
) -> None:
database = _open_db(tmp_path / f"skip-on-away-{bad_flag}.db")
try:
_apply(database)
_child(database)
_insert_rule(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_definition(database, skip_on_away=bad_flag)
definition_id = _insert_definition(database, skip_on_away=0)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_run(
database.execute(
"UPDATE quest_definitions SET skip_on_away = ? "
"WHERE id = ?",
(bad_flag, definition_id),
)
)
finally:
_run(database.close())
def test_schedule_rules_interval_zero_fails(tmp_path) -> None:
database = _open_db(tmp_path / "interval-zero.db")
try: