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

Rename task_* tables to quest_* with legacy-dev migration path (D-007)

Browse Source 
Per design/IMPLEMENTATION-ORDER.md 1.1 and decision D-007: the domain
model takes the quest vocabulary pre-release. SCHEMA_V1 now creates
quest_definitions/quest_instances directly; DAO classes, records and
error messages follow (QuestDefinitionsDao, QuestInstancesDao). The
migration runner gains migration 2, an async-callable step that detects
surviving task_* tables on dev machines and ALTER TABLE RENAMES them
inside the migration transaction, with FK clauses rewritten by SQLite.
Migration 1's content rewrite is legal because nothing was ever
shipped; dev databases stamped version 1 ride migration 2 up to 2.
This commit is contained in:
Joshua King
2026-09-15 09:35:21 -04:00
parent 3ecca2d212
commit c988f6508b
11 changed files with 546 additions and 222 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_dao_instances.py
+15 -15
View File
@@ -1,4 +1,4 @@
"""Tests for dao_instances.py: task_instances + append-only completion_events."""
"""Tests for dao_instances.py: quest_instances + append-only completion_events."""
from __future__ import annotations
import asyncio
@@ -11,10 +11,10 @@ from custom_components.nestquest.dao_children import ChildrenDao
from custom_components.nestquest.dao_instances import (
CompletionEventRecord,
CompletionEventsDao,
TaskInstanceRecord,
TaskInstancesDao,
QuestInstanceRecord,
QuestInstancesDao,
)
from custom_components.nestquest.dao_rules import ScheduleRulesDao, TaskDefinitionsDao
from custom_components.nestquest.dao_rules import ScheduleRulesDao, QuestDefinitionsDao
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
@@ -76,8 +76,8 @@ async def _prepare(path) -> tuple:
await apply_migrations(database)
children = ChildrenDao(database)
rules = ScheduleRulesDao(database)
definitions = TaskDefinitionsDao(database)
instances = TaskInstancesDao(database)
definitions = QuestDefinitionsDao(database)
instances = QuestInstancesDao(database)
events = CompletionEventsDao(database)
child = await children.create("Ada", _now_stamp())
rule = await rules.create("daily", D1)
@@ -111,7 +111,7 @@ def _with_db(tmp_path, name):
# ---------------------------------------------------------------------------
# task_instances: upsert idempotency
# quest_instances: upsert idempotency
# ---------------------------------------------------------------------------
@@ -121,13 +121,13 @@ def test_upsert_creates_then_is_idempotent(tmp_path) -> None:
first = await instances.upsert(
definition.id, child.id, D1, _now_stamp()
)
assert isinstance(first, TaskInstanceRecord)
assert isinstance(first, QuestInstanceRecord)
second = await instances.upsert(
definition.id, child.id, D1, _now_stamp()
)
assert second.id == first.id
rows = await database.fetch_one(
"SELECT COUNT(*) FROM task_instances"
"SELECT COUNT(*) FROM quest_instances"
)
assert rows == (1,)
return second
@@ -323,8 +323,8 @@ def test_upsert_no_past_check_uses_execution_date_not_call_date(
await apply_migrations(database)
children = ChildrenDao(database)
rules = ScheduleRulesDao(database)
definitions = TaskDefinitionsDao(database)
instances = TaskInstancesDao(database)
definitions = QuestDefinitionsDao(database)
instances = QuestInstancesDao(database)
events = CompletionEventsDao(database)
child = await children.create("Ada", _now_stamp())
rule = await rules.create("daily", D1)
@@ -369,7 +369,7 @@ def test_upsert_refuses_immutable_completed_instance(tmp_path) -> None:
# ---------------------------------------------------------------------------
# task_instances: get / list
# quest_instances: get / list
# ---------------------------------------------------------------------------
@@ -453,7 +453,7 @@ def test_list_by_date_range_rejects_inverted_range(tmp_path) -> None:
# ---------------------------------------------------------------------------
# task_instances: delete_future_uncompleted
# quest_instances: delete_future_uncompleted
# ---------------------------------------------------------------------------
@@ -1084,8 +1084,8 @@ def test_append_concurrent_events_serialize(tmp_path) -> None:
async def _prepare_in(database):
children = ChildrenDao(database)
rules = ScheduleRulesDao(database)
definitions = TaskDefinitionsDao(database)
instances = TaskInstancesDao(database)
definitions = QuestDefinitionsDao(database)
instances = QuestInstancesDao(database)
events = CompletionEventsDao(database)
child = await children.create("Ada", _now_stamp())
rule = await rules.create("daily", D1)
tests/test_dao_matrix.py
+13 -13
View File
@@ -27,7 +27,7 @@ from custom_components.nestquest.dao_children import (
)
from custom_components.nestquest.dao_instances import (
CompletionEventsDao,
TaskInstancesDao,
QuestInstancesDao,
)
from custom_components.nestquest.dao_presence import (
PresenceOverridesDao,
@@ -35,7 +35,7 @@ from custom_components.nestquest.dao_presence import (
)
from custom_components.nestquest.dao_rules import (
ScheduleRulesDao,
TaskDefinitionsDao,
QuestDefinitionsDao,
)
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
@@ -70,10 +70,10 @@ ALL_TABLES = (
"children",
"admin_users",
"schedule_rules",
"task_definitions",
"quest_definitions",
"presence_schedules",
"presence_overrides",
"task_instances",
"quest_instances",
"completion_events",
)
@@ -86,10 +86,10 @@ class World:
self.children = ChildrenDao(database)
self.admins = AdminUsersDao(database)
self.rules = ScheduleRulesDao(database)
self.definitions = TaskDefinitionsDao(database)
self.definitions = QuestDefinitionsDao(database)
self.schedules = PresenceSchedulesDao(database)
self.overrides = PresenceOverridesDao(database)
self.instances = TaskInstancesDao(database)
self.instances = QuestInstancesDao(database)
self.events = CompletionEventsDao(database)
async def seed(self) -> None:
@@ -201,7 +201,7 @@ def test_matrix_schedule_rules_crud_and_constraints(tmp_path) -> None:
assert _with_world(tmp_path / "m-rules.db")(_body) is True
def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None:
def test_matrix_quest_definitions_crud_and_constraints(tmp_path) -> None:
async def _body(w: World):
# insert (seeded) + read
fetched = await w.definitions.get(w.definition.id)
@@ -271,7 +271,7 @@ def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
assert _with_world(tmp_path / "m-overrides.db")(_body) is True
def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
def test_matrix_quest_instances_crud_and_constraints(tmp_path) -> None:
async def _body(w: World):
# insert (seeded) + read
fetched = await w.instances.get(w.definition.id, D1)
@@ -292,20 +292,20 @@ def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
assert _with_world(tmp_path / "m-instances.db")(_body) is True
def test_matrix_task_instances_constraints(tmp_path) -> None:
def test_matrix_quest_instances_constraints(tmp_path) -> None:
async def _body(w: World):
# D20 has no instance yet: a raw insert succeeds, then an
# identical second insert must fail on the UNIQUE constraint —
# proving the schema backs the DAO's idempotency.
await w.database.execute(
"INSERT INTO task_instances (definition_id, child_id, "
"INSERT INTO quest_instances (definition_id, child_id, "
"due_date, due_time, generated_at) "
"VALUES (?, ?, ?, NULL, ?)",
(w.definition.id, w.child.id, D20, _stamp()),
)
with pytest.raises(sqlite3.IntegrityError):
await w.database.execute(
"INSERT INTO task_instances (definition_id, child_id, "
"INSERT INTO quest_instances (definition_id, child_id, "
"due_date, due_time, generated_at) "
"VALUES (?, ?, ?, NULL, ?)",
(w.definition.id, w.child.id, D20, _stamp()),
@@ -320,7 +320,7 @@ def test_matrix_task_instances_constraints(tmp_path) -> None:
assert _with_world(tmp_path / "m-instances-constraints.db")(_body) is True
def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None:
def test_matrix_quest_instances_upsert_idempotency(tmp_path) -> None:
async def _body(w: World):
first = await w.instances.upsert(
w.definition.id, w.child.id, D2, _stamp()
@@ -330,7 +330,7 @@ def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None:
)
assert second.id == first.id
count = await w.database.fetch_one(
"SELECT COUNT(*) FROM task_instances"
"SELECT COUNT(*) FROM quest_instances"
)
# Seed created one (D1) plus this one (D2) — no duplicate.
assert count == (2,)
tests/test_dao_rules.py
+25 -25
View File
@@ -1,4 +1,4 @@
"""Tests for dao_rules.py: typed DAO for schedule_rules and task_definitions."""
"""Tests for dao_rules.py: typed DAO for schedule_rules and quest_definitions."""
from __future__ import annotations
import asyncio
@@ -10,8 +10,8 @@ from custom_components.nestquest.dao_children import ChildrenDao
from custom_components.nestquest.dao_rules import (
ScheduleRuleRecord,
ScheduleRulesDao,
TaskDefinitionRecord,
TaskDefinitionsDao,
QuestDefinitionRecord,
QuestDefinitionsDao,
)
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import apply_migrations
@@ -37,7 +37,7 @@ async def _prepare(path) -> tuple:
await database.open(path)
await apply_migrations(database)
rules = ScheduleRulesDao(database)
definitions = TaskDefinitionsDao(database)
definitions = QuestDefinitionsDao(database)
children = ChildrenDao(database)
child = await children.create("Ada", NOW)
return database, rules, definitions, children, child
@@ -322,7 +322,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
await apply_migrations(database)
children = ChildrenDao(database)
rules = ScheduleRulesDao(database)
definitions = TaskDefinitionsDao(database)
definitions = QuestDefinitionsDao(database)
child = await children.create("Ada", NOW)
rule_a = await rules.create("daily", "2026-09-14")
@@ -335,7 +335,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
import custom_components.nestquest.dao_rules as dao_rules
original_validate = (
dao_rules.TaskDefinitionsDao._validate_assignable
dao_rules.QuestDefinitionsDao._validate_assignable
)
release = asyncio.Event()
started = asyncio.Event()
@@ -353,7 +353,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
started.set()
await release.wait()
dao_rules.TaskDefinitionsDao._validate_assignable = (
dao_rules.QuestDefinitionsDao._validate_assignable = (
_pausing_validate
)
try:
@@ -373,7 +373,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
delete_task, create_task, return_exceptions=True
)
finally:
dao_rules.TaskDefinitionsDao._validate_assignable = (
dao_rules.QuestDefinitionsDao._validate_assignable = (
original_validate
)
else:
@@ -405,7 +405,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
assert rule_row is None
else:
# Create won: the delete was rejected, both rows exist.
assert isinstance(creation, TaskDefinitionRecord), (
assert isinstance(creation, QuestDefinitionRecord), (
f"create must succeed, got {creation!r}"
)
assert rule_row is not None
@@ -425,7 +425,7 @@ def test_rule_delete_concurrent_with_definition_create_is_safe(
# ---------------------------------------------------------------------------
# task_definitions: create / get
# quest_definitions: create / get
# ---------------------------------------------------------------------------
@@ -441,7 +441,7 @@ def test_definition_create_returns_typed_record(tmp_path) -> None:
icon="mdi:tooth",
due_time="08:00",
)
assert isinstance(definition, TaskDefinitionRecord)
assert isinstance(definition, QuestDefinitionRecord)
assert definition.title == "Brush teeth"
assert definition.child_id == child.id
assert definition.schedule_rule_id == rule.id
@@ -497,7 +497,7 @@ def test_definition_create_unknown_rule_raises_value_error(
# ---------------------------------------------------------------------------
# task_definitions: list by child / list active
# quest_definitions: list by child / list active
# ---------------------------------------------------------------------------
@@ -539,7 +539,7 @@ def test_definition_list_active_excludes_inactive(tmp_path) -> None:
# ---------------------------------------------------------------------------
# task_definitions: update / set_active / set_assignee
# quest_definitions: update / set_active / set_assignee
# ---------------------------------------------------------------------------
@@ -673,7 +673,7 @@ def test_definition_no_delete_method_exists(tmp_path) -> None:
methods = {
name
for name, _ in inspect.getmembers(
TaskDefinitionsDao, inspect.isfunction
QuestDefinitionsDao, inspect.isfunction
)
}
assert "delete" not in methods
@@ -748,13 +748,13 @@ def test_definition_set_assignee_leaves_history_rows_alone(
# Simulate an already-generated instance (materialization owns
# this table, but its row shape proves reassignment isolation).
await database.execute(
"INSERT INTO task_instances (definition_id, child_id, "
"INSERT INTO quest_instances (definition_id, child_id, "
"due_date, generated_at) VALUES (?, ?, ?, ?)",
(definition.id, child.id, "2026-09-15", NOW),
)
await definitions.set_assignee(definition.id, other.id)
row = await database.fetch_one(
"SELECT child_id FROM task_instances WHERE definition_id = ?",
"SELECT child_id FROM quest_instances WHERE definition_id = ?",
(definition.id,),
)
assert row == (child.id,)
@@ -769,7 +769,7 @@ def test_definition_set_assignee_leaves_history_rows_alone(
def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
"""Guardrail: schedule_rules/task_definitions SQL may appear only
"""Guardrail: schedule_rules/quest_definitions SQL may appear only
in the DAO modules (and the schema DDL declarations). RECURSIVE
scan of package and tests, FROM/INTO/UPDATE/DELETE FROM/JOIN
pattern, with exact repo-relative exclusions justified by role:
@@ -801,7 +801,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|task_definitions)\b",
r"(schedule_rules|quest_definitions)\b",
re.IGNORECASE,
)
offenders: list[str] = []
@@ -813,16 +813,16 @@ def test_rules_and_definitions_sql_lives_only_in_dao_module() -> None:
if sql_pattern.search(py.read_text()):
offenders.append(relative)
assert offenders == [], (
f"SQL touching schedule_rules/task_definitions leaked into: "
f"SQL touching schedule_rules/quest_definitions leaked into: "
f"{offenders}"
)
def test_dao_instances_test_file_uses_dao_not_raw_rules_sql() -> None:
"""Compensating self-scan for the test_dao_instances.py exemption:
that file may query task_instances/completion_events raw (its own
that file may query quest_instances/completion_events raw (its own
guard's scope) but must go through the DAO for schedule_rules and
task_definitions, except its single sanctioned definition-create
quest_definitions, except its single sanctioned definition-create
helper usage and guard spans.
"""
import ast
@@ -853,12 +853,12 @@ 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|task_definitions)\b",
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions)\b",
re.IGNORECASE,
)
assert sql_pattern.search(remaining) is None, (
"test_dao_instances.py must go through the DAO, not raw SQL, "
"for schedule_rules/task_definitions"
"for schedule_rules/quest_definitions"
)
@@ -866,7 +866,7 @@ def test_dao_rules_test_file_uses_dao_not_raw_table_sql() -> None:
"""This test module must exercise the DAO, not raw SQL, for these
tables. Exceptions: the guard functions' own spans (their regexes
mention the names), and the reassignment-isolation test, which
legitimately touches task_instances — a table this DAO does not
legitimately touches quest_instances — a table this DAO does not
own — to prove the guardrail that reassignment leaves it alone.
"""
import ast
@@ -898,7 +898,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|task_definitions)\b",
r"FROM|JOIN)\s+[`'\"]*(\[)?(schedule_rules|quest_definitions)\b",
re.IGNORECASE,
)
assert sql_pattern.search(remaining) is None, (
tests/test_lifecycle.py
+2 -2
View File
@@ -247,12 +247,12 @@ async def test_setup_applies_all_v1_tables(hass, make_entry) -> None:
rows = await database.fetch_all(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name IN ('children', 'admin_users', 'schedule_rules', "
"'task_definitions') ORDER BY name"
"'quest_definitions') ORDER BY name"
)
assert [row[0] for row in rows] == [
"admin_users",
"children",
"quest_definitions",
"schedule_rules",
"task_definitions",
]
await async_unload_entry(hass, entry)
tests/test_migrations.py
+283 -32
View File
@@ -62,10 +62,10 @@ def _counts(database) -> dict[str, int]:
"children",
"admin_users",
"schedule_rules",
"task_definitions",
"quest_definitions",
"presence_schedules",
"presence_overrides",
"task_instances",
"quest_instances",
"completion_events",
)
counts = {}
@@ -76,16 +76,16 @@ def _counts(database) -> dict[str, int]:
# ---------------------------------------------------------------------------
# Fresh file: version 0 -> version 1
# Fresh file: version 0 -> latest version
# ---------------------------------------------------------------------------
def test_fresh_file_migrates_to_version_1(tmp_path) -> None:
def test_fresh_file_migrates_to_latest_version(tmp_path) -> None:
database = _open_db(tmp_path / "fresh.db")
try:
final = _migrate(database)
assert final == 1
assert _version(database) == 1
assert final == len(MIGRATIONS)
assert _version(database) == len(MIGRATIONS)
finally:
_run(database.close())
@@ -98,10 +98,10 @@ def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None:
"children",
"admin_users",
"schedule_rules",
"task_definitions",
"quest_definitions",
"presence_schedules",
"presence_overrides",
"task_instances",
"quest_instances",
"completion_events",
VERSION_TABLE,
}
@@ -135,9 +135,249 @@ def test_empty_version_table_reads_as_version_0(tmp_path) -> None:
_run(database.close())
def test_migration_1_is_the_full_v1_ddl() -> None:
def test_migration_list_shape() -> None:
"""Migration 1 is the v1 DDL; migration 2 is the legacy rename step.
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.
"""
assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS
assert len(MIGRATIONS) == 1
assert callable(MIGRATIONS[1])
assert len(MIGRATIONS) == 2
# ---------------------------------------------------------------------------
# Legacy dev database: task_* tables stamped version 1 -> quest_*
# ---------------------------------------------------------------------------
_LEGACY_DDL = [
"""
CREATE TABLE IF NOT EXISTS children (
id INTEGER PRIMARY KEY AUTOINCREMENT,
display_name TEXT NOT NULL,
colour TEXT,
avatar_ref TEXT,
sort_order INTEGER NOT NULL DEFAULT 0,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS admin_users (
ha_user_id TEXT PRIMARY KEY NOT NULL,
added_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS schedule_rules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
rule_type TEXT NOT NULL,
interval INTEGER NOT NULL DEFAULT 1,
weekday_set TEXT,
day_of_month INTEGER,
nth_weekday INTEGER,
month INTEGER,
start_date TEXT NOT NULL,
end_date TEXT
)
""",
"""
CREATE TABLE IF NOT EXISTS task_definitions (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT NOT NULL,
description TEXT,
icon TEXT,
child_id INTEGER NOT NULL REFERENCES children(id),
schedule_rule_id INTEGER NOT NULL REFERENCES schedule_rules(id),
due_time TEXT,
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS presence_schedules (
id INTEGER PRIMARY KEY AUTOINCREMENT,
child_id INTEGER NOT NULL UNIQUE REFERENCES children(id),
cycle_length_weeks INTEGER NOT NULL,
anchor_date TEXT NOT NULL,
pattern TEXT NOT NULL
)
""",
"""
CREATE TABLE IF NOT EXISTS presence_overrides (
id INTEGER PRIMARY KEY AUTOINCREMENT,
child_id INTEGER NOT NULL REFERENCES children(id),
start_date TEXT NOT NULL,
end_date TEXT NOT NULL,
is_present INTEGER NOT NULL,
note TEXT
)
""",
"""
CREATE TABLE IF NOT EXISTS task_instances (
id INTEGER PRIMARY KEY AUTOINCREMENT,
definition_id INTEGER NOT NULL REFERENCES task_definitions(id),
child_id INTEGER NOT NULL REFERENCES children(id),
due_date TEXT NOT NULL,
due_time TEXT,
generated_at TEXT NOT NULL,
UNIQUE (definition_id, due_date)
)
""",
"""
CREATE TABLE IF NOT EXISTS completion_events (
id INTEGER PRIMARY KEY AUTOINCREMENT,
instance_id INTEGER NOT NULL REFERENCES task_instances(id),
child_id INTEGER NOT NULL REFERENCES children(id),
event_type TEXT NOT NULL,
actor_source TEXT NOT NULL,
actor_user_id TEXT,
occurred_at TEXT NOT NULL,
was_on_time INTEGER
)
""",
]
def _seed_legacy_database(database) -> None:
"""Create a database indistinguishable from a pre-rename dev file.
Applies the legacy task_* DDL, seeds one row in every renamed
table, and stamps version 1 exactly the way the old runner did.
"""
for sql in _LEGACY_DDL:
_run(database.execute(sql))
_run(database.execute(VERSION_TABLE_DDL))
_run(
database.execute(
f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, 1)"
)
)
_run(
database.execute(
"INSERT INTO children (display_name, created_at) "
"VALUES ('Ada', '2026-09-14T00:00:00+00:00')"
)
)
_run(
database.execute(
"INSERT INTO schedule_rules (rule_type, start_date) "
"VALUES ('daily', '2026-09-14')"
)
)
_run(
database.execute(
"INSERT INTO task_definitions (title, child_id, "
"schedule_rule_id, created_at) "
"VALUES ('Dishes', 1, 1, '2026-09-14T00:00:00+00:00')"
)
)
_run(
database.execute(
"INSERT INTO task_instances (definition_id, child_id, "
"due_date, generated_at) "
"VALUES (1, 1, '2026-09-15', '2026-09-14T00:00:00+00:00')"
)
)
def test_legacy_task_tables_are_renamed_and_keep_their_rows(tmp_path) -> None:
database = _open_db(tmp_path / "legacy.db")
try:
_seed_legacy_database(database)
final = _migrate(database)
assert final == len(MIGRATIONS)
assert _version(database) == len(MIGRATIONS)
tables = _tables(database)
assert "task_definitions" not in tables
assert "task_instances" not in tables
assert {"quest_definitions", "quest_instances"} <= tables
# Every seeded row survived the rename.
assert _run(database.fetch_one("SELECT COUNT(*) FROM children")) == (1,)
assert (
_run(
database.fetch_one(
"SELECT title FROM quest_definitions WHERE id = 1"
)
)
== ("Dishes",)
)
assert (
_run(
database.fetch_one(
"SELECT due_date FROM quest_instances WHERE id = 1"
)
)
== ("2026-09-15",)
)
# The untouched completion_events table still exists and is empty.
assert "completion_events" in tables
assert (
_run(database.fetch_one("SELECT COUNT(*) FROM completion_events"))
== (0,)
)
finally:
_run(database.close())
def test_renamed_instances_foreign_key_follows_the_rename(tmp_path) -> None:
database = _open_db(tmp_path / "legacy-fk.db")
try:
_seed_legacy_database(database)
_migrate(database)
fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_instances)"))
referenced = {row[2] for row in fks}
assert "quest_definitions" in referenced
assert "task_definitions" not in referenced
# The renamed table is live: a new row validates its FKs.
_run(
database.execute(
"INSERT INTO children (display_name, created_at) "
"VALUES ('Ben', '2026-09-14T00:00:00+00:00')"
)
)
with pytest.raises(sqlite3.IntegrityError):
_run(
database.execute(
"INSERT INTO quest_instances (definition_id, child_id, "
"due_date, generated_at) "
"VALUES (999, 2, '2026-09-16', "
"'2026-09-14T00:00:00+00:00')"
)
)
finally:
_run(database.close())
def test_legacy_rename_is_idempotent_across_restarts(tmp_path) -> None:
database = _open_db(tmp_path / "legacy-restart.db")
try:
_seed_legacy_database(database)
_migrate(database)
final = _migrate(database)
assert final == len(MIGRATIONS)
tables = _tables(database)
assert "quest_definitions" in tables
assert "task_definitions" not in tables
assert (
_run(database.fetch_one("SELECT COUNT(*) FROM quest_definitions"))
== (1,)
)
finally:
_run(database.close())
def test_fresh_database_never_holds_legacy_table_names(tmp_path) -> None:
database = _open_db(tmp_path / "fresh-legacy.db")
try:
_migrate(database)
tables = _tables(database)
assert not any(name.startswith("task_") for name in tables)
assert _version(database) == len(MIGRATIONS)
finally:
_run(database.close())
# ---------------------------------------------------------------------------
@@ -162,8 +402,8 @@ def test_second_run_applies_nothing_and_keeps_rows(tmp_path) -> None:
)
)
final = _migrate(database)
assert final == 1
assert _version(database) == 1
assert final == len(MIGRATIONS)
assert _version(database) == len(MIGRATIONS)
assert _counts(database)["children"] == 1
assert _counts(database)["admin_users"] == 1
finally:
@@ -228,7 +468,7 @@ def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None:
with caplog.at_level(logging.DEBUG, logger="custom_components.nestquest.migrations"):
_migrate(database)
assert any(
"already at version 1" in r.getMessage()
f"already at version {len(MIGRATIONS)}" in r.getMessage()
for r in caplog.records
)
finally:
@@ -241,9 +481,17 @@ def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None:
def _failing_migrations(prior_statements: list[str]) -> list[list[str]]:
"""Two-migration list where migration 2 fails mid-application."""
"""Three-migration list where migration 3 fails mid-application.
Migration 2 is a harmless no-op filler so the real runner — which
has already stamped version 2 by the time these tests re-run it —
still has a pending migration to fail on.
"""
return [
prior_statements,
[
"CREATE TABLE IF NOT EXISTS filler_table (id INTEGER PRIMARY KEY)"
],
[
"CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)",
"CREATE TABLE broken (",
@@ -254,20 +502,20 @@ def _failing_migrations(prior_statements: list[str]) -> list[list[str]]:
def test_failing_migration_rolls_back_to_prior_version(tmp_path) -> None:
database = _open_db(tmp_path / "rollback.db")
try:
# Manually simulate a database at version 1: apply migration 1's
# statements directly and stamp version 1 through the runner.
# Manually simulate a fully-migrated database: run the real
# migration list, which stamps the latest version.
_migrate(database)
version_row = _run(
database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
)
assert version_row == (1,)
assert version_row == (len(MIGRATIONS),)
with pytest.raises(sqlite3.OperationalError):
_migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS))
# Version stamp unchanged, statements from the failed migration
# absent, and the v1 tables untouched.
assert _version(database) == 1
assert _version(database) == len(MIGRATIONS)
assert "later_table" not in _tables(database)
assert _counts(database)["children"] == 0
assert _counts(database)["completion_events"] == 0
@@ -275,21 +523,24 @@ def test_failing_migration_rolls_back_to_prior_version(tmp_path) -> None:
_run(database.close())
def test_failing_migration_on_fresh_file_leaves_version_0(tmp_path) -> None:
def test_failing_migration_on_fresh_file_leaves_last_good_version(
tmp_path,
) -> None:
database = _open_db(tmp_path / "fresh-rollback.db")
try:
# The fresh file still reaches migration 2 (the failing one):
# migration 1 commits, stamping version 1, before migration 2
# fails and rolls back. The run aborts at version 1 — the last
# fully-applied migration — not at version 0.
# The fresh file reaches the failing migration (the last one):
# earlier migrations commit and stamp before it fails and rolls
# back. The run aborts at the last fully-applied migration —
# not at version 0.
broken = _failing_migrations(SCHEMA_V1_STATEMENTS)
with pytest.raises(sqlite3.OperationalError):
_migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS))
assert _version(database) == 1
# Migration 2's tables must not survive the failed attempt.
_migrate(database, broken)
assert _version(database) == len(broken) - 1
# The failing migration's tables must not survive the attempt.
assert "later_table" not in _tables(database)
# Migration 1's tables do: it committed before the failure.
assert "children" in _tables(database)
assert "task_instances" in _tables(database)
assert "quest_instances" in _tables(database)
finally:
_run(database.close())
@@ -302,12 +553,12 @@ def test_failed_migration_can_be_retried_after_fix(tmp_path) -> None:
with pytest.raises(sqlite3.OperationalError):
_migrate(database, broken)
# "Fix" the migration and retry: it now applies cleanly.
broken[1] = [
broken[-1] = [
"CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)"
]
final = _migrate(database, broken)
assert final == 2
assert _version(database) == 2
assert final == len(broken)
assert _version(database) == len(broken)
assert "later_table" in _tables(database)
finally:
_run(database.close())
@@ -575,7 +826,7 @@ async def test_setup_entry_migrates_fresh_database(hass, make_entry) -> None:
entry = await _setup_entry(hass, make_entry(), hass.registry)
database = entry.runtime_data.database
assert await read_schema_version(database) == 1
assert await read_schema_version(database) == len(MIGRATIONS)
# The v1 tables exist because the runner applied migration 1.
names = {
row[0]
@@ -626,7 +877,7 @@ async def test_options_reload_reopens_and_rechecks_version(
database_b = entry_reloaded.runtime_data.database
assert database_b is not database_a
try:
assert await read_schema_version(database_b) == 1
assert await read_schema_version(database_b) == len(MIGRATIONS)
row = await database_b.fetch_one("SELECT COUNT(*) FROM children")
assert row == (1,)
finally:
tests/test_schema.py
+56 -56
View File
@@ -15,8 +15,8 @@ from custom_components.nestquest.schema import (
SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
SCHEMA_V1_SCHEDULE_RULES_DDL,
SCHEMA_V1_STATEMENTS,
SCHEMA_V1_TASK_DEFINITIONS_DDL,
SCHEMA_V1_TASK_INSTANCES_DDL,
SCHEMA_V1_QUEST_DEFINITIONS_DDL,
SCHEMA_V1_QUEST_INSTANCES_DDL,
async_apply_ddl,
)
@@ -91,7 +91,7 @@ def _insert_definition(database, child_id=1, rule_id=1, **overrides):
placeholders = ", ".join("?" for _ in values)
result = _run(
database.execute(
f"INSERT INTO task_definitions ({columns}) VALUES ({placeholders})",
f"INSERT INTO quest_definitions ({columns}) VALUES ({placeholders})",
tuple(values.values()),
)
)
@@ -161,10 +161,10 @@ def test_schema_v1_statements_compose_the_six_tables() -> None:
*SCHEMA_V1_CHILDREN_DDL,
*SCHEMA_V1_ADMIN_USERS_DDL,
*SCHEMA_V1_SCHEDULE_RULES_DDL,
*SCHEMA_V1_TASK_DEFINITIONS_DDL,
*SCHEMA_V1_QUEST_DEFINITIONS_DDL,
*SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
*SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
*SCHEMA_V1_TASK_INSTANCES_DDL,
*SCHEMA_V1_QUEST_INSTANCES_DDL,
*SCHEMA_V1_COMPLETION_EVENTS_DDL,
]
@@ -256,17 +256,17 @@ def test_schedule_rules_table_exists_after_applying_ddl(tmp_path) -> None:
_run(database.close())
def test_task_definitions_table_exists_after_applying_ddl(tmp_path) -> None:
def test_quest_definitions_table_exists_after_applying_ddl(tmp_path) -> None:
database = _open_db(tmp_path / "definitions.db")
try:
_apply(database)
row = _run(
database.fetch_one(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name = 'task_definitions'"
"AND name = 'quest_definitions'"
)
)
assert row is not None and row[0] == "task_definitions"
assert row is not None and row[0] == "quest_definitions"
finally:
_run(database.close())
@@ -292,11 +292,11 @@ def test_schedule_rules_columns_types_and_constraints(tmp_path) -> None:
_run(database.close())
def test_task_definitions_columns_types_and_constraints(tmp_path) -> None:
def test_quest_definitions_columns_types_and_constraints(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-columns.db")
try:
_apply(database)
columns = _run(database.fetch_all("PRAGMA table_info(task_definitions)"))
columns = _run(database.fetch_all("PRAGMA table_info(quest_definitions)"))
assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
("id", "INTEGER", 0, None, 1),
("title", "TEXT", 1, None, 0),
@@ -312,11 +312,11 @@ def test_task_definitions_columns_types_and_constraints(tmp_path) -> None:
_run(database.close())
def test_task_definitions_foreign_keys_declared(tmp_path) -> None:
def test_quest_definitions_foreign_keys_declared(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-fks.db")
try:
_apply(database)
fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_definitions)"))
fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_definitions)"))
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {
("children", "child_id", "id"),
@@ -519,7 +519,7 @@ def test_failed_statement_rolls_back_whole_application(tmp_path) -> None:
# ---------------------------------------------------------------------------
# schedule_rules and task_definitions constraint enforcement
# schedule_rules and quest_definitions constraint enforcement
# ---------------------------------------------------------------------------
@@ -813,7 +813,7 @@ def test_schedule_rules_start_date_not_null_enforced(tmp_path) -> None:
_run(database.close())
def test_task_definitions_insert_requires_existing_child_and_rule(
def test_quest_definitions_insert_requires_existing_child_and_rule(
tmp_path,
) -> None:
database = _open_db(tmp_path / "definitions-fk-unknown.db")
@@ -831,7 +831,7 @@ def test_task_definitions_insert_requires_existing_child_and_rule(
)
row = _run(
database.fetch_one(
"SELECT title, child_id, schedule_rule_id FROM task_definitions "
"SELECT title, child_id, schedule_rule_id FROM quest_definitions "
"WHERE id = ?",
(definition_id,),
)
@@ -841,7 +841,7 @@ def test_task_definitions_insert_requires_existing_child_and_rule(
_run(database.close())
def test_task_definitions_unknown_child_id_fails(tmp_path) -> None:
def test_quest_definitions_unknown_child_id_fails(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-bad-child.db")
try:
_apply(database)
@@ -852,7 +852,7 @@ def test_task_definitions_unknown_child_id_fails(tmp_path) -> None:
_run(database.close())
def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
def test_quest_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-bad-rule.db")
try:
_apply(database)
@@ -868,7 +868,7 @@ def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None:
_run(database.close())
def test_task_definitions_is_active_check_enforced(tmp_path) -> None:
def test_quest_definitions_is_active_check_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-is-active.db")
try:
_apply(database)
@@ -880,7 +880,7 @@ def test_task_definitions_is_active_check_enforced(tmp_path) -> None:
@pytest.mark.parametrize("bad_is_active", ["abc", 1.5])
def test_task_definitions_is_active_non_integer_fails(
def test_quest_definitions_is_active_non_integer_fails(
tmp_path, bad_is_active
) -> None:
database = _open_db(tmp_path / "definitions-is-active-typeof.db")
@@ -901,7 +901,7 @@ def test_task_definitions_is_active_non_integer_fails(
_run(database.close())
def test_task_definitions_title_not_null_enforced(tmp_path) -> None:
def test_quest_definitions_title_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-title.db")
try:
_apply(database)
@@ -912,7 +912,7 @@ def test_task_definitions_title_not_null_enforced(tmp_path) -> None:
_run(database.close())
def test_task_definitions_created_at_not_null_enforced(tmp_path) -> None:
def test_quest_definitions_created_at_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-created-at.db")
try:
_apply(database)
@@ -925,7 +925,7 @@ def test_task_definitions_created_at_not_null_enforced(tmp_path) -> None:
_run(database.close())
def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None:
def test_quest_definitions_defaults_apply_on_insert(tmp_path) -> None:
database = _open_db(tmp_path / "definitions-defaults.db")
try:
_apply(database)
@@ -941,7 +941,7 @@ def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None:
)
row = _run(
database.fetch_one(
"SELECT is_active, due_time FROM task_definitions WHERE id = ?",
"SELECT is_active, due_time FROM quest_definitions WHERE id = ?",
(definition_id,),
)
)
@@ -969,7 +969,7 @@ def test_applying_extended_ddl_twice_keeps_new_table_rows(tmp_path) -> None:
_insert_definition(database, child_id=1, rule_id=1)
_apply(database)
counts = []
for table in ("children", "schedule_rules", "task_definitions"):
for table in ("children", "schedule_rules", "quest_definitions"):
row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
counts.append(row[0])
assert counts == [1, 1, 1]
@@ -1406,7 +1406,7 @@ def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
# ---------------------------------------------------------------------------
# task_instances and completion_events: helpers
# quest_instances and completion_events: helpers
# ---------------------------------------------------------------------------
@@ -1428,7 +1428,7 @@ def _insert_instance(database, definition_id=1, **overrides):
placeholders = ", ".join("?" for _ in values)
result = _run(
database.execute(
f"INSERT INTO task_instances ({columns}) VALUES ({placeholders})",
f"INSERT INTO quest_instances ({columns}) VALUES ({placeholders})",
tuple(values.values()),
)
)
@@ -1466,28 +1466,28 @@ def _insert_event(database, instance_id=1, **overrides):
def _setup_definition(database) -> None:
"""Create one child, one schedule rule and one task definition."""
"""Create one child, one schedule rule and one quest definition."""
_child(database)
_insert_rule(database)
_insert_definition(database, child_id=1, rule_id=1)
# ---------------------------------------------------------------------------
# task_instances and completion_events: DDL shape
# quest_instances and completion_events: DDL shape
# ---------------------------------------------------------------------------
def test_task_instances_table_exists_after_applying_ddl(tmp_path) -> None:
def test_quest_instances_table_exists_after_applying_ddl(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances.db")
try:
_apply(database)
row = _run(
database.fetch_one(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name = 'task_instances'"
"AND name = 'quest_instances'"
)
)
assert row is not None and row[0] == "task_instances"
assert row is not None and row[0] == "quest_instances"
finally:
_run(database.close())
@@ -1507,11 +1507,11 @@ def test_completion_events_table_exists_after_applying_ddl(tmp_path) -> None:
_run(database.close())
def test_task_instances_columns_types_and_constraints(tmp_path) -> None:
def test_quest_instances_columns_types_and_constraints(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-columns.db")
try:
_apply(database)
columns = _run(database.fetch_all("PRAGMA table_info(task_instances)"))
columns = _run(database.fetch_all("PRAGMA table_info(quest_instances)"))
# cid, name, type, notnull, dflt_value, pk
assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
("id", "INTEGER", 0, None, 1),
@@ -1544,14 +1544,14 @@ def test_completion_events_columns_types_and_constraints(tmp_path) -> None:
_run(database.close())
def test_task_instances_foreign_keys_declared(tmp_path) -> None:
def test_quest_instances_foreign_keys_declared(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-fks.db")
try:
_apply(database)
fks = _run(database.fetch_all("PRAGMA foreign_key_list(task_instances)"))
fks = _run(database.fetch_all("PRAGMA foreign_key_list(quest_instances)"))
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {
("task_definitions", "definition_id", "id"),
("quest_definitions", "definition_id", "id"),
("children", "child_id", "id"),
}
finally:
@@ -1565,14 +1565,14 @@ def test_completion_events_foreign_keys_declared(tmp_path) -> None:
fks = _run(database.fetch_all("PRAGMA foreign_key_list(completion_events)"))
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {
("task_instances", "instance_id", "id"),
("quest_instances", "instance_id", "id"),
("children", "child_id", "id"),
}
finally:
_run(database.close())
def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
def test_quest_instances_unique_index_on_definition_and_date(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-unique.db")
try:
_apply(database)
@@ -1580,7 +1580,7 @@ def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
rows = _run(
database.fetch_all(
"SELECT name, sql FROM sqlite_master WHERE type = 'index' "
"AND tbl_name = 'task_instances'"
"AND tbl_name = 'quest_instances'"
)
)
implicit = [(name, sql) for name, sql in rows if sql is None]
@@ -1599,7 +1599,7 @@ def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
row = _run(
database.fetch_one(
"SELECT COUNT(DISTINCT name) FROM pragma_index_list"
"('task_instances') WHERE \"unique\" = 1"
"('quest_instances') WHERE \"unique\" = 1"
)
)
assert row == (1,)
@@ -1608,11 +1608,11 @@ def test_task_instances_unique_index_on_definition_and_date(tmp_path) -> None:
# ---------------------------------------------------------------------------
# task_instances: constraint enforcement
# quest_instances: constraint enforcement
# ---------------------------------------------------------------------------
def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
def test_quest_instances_valid_insert_round_trips(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-valid.db")
try:
_apply(database)
@@ -1621,7 +1621,7 @@ def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
row = _run(
database.fetch_one(
"SELECT definition_id, child_id, due_date, due_time, generated_at "
"FROM task_instances WHERE id = ?",
"FROM quest_instances WHERE id = ?",
(instance_id,),
)
)
@@ -1630,7 +1630,7 @@ def test_task_instances_valid_insert_round_trips(tmp_path) -> None:
_run(database.close())
def test_task_instances_due_time_round_trips(tmp_path) -> None:
def test_quest_instances_due_time_round_trips(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-due-time.db")
try:
_apply(database)
@@ -1640,7 +1640,7 @@ def test_task_instances_due_time_round_trips(tmp_path) -> None:
)
row = _run(
database.fetch_one(
"SELECT due_time FROM task_instances WHERE id = ?",
"SELECT due_time FROM quest_instances WHERE id = ?",
(instance_id,),
)
)
@@ -1649,7 +1649,7 @@ def test_task_instances_due_time_round_trips(tmp_path) -> None:
_run(database.close())
def test_task_instances_duplicate_definition_date_fails(tmp_path) -> None:
def test_quest_instances_duplicate_definition_date_fails(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-duplicate.db")
try:
_apply(database)
@@ -1661,7 +1661,7 @@ def test_task_instances_duplicate_definition_date_fails(tmp_path) -> None:
_run(database.close())
def test_task_instances_same_date_different_definitions_allowed(
def test_quest_instances_same_date_different_definitions_allowed(
tmp_path,
) -> None:
database = _open_db(tmp_path / "task-instances-multi-def.db")
@@ -1678,26 +1678,26 @@ def test_task_instances_same_date_different_definitions_allowed(
_insert_definition(database, child_id=2, rule_id=rule_id)
_insert_instance(database, definition_id=1, due_date="2026-09-14")
_insert_instance(database, definition_id=2, due_date="2026-09-14")
row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
assert row == (2,)
finally:
_run(database.close())
def test_task_instances_same_definition_different_dates_allowed(tmp_path) -> None:
def test_quest_instances_same_definition_different_dates_allowed(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-multi-date.db")
try:
_apply(database)
_setup_definition(database)
_insert_instance(database, definition_id=1, due_date="2026-09-14")
_insert_instance(database, definition_id=1, due_date="2026-09-15")
row = _run(database.fetch_one("SELECT COUNT(*) FROM task_instances"))
row = _run(database.fetch_one("SELECT COUNT(*) FROM quest_instances"))
assert row == (2,)
finally:
_run(database.close())
def test_task_instances_unknown_definition_fails(tmp_path) -> None:
def test_quest_instances_unknown_definition_fails(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-bad-def.db")
try:
_apply(database)
@@ -1708,7 +1708,7 @@ def test_task_instances_unknown_definition_fails(tmp_path) -> None:
_run(database.close())
def test_task_instances_unknown_child_fails(tmp_path) -> None:
def test_quest_instances_unknown_child_fails(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-bad-child.db")
try:
_apply(database)
@@ -1719,7 +1719,7 @@ def test_task_instances_unknown_child_fails(tmp_path) -> None:
_run(database.close())
def test_task_instances_due_date_not_null_enforced(tmp_path) -> None:
def test_quest_instances_due_date_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-due-null.db")
try:
_apply(database)
@@ -1730,7 +1730,7 @@ def test_task_instances_due_date_not_null_enforced(tmp_path) -> None:
_run(database.close())
def test_task_instances_generated_at_not_null_enforced(tmp_path) -> None:
def test_quest_instances_generated_at_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "task-instances-generated-null.db")
try:
_apply(database)
@@ -2026,8 +2026,8 @@ def test_applying_full_ddl_twice_keeps_rows(tmp_path) -> None:
for table in (
"children",
"schedule_rules",
"task_definitions",
"task_instances",
"quest_definitions",
"quest_instances",
"completion_events",
):
row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
tests/test_startup_db.py
+6 -4
View File
@@ -18,6 +18,7 @@ from custom_components.nestquest import (
)
from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.migrations import (
MIGRATIONS,
VERSION_TABLE,
apply_migrations,
)
@@ -114,12 +115,13 @@ def test_missing_file_opens_fresh(tmp_path) -> None:
assert not db_path.exists()
database = _run(_async_open_database(_make_hass_mock(), db_path))
try:
# The file was created fresh and migrations stamped version 1.
# The file was created fresh and migrations stamped the
# latest schema version.
assert db_path.exists()
row = _run(
database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
)
assert row == (1,)
assert row == (len(MIGRATIONS),)
finally:
_run(database.close())
@@ -151,7 +153,7 @@ def test_valid_file_opens_and_migrations_noop(tmp_path) -> None:
row = _run(
database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
)
assert row == (1,)
assert row == (len(MIGRATIONS),)
finally:
_run(database.close())
# Opening a valid file must not rewrite it (the no-op path).
@@ -401,7 +403,7 @@ async def test_setup_with_missing_file_creates_and_loads(
row = await database.fetch_one(
f"SELECT version FROM {VERSION_TABLE}"
)
assert row == (1,)
assert row == (len(MIGRATIONS),)
finally:
await database.close()
assert await async_unload_entry(hass, entry) is True