diff --git a/custom_components/nestquest/schema.py b/custom_components/nestquest/schema.py index ffd3278..aa329ad 100644 --- a/custom_components/nestquest/schema.py +++ b/custom_components/nestquest/schema.py @@ -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, ] diff --git a/tests/test_schema.py b/tests/test_schema.py index 689bbdc..247b9f7 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -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()) \ No newline at end of file