From 8e0baaa5d91e4f7653024e868afa04459df7ce00 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:41:31 -0400 Subject: [PATCH] feat: add schema v1 ddl for schedule_rules and task_definitions --- custom_components/nestquest/schema.py | 59 +++ tests/test_lifecycle.py | 21 +- tests/test_schema.py | 497 +++++++++++++++++++++++++- 3 files changed, 575 insertions(+), 2 deletions(-) diff --git a/custom_components/nestquest/schema.py b/custom_components/nestquest/schema.py index f9d8322..163e98f 100644 --- a/custom_components/nestquest/schema.py +++ b/custom_components/nestquest/schema.py @@ -12,6 +12,24 @@ caller/DAO rather than a DB default, per the feature guardrails. No produced by callers is not fixed yet (``datetime.isoformat()`` may include microseconds, changing the length), so enforcing a fixed length here would reject valid values. Only non-null is enforced for now. + +Date policy: ``start_date``/``end_date`` are ISO date strings (YYYY-MM-DD, +no time component) compared lexically by the ``end_date >= start_date`` +CHECK; due_date/due_time string shapes are caller-validated by the rule +engine and DAO layers, not by DDL. + +``weekday_set`` format: comma-separated weekday integers (CSV), 0 = +Monday .. 6 = Sunday, e.g. ``'0,2,4'`` for Mon/Wed/Fri. Chosen over JSON +because it sorts, compares and parses trivially in SQL. + +Rule-type coherence: two CHECKs enforce that ``weekly`` carries a +non-null ``weekday_set``, ``monthly`` carries a ``day_of_month`` or an +``nth_weekday``, ``yearly`` carries a ``month``, and ``daily``/``custom`` +carry none of the month-specific fields. Relaxed deliberately: the DDL +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. """ from __future__ import annotations @@ -40,9 +58,50 @@ SCHEMA_V1_ADMIN_USERS_DDL: list[str] = [ """, ] +SCHEMA_V1_SCHEDULE_RULES_DDL: list[str] = [ + """ + CREATE TABLE IF NOT EXISTS schedule_rules ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + rule_type TEXT NOT NULL CHECK (rule_type IN ('daily', 'weekly', 'monthly', 'yearly', 'custom')), + interval INTEGER NOT NULL DEFAULT 1 CHECK (interval >= 1), + weekday_set TEXT, + day_of_month INTEGER CHECK (day_of_month IS NULL OR (day_of_month >= 1 AND day_of_month <= 31)), + nth_weekday INTEGER CHECK (nth_weekday IN (-1, 1, 2, 3, 4, 5)), + month INTEGER CHECK (month IS NULL OR (month >= 1 AND month <= 12)), + start_date TEXT NOT NULL, + end_date TEXT, + CHECK (end_date IS NULL OR end_date >= start_date), + CHECK ( + (rule_type = 'weekly' AND weekday_set IS NOT NULL) + OR (rule_type = 'monthly' AND (day_of_month IS NOT NULL OR nth_weekday IS NOT NULL)) + OR (rule_type = 'yearly' AND month IS NOT NULL) + OR (rule_type IN ('daily', 'custom') AND day_of_month IS NULL AND nth_weekday IS NULL AND month IS NULL) + ) + ) + """, +] + +SCHEMA_V1_TASK_DEFINITIONS_DDL: list[str] = [ + """ + 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 CHECK (is_active IN (0, 1)), + created_at TEXT NOT NULL + ) + """, +] + SCHEMA_V1_STATEMENTS: list[str] = [ *SCHEMA_V1_CHILDREN_DDL, *SCHEMA_V1_ADMIN_USERS_DDL, + *SCHEMA_V1_SCHEDULE_RULES_DDL, + *SCHEMA_V1_TASK_DEFINITIONS_DDL, ] diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index b145aad..246f8b5 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -236,4 +236,23 @@ async def test_unload_closes_database_even_when_listener_removal_raises() -> Non await async_unload_entry(hass, entry) assert database.connected is False, "database leaked despite remover error" assert entry.runtime_data is None - assert DOMAIN not in hass.data \ No newline at end of file + assert DOMAIN not in hass.data + + +async def test_setup_applies_all_v1_tables(hass, make_entry) -> None: + """Setup applies the full v1 DDL: all four v1 tables exist afterwards.""" + entry = _wire(make_entry(), hass.registry) + assert await async_setup_entry(hass, entry) is True + database = entry.runtime_data.database + 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" + ) + assert [row[0] for row in rows] == [ + "admin_users", + "children", + "schedule_rules", + "task_definitions", + ] + await async_unload_entry(hass, entry) \ No newline at end of file diff --git a/tests/test_schema.py b/tests/test_schema.py index b1714ab..32650da 100644 --- a/tests/test_schema.py +++ b/tests/test_schema.py @@ -10,7 +10,9 @@ from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.schema import ( SCHEMA_V1_ADMIN_USERS_DDL, SCHEMA_V1_CHILDREN_DDL, + SCHEMA_V1_SCHEDULE_RULES_DDL, SCHEMA_V1_STATEMENTS, + SCHEMA_V1_TASK_DEFINITIONS_DDL, async_apply_ddl, ) @@ -37,15 +39,73 @@ def _apply(database) -> None: _run(async_apply_ddl(database, SCHEMA_V1_STATEMENTS)) +# Sentinel used by the insert helpers: omit the column from the INSERT so +# its DDL DEFAULT applies (explicit NULL would not trigger a DEFAULT). +_OMIT = object() + + +def _insert_rule(database, **overrides): + """Insert one valid schedule_rule, returning the rule id. + + Pass ``_OMIT`` as a value to leave that column out of the INSERT so + its DDL DEFAULT applies. + """ + values: dict[str, object] = { + "rule_type": "weekly", + "weekday_set": "0,2,4", + "start_date": "2026-09-13", + } + 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 schedule_rules ({columns}) VALUES ({placeholders})", + tuple(values.values()), + ) + ) + assert result.lastrowid is not None + return result.lastrowid + + +def _insert_definition(database, child_id=1, rule_id=1, **overrides): + """Insert one valid task_definition, returning the definition id. + + Pass ``_OMIT`` as a value to leave that column out of the INSERT so + its DDL DEFAULT applies. + """ + values: dict[str, object] = { + "title": "Brush teeth", + "child_id": child_id, + "schedule_rule_id": rule_id, + "created_at": "2026-09-13T00:00:00+00:00", + } + 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 task_definitions ({columns}) VALUES ({placeholders})", + tuple(values.values()), + ) + ) + assert result.lastrowid is not None + return result.lastrowid + + # --------------------------------------------------------------------------- # DDL constants # --------------------------------------------------------------------------- -def test_schema_v1_statements_compose_the_two_tables() -> None: +def test_schema_v1_statements_compose_the_four_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, ] @@ -121,6 +181,91 @@ def test_admin_users_columns_types_and_constraints(tmp_path) -> None: _run(database.close()) +def test_schedule_rules_table_exists_after_applying_ddl(tmp_path) -> None: + database = _open_db(tmp_path / "rules.db") + try: + _apply(database) + row = _run( + database.fetch_one( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name = 'schedule_rules'" + ) + ) + assert row is not None and row[0] == "schedule_rules" + finally: + _run(database.close()) + + +def test_task_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'" + ) + ) + assert row is not None and row[0] == "task_definitions" + finally: + _run(database.close()) + + +def test_schedule_rules_columns_types_and_constraints(tmp_path) -> None: + database = _open_db(tmp_path / "rules-columns.db") + try: + _apply(database) + columns = _run(database.fetch_all("PRAGMA table_info(schedule_rules)")) + # 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), + ("rule_type", "TEXT", 1, None, 0), + ("interval", "INTEGER", 1, "1", 0), + ("weekday_set", "TEXT", 0, None, 0), + ("day_of_month", "INTEGER", 0, None, 0), + ("nth_weekday", "INTEGER", 0, None, 0), + ("month", "INTEGER", 0, None, 0), + ("start_date", "TEXT", 1, None, 0), + ("end_date", "TEXT", 0, None, 0), + ] + finally: + _run(database.close()) + + +def test_task_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)")) + 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), + ("description", "TEXT", 0, None, 0), + ("icon", "TEXT", 0, None, 0), + ("child_id", "INTEGER", 1, None, 0), + ("schedule_rule_id", "INTEGER", 1, None, 0), + ("due_time", "TEXT", 0, None, 0), + ("is_active", "INTEGER", 1, "1", 0), + ("created_at", "TEXT", 1, None, 0), + ] + finally: + _run(database.close()) + + +def test_task_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)")) + declared = {(row[2], row[3], row[4]) for row in fks} + assert declared == { + ("children", "child_id", "id"), + ("schedule_rules", "schedule_rule_id", "id"), + } + finally: + _run(database.close()) + + # --------------------------------------------------------------------------- # Constraint enforcement # --------------------------------------------------------------------------- @@ -282,5 +427,355 @@ def test_failed_statement_rolls_back_whole_application(tmp_path) -> None: ) ) assert row is None, "partial DDL must not survive a failed application" + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# schedule_rules and task_definitions constraint enforcement +# --------------------------------------------------------------------------- + + +def test_schedule_rules_valid_insert_round_trips(tmp_path) -> None: + database = _open_db(tmp_path / "valid-rule.db") + try: + _apply(database) + rule_id = _insert_rule(database) + row = _run( + database.fetch_one( + "SELECT rule_type, interval, weekday_set, start_date " + "FROM schedule_rules WHERE id = ?", + (rule_id,), + ) + ) + assert row == ("weekly", 1, "0,2,4", "2026-09-13") + finally: + _run(database.close()) + + +def test_schedule_rules_interval_default_applies(tmp_path) -> None: + database = _open_db(tmp_path / "interval-default.db") + try: + _apply(database) + rule_id = _insert_rule(database, interval=_OMIT) + row = _run( + database.fetch_one( + "SELECT interval FROM schedule_rules WHERE id = ?", (rule_id,) + ) + ) + assert row == (1,) + finally: + _run(database.close()) + + +def test_schedule_rules_rule_type_check_enforced(tmp_path) -> None: + database = _open_db(tmp_path / "rule-type.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule(database, rule_type="biweekly", weekday_set=None) + finally: + _run(database.close()) + + +def test_schedule_rules_interval_zero_fails(tmp_path) -> None: + database = _open_db(tmp_path / "interval-zero.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule(database, interval=0) + finally: + _run(database.close()) + + +def test_schedule_rules_day_of_month_out_of_range_fails(tmp_path) -> None: + database = _open_db(tmp_path / "day-of-month.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule( + database, + rule_type="monthly", + weekday_set=None, + day_of_month=32, + ) + finally: + _run(database.close()) + + +def test_schedule_rules_nth_weekday_zero_fails(tmp_path) -> None: + database = _open_db(tmp_path / "nth-weekday.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule( + database, + rule_type="monthly", + weekday_set=None, + day_of_month=None, + nth_weekday=0, + ) + finally: + _run(database.close()) + + +def test_schedule_rules_nth_weekday_minus_one_last_week_allowed(tmp_path) -> None: + database = _open_db(tmp_path / "nth-weekday-last.db") + try: + _apply(database) + rule_id = _insert_rule( + database, + rule_type="monthly", + weekday_set=None, + day_of_month=None, + nth_weekday=-1, + ) + row = _run( + database.fetch_one( + "SELECT nth_weekday FROM schedule_rules WHERE id = ?", (rule_id,) + ) + ) + assert row == (-1,) + finally: + _run(database.close()) + + +def test_schedule_rules_month_out_of_range_fails(tmp_path) -> None: + database = _open_db(tmp_path / "month.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule( + database, rule_type="yearly", weekday_set=None, month=13 + ) + finally: + _run(database.close()) + + +def test_schedule_rules_weekly_without_weekday_set_fails(tmp_path) -> None: + database = _open_db(tmp_path / "weekly-no-weekdays.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule(database, rule_type="weekly", weekday_set=None) + finally: + _run(database.close()) + + +def test_schedule_rules_monthly_without_day_or_nth_fails(tmp_path) -> None: + database = _open_db(tmp_path / "monthly-no-fields.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule( + database, + rule_type="monthly", + weekday_set=None, + day_of_month=None, + nth_weekday=None, + ) + finally: + _run(database.close()) + + +def test_schedule_rules_yearly_without_month_fails(tmp_path) -> None: + database = _open_db(tmp_path / "yearly-no-month.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule( + database, + rule_type="yearly", + weekday_set=None, + day_of_month=None, + month=None, + ) + finally: + _run(database.close()) + + +def test_schedule_rules_daily_with_month_fields_fails(tmp_path) -> None: + database = _open_db(tmp_path / "daily-month-fields.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule(database, rule_type="daily", month=3) + finally: + _run(database.close()) + + +def test_schedule_rules_end_date_before_start_date_fails(tmp_path) -> None: + database = _open_db(tmp_path / "end-before-start.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_rule(database, end_date="2026-09-12") + finally: + _run(database.close()) + + +def test_schedule_rules_end_date_equal_to_start_date_allowed(tmp_path) -> None: + database = _open_db(tmp_path / "end-equals-start.db") + try: + _apply(database) + rule_id = _insert_rule(database, end_date="2026-09-13") + row = _run( + database.fetch_one( + "SELECT end_date FROM schedule_rules WHERE id = ?", (rule_id,) + ) + ) + assert row == ("2026-09-13",) + finally: + _run(database.close()) + + +def test_schedule_rules_start_date_not_null_enforced(tmp_path) -> None: + database = _open_db(tmp_path / "start-date-null.db") + try: + _apply(database) + with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"): + _insert_rule(database, start_date=None) + finally: + _run(database.close()) + + +def test_task_definitions_insert_requires_existing_child_and_rule( + tmp_path, +) -> None: + database = _open_db(tmp_path / "definitions-fk-unknown.db") + try: + _apply(database) + _run( + database.execute( + "INSERT INTO children (display_name, created_at) VALUES (?, ?)", + ("Ada", "2026-09-13T00:00:00+00:00"), + ) + ) + rule_id = _insert_rule(database) + definition_id = _insert_definition( + database, child_id=1, rule_id=rule_id + ) + row = _run( + database.fetch_one( + "SELECT title, child_id, schedule_rule_id FROM task_definitions " + "WHERE id = ?", + (definition_id,), + ) + ) + assert row == ("Brush teeth", 1, rule_id) + finally: + _run(database.close()) + + +def test_task_definitions_unknown_child_id_fails(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-bad-child.db") + try: + _apply(database) + _insert_rule(database) + with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"): + _insert_definition(database, child_id=999) + finally: + _run(database.close()) + + +def test_task_definitions_unknown_schedule_rule_id_fails(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-bad-rule.db") + try: + _apply(database) + _run( + database.execute( + "INSERT INTO children (display_name, created_at) VALUES (?, ?)", + ("Ada", "2026-09-13T00:00:00+00:00"), + ) + ) + with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"): + _insert_definition(database, child_id=1, rule_id=999) + finally: + _run(database.close()) + + +def test_task_definitions_is_active_check_enforced(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-is-active.db") + try: + _apply(database) + _insert_rule(database) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _insert_definition(database, child_id=1, rule_id=1, is_active=2) + finally: + _run(database.close()) + + +def test_task_definitions_title_not_null_enforced(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-title.db") + try: + _apply(database) + _insert_rule(database) + with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"): + _insert_definition(database, child_id=1, rule_id=1, title=None) + finally: + _run(database.close()) + + +def test_task_definitions_created_at_not_null_enforced(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-created-at.db") + try: + _apply(database) + _insert_rule(database) + with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"): + _insert_definition( + database, child_id=1, rule_id=1, created_at=None + ) + finally: + _run(database.close()) + + +def test_task_definitions_defaults_apply_on_insert(tmp_path) -> None: + database = _open_db(tmp_path / "definitions-defaults.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, child_id=1, rule_id=1, is_active=_OMIT + ) + row = _run( + database.fetch_one( + "SELECT is_active, due_time FROM task_definitions WHERE id = ?", + (definition_id,), + ) + ) + assert row == (1, None) + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Idempotence across the extended v1 list +# --------------------------------------------------------------------------- + + +def test_applying_extended_ddl_twice_keeps_new_table_rows(tmp_path) -> None: + database = _open_db(tmp_path / "idempotent-extended.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) + _insert_definition(database, child_id=1, rule_id=1) + _apply(database) + counts = [] + for table in ("children", "schedule_rules", "task_definitions"): + row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}")) + counts.append(row[0]) + assert counts == [1, 1, 1] finally: _run(database.close()) \ No newline at end of file