diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index d6d97be..d4c16f5 100644 --- a/custom_components/nestquest/__init__.py +++ b/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() diff --git a/custom_components/nestquest/schema.py b/custom_components/nestquest/schema.py new file mode 100644 index 0000000..f9d8322 --- /dev/null +++ b/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) \ No newline at end of file diff --git a/tests/test_schema.py b/tests/test_schema.py new file mode 100644 index 0000000..b1714ab --- /dev/null +++ b/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()) \ No newline at end of file