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

feat: add schema v1 ddl for presence schedules and overrides

Browse Source
This commit is contained in:
Joshua King
2026-09-13 21:56:30 -04:00
parent 30748f786a
commit 2e22291402
2 changed files with 566 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files
custom_components/nestquest/schema.py
+81
View File
@@ -42,6 +42,49 @@ does NOT forbid ``month`` on weekly/monthly rules (an "every 2 weeks"
rule or a monthly rule could later be pinned to a month) nor extra
fields on weekly/monthly/yearly rules generally, so the CHECKs only
reject what the rule engine definitively cannot express now.
Presence pattern format: ``presence_schedules.pattern`` encodes the
whole N-week repeating cycle in one TEXT column as pipe-separated
segments, one per week of the cycle: segment ``i`` is a ``weekday_set``
CSV (same single-digit 0-6 convention as ``schedule_rules.weekday_set``)
for week ``i``, e.g. ``'0,2,4|1,3'`` = Mon/Wed/Fri in week 0, Tue/Thu in
week 1. An empty segment means the child is absent every day of that
week; a wholly empty pattern with ``cycle_length_weeks = 1`` means
absent every day of every week, which is distinct from having no
schedule row at all (a child with no schedule is present every day, per
the feature guardrails). Chosen over JSON or per-week rows to stay
consistent with the CSV ``weekday_set`` convention and fully checkable
by plain CHECKs. DB-level validation is pure CHECK, no triggers or
custom functions: char-class GLOBs reject every character outside
digits 0-6, comma and pipe; adjacency GLOBs reject empty elements
(double comma, comma next to a pipe, leading/trailing comma) and
two-digit runs; a further GLOB rejects contiguous runs of 8+ comma
separated digits, which caps a segment at 7 elements — the GLOB cannot
match across a ``|``, so cross-segment boundaries stay safe; and the
segment count is pinned to ``cycle_length_weeks`` by
``LENGTH(pattern) - LENGTH(REPLACE(pattern, '|', '')) + 1``. Per
weekday_set this is not full CSV-set normalisation (duplicates like
``'0,0'`` or unsorted sets stay storable, as with ``schedule_rules``);
the engine treats the segment as a set. A
one-row-per-``week_index`` normalization was considered and rejected:
the flat single-column encoding keeps the done-condition's table shape
and proved fully checkable with plain CHECKs.
``cycle_length_weeks`` is capped at 4: 2 (alternating weeks) is the
common case, and a small cap keeps anchor arithmetic and the encoded
pattern human-readable. Revisit the cap — and re-run the pattern
validation matrix in tests — if a longer cycle is ever needed.
``anchor_date`` is an ISO date string pinning the cycle: the week index
of a date is (days since anchor) // 7 mod ``cycle_length_weeks``,
computed by the presence engine (Feature 05) from anchor-date
arithmetic, never ISO week parity. Like other date columns it carries
no shape CHECK, matching the established date policy.
Presence overrides: ``presence_overrides.end_date`` is NOT NULL,
unlike ``schedule_rules.end_date`` — an override is always a concrete
date range; a single-day override stores the same date in both columns.
``is_present`` 0/1 marks the child absent/present for the whole range;
``note`` is optional free text.
"""
from __future__ import annotations
@@ -130,11 +173,49 @@ SCHEMA_V1_TASK_DEFINITIONS_DDL: list[str] = [
""",
]
SCHEMA_V1_PRESENCE_SCHEDULES_DDL: list[str] = [
f"""
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 CHECK (typeof(cycle_length_weeks) = 'integer' AND cycle_length_weeks >= 1 AND cycle_length_weeks <= 4),
anchor_date TEXT NOT NULL,
pattern TEXT NOT NULL CHECK (
NOT pattern GLOB '*[^0-6,|]*'
AND NOT pattern GLOB '*[0-6][0-6]*'
AND NOT pattern GLOB '*,,*'
AND NOT pattern GLOB '*,|*'
AND NOT pattern GLOB '*|,*'
AND NOT pattern GLOB ',*'
AND NOT pattern GLOB '*,'
AND NOT pattern GLOB '*[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6],[0-6]*'
AND (LENGTH(pattern) - LENGTH(REPLACE(pattern, '|', '')) + 1) = cycle_length_weeks
)
)
""",
]
SCHEMA_V1_PRESENCE_OVERRIDES_DDL: list[str] = [
"""
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 CHECK (typeof(is_present) = 'integer' AND is_present IN (0, 1)),
note TEXT,
CHECK (end_date >= start_date)
)
""",
]
SCHEMA_V1_STATEMENTS: list[str] = [
*SCHEMA_V1_CHILDREN_DDL,
*SCHEMA_V1_ADMIN_USERS_DDL,
*SCHEMA_V1_SCHEDULE_RULES_DDL,
*SCHEMA_V1_TASK_DEFINITIONS_DDL,
*SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
*SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
]
tests/test_schema.py
+485 -1
View File
@@ -10,6 +10,8 @@ from custom_components.nestquest.db import NestQuestDatabase
from custom_components.nestquest.schema import (
SCHEMA_V1_ADMIN_USERS_DDL,
SCHEMA_V1_CHILDREN_DDL,
SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
SCHEMA_V1_SCHEDULE_RULES_DDL,
SCHEMA_V1_STATEMENTS,
SCHEMA_V1_TASK_DEFINITIONS_DDL,
@@ -95,17 +97,71 @@ def _insert_definition(database, child_id=1, rule_id=1, **overrides):
return result.lastrowid
def _insert_schedule(database, child_id, **overrides):
"""Insert one valid presence_schedule, returning the schedule id.
Pass ``_OMIT`` as a value to leave that column out of the INSERT so
its DDL DEFAULT applies.
"""
values: dict[str, object] = {
"child_id": child_id,
"cycle_length_weeks": 2,
"anchor_date": "2026-09-13",
"pattern": "0,2,4|1,3",
}
values.update(overrides)
values = {k: v for k, v in values.items() if v is not _OMIT}
columns = ", ".join(values)
placeholders = ", ".join("?" for _ in values)
result = _run(
database.execute(
f"INSERT INTO presence_schedules ({columns}) VALUES ({placeholders})",
tuple(values.values()),
)
)
assert result.lastrowid is not None
return result.lastrowid
def _insert_override(database, child_id, **overrides):
"""Insert one valid presence_override, returning the override id.
Pass ``_OMIT`` as a value to leave that column out of the INSERT so
its DDL DEFAULT applies.
"""
values: dict[str, object] = {
"child_id": child_id,
"start_date": "2026-09-13",
"end_date": "2026-09-13",
"is_present": 1,
}
values.update(overrides)
values = {k: v for k, v in values.items() if v is not _OMIT}
columns = ", ".join(values)
placeholders = ", ".join("?" for _ in values)
result = _run(
database.execute(
f"INSERT INTO presence_overrides ({columns}) VALUES ({placeholders})",
tuple(values.values()),
)
)
assert result.lastrowid is not None
return result.lastrowid
# ---------------------------------------------------------------------------
# DDL constants
# ---------------------------------------------------------------------------
def test_schema_v1_statements_compose_the_four_tables() -> None:
def test_schema_v1_statements_compose_the_six_tables() -> None:
assert SCHEMA_V1_STATEMENTS == [
*SCHEMA_V1_CHILDREN_DDL,
*SCHEMA_V1_ADMIN_USERS_DDL,
*SCHEMA_V1_SCHEDULE_RULES_DDL,
*SCHEMA_V1_TASK_DEFINITIONS_DDL,
*SCHEMA_V1_PRESENCE_SCHEDULES_DDL,
*SCHEMA_V1_PRESENCE_OVERRIDES_DDL,
]
@@ -913,5 +969,433 @@ def test_applying_extended_ddl_twice_keeps_new_table_rows(tmp_path) -> None:
row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
counts.append(row[0])
assert counts == [1, 1, 1]
finally:
_run(database.close())
# ---------------------------------------------------------------------------
# presence_schedules and presence_overrides constraint enforcement
# ---------------------------------------------------------------------------
def _child(database, name="Ada") -> int:
_run(
database.execute(
"INSERT INTO children (display_name, created_at) VALUES (?, ?)",
(name, "2026-09-13T00:00:00+00:00"),
)
)
return 1
def test_presence_schedules_table_exists_after_applying_ddl(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules.db")
try:
_apply(database)
row = _run(
database.fetch_one(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name = 'presence_schedules'"
)
)
assert row is not None and row[0] == "presence_schedules"
finally:
_run(database.close())
def test_presence_overrides_table_exists_after_applying_ddl(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides.db")
try:
_apply(database)
row = _run(
database.fetch_one(
"SELECT name FROM sqlite_master WHERE type = 'table' "
"AND name = 'presence_overrides'"
)
)
assert row is not None and row[0] == "presence_overrides"
finally:
_run(database.close())
def test_presence_schedules_columns_types_and_constraints(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-columns.db")
try:
_apply(database)
columns = _run(database.fetch_all("PRAGMA table_info(presence_schedules)"))
# 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),
("child_id", "INTEGER", 1, None, 0),
("cycle_length_weeks", "INTEGER", 1, None, 0),
("anchor_date", "TEXT", 1, None, 0),
("pattern", "TEXT", 1, None, 0),
]
finally:
_run(database.close())
def test_presence_overrides_columns_types_and_constraints(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-columns.db")
try:
_apply(database)
columns = _run(database.fetch_all("PRAGMA table_info(presence_overrides)"))
assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
("id", "INTEGER", 0, None, 1),
("child_id", "INTEGER", 1, None, 0),
("start_date", "TEXT", 1, None, 0),
("end_date", "TEXT", 1, None, 0),
("is_present", "INTEGER", 1, None, 0),
("note", "TEXT", 0, None, 0),
]
finally:
_run(database.close())
def test_presence_schedules_foreign_key_declared(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-fks.db")
try:
_apply(database)
fks = _run(database.fetch_all("PRAGMA foreign_key_list(presence_schedules)"))
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {("children", "child_id", "id")}
finally:
_run(database.close())
def test_presence_overrides_foreign_key_declared(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-fks.db")
try:
_apply(database)
fks = _run(database.fetch_all("PRAGMA foreign_key_list(presence_overrides)"))
declared = {(row[2], row[3], row[4]) for row in fks}
assert declared == {("children", "child_id", "id")}
finally:
_run(database.close())
def test_presence_schedules_valid_insert_round_trips(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-valid.db")
try:
_apply(database)
_child(database)
schedule_id = _insert_schedule(database, child_id=1)
row = _run(
database.fetch_one(
"SELECT child_id, cycle_length_weeks, anchor_date, pattern "
"FROM presence_schedules WHERE id = ?",
(schedule_id,),
)
)
assert row == (1, 2, "2026-09-13", "0,2,4|1,3")
finally:
_run(database.close())
def test_presence_schedules_second_schedule_for_same_child_fails(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-unique.db")
try:
_apply(database)
_child(database)
_insert_schedule(database, child_id=1)
with pytest.raises(sqlite3.IntegrityError, match="UNIQUE"):
_insert_schedule(database, child_id=1)
finally:
_run(database.close())
def test_presence_schedules_unknown_child_fails(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-bad-child.db")
try:
_apply(database)
with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
_insert_schedule(database, child_id=999)
finally:
_run(database.close())
@pytest.mark.parametrize("bad_cycle", [0, 5, "abc", 1.5])
def test_presence_schedules_cycle_length_out_of_range_or_type_fails(
tmp_path, bad_cycle
) -> None:
database = _open_db(tmp_path / f"cycle-{type(bad_cycle).__name__}.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_schedule(database, child_id=1, cycle_length_weeks=bad_cycle)
finally:
_run(database.close())
@pytest.mark.parametrize(
("cycle", "pattern"),
[
(1, ""),
(1, "0"),
(1, "0,2,4"),
(2, "0,2,4|1,3"),
(2, "|0"),
(2, "0|"),
(2, "0,1,2,3,4,5,6|"),
(3, "0||1"),
(4, "|||"),
(4, "0|1|2|3"),
(4, "0,1|0,1|0,1|0,1"),
],
)
def test_presence_schedules_valid_patterns_succeed(tmp_path, cycle, pattern) -> None:
database = _open_db(tmp_path / "presence-schedules-patterns-valid.db")
try:
_apply(database)
_child(database)
schedule_id = _insert_schedule(
database, child_id=1, cycle_length_weeks=cycle, pattern=pattern
)
row = _run(
database.fetch_one(
"SELECT pattern FROM presence_schedules WHERE id = ?",
(schedule_id,),
)
)
assert row == (pattern,)
finally:
_run(database.close())
@pytest.mark.parametrize(
("cycle", "pattern"),
[
(2, "0,2,4"), # too few segments
(1, "0|1"), # too many segments
(4, "0|1|2"), # too few segments
(2, "0,2,4||1,3"), # empty middle segment is only valid if empty everywhere
(2, "x|1"),
(2, "7|1"),
(2, "0,8|1"),
(2, "0,|1"),
(2, "0|,1"),
(2, "0|1,"),
(2, ",0|1"),
(2, "0,,1|2"),
(2, "00,1|2"),
(2, "0;1|2"),
(2, "0 1|2"),
(2, "0,2,4|1,7"),
(2, "0,2,4|10,3"),
(1, "0,1,2,3,4,5,6,0"), # more than 7 elements
],
)
def test_presence_schedules_invalid_patterns_fail(tmp_path, cycle, pattern) -> None:
database = _open_db(tmp_path / "presence-schedules-patterns-invalid.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_schedule(
database, child_id=1, cycle_length_weeks=cycle, pattern=pattern
)
finally:
_run(database.close())
def test_presence_schedules_cycle_length_bounds_pass(tmp_path) -> None:
database = _open_db(tmp_path / "presence-schedules-cycle-bounds.db")
try:
_apply(database)
_child(database)
_insert_schedule(database, child_id=1, cycle_length_weeks=1, pattern="0")
_run(
database.execute(
"INSERT INTO children (display_name, created_at) VALUES (?, ?)",
("Bo", "2026-09-13T00:00:00+00:00"),
)
)
_insert_schedule(database, child_id=2, cycle_length_weeks=4, pattern="|||")
finally:
_run(database.close())
def test_presence_overrides_valid_insert_round_trips(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-valid.db")
try:
_apply(database)
_child(database)
override_id = _insert_override(
database,
child_id=1,
note="Grandma's birthday",
)
row = _run(
database.fetch_one(
"SELECT child_id, start_date, end_date, is_present, note "
"FROM presence_overrides WHERE id = ?",
(override_id,),
)
)
assert row == (1, "2026-09-13", "2026-09-13", 1, "Grandma's birthday")
finally:
_run(database.close())
def test_presence_overrides_range_and_absent_round_trip(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-range.db")
try:
_apply(database)
_child(database)
override_id = _insert_override(
database,
child_id=1,
start_date="2026-12-21",
end_date="2027-01-04",
is_present=0,
)
row = _run(
database.fetch_one(
"SELECT start_date, end_date, is_present, note "
"FROM presence_overrides WHERE id = ?",
(override_id,),
)
)
assert row == ("2026-12-21", "2027-01-04", 0, None)
finally:
_run(database.close())
def test_presence_overrides_unknown_child_fails(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-bad-child.db")
try:
_apply(database)
with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"):
_insert_override(database, child_id=999)
finally:
_run(database.close())
def test_presence_overrides_is_present_check_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-is-present.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_override(database, child_id=1, is_present=2)
finally:
_run(database.close())
@pytest.mark.parametrize(
"bad_is_present",
["abc", 1.5, 2.5],
)
def test_presence_overrides_is_present_non_integer_fails(
tmp_path, bad_is_present
) -> None:
database = _open_db(tmp_path / "presence-overrides-is-present-typeof.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_override(database, child_id=1, is_present=bad_is_present)
finally:
_run(database.close())
def test_presence_overrides_end_date_before_start_date_fails(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-end-before-start.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
_insert_override(
database, child_id=1, start_date="2026-09-13", end_date="2026-09-12"
)
finally:
_run(database.close())
def test_presence_overrides_end_date_equal_to_start_date_allowed(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-end-equals-start.db")
try:
_apply(database)
_child(database)
override_id = _insert_override(
database, child_id=1, start_date="2026-09-13", end_date="2026-09-13"
)
row = _run(
database.fetch_one(
"SELECT end_date FROM presence_overrides WHERE id = ?",
(override_id,),
)
)
assert row == ("2026-09-13",)
finally:
_run(database.close())
def test_presence_overrides_start_date_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-start-null.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
_insert_override(database, child_id=1, start_date=None)
finally:
_run(database.close())
def test_presence_overrides_end_date_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-end-null.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
_insert_override(database, child_id=1, end_date=None)
finally:
_run(database.close())
def test_presence_overrides_is_present_not_null_enforced(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-is-present-null.db")
try:
_apply(database)
_child(database)
with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
_insert_override(database, child_id=1, is_present=None)
finally:
_run(database.close())
def test_presence_overrides_multiple_children_allowed(tmp_path) -> None:
database = _open_db(tmp_path / "presence-overrides-multi-child.db")
try:
_apply(database)
_child(database, "Ada")
_run(
database.execute(
"INSERT INTO children (display_name, created_at) VALUES (?, ?)",
("Bo", "2026-09-13T00:00:00+00:00"),
)
)
_insert_override(database, child_id=1)
_insert_override(database, child_id=2)
row = _run(database.fetch_one("SELECT COUNT(*) FROM presence_overrides"))
assert row == (2,)
finally:
_run(database.close())
def test_applying_presence_ddl_twice_keeps_rows(tmp_path) -> None:
database = _open_db(tmp_path / "presence-idempotent.db")
try:
_apply(database)
_child(database)
_insert_schedule(database, child_id=1)
_insert_override(database, child_id=1)
_apply(database)
counts = []
for table in ("presence_schedules", "presence_overrides"):
row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}"))
counts.append(row[0])
assert counts == [1, 1]
finally:
_run(database.close())