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

Feature 22: Admin Definitions Read and skip_on_away (#193, #194) (#195)

Browse Source 
Co-authored-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
Co-committed-by: dev_agent <17+dev_agent@code.cubecraftcreations.com>
This commit was merged in pull request #195.
This commit is contained in:
dev_agent
2026-09-23 20:52:06 -04:00
committed by approver_agent
parent a808f89993
commit 42ea5710d7
12 changed files with 925 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_api_admin_questdef.py
+123
View File
@@ -215,6 +215,7 @@ async def test_create_persists_definition_with_assignees_and_windows(
"description",
"icon",
"is_active",
"skip_on_away",
"rule",
"assignees",
"windows",
@@ -223,6 +224,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",
@@ -521,9 +523,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_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_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: