From 43b20f2b126cedc46d006ad887fc9e9df1a9d858 Mon Sep 17 00:00:00 2001 From: orchestrator-agent <20+orchestrator-agent@code.cubecraftcreations.com> Date: Mon, 14 Sep 2026 07:26:34 -0400 Subject: [PATCH] DAO layer for children and admin_users (#18) --- custom_components/nestquest/dao_children.py | 289 ++++++++++ tests/test_dao_children.py | 553 ++++++++++++++++++++ 2 files changed, 842 insertions(+) create mode 100644 custom_components/nestquest/dao_children.py create mode 100644 tests/test_dao_children.py diff --git a/custom_components/nestquest/dao_children.py b/custom_components/nestquest/dao_children.py new file mode 100644 index 0000000..88eb62c --- /dev/null +++ b/custom_components/nestquest/dao_children.py @@ -0,0 +1,289 @@ +"""Typed DAO layer for the children and admin_users tables. + +All SQL for ``children`` and ``admin_users`` lives in this module per +the feature guardrail: callers get typed dataclasses back and never see +raw rows or SQL. Every method is async and runs through +:class:`~.db.NestQuestDatabase`, so each statement executes on the HA +executor and the event loop never blocks. + +Error contract: violations surface as the underlying +:class:`sqlite3.IntegrityError` (duplicate, foreign-key, or CHECK +failure) from the execute call; the DAO adds no wrapping exception +type. A get that matches nothing returns ``None``; a mutation that +matches nothing is reported by :class:`~.db.ExecutionResult.rowcount`, +which each method returns where the caller may need it (reorder, and +the admin remove/set paths). +""" +from __future__ import annotations + +import asyncio +from dataclasses import dataclass + +from .db import NestQuestDatabase + + +@dataclass(frozen=True) +class ChildRecord: + """One row of ``children``.""" + + id: int + display_name: str + colour: str | None + avatar_ref: str | None + sort_order: int + is_active: bool + created_at: str + + +@dataclass(frozen=True) +class AdminUserRecord: + """One row of ``admin_users``.""" + + ha_user_id: str + added_at: str + + +def _child_from_row(row: tuple) -> ChildRecord: + """Build a :class:`ChildRecord` from a SELECT row.""" + return ChildRecord( + id=row[0], + display_name=row[1], + colour=row[2], + avatar_ref=row[3], + sort_order=row[4], + is_active=bool(row[5]), + created_at=row[6], + ) + + +_CHILD_COLUMNS = ( + "id, display_name, colour, avatar_ref, sort_order, is_active, created_at" +) + + +#: One asyncio.Lock per (connection wrapper, running loop), shared by +#: every ChildrenDao instance over the same database: the wrapper +#: rejects a second concurrent transaction with RuntimeError, so ALL +#: transactional DAO operations on one connection must queue here, not +#: per-instance. Keyed like the migration runner's lock; entries are +#: never evicted (a config entry holds one wrapper per process). +_CONNECTION_LOCKS: dict[tuple[int, int], asyncio.Lock] = {} + + +def _connection_lock(database) -> asyncio.Lock: + """Return the shared transaction-serializing lock for ``database``.""" + loop = asyncio.get_running_loop() + key = (id(database), id(loop)) + lock = _CONNECTION_LOCKS.get(key) + if lock is None: + lock = asyncio.Lock() + _CONNECTION_LOCKS[key] = lock + return lock + + +class ChildrenDao: + """Typed async data access for the ``children`` table. + + Deletion is deliberately absent: children are deactivated, never + deleted, because completion history references them (feature + guardrail). + """ + + def __init__(self, database) -> None: + self._database = database + + async def create( + self, + display_name: str, + created_at: str, + *, + colour: str | None = None, + avatar_ref: str | None = None, + sort_order: int = 0, + is_active: bool = True, + ) -> ChildRecord: + """Insert one child and return the record as stored.""" + result = await self._database.execute( + "INSERT INTO children (display_name, colour, avatar_ref, " + "sort_order, is_active, created_at) VALUES (?, ?, ?, ?, ?, ?)", + ( + display_name, + colour, + avatar_ref, + sort_order, + int(is_active), + created_at, + ), + ) + child = await self.get(result.lastrowid) + assert child is not None + return child + + async def get(self, child_id: int) -> ChildRecord | None: + """Return the child with ``child_id``, or None.""" + row = await self._database.fetch_one( + f"SELECT {_CHILD_COLUMNS} FROM children WHERE id = ?", + (child_id,), + ) + return _child_from_row(row) if row is not None else None + + async def list_active(self) -> list[ChildRecord]: + """Return active children ordered by sort_order, then id.""" + rows = await self._database.fetch_all( + f"SELECT {_CHILD_COLUMNS} FROM children WHERE is_active = 1 " + "ORDER BY sort_order, id" + ) + return [_child_from_row(row) for row in rows] + + async def list_all(self) -> list[ChildRecord]: + """Return all children ordered by sort_order, then id.""" + rows = await self._database.fetch_all( + f"SELECT {_CHILD_COLUMNS} FROM children " + "ORDER BY sort_order, id" + ) + return [record for record in map(_child_from_row, rows)] + + async def update( + self, + child_id: int, + *, + display_name: str | None = None, + colour: str | None = None, + avatar_ref: str | None = None, + sort_order: int | None = None, + ) -> int: + """Update the given fields; returns rows updated (0 if absent). + + Only non-None arguments are written, so a caller cannot + accidentally blank a column by omitting it. + """ + assignments: list[str] = [] + parameters: list[object] = [] + if display_name is not None: + assignments.append("display_name = ?") + parameters.append(display_name) + if colour is not None: + assignments.append("colour = ?") + parameters.append(colour) + if avatar_ref is not None: + assignments.append("avatar_ref = ?") + parameters.append(avatar_ref) + if sort_order is not None: + assignments.append("sort_order = ?") + parameters.append(sort_order) + if not assignments: + return 0 + parameters.append(child_id) + result = await self._database.execute( + f"UPDATE children SET {', '.join(assignments)} WHERE id = ?", + tuple(parameters), + ) + return result.rowcount + + async def set_active(self, child_id: int, is_active: bool) -> int: + """Set the child's active flag; returns rows updated (0 if absent).""" + result = await self._database.execute( + "UPDATE children SET is_active = ? WHERE id = ?", + (int(is_active), child_id), + ) + return result.rowcount + + async def reorder(self, ordered_ids: list[int]) -> None: + """Persist a new sort order: ``ordered_ids[i]`` gets sort_order i. + + Runs inside one transaction; positions are staged at offset + -len (all negative, so any permutation of the same ids never + collides mid-reorder under the wrapper's single-connection + serialization), then finalized 0..n-1. Duplicate ids are + rejected up front: they cannot all hold distinct positions, and + silently assigning the last occurrence's index would hide a + caller bug. + + Concurrent reorder() calls on one connection are serialized by + a connection-scoped asyncio lock (shared across every ChildrenDao + over the same wrapper, however many instances): the wrapper's + transaction() rejects a second concurrent transaction with + RuntimeError, so racing callers would crash instead of + queueing. The lock makes the second call wait, then run + against the first call's final state. + """ + if not ordered_ids: + return + if len(set(ordered_ids)) != len(ordered_ids): + raise ValueError( + "reorder() received duplicate child ids; every id must " + "appear at most once" + ) + async with _connection_lock(self._database): + async with self._database.transaction(): + offset = len(ordered_ids) + await self._database.execute_many( + "UPDATE children SET sort_order = ? WHERE id = ?", + [ + (-(offset + index), child_id) + for index, child_id in enumerate(ordered_ids) + ], + ) + await self._database.execute_many( + "UPDATE children SET sort_order = ? WHERE id = ?", + [ + (index, child_id) + for index, child_id in enumerate(ordered_ids) + ], + ) + + +class AdminUsersDao: + """Typed async access to the ``admin_users`` allowlist. + + The allowlist stores HA user IDs, never usernames (feature + guardrail); an empty allowlist means no admin, enforced by callers + resolving through :meth:`exists`. + """ + + def __init__(self, database) -> None: + self._database = database + + async def add(self, ha_user_id: str, added_at: str) -> bool: + """Add a user to the allowlist; True if inserted, False if present. + + Idempotent under concurrency: a single INSERT ... ON CONFLICT DO + NOTHING decides and inserts in one statement (the wrapper + serializes statements, not check-then-insert sequences, so a + SELECT/INSERT pair would let two concurrent adds both see no + row and the second one raise IntegrityError). rowcount reports + whether the insert actually happened. + """ + result = await self._database.execute( + "INSERT INTO admin_users (ha_user_id, added_at) VALUES (?, ?) " + "ON CONFLICT (ha_user_id) DO NOTHING", + (ha_user_id, added_at), + ) + return result.rowcount > 0 + + async def remove(self, ha_user_id: str) -> bool: + """Remove a user from the allowlist; True if a row was removed.""" + result = await self._database.execute( + "DELETE FROM admin_users WHERE ha_user_id = ?", + (ha_user_id,), + ) + return result.rowcount > 0 + + async def list(self) -> list[AdminUserRecord]: + """Return every allowlisted user, oldest first.""" + rows = await self._database.fetch_all( + "SELECT ha_user_id, added_at FROM admin_users " + "ORDER BY added_at, ha_user_id" + ) + return [ + AdminUserRecord(ha_user_id=row[0], added_at=row[1]) + for row in rows + ] + + async def exists(self, ha_user_id: str) -> bool: + """Return True if ``ha_user_id`` is allowlisted.""" + row = await self._database.fetch_one( + "SELECT 1 FROM admin_users WHERE ha_user_id = ?", + (ha_user_id,), + ) + return row is not None \ No newline at end of file diff --git a/tests/test_dao_children.py b/tests/test_dao_children.py new file mode 100644 index 0000000..d2ca721 --- /dev/null +++ b/tests/test_dao_children.py @@ -0,0 +1,553 @@ +"""Tests for dao_children.py: typed DAO for children and admin_users.""" +from __future__ import annotations + +import asyncio +import sqlite3 + +import pytest + +from custom_components.nestquest.dao_children import ( + AdminUserRecord, + AdminUsersDao, + ChildRecord, + ChildrenDao, +) +from custom_components.nestquest.db import NestQuestDatabase +from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS +from custom_components.nestquest.migrations import apply_migrations + + +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)) + _run(apply_migrations(database)) + return database + + +def _daos(database) -> tuple[ChildrenDao, AdminUsersDao]: + return ChildrenDao(database), AdminUsersDao(database) + + +NOW = "2026-09-14T11:00:00+00:00" + + +def _make_child(dao, name="Ada", **kwargs) -> ChildRecord: + return _run(dao.create(name, NOW, **kwargs)) + + +# --------------------------------------------------------------------------- +# children: create and get +# --------------------------------------------------------------------------- + + +def test_create_returns_typed_record(tmp_path) -> None: + database = _open_db(tmp_path / "create.db") + try: + dao, _ = _daos(database) + child = _make_child(dao, colour="#ff0000", avatar_ref="avatars/1") + assert isinstance(child, ChildRecord) + assert child.id == 1 + assert child.display_name == "Ada" + assert child.colour == "#ff0000" + assert child.avatar_ref == "avatars/1" + assert child.sort_order == 0 + assert child.is_active is True + assert child.created_at == NOW + finally: + _run(database.close()) + + +def test_get_returns_none_for_missing_child(tmp_path) -> None: + database = _open_db(tmp_path / "get-missing.db") + try: + dao, _ = _daos(database) + assert _run(dao.get(999)) is None + finally: + _run(database.close()) + + +def test_create_duplicate_name_is_allowed(tmp_path) -> None: + database = _open_db(tmp_path / "dup-name.db") + try: + dao, _ = _daos(database) + first = _make_child(dao, "Ada") + second = _make_child(dao, "Ada") + assert first.id != second.id + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# children: list active / list all +# --------------------------------------------------------------------------- + + +def test_list_active_excludes_inactive_and_sorts(tmp_path) -> None: + database = _open_db(tmp_path / "list-active.db") + try: + dao, _ = _daos(database) + _make_child(dao, "Bo", sort_order=2) + ada = _make_child(dao, "Ada", sort_order=1) + cy = _make_child(dao, "Cy", sort_order=1) + _make_child(dao, "Dee", sort_order=0) + _run(dao.set_active(ada.id, False)) + active = _run(dao.list_active()) + assert [c.display_name for c in active] == ["Dee", "Cy", "Bo"] + assert all(c.is_active for c in active) + finally: + _run(database.close()) + + +def test_list_all_includes_inactive(tmp_path) -> None: + database = _open_db(tmp_path / "list-all.db") + try: + dao, _ = _daos(database) + ada = _make_child(dao, "Ada", sort_order=1) + _make_child(dao, "Bo", sort_order=0) + _run(dao.set_active(ada.id, False)) + all_children = _run(dao.list_all()) + assert [c.display_name for c in all_children] == ["Bo", "Ada"] + assert all_children[1].is_active is False + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# children: update, set_active, reorder +# --------------------------------------------------------------------------- + + +def test_update_fields_round_trip(tmp_path) -> None: + database = _open_db(tmp_path / "update.db") + try: + dao, _ = _daos(database) + child = _make_child(dao, "Ada", colour="#111111") + updated = _run( + dao.update( + child.id, + display_name="Ada L.", + colour="#222222", + avatar_ref="avatars/x.png", + ) + ) + assert updated == 1 + fetched = _run(dao.get(child.id)) + assert fetched is not None + assert fetched.display_name == "Ada L." + assert fetched.colour == "#222222" + assert fetched.avatar_ref == "avatars/x.png" + assert fetched.sort_order == 0 + finally: + _run(database.close()) + + +def test_update_omitted_fields_untouched(tmp_path) -> None: + database = _open_db(tmp_path / "update-partial.db") + try: + dao, _ = _daos(database) + child = _make_child(dao, "Ada", colour="#111111") + _run(dao.update(child.id, display_name="Ada L.")) + fetched = _run(dao.get(child.id)) + assert fetched.colour == "#111111" + finally: + _run(database.close()) + + +def test_update_no_fields_returns_zero_and_changes_nothing(tmp_path) -> None: + database = _open_db(tmp_path / "update-empty.db") + try: + dao, _ = _daos(database) + child = _make_child(dao, "Ada") + assert _run(dao.update(child.id)) == 0 + fetched = _run(dao.get(child.id)) + assert fetched is not None and fetched.display_name == "Ada" + finally: + _run(database.close()) + + +def test_update_missing_child_returns_zero(tmp_path) -> None: + database = _open_db(tmp_path / "update-missing.db") + try: + dao, _ = _daos(database) + assert _run(dao.update(999, display_name="X")) == 0 + finally: + _run(database.close()) + + +def test_set_active_missing_child_returns_zero(tmp_path) -> None: + database = _open_db(tmp_path / "set-active-missing.db") + try: + dao, _ = _daos(database) + assert _run(dao.set_active(999, False)) == 0 + finally: + _run(database.close()) + + +def test_reorder_assigns_positions_in_order(tmp_path) -> None: + database = _open_db(tmp_path / "reorder.db") + try: + dao, _ = _daos(database) + a = _make_child(dao, "Ada") + b = _make_child(dao, "Bo") + c = _make_child(dao, "Cy") + _run(dao.reorder([c.id, a.id, b.id])) + children = _run(dao.list_all()) + assert [x.display_name for x in children] == ["Cy", "Ada", "Bo"] + assert [x.sort_order for x in children] == [0, 1, 2] + finally: + _run(database.close()) + + +def test_reorder_is_transactional_and_idempotent(tmp_path) -> None: + database = _open_db(tmp_path / "reorder-idempotent.db") + try: + dao, _ = _daos(database) + a = _make_child(dao, "Ada") + b = _make_child(dao, "Bo") + _run(dao.reorder([b.id, a.id])) + _run(dao.reorder([b.id, a.id])) + children = _run(dao.list_all()) + assert [(x.display_name, x.sort_order) for x in children] == [ + ("Bo", 0), + ("Ada", 1), + ] + finally: + _run(database.close()) + + +def test_reorder_empty_list_is_noop(tmp_path) -> None: + database = _open_db(tmp_path / "reorder-empty.db") + try: + dao, _ = _daos(database) + _make_child(dao, "Ada") + _run(dao.reorder([])) + children = _run(dao.list_all()) + assert children[0].sort_order == 0 + finally: + _run(database.close()) + + +def test_reorder_unknown_id_leaves_it_untouched(tmp_path) -> None: + database = _open_db(tmp_path / "reorder-unknown.db") + try: + dao, _ = _daos(database) + a = _make_child(dao, "Ada") + _run(dao.reorder([a.id, 999])) + children = _run(dao.list_all()) + assert children[0].sort_order == 0 + finally: + _run(database.close()) + + +def test_reorder_rolls_back_to_original_order_on_failure(tmp_path) -> None: + """A failing finalize batch rolls the whole reorder back. + + The two-phase staging is only meaningful if a crash between the + batches leaves the original order intact: monkeypatch execute_many + so the finalize (non-negative) batch raises, then assert every + sort_order is unchanged. + """ + database = _open_db(tmp_path / "reorder-rollback.db") + try: + dao, _ = _daos(database) + a = _make_child(dao, "Ada", sort_order=0) + b = _make_child(dao, "Bo", sort_order=1) + + original_execute_many = database.execute_many + + async def _failing_execute_many(sql, parameters): + # The staging batch writes negative positions; the finalize + # batch writes 0..n-1. Fail the finalize batch. + if all(position >= 0 for position, _ in parameters): + raise sqlite3.OperationalError("boom") + return await original_execute_many(sql, parameters) + + database.execute_many = _failing_execute_many + with pytest.raises(sqlite3.OperationalError, match="boom"): + _run(dao.reorder([b.id, a.id])) + children = _run(dao.list_all()) + assert [(c.display_name, c.sort_order) for c in children] == [ + ("Ada", 0), + ("Bo", 1), + ], "failed reorder must leave original sort orders intact" + finally: + _run(database.close()) + + +def test_reorder_rejects_duplicate_ids(tmp_path) -> None: + database = _open_db(tmp_path / "reorder-duplicate.db") + try: + dao, _ = _daos(database) + a = _make_child(dao, "Ada") + with pytest.raises(ValueError, match="duplicate"): + _run(dao.reorder([a.id, a.id])) + fetched = _run(dao.get(a.id)) + assert fetched is not None and fetched.sort_order == 0 + finally: + _run(database.close()) + + +def test_reorder_concurrent_calls_serialize(tmp_path) -> None: + """Concurrent reorders from DIFFERENT DAO instances queue safely. + + Two ChildrenDao objects over one connection race two reorders: the + connection-scoped lock makes the second wait, then run against the + first call's final state. Without the lock the wrapper's + transaction() would reject the second concurrent transaction with + RuntimeError. + """ + async def _main() -> None: + database = NestQuestDatabase(_make_hass_mock()) + await database.open(tmp_path / "reorder-concurrent.db") + try: + await apply_migrations(database) + dao_a, _ = _daos(database) + a = await dao_a.create("Ada", NOW) + b = await dao_a.create("Bo", NOW) + dao_b, _ = _daos(database) # second instance, same connection + task_1 = asyncio.ensure_future(dao_a.reorder([b.id, a.id])) + await asyncio.sleep(0) + task_2 = asyncio.ensure_future(dao_b.reorder([a.id, b.id])) + await asyncio.gather(task_1, task_2) + children = await dao_a.list_all() + orders = {c.display_name: c.sort_order for c in children} + assert sorted(orders.values()) == [0, 1] + finally: + await database.close() + + _run(_main()) + + +# --------------------------------------------------------------------------- +# admin_users: add, remove, list, exists +# --------------------------------------------------------------------------- + + +def test_admin_add_then_exists(tmp_path) -> None: + database = _open_db(tmp_path / "admin-add.db") + try: + _, dao = _daos(database) + assert _run(dao.add("user-1", NOW)) is True + assert _run(dao.exists("user-1")) is True + assert _run(dao.exists("user-2")) is False + finally: + _run(database.close()) + + +def test_admin_add_is_idempotent(tmp_path) -> None: + database = _open_db(tmp_path / "admin-idempotent.db") + try: + _, dao = _daos(database) + assert _run(dao.add("user-1", NOW)) is True + assert _run(dao.add("user-1", NOW)) is False + records = _run(dao.list()) + assert len(records) == 1 + assert records[0].added_at == NOW + finally: + _run(database.close()) + + +def test_admin_add_concurrent_calls_are_idempotent(tmp_path) -> None: + """Concurrent add() calls of the same user: one inserts, rest False. + + The single-statement ON CONFLICT DO NOTHING makes the decision and + the write atomic per statement, so racing coroutines cannot both + 'see no row' and one of them raise IntegrityError. + """ + async def _main() -> None: + database = NestQuestDatabase(_make_hass_mock()) + await database.open(tmp_path / "admin-race.db") + try: + await apply_migrations(database) + _, dao = _daos(database) + tasks = [ + asyncio.ensure_future(dao.add("user-1", NOW)) + for _ in range(5) + ] + await asyncio.sleep(0) + results = list(await asyncio.gather(*tasks)) + assert sorted(results) == [False, False, False, False, True] + records = await dao.list() + assert len(records) == 1 + assert records[0].added_at == NOW + finally: + await database.close() + + _run(_main()) + + +def test_admin_add_preserves_original_added_at(tmp_path) -> None: + database = _open_db(tmp_path / "admin-added-at.db") + try: + _, dao = _daos(database) + _run(dao.add("user-1", NOW)) + _run(dao.add("user-1", "2026-09-15T00:00:00+00:00")) + records = _run(dao.list()) + assert records[0].added_at == NOW + finally: + _run(database.close()) + + +def test_admin_remove_returns_true_then_false(tmp_path) -> None: + database = _open_db(tmp_path / "admin-remove.db") + try: + _, dao = _daos(database) + _run(dao.add("user-1", NOW)) + assert _run(dao.remove("user-1")) is True + assert _run(dao.remove("user-1")) is False + assert _run(dao.exists("user-1")) is False + finally: + _run(database.close()) + + +def test_admin_list_returns_typed_records_ordered(tmp_path) -> None: + database = _open_db(tmp_path / "admin-list.db") + try: + _, dao = _daos(database) + _run(dao.add("user-2", "2026-09-14T10:00:00+00:00")) + _run(dao.add("user-1", "2026-09-14T09:00:00+00:00")) + records = _run(dao.list()) + assert isinstance(records[0], AdminUserRecord) + assert [r.ha_user_id for r in records] == ["user-1", "user-2"] + assert records[0].added_at == "2026-09-14T09:00:00+00:00" + finally: + _run(database.close()) + + +def test_admin_exists_empty_allowlist_fails_closed(tmp_path) -> None: + database = _open_db(tmp_path / "admin-empty.db") + try: + _, dao = _daos(database) + assert _run(dao.exists("user-1")) is False + assert _run(dao.list()) == [] + finally: + _run(database.close()) + + +def test_admin_records_are_typed_not_raw_rows(tmp_path) -> None: + database = _open_db(tmp_path / "admin-typed.db") + try: + _, dao = _daos(database) + _run(dao.add("user-1", NOW)) + records = _run(dao.list()) + assert records == [AdminUserRecord("user-1", NOW)] + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# No SQL for these tables outside the DAO module +# --------------------------------------------------------------------------- + + +def test_children_and_admin_sql_lives_only_in_dao_module() -> None: + """Guardrail: children/admin_users SQL may appear only in the DAO + (and the schema DDL declarations). Scans the integration package + AND the repo's test files RECURSIVELY, matching FROM/INTO/UPDATE/ + DELETE FROM/JOIN against either table — including DELETE FROM + children, the hard-delete path the guardrail forbids. + + Exclusions are exact repo-relative paths, narrowly justified: + schema.py declares the tables' DDL; migrations.py applies that DDL; + test_schema.py and test_migrations.py test exactly those + DDL/migration layers. None of them touch the tables' data paths — + that is the DAO's job. + """ + import re + from pathlib import Path + + package = Path( + __import__( + "custom_components.nestquest", fromlist=["__file__"] + ).__file__ + ).parent + repo_root = package.parent.parent + scan_roots = [package, repo_root / "tests"] + + allowed = { + "custom_components/nestquest/dao_children.py", # the DAO itself + "custom_components/nestquest/schema.py", # declares the DDL + "custom_components/nestquest/migrations.py", # applies the DDL + "tests/test_schema.py", # tests the DDL + "tests/test_migrations.py", # tests migration application + "tests/test_dao_children.py", # this file, scanned separately + } + sql_pattern = re.compile( + r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+" + r'["\`]*(children|admin_users)["\`]*\b', + re.IGNORECASE, + ) + offenders: list[str] = [] + for root in scan_roots: + for py in sorted(root.rglob("*.py")): + relative = py.relative_to(repo_root).as_posix() + if relative in allowed: + continue + if sql_pattern.search(py.read_text()): + offenders.append(relative) + assert offenders == [], ( + f"SQL touching children/admin_users leaked into: {offenders}" + ) + + +def test_dao_test_file_uses_dao_not_raw_table_sql() -> None: + """This test module must exercise the DAO, not raw SQL, for these + tables. The two guard functions in this file necessarily mention + the table names inside their regex patterns; their source spans are + removed BEFORE the scan, and the remaining lines are then scanned + as ONE combined blob (not line-by-line), so SQL split across + physical lines — e.g. an implicitly concatenated string ending in + FROM children — is still caught. The guard regexes themselves + cannot match the SQL keyword pattern, so removing the spans only + spares regex text, not executable SQL. + """ + import ast + import re + from pathlib import Path + + path = Path(__file__) + text = path.read_text() + lines = text.splitlines(keepends=True) + tree = ast.parse(text) + guard_names = { + "test_children_and_admin_sql_lives_only_in_dao_module", + "test_dao_test_file_uses_dao_not_raw_table_sql", + } + excluded: set[int] = set() + for node in ast.walk(tree): + if ( + isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and node.name in guard_names + ): + excluded.update( + range(node.lineno, (node.end_lineno or node.lineno) + 1) + ) + + # Scan the source minus excluded lines as ONE blob so SQL split + # across physical lines cannot evade the pattern. + remaining = "".join( + line + for number, line in enumerate(lines, start=1) + if number not in excluded + ) + sql_pattern = re.compile( + r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|" + r"DELETE\s+FROM|FROM|JOIN)\s+[`'\"]*(\[)?" + r"(children|admin_users)\b", + re.IGNORECASE, + ) + assert sql_pattern.search(remaining) is None, ( + "tests must go through the DAO, not raw SQL, for these tables" + )