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

Quest windows via quest_definition_windows (D-008)

Browse Source 
Per IMPLEMENTATION-ORDER.md 1.4: a quest_definition_windows table
(definition_id, window, due_time) declares which day windows a
definition spans, one row per definition+window via the composite
primary key; window is CHECK-constrained to morning/afternoon/evening.

- const.py: WINDOW_MORNING/AFTERNOON/EVENING, QUEST_WINDOWS canonical
  order, WINDOW_CLOCK_RANGES with the documented boundaries and the
  boundary-to-later-window rule documented for Feature 07.
- schema.py: windows DDL joins SCHEMA_V1_STATEMENTS; due_time is
  caller-validated strict HH:MM (DAO layer), matching the date policy.
- migrations.py: migration 4 is a plain CREATE IF NOT EXISTS statement
  list (no rebuild needed for a new table).
- dao_rules.py: QuestDefinitionsDao gains upsert_window (idempotent
  per window, due_time refresh in place), list_windows (const order),
  remove_window; _validate_time/_validate_window helpers.
- tests: windows DDL shape/constraints (CHECK matrix, duplicate,
  unknown definition), DAO round-trips, const range shapes; leak
  guards extended to the new table.
This commit is contained in:
Joshua King
2026-09-15 10:05:43 -04:00
parent 25f0a43e16
commit ebc2064222
8 changed files with 486 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_const.py
+30
View File
@@ -73,6 +73,36 @@ def test_panel_idle_timeout_config() -> None:
assert const.DEFAULT_PANEL_IDLE_TIMEOUT > 0
def test_quest_windows_canonical_order() -> None:
"""QUEST_WINDOWS is exactly the three spellings in canonical order."""
assert const.QUEST_WINDOWS == ("morning", "afternoon", "evening")
assert const.WINDOW_MORNING == "morning"
assert const.WINDOW_AFTERNOON == "afternoon"
assert const.WINDOW_EVENING == "evening"
# No duplicates, every name covered by its individual constant.
assert len(set(const.QUEST_WINDOWS)) == 3
def test_window_clock_ranges_cover_every_window() -> None:
"""Each window has a strict HH:MM (start, end) range.
The ranges follow the documented boundaries (D-008): morning ends
11:59, afternoon spans 12:00-17:00, evening 17:00-21:00, and each
window's end equals the next window's start (the boundary belongs
to the later window at classification time).
"""
assert const.WINDOW_CLOCK_RANGES == {
"morning": ("00:00", "11:59"),
"afternoon": ("12:00", "17:00"),
"evening": ("17:00", "21:00"),
}
for window in const.QUEST_WINDOWS:
start, end = const.WINDOW_CLOCK_RANGES[window]
for value in (start, end):
parsed = datetime.strptime(value, "%H:%M").time()
assert parsed.strftime("%H:%M") == value
def test_no_hardcoded_domain_or_db_filename_in_modules() -> None:
"""Ensure no module other than const.py hardcodes DOMAIN string or SQLITE_DB_FILENAME."""
pkg_dir = Path(const.__file__).parent
tests/test_dao_rules.py
+94 -3
View File
@@ -11,6 +11,7 @@ from custom_components.nestquest.dao_rules import (
ScheduleRuleRecord,
ScheduleRulesDao,
QuestDefinitionRecord,
QuestDefinitionWindowRecord,
QuestDefinitionsDao,
)
from custom_components.nestquest.db import NestQuestDatabase
@@ -790,6 +791,96 @@ def test_definition_list_assignees_orders_by_child_sort_order(
_with_db(tmp_path, "definition-assignees-order.db")(_body)
def test_definition_upsert_window_round_trip(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)
window = await definitions.upsert_window(
definition.id, "morning", due_time="09:00"
)
assert window == QuestDefinitionWindowRecord(
definition.id, "morning", "09:00"
)
# Upsert: re-declaring the same window updates in place, never
# duplicates (one row per definition+window).
updated = await definitions.upsert_window(
definition.id, "morning", due_time="09:30"
)
assert updated.due_time == "09:30"
rows = await definitions.list_windows(definition.id)
assert [w.window for w in rows] == ["morning"]
# All three windows list in const order regardless of insertion.
await definitions.upsert_window(definition.id, "evening")
await definitions.upsert_window(definition.id, "afternoon")
rows = await definitions.list_windows(definition.id)
assert [w.window for w in rows] == [
"morning",
"afternoon",
"evening",
]
assert rows[1].due_time is None
return rows
_with_db(tmp_path, "definition-window-roundtrip.db")(_body)
def test_definition_window_unknown_name_raises(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)
with pytest.raises(ValueError, match="window must be one of"):
await definitions.upsert_window(definition.id, "noon")
with pytest.raises(ValueError, match="window must be one of"):
await definitions.remove_window(definition.id, "MORNING")
return None
_with_db(tmp_path, "definition-window-bad-name.db")(_body)
def test_definition_window_bad_due_time_raises(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)
with pytest.raises(ValueError, match="HH:MM"):
await definitions.upsert_window(
definition.id, "morning", due_time="9:30"
)
with pytest.raises(ValueError, match="HH:MM"):
await definitions.upsert_window(
definition.id, "morning", due_time="25:00"
)
return None
_with_db(tmp_path, "definition-window-bad-time.db")(_body)
def test_definition_window_missing_definition_raises(tmp_path) -> None:
async def _body(database, rules, definitions, children, child):
with pytest.raises(ValueError, match="does not exist"):
await definitions.upsert_window(999, "morning")
return None
_with_db(tmp_path, "definition-window-missing.db")(_body)
def test_definition_remove_window_round_trip(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)
await definitions.upsert_window(definition.id, "morning")
assert await definitions.remove_window(
definition.id, "morning"
) is True
assert await definitions.list_windows(definition.id) == []
# Removing an absent window reports False.
assert await definitions.remove_window(
definition.id, "morning"
) is False
return None
_with_db(tmp_path, "definition-window-remove.db")(_body)
def test_definition_unassigned_has_no_assignees(tmp_path) -> None:
async def _body(database, rules, definitions, children, child):
rule = await rules.create("daily", "2026-09-14")
@@ -889,7 +980,7 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
}
sql_pattern = re.compile(
r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+[`'\"]*(\[)?"
r"(schedule_rules|quest_definitions|quest_definition_assignees)\b",
r"(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
re.IGNORECASE,
)
offenders: list[str] = []
@@ -941,7 +1032,7 @@ def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
)
sql_pattern = re.compile(
r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees)\b",
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
re.IGNORECASE,
)
assert sql_pattern.search(remaining) is None, (
@@ -986,7 +1077,7 @@ def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
)
sql_pattern = re.compile(
r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees)\b",
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions|quest_definition_assignees|quest_definition_windows)\b",
re.IGNORECASE,
)
assert sql_pattern.search(remaining) is None, (
tests/test_migrations.py
+11 -5
View File
@@ -17,7 +17,10 @@ from custom_components.nestquest.migrations import (
apply_migrations,
read_schema_version,
)
from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS
from custom_components.nestquest.schema import (
SCHEMA_V1_STATEMENTS,
SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
)
def _run(coro):
@@ -64,6 +67,7 @@ def _counts(database) -> dict[str, int]:
"schedule_rules",
"quest_definitions",
"quest_definition_assignees",
"quest_definition_windows",
"presence_schedules",
"presence_overrides",
"quest_instances",
@@ -101,6 +105,7 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
"schedule_rules",
"quest_definitions",
"quest_definition_assignees",
"quest_definition_windows",
"presence_schedules",
"presence_overrides",
"quest_instances",
@@ -138,18 +143,19 @@ def test_empty_version_table_reads_as_version_0(tmp_path) -> None:
def test_migration_list_shape() -> None:
"""Migration 1 is the v1 DDL; 2 and 3 are the callable steps.
"""Migration 1 is the v1 DDL; 2 and 3 are callable steps.
Nothing ever shipped, so migration 1 was rewritten pre-release
(D-007) to create the quest_* tables directly; migration 2 exists
for dev databases stamped version 1 by the pre-rename runner, and
for dev databases stamped version 1 by the pre-rename runner,
migration 3 brings pre-D-008 definitions to the multi-assignee
model.
model, and migration 4 adds the windows table.
"""
assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
assert callable(MIGRATIONS[1])
assert callable(MIGRATIONS[2])
assert len(MIGRATIONS) == 3
assert MIGRATIONS[3] == SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL
assert len(MIGRATIONS) == 4
# ---------------------------------------------------------------------------
tests/test_schema.py
+163 -1
View File
@@ -17,6 +17,7 @@ from custom_components.nestquest.schema import (
SCHEMA_V1_STATEMENTS,
SCHEMA_V1_QUEST_DEFINITIONS_DDL,
SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
SCHEMA_V1_QUEST_INSTANCES_DDL,
async_apply_ddl,
)
@@ -167,13 +168,14 @@ def _insert_override(database, child_id, **overrides):
# ---------------------------------------------------------------------------
def test_schema_v1_statements_compose_the_seven_tables() -> None:
def test_schema_v1_statements_compose_the_eight_tables() -> None:
assert SCHEMA_V1_STATEMENTS == [
*SCHEMA_V1_CHILDREN_DDL,
*SCHEMA_V1_ADMIN_USERS_DDL,
*SCHEMA_V1_SCHEDULE_RULES_DDL,
*SCHEMA_V1_QUEST_DEFINITIONS_DDL,
*SCHEMA_V1_QUEST_DEFINITION_ASSIGNEES_DDL,
*SCHEMA_V1_QUEST_DEFINITION_WINDOWS_DDL,
*SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
*SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
*SCHEMA_V1_QUEST_INSTANCES_DDL,
@@ -974,6 +976,166 @@ def test_quest_definition_assignees_duplicate_pair_fails(tmp_path) -> None:
_run(database.close())
def test_quest_definition_windows_table_exists_after_applying_ddl(
tmp_path,
) -> None:
database = _open_db(tmp_path / "windows.db")
try:
_apply(database)
row = _run(
database.fetch_one(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name = 'quest_definition_windows'"
)
)
assert row is not None and row[0] == "quest_definition_windows"
finally:
_run(database.close())
def test_quest_definition_windows_columns_types_and_constraints(
tmp_path,
) -> None:
database = _open_db(tmp_path / "windows-columns.db")
try:
_apply(database)
columns = _run(
database.fetch_all("PRAGMA table_info(quest_definition_windows)")
)
# cid, name, type, notnull, dflt_value, pk — the composite
# primary key spans definition_id and window.
assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
("definition_id", "INTEGER", 1, None, 1),
("window", "TEXT", 1, None, 2),
("due_time", "TEXT", 0, None, 0),
]
finally:
_run(database.close())
def test_quest_definition_windows_foreign_keys_declared(tmp_path) -> None:
database = _open_db(tmp_path / "windows-fks.db")
try:
_apply(database)
fks = _run(
database.fetch_all(
"PRAGMA foreign_key_list(quest_definition_windows)"
)
)
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {
("quest_definitions", "definition_id", "id"),
}
finally:
_run(database.close())
def test_quest_definition_windows_valid_rows_round_trip(tmp_path) -> None:
database = _open_db(tmp_path / "windows-roundtrip.db")
try:
_apply(database)
_run(
database.execute(
"INSERT INTO children (display_name, created_at) VALUES (?, ?)",
("Ada", "2026-09-13T00:00:00+00:00"),
)
)
_insert_rule(database)
definition_id = _insert_definition(database, assignees=None)
for window, due in (
("morning", "09:00"),
("afternoon", None),
("evening", "19:30"),
):
_run(
database.execute(
"INSERT INTO quest_definition_windows "
"(definition_id, window, due_time) VALUES (?, ?, ?)",
(definition_id, window, due),
)
)
rows = _run(
database.fetch_all(
"SELECT window, due_time FROM quest_definition_windows "
"WHERE definition_id = ? ORDER BY window",
(definition_id,),
)
)
assert rows == [
("afternoon", None),
("evening", "19:30"),
("morning", "09:00"),
]
finally:
_run(database.close())
@pytest.mark.parametrize(
"window", ["Midnight", "MORNING", "noon", "", "evenings"]
)
def test_quest_definition_windows_window_check_enforced(
tmp_path, window
) -> None:
database = _open_db(tmp_path / "windows-check.db")
try:
_apply(database)
_insert_rule(database)
definition_id = _insert_definition(database, assignees=None)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_run(
database.execute(
"INSERT INTO quest_definition_windows "
"(definition_id, window, due_time) VALUES (?, ?, NULL)",
(definition_id, window),
)
)
finally:
_run(database.close())
def test_quest_definition_windows_duplicate_window_fails(tmp_path) -> None:
database = _open_db(tmp_path / "windows-duplicate.db")
try:
_apply(database)
_insert_rule(database)
definition_id = _insert_definition(database, assignees=None)
_run(
database.execute(
"INSERT INTO quest_definition_windows "
"(definition_id, window, due_time) "
"VALUES (?, 'morning', NULL)",
(definition_id,),
)
)
with pytest.raises(sqlite3.IntegrityError):
_run(
database.execute(
"INSERT INTO quest_definition_windows "
"(definition_id, window, due_time) "
"VALUES (?, 'morning', '09:00')",
(definition_id,),
)
)
finally:
_run(database.close())
def test_quest_definition_windows_unknown_definition_fails(tmp_path) -> None:
database = _open_db(tmp_path / "windows-bad-definition.db")
try:
_apply(database)
with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
_run(
database.execute(
"INSERT INTO quest_definition_windows "
"(definition_id, window, due_time) "
"VALUES (999, 'morning', NULL)"
)
)
finally:
_run(database.close())
def test_quest_definition_assignees_multiple_children_allowed(
tmp_path,
) -> None: