fix: serialize concurrent migration runners with a per-connection lock

NestQuestDatabase rejects a second transaction while one is active, so
two concurrent apply_migrations callers on one connection crashed with
RuntimeError instead of queueing. apply_migrations now holds a
per-(connection, loop) asyncio lock across the whole decide-and-apply
span; the queued runner re-reads the stamped version and applies
nothing. Deterministic concurrent-runner regression test added.

custom_components/nestquest/dao_children.py

@@ -1,289 +0,0 @@
-"""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

tests/test_dao_children.py

@@ -1,553 +0,0 @@
-"""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"
-    )