feat: add schema v1 ddl for children and admin_users

custom_components/nestquest/__init__.py

@@ -11,6 +11,7 @@ from homeassistant.core import HomeAssistant
 
 from .const import DOMAIN, LOGGER
 from .db import NestQuestDatabase
+from .schema import SCHEMA_V1_STATEMENTS, async_apply_ddl
 from .store import async_get_db_path
 
 _LOGGER = LOGGER
@@ -51,6 +52,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
 
     database = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
     try:
+        await async_apply_ddl(database, SCHEMA_V1_STATEMENTS)
         remove_update_listener = entry.add_update_listener(_async_update_listener)
     except BaseException:
         await database.close()

custom_components/nestquest/schema.py

@@ -0,0 +1,59 @@
+"""Declarative schema DDL for the NestQuest SQLite database.
+
+Every v1 table is declared here as plain SQL statements so later schema
+tasks can append their own DDL to :data:`SCHEMA_V1_STATEMENTS` without
+touching this module's tables.  All statements use
+``CREATE TABLE IF NOT EXISTS`` so application is idempotent; the versioned
+migration runner is a separate later concern.
+
+Timestamp policy: all timestamps are UTC ISO-8601 strings, set by the
+caller/DAO rather than a DB default, per the feature guardrails.  No
+``length(created_at) = 25`` CHECK is added: the exact string shape
+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.
+"""
+from __future__ import annotations
+
+from .db import NestQuestDatabase
+
+SCHEMA_V1_CHILDREN_DDL: list[str] = [
+    """
+    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 CHECK (is_active IN (0, 1)),
+        created_at TEXT NOT NULL
+    )
+    """,
+]
+
+SCHEMA_V1_ADMIN_USERS_DDL: list[str] = [
+    """
+    CREATE TABLE IF NOT EXISTS admin_users (
+        ha_user_id TEXT PRIMARY KEY NOT NULL,
+        added_at TEXT NOT NULL
+    )
+    """,
+]
+
+SCHEMA_V1_STATEMENTS: list[str] = [
+    *SCHEMA_V1_CHILDREN_DDL,
+    *SCHEMA_V1_ADMIN_USERS_DDL,
+]
+
+
+async def async_apply_ddl(
+    database: NestQuestDatabase, statements: list[str]
+) -> None:
+    """Apply each DDL statement inside one transaction on ``database``.
+
+    Idempotent: every statement is ``CREATE TABLE IF NOT EXISTS``, so
+    applying the same list twice leaves the schema and its rows intact.
+    """
+    async with database.transaction():
+        for sql in statements:
+            await database.execute(sql)

tests/test_schema.py

@@ -0,0 +1,286 @@
+"""Tests for schema.py: v1 DDL for children and admin_users tables."""
+from __future__ import annotations
+
+import asyncio
+import sqlite3
+
+import pytest
+
+from custom_components.nestquest.db import NestQuestDatabase
+from custom_components.nestquest.schema import (
+    SCHEMA_V1_ADMIN_USERS_DDL,
+    SCHEMA_V1_CHILDREN_DDL,
+    SCHEMA_V1_STATEMENTS,
+    async_apply_ddl,
+)
+
+
+def _run(coro):
+    return asyncio.new_event_loop().run_until_complete(coro)
+
+
+def _make_hass_mock():
+    from unittest.mock import AsyncMock, MagicMock
+
+    hass = MagicMock()
+    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
+    return hass
+
+
+def _open_db(path) -> NestQuestDatabase:
+    database = NestQuestDatabase(_make_hass_mock())
+    _run(database.open(path))
+    return database
+
+
+def _apply(database) -> None:
+    _run(async_apply_ddl(database, SCHEMA_V1_STATEMENTS))
+
+
+# ---------------------------------------------------------------------------
+# DDL constants
+# ---------------------------------------------------------------------------
+
+
+def test_schema_v1_statements_compose_the_two_tables() -> None:
+    assert SCHEMA_V1_STATEMENTS == [
+        *SCHEMA_V1_CHILDREN_DDL,
+        *SCHEMA_V1_ADMIN_USERS_DDL,
+    ]
+
+
+def test_every_statement_is_create_if_not_exists() -> None:
+    for sql in SCHEMA_V1_STATEMENTS:
+        assert sql.lstrip().upper().startswith("CREATE TABLE IF NOT EXISTS")
+
+
+# ---------------------------------------------------------------------------
+# Tables after applying DDL
+# ---------------------------------------------------------------------------
+
+
+def test_children_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "children.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'children'"
+            )
+        )
+        assert row is not None and row[0] == "children"
+    finally:
+        _run(database.close())
+
+
+def test_admin_users_table_exists_after_applying_ddl(tmp_path) -> None:
+    database = _open_db(tmp_path / "admins.db")
+    try:
+        _apply(database)
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'admin_users'"
+            )
+        )
+        assert row is not None and row[0] == "admin_users"
+    finally:
+        _run(database.close())
+
+
+def test_children_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "children-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(children)"))
+        # 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),
+            ("display_name", "TEXT", 1, None, 0),
+            ("colour", "TEXT", 0, None, 0),
+            ("avatar_ref", "TEXT", 0, None, 0),
+            ("sort_order", "INTEGER", 1, "0", 0),
+            ("is_active", "INTEGER", 1, "1", 0),
+            ("created_at", "TEXT", 1, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+def test_admin_users_columns_types_and_constraints(tmp_path) -> None:
+    database = _open_db(tmp_path / "admins-columns.db")
+    try:
+        _apply(database)
+        columns = _run(database.fetch_all("PRAGMA table_info(admin_users)"))
+        assert [(row[1], row[2], row[3], row[4], row[5]) for row in columns] == [
+            ("ha_user_id", "TEXT", 1, None, 1),
+            ("added_at", "TEXT", 1, None, 0),
+        ]
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Constraint enforcement
+# ---------------------------------------------------------------------------
+
+
+def test_children_display_name_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "display-name.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _run(
+                database.execute(
+                    "INSERT INTO children (display_name, created_at) "
+                    "VALUES (NULL, ?)",
+                    ("2026-09-13T00:00:00+00:00",),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_children_is_active_check_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "is-active.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="CHECK"):
+            _run(
+                database.execute(
+                    "INSERT INTO children (display_name, is_active, created_at) "
+                    "VALUES (?, ?, ?)",
+                    ("Ada", 2, "2026-09-13T00:00:00+00:00"),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_children_created_at_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "created-at.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _run(
+                database.execute(
+                    "INSERT INTO children (display_name, created_at) "
+                    "VALUES (?, NULL)",
+                    ("Ada",),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_admin_users_primary_key_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "pk.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO admin_users (ha_user_id, added_at) VALUES (?, ?)",
+                ("user-1", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        with pytest.raises(sqlite3.IntegrityError, match="UNIQUE"):
+            _run(
+                database.execute(
+                    "INSERT INTO admin_users (ha_user_id, added_at) VALUES (?, ?)",
+                    ("user-1", "2026-09-13T00:00:01+00:00"),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_admin_users_added_at_not_null_enforced(tmp_path) -> None:
+    database = _open_db(tmp_path / "added-at.db")
+    try:
+        _apply(database)
+        with pytest.raises(sqlite3.IntegrityError, match="NOT NULL"):
+            _run(
+                database.execute(
+                    "INSERT INTO admin_users (ha_user_id, added_at) "
+                    "VALUES (?, NULL)",
+                    ("user-1",),
+                )
+            )
+    finally:
+        _run(database.close())
+
+
+def test_children_defaults_apply_on_insert(tmp_path) -> None:
+    database = _open_db(tmp_path / "defaults.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        row = _run(
+            database.fetch_one(
+                "SELECT sort_order, is_active FROM children "
+                "WHERE display_name = 'Ada'"
+            )
+        )
+        assert row == (0, 1)
+    finally:
+        _run(database.close())
+
+
+# ---------------------------------------------------------------------------
+# Idempotence and transactional application
+# ---------------------------------------------------------------------------
+
+
+def test_applying_ddl_twice_succeeds_and_keeps_rows(tmp_path) -> None:
+    database = _open_db(tmp_path / "idempotent.db")
+    try:
+        _apply(database)
+        _run(
+            database.execute(
+                "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
+                ("Ada", "2026-09-13T00:00:00+00:00"),
+            )
+        )
+        _apply(database)
+        row = _run(
+            database.fetch_one("SELECT COUNT(*) FROM children")
+        )
+        assert row == (1,)
+    finally:
+        _run(database.close())
+
+
+def test_failed_statement_rolls_back_whole_application(tmp_path) -> None:
+    from unittest.mock import AsyncMock
+
+    database = _open_db(tmp_path / "rollback.db")
+    try:
+        bad_statements = [
+            "CREATE TABLE IF NOT EXISTS children (id INTEGER PRIMARY KEY)",
+            "CREATE TABLE nope (",
+        ]
+        original = database.execute
+
+        async def _failing_execute(sql, parameters=()):
+            if "CREATE TABLE nope" in sql:
+                raise sqlite3.OperationalError("near \"(\": syntax error")
+            return await original(sql, parameters)
+
+        database.execute = AsyncMock(side_effect=_failing_execute)
+        with pytest.raises(sqlite3.OperationalError):
+            _run(async_apply_ddl(database, bad_statements))
+        row = _run(
+            database.fetch_one(
+                "SELECT name FROM sqlite_master WHERE type = 'table' "
+                "AND name = 'children'"
+            )
+        )
+        assert row is None, "partial DDL must not survive a failed application"
+    finally:
+        _run(database.close())