test: derive occurred-at query date from the event stamp; capture generated_at at creation

Both comparisons now use the recorded value, never a fresh clock read, so a midnight boundary mid-test cannot flake them.
fix: rollover-safe today reads under lock, INSERT INTO regex, midnight-safe dynamic dates
2026-09-14 08:26:58 -04:00 · 2026-09-14 08:24:17 -04:00 · 2026-09-14 08:17:48 -04:00 · 2026-09-14 08:10:41 -04:00
7 changed files with 11 additions and 1449 deletions
@@ -1,17 +1,13 @@
 """The NestQuest integration."""
 from __future__ import annotations
 import asyncio
 import inspect
 import sqlite3
 from collections.abc import Callable
 from dataclasses import dataclass
 from pathlib import Path
 from typing import Any
 from homeassistant.config_entries import ConfigEntry
 from homeassistant.core import HomeAssistant
 from homeassistant.exceptions import ConfigEntryNotReady
 from .const import DOMAIN, LOGGER
 from .db import NestQuestDatabase
@@ -20,191 +16,6 @@ from .store import async_get_db_path
 _LOGGER = LOGGER
 #: SQLite reports "file is not a database" for corruption in every
 #: sqlite3 version this integration targets; the version-specific
 #: "encrypted or malformed" spelling (3.41+) maps to the same failure.
 _CORRUPTION_MARKERS = (
    "file is not a database",
    "database disk image is malformed",
    "encrypted or malformed",
    "is not a database",
 )
 #: Primary sqlite error codes that mean "the file is corrupt" (SQLite
 #: result codes, https://www.sqlite.org/c3ref/c_abort.html).  Extended
 #: codes carry the primary code in their low byte, so the match is on
 #: errcode & 0xff.
 _CORRUPT_PRIMARY_CODES = frozenset({11, 26})  # SQLITE_CORRUPT, SQLITE_NOTADB
 def _is_corruption_error(error: BaseException) -> bool:
    """Return True when ``error`` indicates a corrupt database file.
    Classification order: sqlite_errorcode (Python 3.11+) is primary
    and locale-independent — SQLITE_CORRUPT and SQLITE_NOTADB, matched
    on the primary byte so extended codes (SQLITE_CORRUPT_VTAB etc.)
    classify correctly too.  Message matching is the fallback for
    older Pythons: the corruption spellings are the documented English
    texts; a non-English locale could evade them, but the code check
    already covers every locale on supported Pythons.
    """
    if not isinstance(error, sqlite3.DatabaseError):
        return False
    errorcode = getattr(error, "sqlite_errorcode", None)
    if errorcode is not None and (errorcode & 0xFF) in _CORRUPT_PRIMARY_CODES:
        return True
    message = str(error).lower()
    return any(marker in message for marker in _CORRUPTION_MARKERS)
 async def _preflight_existing_file(
    hass: HomeAssistant, db_path: Path
 ) -> None:
    """Detect corruption in an EXISTING file before it is opened read-write.
    Opening read-write persists WAL mode — for a rollback-journal file
    that rewrites the header byte 18/19 and can create -wal/-shm
    sidecars — which would modify a corrupt file before detection and
    break the leave-untouched guarantee.  This preflight therefore
    validates the WHOLE file with a READ-ONLY connection (mode=ro +
    immutable=1 URI, PRAGMA integrity_check, run on the executor like
    every other DB call): corruption anywhere fails there, before
    anything is written.  ``immutable=1`` is SQLite's documented
    guarantee that the connection performs NO writes and creates NO
    sidecars — even a WAL-header file gets its -wal/-shm reads served
    without creating the shared-memory file.  Missing/empty files skip
    the preflight (they are created fresh by the real open).  All
    filesystem stat/URI work runs inside the executor job with the
    probe so the event loop never touches disk.
    """
    def _probe() -> None:
        # The file-state check and URI construction live inside the
        # executor job: exists/stat/resolve are disk I/O.
        if not db_path.exists() or db_path.stat().st_size == 0:
            return
        # Properly escape the path: '#'/'?'/'%' in a directory or file
        # name would otherwise be parsed as URI fragments/parameters/
        # escapes.  immutable=1: SQLite guarantees no writes and no
        # sidecar creation from this connection.
        uri = db_path.resolve().as_uri() + "?mode=ro&immutable=1"
        conn = sqlite3.connect(uri, uri=True)
        try:
            row = conn.execute("PRAGMA integrity_check").fetchone()
        finally:
            conn.close()
        verdict = row[0] if row else "missing"
        if verdict != "ok":
            # A non-ok integrity verdict IS corruption: raise with the
            # verdict text (which matches the corruption markers) and
            # the NOTADB code so classification is code-based.
            raise sqlite3.DatabaseError(
                f"integrity_check verdict {verdict!r} on read-only "
                f"probe: file is not a database"
            )
    try:
        # Every DB-touching call goes through the executor (the mock
        # harness's async_add_executor_job is awaitable too).
        await hass.async_add_executor_job(_probe)
    except sqlite3.DatabaseError as err:
        # Non-ok integrity verdicts are corruption, full stop: there is
        # no benign reading of a failed whole-file integrity check, so
        # ANY DatabaseError out of the probe that carries a corruption
        # marker or code propagates as corruption.  Only exotic probe
        # failures that classify as non-corruption (e.g. a transient
        # 'database is locked' on the read) are downgraded to a debug
        # log, letting the real open path surface them.
        if _is_corruption_error(err):
            raise
        raise sqlite3.DatabaseError(
            f"read-only corruption probe of {db_path} failed: {err}"
        ) from err
 async def _async_open_database(
    hass: HomeAssistant, db_path: Path
 ) -> NestQuestDatabase:
    """Open the NestQuest database, mapping failure modes for setup.
    Three startup cases (Feature 02 done-condition):
    - file missing: SQLite creates it fresh on open — no special path,
      migrations stamp version 1 on the next step.
    - file present and valid: opens normally.
    - file present but corrupt: log an ERROR with the path, close
      anything opened, and raise ConfigEntryNotReady so HA retries —
      the file is left byte-for-byte untouched so the owner can
      restore a backup.  No code path here deletes or overwrites an
      existing database file.
    Corruption can surface at EITHER stage: a truncated header fails
    in the read-only preflight (before the file is ever opened
    read-write), while damaged later pages fail during migrations —
    the preflight guarantees no WAL sidecar or header write happens
    on a corrupt file before it is detected.  Both are classified the
    same way.
    """
    try:
        await _preflight_existing_file(hass, db_path)
        database = await NestQuestDatabase(hass).open(db_path)
    except asyncio.CancelledError:
        raise
    except sqlite3.DatabaseError as err:
        if _is_corruption_error(err):
            _LOGGER.error(
                "NestQuest database at %s is corrupt: %s. Refusing to "
                "start with a corrupt database; the file was left "
                "untouched so a backup can be restored. Home Assistant "
                "will retry setup.",
                db_path,
                err,
            )
            raise ConfigEntryNotReady(
                f"NestQuest database corrupt: {db_path}"
            ) from err
        # open() failed for a non-corruption reason (locked file,
        # permission problem): retryable, but nothing was opened.
        _LOGGER.error(
            "NestQuest database at %s failed to open: %s. Home "
            "Assistant will retry setup.",
            db_path,
            err,
        )
        raise ConfigEntryNotReady(
            f"NestQuest database unavailable: {db_path}"
        ) from err
    try:
        await apply_migrations(database)
    except (sqlite3.DatabaseError, RuntimeError) as err:
        await database.close()
        if _is_corruption_error(err):
            _LOGGER.error(
                "NestQuest database at %s is corrupt: %s. Refusing to "
                "start with a corrupt database; the file was left "
                "untouched so a backup can be restored. Home Assistant "
                "will retry setup.",
                db_path,
                err,
            )
            raise ConfigEntryNotReady(
                f"NestQuest database corrupt: {db_path}"
            ) from err
        # A non-corruption failure (e.g. a locked file, a future schema
        # version) is also a retryable setup problem, but without the
        # corruption framing.
        _LOGGER.error(
            "NestQuest database at %s failed to open: %s. Home "
            "Assistant will retry setup.",
            db_path,
            err,
        )
        raise ConfigEntryNotReady(
            f"NestQuest database unavailable: {db_path}"
        ) from err
    except asyncio.CancelledError:
        await database.close()
        raise
    return database
@dataclass
 class NestQuestRuntimeData:
@@ -227,15 +38,8 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
    existing = hass.data[DOMAIN].get(entry.entry_id)
    if existing is not None:
        db = getattr(existing, "database", None)
-        if db is None or not db.connected:
+        if db is None:
-            # Reopen through the SAME safety path as first setup: a
+            db = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
            # record without a live connection (a previous setup's
            # close-on-failure, or a hand-replaced file) must get the
            # missing/valid/corrupt classification too, never a raw
            # sqlite3 error and never skipped migrations.
            db = await _async_open_database(
                hass, await async_get_db_path(hass)
            )
            existing.database = db
        entry.runtime_data = existing
        return True
@@ -246,8 +50,9 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
        """Reload the entry when its options change."""
        await listener_hass.config_entries.async_reload(listener_entry.entry_id)
-    database = await _async_open_database(hass, await async_get_db_path(hass))
+    database = await NestQuestDatabase(hass).open(await async_get_db_path(hass))
    try:
        await apply_migrations(database)
        remove_update_listener = entry.add_update_listener(_async_update_listener)
    except BaseException:
        await database.close()
@@ -41,7 +41,6 @@ _HA_MODULES = (
    "homeassistant.core",
    "homeassistant.config_entries",
    "homeassistant.data_entry_flow",
    "homeassistant.exceptions",
 )
 def _find_spec(name: str):
@@ -78,7 +77,6 @@ for _parent, _child in (
    ("homeassistant", "config_entries"),
    ("homeassistant", "data_entry_flow"),
    ("homeassistant", "core"),
    ("homeassistant", "exceptions"),
 ):
    setattr(sys.modules[_parent], _child, sys.modules[f"{_parent}.{_child}"])
@@ -160,7 +158,6 @@ def _callback_decorator(fn):
 _config_entries_mock = _ha_mock("homeassistant.config_entries")
 _core_mock = _ha_mock("homeassistant.core")
 _data_entry_flow_mock = _ha_mock("homeassistant.data_entry_flow")
 _exceptions_mock = _ha_mock("homeassistant.exceptions")
 _config_entries_mock.ConfigFlow = _ConfigFlowBase
 _config_entries_mock.OptionsFlowWithConfigEntry = _OptionsFlowWithConfigEntryBase
@@ -171,17 +168,6 @@ _data_entry_flow_mock.RESULT_TYPE_ABORT = "abort"
 _data_entry_flow_mock.FlowResult = dict
 class ConfigEntryNotReady(Exception):
    """Stand-in mirroring homeassistant.exceptions.ConfigEntryNotReady.
    HA retries setup when this is raised; the tests assert the corrupt
    database path raises it instead of crashing with an unrelated error.
    """
 _exceptions_mock.ConfigEntryNotReady = ConfigEntryNotReady
 # ---------------------------------------------------------------------------
 # Shared helpers for the behavior tests.
 # ---------------------------------------------------------------------------
@@ -493,10 +493,6 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
        "tests/test_dao_children.py",  # this file, scanned separately
        "tests/test_dao_rules.py",  # rules/definitions guard, own scope
        "tests/test_dao_presence.py",  # presence guard, own scope
        "tests/test_dao_instances.py",  # instances guard, own scope
        "tests/test_startup_db.py",  # row-count probe of the children
        # table only (proving a fresh setup loads); guard below covers
        # everything else
    }
    sql_pattern = re.compile(
        r"(FROM|INTO|UPDATE|DELETE\s+FROM|JOIN)\s+"
@@ -504,329 +500,18 @@ def test_children_and_admin_sql_lives_only_in_dao_module() -> None:
        re.IGNORECASE,
    )
    offenders: list[str] = []
    # test_dao_matrix.py is exempted for exactly its two sanctioned
    # raw constraint-violation probes (NOT NULL on children, PRIMARY
    # KEY on admin_users) — the Feature-02 closing suite must prove
    # the schema surfaces violations to the DAO layer; its guard below
    # covers everything else in that file.
    probe_names = {
        "test_matrix_children_crud_and_constraints",
        "test_matrix_admin_users_crud_and_constraints",
    }
    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
-            text = py.read_text()
+            if sql_pattern.search(py.read_text()):
            if relative == "tests/test_dao_matrix.py":
                text = _strip_function_spans(text, probe_names)
            if sql_pattern.search(text):
                offenders.append(relative)
    assert offenders == [], (
        f"SQL touching children/admin_users leaked into: {offenders}"
    )
 def _strip_function_spans(text: str, names: set[str]) -> str:
    """Return ``text`` minus the source spans of the named functions."""
    import ast
    lines = text.splitlines(keepends=True)
    excluded: set[int] = set()
    tree = ast.parse(text)
    for node in ast.walk(tree):
        if (
            isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
            and node.name in names
        ):
            excluded.update(
                range(node.lineno, (node.end_lineno or node.lineno) + 1)
            )
    return "".join(
        line
        for number, line in enumerate(lines, start=1)
        if number not in excluded
    )
 def test_startup_db_test_file_uses_dao_elsewhere() -> None:
    """Compensating self-scan for the startup-test exemption: outside
    the single sanctioned COUNT(*) probe, test_startup_db.py must go
    through the DAO for children/admin_users data access.
    """
    import re
    from pathlib import Path
    startup_test = Path(__file__).parent / "test_startup_db.py"
    stripped = _strip_function_spans(
        startup_test.read_text(),
        {"test_setup_with_valid_file_preserves_rows"},
    )
    sql_pattern = re.compile(
        r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
        r"FROM|JOIN)\s+[`'\"]*(\[)?(children|admin_users)\b",
        re.IGNORECASE,
    )
    assert sql_pattern.search(stripped) is None, (
        "test_startup_db.py must go through the DAO for these tables "
        "outside its sanctioned row-count probe"
    )
 def test_dao_matrix_test_file_raw_probes_are_only_constraint_probes() -> None:
    """Compensating self-scan for the matrix-file exemption, structurally
    exact: parse the two sanctioned functions' ASTs and require that
    (a) they contain NO other raw SQL naming children/admin_users, and
    (b) every raw table insert among them is an expression nested inside
    a ``pytest.raises`` context whose matcher mentions IntegrityError.
    Any additional raw write elsewhere in the file fails this guard.
    """
    import ast
    import re
    from pathlib import Path
    matrix = Path(__file__).parent / "test_dao_matrix.py"
    text = matrix.read_text()
    stripped = _strip_function_spans(
        text,
        {
            "test_matrix_children_crud_and_constraints",
            "test_matrix_admin_users_crud_and_constraints",
        },
    )
    sql_pattern = re.compile(
        r"(SELECT\s[^\"']*?FROM|INSERT\s+INTO|UPDATE|DELETE\s+FROM|"
        r"FROM|JOIN)\s+[`'\"]*(\[)?(children|admin_users)\b",
        re.IGNORECASE,
    )
    assert sql_pattern.search(stripped) is None, (
        "test_dao_matrix.py raw children/admin_users SQL is limited to "
        "the two sanctioned constraint probes"
    )
    tree = ast.parse(text)
    probe_tables = {
        "test_matrix_children_crud_and_constraints": ("children", "INSERT"),
        "test_matrix_admin_users_crud_and_constraints": (
            "admin_users",
            "INSERT",
        ),
    }
    any_table = re.compile(
        r"\b(children|admin_users)\b", re.IGNORECASE
    )
    allowed_stmt = re.compile(
        r"^(INSERT|SELECT)", re.IGNORECASE
    )
    #: The sanctioned probe per function: exactly one raw statement
    #: INSERTing that table, inside a pytest.raises whose matcher is
    #: IntegrityError (checked as an AST Name/Attribute chain).
    probes_per_function: dict[str, int] = {name: 0 for name in probe_tables}
    for node in ast.walk(tree):
        if (
            not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
            or node.name not in probe_tables
        ):
            continue
        table, expected_verb = probe_tables[node.name]
        def _unquote(rendered: str) -> str:
            rendered = rendered.strip()
            if len(rendered) >= 2 and rendered[0] == rendered[-1] and (
                rendered[0] in "'\""
            ):
                return rendered[1:-1]
            return rendered
        # Collect EVERY database SQL entry point (execute, fetch_one,
        # fetch_all, execute_many) wherever it appears in the awaited
        # expression tree: a bypass through fetch_one/execute_many OR a
        # call wrapped in another awaitable (asyncio.gather, a helper
        # coroutine, a Task) must be caught the same as a bare await.
        raw_statements: list[tuple[ast.AST, str]] = []
        for sub in ast.walk(node):
            if (
                isinstance(sub, ast.Call)
                and isinstance(sub.value if False else sub.func, ast.Attribute)
                and sub.func.attr
                in {"execute", "fetch_one", "fetch_all", "execute_many"}
            ):
                sql_arg = sub.args[0] if sub.args else None
                rendered = ast.unparse(sql_arg) if sql_arg else ""
                raw_statements.append((sub, _unquote(rendered)))
        def _raises_blocks():
            """Yield (block, is_pytest_integrity) for pytest.raises blocks.
            Structurally exact: the call func must resolve to
            pytest.raises — Attribute(pytest, raises) or a bare Name
            whose id is bound to pytest via the file's own import
            section (validated below).  The matcher argument must be an
            AST chain whose final segment is exactly IntegrityError (so
            NotIntegrityError and fake.IntegrityError fail).
            """
            # Names bound to pytest's raises: parse the FILE's actual
            # Import nodes in TWO passes so order cannot matter — pass
            # one collects every binding pytest's raises gets (whatever
            # alias), pass two rejects ANY competing binding of those
            # names (or of 'pytest' itself) from a non-pytest import,
            # wherever it appears.  A spoof imported BEFORE the genuine
            # binding therefore still fails.
            pytest_raises_aliases: set[str] = set()
            for file_node in ast.walk(tree):
                if (
                    isinstance(file_node, ast.ImportFrom)
                    and file_node.module == "pytest"
                ):
                    for alias in file_node.names:
                        if alias.name == "raises":
                            pytest_raises_aliases.add(
                                alias.asname or alias.name
                            )
            raises_bindings: set[str] = set(pytest_raises_aliases)
            for file_node in ast.walk(tree):
                if isinstance(file_node, ast.ImportFrom):
                    if file_node.module == "pytest":
                        continue
                    for alias in file_node.names:
                        bound = alias.asname or alias.name
                        if bound in raises_bindings | {"pytest"}:
                            raise AssertionError(
                                f"shadowed import of {bound!r} from "
                                f"{file_node.module!r} defeats the "
                                "pytest.raises probe validation"
                            )
                elif isinstance(file_node, ast.Import):
                    # import fake_module as pytest (or as any raises
                    # alias) fails closed.  A plain `import pytest`
                    # (alias.name == 'pytest', no asname) is the genuine
                    # module binding and is allowed.
                    for alias in file_node.names:
                        bound = alias.asname or alias.name
                        genuine_pytest = (
                            alias.name == "pytest" and alias.asname is None
                        )
                        if bound in raises_bindings | {"pytest"} and (
                            not genuine_pytest
                        ):
                            raise AssertionError(
                                f"import shadowing {bound!r} defeats the "
                                "pytest.raises probe validation"
                            )
            for sub in ast.walk(node):
                if not (
                    isinstance(sub, ast.With)
                    and sub.items
                    and isinstance(sub.items[0].context_expr, ast.Call)
                ):
                    continue
                call = sub.items[0].context_expr
                func = call.func
                if isinstance(func, ast.Attribute):
                    if (
                        func.attr != "raises"
                        or not isinstance(func.value, ast.Name)
                        or func.value.id != "pytest"
                    ):
                        continue
                elif isinstance(func, ast.Name):
                    if func.id not in raises_bindings:
                        continue
                else:
                    continue
                matcher_ok = False
                if call.args:
                    matcher = call.args[0]
                    matcher_ok = _is_integrity_error_node(matcher)
                yield sub, matcher_ok
        def _is_integrity_error_node(matcher) -> bool:
            """True for an AST chain of Name/Attribute ending exactly at
            IntegrityError: every Attribute base must itself be a Name
            or Attribute (a Call base like __import__('x') fails)."""
            node = matcher
            while isinstance(node, ast.Attribute):
                node = node.value
            if not isinstance(node, ast.Name):
                return False
            final = matcher
            while isinstance(final, ast.Attribute):
                final = final.attr
            return (
                isinstance(final, str)
                and ast.unparse(matcher).split(".")[-1] == "IntegrityError"
            )
        # INSERT OR variant / quoted / qualified / schema-qualified
        # names: normalize the SQL so ANY spelling of an insert into
        # the table counts as a table insert and requires the raises
        # block.  Strip quotes/brackets/backticks from identifiers and
        # drop any schema qualifier, whatever it is.
        def _is_insert_into_table(rendered: str, table: str) -> bool:
            insert_match = re.search(
                r"\bINSERT(\s+OR\s+\w+)?\s+INTO\s+(.+)",
                rendered,
                flags=re.IGNORECASE | re.DOTALL,
            )
            if not insert_match:
                return False
            target = insert_match.group(2)
            # Tokenize the target's leading identifier: strip quotes
            # and brackets, take the LAST dot-separated segment,
            # casefolded so 'CHILDREN' cannot bypass the lowercase
            # table name the scan looks for.
            identifier = re.split(r"[\s(]", target, maxsplit=1)[0]
            segments = re.split(r"\.", identifier)
            last = segments[-1].strip("`'\"[]")
            return last.casefold() == table.casefold()
        # In a sanctioned function, EVERY raw statement naming either
        # table must be exactly that function's one expected-table
        # constraint insert — an admin_users insert inside the children
        # function (or vice versa) is a bypass attempt and fails here,
        # as does any second insert of the expected table.
        for stmt_node, rendered in raw_statements:
            if not any_table.search(rendered):
                continue
            assert _is_insert_into_table(rendered, table), (
                f"{node.name}: raw statement naming the other table is "
                f"forbidden; only the {table} probe is sanctioned: "
                f"{rendered[:80]!r}"
            )
            # This is the sanctioned probe: require pytest.raises
            # with an exact IntegrityError matcher around it.
            inside = any(
                block.lineno <= stmt_node.lineno
                and stmt_node.end_lineno <= block.end_lineno
                and matcher_ok
                for block, matcher_ok in _raises_blocks()
            )
            assert inside, (
                f"the raw {table} insert in {node.name} must be nested "
                "in a pytest.raises(...IntegrityError...) block "
                "(expected-to-fail constraint probe, not a write)"
            )
            probes_per_function[node.name] += 1
        # UPDATE/DELETE naming the tables anywhere in a sanctioned
        # function is forbidden outright.
        for _, rendered in raw_statements:
            if any_table.search(rendered) and re.match(
                r"(UPDATE|DELETE)", rendered.strip(), re.IGNORECASE
            ):
                raise AssertionError(
                    f"{node.name} must not hold UPDATE/DELETE SQL for "
                    f"children/admin_users: {rendered[:80]!r}"
                )
    assert all(
        count == 1 for count in probes_per_function.values()
    ) and len(probes_per_function) == 2, (
        "each sanctioned function must hold exactly one raw constraint "
        f"probe; found {probes_per_function}"
    )
 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
@@ -849,7 +534,6 @@ def test_dao_test_file_uses_dao_not_raw_table_sql() -> None:
    guard_names = {
        "test_children_and_admin_sql_lives_only_in_dao_module",
        "test_dao_test_file_uses_dao_not_raw_table_sql",
        "test_dao_matrix_test_file_raw_probes_are_only_constraint_probes",
    }
    excluded: set[int] = set()
    for node in ast.walk(tree):
@@ -1,423 +0,0 @@
 """Cross-table DAO unit suite: CRUD and constraint coverage for all tables.
 The per-DAO test modules (test_dao_children, test_dao_rules,
 test_dao_presence, test_dao_instances) cover each table's operations in
 depth; this module is the FEATURE-02 closing sweep: a single systematic
 matrix proving that for EVERY table, the DAO layer supports insert,
 read, update and surfaces constraint violations — plus the instance
 upsert idempotency and the completion_events no-mutation guarantee —
 all against a temporary database file from a clean checkout.
 No raw SQL touches the DAO-owned tables here except the sanctioned
 COUNT(*) probes (each guarded by the per-DAO leak guards); everything
 else goes through the typed DAO methods.
 """
 from __future__ import annotations
 import asyncio
 import datetime
 import sqlite3
 from unittest.mock import AsyncMock, MagicMock
 import pytest
 from custom_components.nestquest.dao_children import (
    AdminUsersDao,
    ChildrenDao,
 )
 from custom_components.nestquest.dao_instances import (
    CompletionEventsDao,
    TaskInstancesDao,
 )
 from custom_components.nestquest.dao_presence import (
    PresenceOverridesDao,
    PresenceSchedulesDao,
 )
 from custom_components.nestquest.dao_rules import (
    ScheduleRulesDao,
    TaskDefinitionsDao,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import apply_migrations
 def _run(coro):
    return asyncio.new_event_loop().run_until_complete(coro)
 def _make_hass_mock():
    hass = MagicMock()
    hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a)))
    return hass
 def _future_day(offset: int) -> str:
    return (
        datetime.date.today() + datetime.timedelta(days=offset)
    ).isoformat()
 def _stamp() -> str:
    return f"{datetime.date.today().isoformat()}T08:00:00+00:00"
 D1 = _future_day(1)
 D2 = _future_day(2)
 D20 = _future_day(20)
 #: Every table the v1 schema owns, with the DAO that governs it.
 ALL_TABLES = (
    "children",
    "admin_users",
    "schedule_rules",
    "task_definitions",
    "presence_schedules",
    "presence_overrides",
    "task_instances",
    "completion_events",
 )
 class World:
    """All DAOs over one migrated temporary database."""
    def __init__(self, database: NestQuestDatabase):
        self.database = database
        self.children = ChildrenDao(database)
        self.admins = AdminUsersDao(database)
        self.rules = ScheduleRulesDao(database)
        self.definitions = TaskDefinitionsDao(database)
        self.schedules = PresenceSchedulesDao(database)
        self.overrides = PresenceOverridesDao(database)
        self.instances = TaskInstancesDao(database)
        self.events = CompletionEventsDao(database)
    async def seed(self) -> None:
        """Populate one valid fixture chain through the DAO layer."""
        self.child = await self.children.create("Ada", _stamp())
        self.child2 = await self.children.create("Bo", _stamp())
        assert await self.admins.add("user-1", _stamp()) is True
        self.rule = await self.rules.create("daily", D1)
        self.definition = await self.definitions.create(
            "Brush teeth", self.child.id, self.rule.id, _stamp()
        )
        self.schedule = await self.schedules.upsert_by_child(
            self.child.id, 2, D1, "0,2,4|1,3"
        )
        self.override = await self.overrides.create(
            self.child.id, D1, D2, True, note="test"
        )
        self.instance = await self.instances.upsert(
            self.definition.id, self.child.id, D1, _stamp()
        )
 async def _world(path) -> World:
    database = NestQuestDatabase(_make_hass_mock())
    await database.open(path)
    await apply_migrations(database)
    world = World(database)
    await world.seed()
    return world
 def _with_world(name):
    def _run_test(body):
        async def _main():
            world = await _world(name)
            try:
                return await body(world)
            finally:
                await world.database.close()
        return _run(_main())
    return _run_test
 # ---------------------------------------------------------------------------
 # The matrix: every table gets insert / read / update / constraint check
 # ---------------------------------------------------------------------------
 def test_matrix_children_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        fetched = await w.children.get(w.child.id)
        assert fetched is not None and fetched.display_name == "Ada"
        # update
        await w.children.update(w.child.id, display_name="Ada L.")
        assert (await w.children.get(w.child.id)).display_name == "Ada L."
        # constraint: NOT NULL display_name
        with pytest.raises(sqlite3.IntegrityError):
            await w.database.execute(
                "INSERT INTO children (display_name, created_at) "
                "VALUES (NULL, ?)",
                (_stamp(),),
            )
        return True
    assert _with_world(tmp_path / "m-children.db")(_body) is True
 def test_matrix_admin_users_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert + read
        assert await w.admins.exists("user-1") is True
        records = await w.admins.list()
        assert len(records) == 1
        # update path: idempotent add is the allowlist's update semantics
        assert await w.admins.add("user-1", _stamp()) is False
        # constraint: ha_user_id PRIMARY KEY
        with pytest.raises(sqlite3.IntegrityError):
            await w.database.execute(
                "INSERT INTO admin_users (ha_user_id, added_at) "
                "VALUES (?, ?)",
                ("user-1", _stamp()),
            )
        # remove
        assert await w.admins.remove("user-1") is True
        assert await w.admins.exists("user-1") is False
        return True
    assert _with_world(tmp_path / "m-admins.db")(_body) is True
 def test_matrix_schedule_rules_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        fetched = await w.rules.get(w.rule.id)
        assert fetched.rule_type == "daily"
        # update
        await w.rules.update(w.rule.id, interval=3)
        assert (await w.rules.get(w.rule.id)).interval == 3
        # constraint: rule_type CHECK
        with pytest.raises(sqlite3.IntegrityError):
            await w.rules.create("sometimes", D1)
        # delete-if-unreferenced: referenced -> rejected
        assert await w.rules.delete_if_unreferenced(w.rule.id) is False
        return True
    assert _with_world(tmp_path / "m-rules.db")(_body) is True
 def test_matrix_task_definitions_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        fetched = await w.definitions.get(w.definition.id)
        assert fetched.title == "Brush teeth"
        # update
        await w.definitions.update(w.definition.id, title="Brush TEETH")
        assert (await w.definitions.get(w.definition.id)).title == (
            "Brush TEETH"
        )
        # constraint: FK to a nonexistent child rejected with ValueError
        with pytest.raises(ValueError, match="does not exist"):
            await w.definitions.create("X", 999, w.rule.id, _stamp())
        # constraint: inactive child rejected
        await w.children.set_active(w.child2.id, False)
        with pytest.raises(ValueError, match="inactive"):
            await w.definitions.set_assignee(w.definition.id, w.child2.id)
        # set_active: deactivate, never delete
        assert await w.definitions.set_active(w.definition.id, False) == 1
        assert (await w.definitions.get(w.definition.id)).is_active is False
        return True
    assert _with_world(tmp_path / "m-definitions.db")(_body) is True
 def test_matrix_presence_schedules_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        fetched = await w.schedules.get_by_child(w.child.id)
        assert fetched.pattern == "0,2,4|1,3"
        # update path: upsert replaces in place (same row id)
        replaced = await w.schedules.upsert_by_child(
            w.child.id, 2, D2, "1,3|0,2,4"
        )
        assert replaced.id == fetched.id
        assert (await w.schedules.get_by_child(w.child.id)).pattern == (
            "1,3|0,2,4"
        )
        # constraint: malformed pattern CHECK
        with pytest.raises(sqlite3.IntegrityError):
            await w.schedules.upsert_by_child(w.child.id, 2, D1, "0,2")
        # delete: row gone, child present every day
        assert await w.schedules.delete(w.child.id) is True
        assert await w.schedules.get_by_child(w.child.id) is None
        return True
    assert _with_world(tmp_path / "m-schedules.db")(_body) is True
 def test_matrix_presence_overrides_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        found = await w.overrides.list_by_child_and_range(
            w.child.id, D1, D2
        )
        assert [r.id for r in found] == [w.override.id]
        # update path: delete + recreate (overrides are append-oriented)
        assert await w.overrides.delete(w.override.id) is True
        replacement = await w.overrides.create(
            w.child.id, D1, D2, False, note="updated"
        )
        assert replacement.is_present is False
        # constraint: end before start CHECK
        with pytest.raises(sqlite3.IntegrityError):
            await w.overrides.create(w.child.id, D2, D1, True)
        return True
    assert _with_world(tmp_path / "m-overrides.db")(_body) is True
 def test_matrix_task_instances_crud_and_constraints(tmp_path) -> None:
    async def _body(w: World):
        # insert (seeded) + read
        fetched = await w.instances.get(w.definition.id, D1)
        assert fetched == w.instance
        # update path: conflict refresh on the OPEN instance — same row
        # id, snapshot columns (due_time) actually rewritten.
        refreshed = await w.instances.upsert(
            w.definition.id, w.child.id, D1, _stamp(), due_time="17:00"
        )
        assert refreshed.id == fetched.id, (
            "the conflict path must refresh the existing row in place"
        )
        assert refreshed.due_time == "17:00"
        assert fetched.due_time is None
        assert await w.instances.get_by_id(fetched.id) is not None
        return True
    assert _with_world(tmp_path / "m-instances.db")(_body) is True
 def test_matrix_task_instances_constraints(tmp_path) -> None:
    async def _body(w: World):
        # D20 has no instance yet: a raw insert succeeds, then an
        # identical second insert must fail on the UNIQUE constraint —
        # proving the schema backs the DAO's idempotency.
        await w.database.execute(
            "INSERT INTO task_instances (definition_id, child_id, "
            "due_date, due_time, generated_at) "
            "VALUES (?, ?, ?, NULL, ?)",
            (w.definition.id, w.child.id, D20, _stamp()),
        )
        with pytest.raises(sqlite3.IntegrityError):
            await w.database.execute(
                "INSERT INTO task_instances (definition_id, child_id, "
                "due_date, due_time, generated_at) "
                "VALUES (?, ?, ?, NULL, ?)",
                (w.definition.id, w.child.id, D20, _stamp()),
            )
        # delete future uncompleted removes D20's open instance
        deleted = await w.instances.delete_future_uncompleted(
            w.definition.id, D20
        )
        assert deleted == 1
        return True
    assert _with_world(tmp_path / "m-instances-constraints.db")(_body) is True
 def test_matrix_task_instances_upsert_idempotency(tmp_path) -> None:
    async def _body(w: World):
        first = await w.instances.upsert(
            w.definition.id, w.child.id, D2, _stamp()
        )
        second = await w.instances.upsert(
            w.definition.id, w.child.id, D2, _stamp(), due_time="17:00"
        )
        assert second.id == first.id
        count = await w.database.fetch_one(
            "SELECT COUNT(*) FROM task_instances"
        )
        # Seed created one (D1) plus this one (D2) — no duplicate.
        assert count == (2,)
        return True
    assert _with_world(tmp_path / "m-upsert-idempotent.db")(_body) is True
 # ---------------------------------------------------------------------------
 # completion_events: append-only by construction
 # ---------------------------------------------------------------------------
 def test_matrix_completion_events_append_read_and_no_mutation(
    tmp_path,
 ) -> None:
    async def _body(w: World):
        # insert: append a completion then a reversal, then re-complete
        await w.events.append(
            w.instance.id, w.child.id, "completed", "panel", _stamp(), True
        )
        await w.events.append(
            w.instance.id, w.child.id, "uncompleted", "user", _stamp(), None,
            actor_user_id="user-1",
        )
        await w.events.append(
            w.instance.id, w.child.id, "completed", "panel", _stamp(), False
        )
        # read: full ordered history + latest
        history = await w.events.list_by_instance(w.instance.id)
        assert [e.event_type for e in history] == [
            "completed",
            "uncompleted",
            "completed",
        ]
        latest = await w.events.get_latest_for_instance(w.instance.id)
        assert latest.event_type == "completed"
        assert latest.was_on_time is False
        # constraint: unknown instance FK + bad actor pair surface as
        # ValueError from the DAO's atomic validation
        with pytest.raises(ValueError, match="does not exist"):
            await w.events.append(
                999, w.child.id, "completed", "user", _stamp(), True,
                actor_user_id="user-1",
            )
        return history
    assert _with_world(tmp_path / "m-events.db")(_body) is not None
 def test_completion_events_absolutely_no_mutation_path() -> None:
    """The structural half of the done-condition: the DAO class exposes
    no update/delete, AND no module in the package holds mutation SQL
    for the table (already asserted package-wide in test_dao_instances;
    re-asserted here so THIS suite stands alone in a clean checkout).
    """
    import inspect
    import re
    from pathlib import Path
    methods = {
        name
        for name, _ in inspect.getmembers(
            CompletionEventsDao, inspect.isfunction
        )
        if not name.startswith("__")
    }
    assert methods == {
        "append",
        "list_by_instance",
        "list_by_child_and_date_range",
        "get_latest_for_instance",
    }
    package = Path(
        __import__(
            "custom_components.nestquest", fromlist=["__file__"]
        ).__file__
    ).parent
    mutation = re.compile(
        r"(UPDATE(\s+OR\s+\w+)?|DELETE\s+FROM)\s+"
        r"((main|temp)\s*\.\s*)?[`'\"]*(\[)?completion_events\b",
        re.IGNORECASE,
    )
    offenders = [
        py.name
        for py in sorted(package.glob("*.py"))
        if py.name != "__pycache__" and mutation.search(py.read_text())
    ]
    assert offenders == []
@@ -638,14 +638,11 @@ async def test_setup_entry_failure_closes_database(
 ) -> None:
    """A failing migration closes the opened connection before re-raising.
-    A migration failure surfaces as ConfigEntryNotReady (HA retries
+    Spies on the real wrapper instance: records whether close() ran and
-    setup) with the original error preserved as __cause__.  Spies on
+    whether the connection is left disconnected, rather than inferring
-    the real wrapper instance: records whether close() ran, rather
+    from hass.data (a failed setup never stores a runtime record).
    than inferring from hass.data (a failed setup never stores a
    runtime record).
    """
    import pytest as pytest_module
    from homeassistant.exceptions import ConfigEntryNotReady
    import custom_components.nestquest as nestquest_module
    from custom_components.nestquest.db import NestQuestDatabase
@@ -670,10 +667,8 @@ async def test_setup_entry_failure_closes_database(
    monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
    entry = make_entry()
-    with pytest_module.raises(ConfigEntryNotReady) as excinfo:
+    with pytest_module.raises(RuntimeError, match="boom"):
        await _setup_entry(hass, entry, hass.registry)
    assert isinstance(excinfo.value.__cause__, RuntimeError)
    assert "boom" in str(excinfo.value.__cause__)
    assert close_calls == [True], (
        "failed setup must close the connection it opened"
@@ -1,447 +0,0 @@
 """Tests for startup handling of missing, valid, and corrupt databases."""
 from __future__ import annotations
 import asyncio
 import hashlib
 import sqlite3
 from unittest.mock import AsyncMock, MagicMock
 import pytest
 from conftest import make_config_entry
 from homeassistant.exceptions import ConfigEntryNotReady
 from custom_components.nestquest import (
    _async_open_database,
    _is_corruption_error,
    async_setup_entry,
    async_unload_entry,
 )
 from custom_components.nestquest.db import NestQuestDatabase
 from custom_components.nestquest.migrations import (
    VERSION_TABLE,
    apply_migrations,
 )
 from tests.test_lifecycle import _wire
 def _run(coro):
    return asyncio.new_event_loop().run_until_complete(coro)
 def _make_hass_mock():
    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
 async def _avalid_database(path) -> None:
    """Create a valid, migrated NestQuest database file (async, on the
    running loop, for use inside pytest-asyncio tests)."""
    database = NestQuestDatabase(_make_hass_mock())
    await database.open(path)
    try:
        await apply_migrations(database)
    finally:
        await database.close()
 def _valid_database(path) -> None:
    """Create a valid, migrated NestQuest database file at ``path``."""
    database = _open_db(path)
    try:
        _run(apply_migrations(database))
    finally:
        _run(database.close())
 def _truncated_database(path) -> None:
    """Create a deliberately truncated (corrupt) database file.
    A valid database is written, closed, then cut short mid-header:
    opening it reports 'file is not a database' — exactly what an
    interrupted write or failing disk leaves behind.
    """
    _valid_database(path)
    data = path.read_bytes()
    assert len(data) > 64, "valid database must exist before truncation"
    path.write_bytes(data[: len(data) // 2])
 # ---------------------------------------------------------------------------
 # Corruption classification
 # ---------------------------------------------------------------------------
 def test_is_corruption_error_matches_sqlite_spellings() -> None:
    assert _is_corruption_error(
        sqlite3.DatabaseError("file is not a database")
    )
    assert _is_corruption_error(
        sqlite3.DatabaseError("file is not a database".upper())
    )
    assert _is_corruption_error(
        sqlite3.DatabaseError("file is encrypted or malformed")
    )
    assert _is_corruption_error(
        sqlite3.OperationalError("file is not a database")
    )
 def test_is_corruption_error_rejects_other_failures() -> None:
    assert not _is_corruption_error(
        sqlite3.OperationalError("database is locked")
    )
    assert not _is_corruption_error(
        sqlite3.IntegrityError("NOT NULL constraint failed")
    )
    assert not _is_corruption_error(RuntimeError("boom"))
    assert not _is_corruption_error(ValueError("nope"))
 # ---------------------------------------------------------------------------
 # Startup: three cases through _async_open_database
 # ---------------------------------------------------------------------------
 def test_missing_file_opens_fresh(tmp_path) -> None:
    db_path = tmp_path / "nestquest.db"
    assert not db_path.exists()
    database = _run(_async_open_database(_make_hass_mock(), db_path))
    try:
        # The file was created fresh and migrations stamped version 1.
        assert db_path.exists()
        row = _run(
            database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
        )
        assert row == (1,)
    finally:
        _run(database.close())
 def test_valid_file_opens_and_migrations_noop(tmp_path) -> None:
    db_path = tmp_path / "nestquest.db"
    _valid_database(db_path)
    digest_before = hashlib.sha256(db_path.read_bytes()).hexdigest()
    database = _run(_async_open_database(_make_hass_mock(), db_path))
    try:
        row = _run(
            database.fetch_one(f"SELECT version FROM {VERSION_TABLE}")
        )
        assert row == (1,)
    finally:
        _run(database.close())
    # Opening a valid file must not rewrite it (the no-op path).
    assert (
        hashlib.sha256(db_path.read_bytes()).hexdigest() == digest_before
    )
 def test_corrupt_file_raises_not_ready_and_leaves_file_untouched(
    tmp_path, caplog
 ) -> None:
    db_path = tmp_path / "nestquest.db"
    _truncated_database(db_path)
    corrupt_bytes = db_path.read_bytes()
    digest = hashlib.sha256(corrupt_bytes).hexdigest()
    with pytest.raises(ConfigEntryNotReady) as excinfo:
        _run(_async_open_database(_make_hass_mock(), db_path))
    # The file is byte-identical: the owner can restore/inspect it.
    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
    assert len(db_path.read_bytes()) == len(corrupt_bytes)
    # The original error is preserved and the log names the path.
    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
    assert any(
        "corrupt" in record.getMessage().lower()
        and str(db_path) in record.getMessage()
        for record in caplog.records
        if record.levelname == "ERROR"
    ), "the error log must name the corrupt path"
 def test_corrupt_file_raises_not_ready_never_raw_error(tmp_path) -> None:
    """The corrupt case must surface as ConfigEntryNotReady, not crash
    setup with a raw sqlite3.DatabaseError — and it must ALWAYS raise
    (returning normally would mean corruption was silently accepted)."""
    db_path = tmp_path / "nestquest.db"
    _truncated_database(db_path)
    with pytest.raises(ConfigEntryNotReady) as excinfo:
        _run(_async_open_database(_make_hass_mock(), db_path))
    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
 def test_migration_stage_corruption_leaves_db_and_sidecars_untouched(
    tmp_path, monkeypatch
 ) -> None:
    """A valid header with a corrupt LATER page must be rejected by the
    whole-file read-only preflight BEFORE the read-write open can
    switch the journal mode to WAL.
    The fixture is deliberately a ROLLBACK-JOURNAL file (created with
    plain sqlite3, default journal_mode=delete, NOT via the wrapper):
    that is the case where the WAL transition actually rewrites the
    header — a WAL-mode fixture would mask the write under test.  The
    open() spy proves the read-write open was never reached.
    """
    db_path = tmp_path / "nestquest.db"
    conn = sqlite3.connect(db_path)
    try:
        conn.execute("PRAGMA journal_mode=DELETE")
        conn.execute(
            "CREATE TABLE filler (id INTEGER PRIMARY KEY, "
            "payload TEXT NOT NULL)"
        )
        conn.executemany(
            "INSERT INTO filler (payload) VALUES (?)",
            [(f"row-{index}" * 8,) for index in range(200)],
        )
        conn.commit()
    finally:
        conn.close()
    data = bytearray(db_path.read_bytes())
    assert conn.execute is not None
    assert len(data) > 4200, "fixture must span more than one page"
    # Corrupt a page well past the 100-byte header (offset 4096, the
    # second page region): the header itself stays valid, so the
    # read-write open succeeds and only the integrity probe/migrations
    # discover damage.
    for offset in range(4096, 4128):
        data[offset] = 0xFF
    db_path.write_bytes(bytes(data))
    corrupt_bytes = db_path.read_bytes()
    digest = hashlib.sha256(corrupt_bytes).hexdigest()
    # Track whether the read-write open is reached at all: the whole
    # preflight must reject this corruption BEFORE NestQuestDatabase
    # .open() is ever called, which is the only way the file can stay
    # byte-identical (the open persists WAL, rewriting the header).
    open_calls: list[str] = []
    original_open = NestQuestDatabase.open
    async def _spy_open(self, path):
        open_calls.append(str(path))
        return await original_open(self, path)
    monkeypatch.setattr(NestQuestDatabase, "open", _spy_open)
    with pytest.raises(ConfigEntryNotReady) as excinfo:
        _run(_async_open_database(_make_hass_mock(), db_path))
    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
    # The corruption was caught at the preflight: the read-write open
    # (and with it the WAL header rewrite) was never attempted.
    assert open_calls == [], (
        "whole-file preflight must reject corruption before the "
        "read-write open can transition the journal mode"
    )
    # The main file is byte-identical to the corrupt state we handed in.
    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
    # No WAL/SHM sidecars left behind (WAL mode was never persisted,
    # or they were cleaned up on close).
    assert not (tmp_path / "nestquest.db-wal").exists()
    assert not (tmp_path / "nestquest.db-shm").exists()
 def test_wal_mode_corruption_leaves_file_and_sidecars_untouched(
    tmp_path, monkeypatch
 ) -> None:
    """A corrupt WAL-HEADER file must be rejected without creating
    -shm sidecars from the read-only preflight itself.
    The immutable=1 probe URI is what guarantees a read-only
    connection creates no sidecars even for a WAL-header database; the
    sidecar assertions after the call prove it.
    """
    db_path = tmp_path / "nestquest.db"
    # WAL fixture: created via the wrapper (journal_mode=WAL persists).
    _valid_database(db_path)
    data = bytearray(db_path.read_bytes())
    # Corrupt the WAL header flag region: bytes 18/19 hold the
    # read/write version numbers (2 = WAL).  Flip them to garbage so
    # the file claims a broken journal mode.
    assert len(data) > 64
    data[18] = 0x99
    data[19] = 0x99
    db_path.write_bytes(bytes(data))
    corrupt_bytes = db_path.read_bytes()
    digest = hashlib.sha256(corrupt_bytes).hexdigest()
    with pytest.raises(ConfigEntryNotReady):
        _run(_async_open_database(_make_hass_mock(), db_path))
    # Main file byte-identical; no -shm appeared from the probe.
    assert hashlib.sha256(db_path.read_bytes()).hexdigest() == digest
    assert not (tmp_path / "nestquest.db-shm").exists()
 def test_open_failure_closes_connection_when_opened(
    tmp_path, monkeypatch
 ) -> None:
    """A failure DURING MIGRATIONS closes the opened connection; a
    failure inside open() itself has nothing to close (the wrapper
    never adopted a connection), so close must not be called there.
    """
    db_path = tmp_path / "nestquest.db"
    _truncated_database(db_path)
    close_calls: list[bool] = []
    original_close = NestQuestDatabase.close
    async def _spy_close(self):
        close_calls.append(True)
        return await original_close(self)
    monkeypatch.setattr(NestQuestDatabase, "close", _spy_close)
    with pytest.raises(ConfigEntryNotReady):
        _run(_async_open_database(_make_hass_mock(), db_path))
    # Corruption hit at PRAGMA time inside open(): no connection was
    # adopted, so there is nothing to close and close() must NOT have
    # been called (calling close on a never-opened wrapper would be
    # harmless but the helper deliberately skips it).
    assert close_calls == []
    # Contrast: a failure DURING migrations (file opens fine) closes.
    close_calls.clear()
    valid_path = tmp_path / "migrations-fail.db"
    _valid_database(valid_path)
    # _valid_database's own close went through the spy; only the
    # failure path's close counts from here.
    close_calls.clear()
    from custom_components.nestquest import migrations as migrations_mod
    from custom_components.nestquest.db import NestQuestDatabase as _DB
    async def _failing(database, migrations=None):
        raise sqlite3.DatabaseError("database disk image is malformed")
    monkeypatch.setattr(migrations_mod, "apply_migrations", _failing)
    monkeypatch.setattr(
        "custom_components.nestquest.apply_migrations", _failing
    )
    with pytest.raises(ConfigEntryNotReady) as excinfo:
        _run(_async_open_database(_make_hass_mock(), valid_path))
    # Exactly one close: the migrations-failure cleanup path.  (The
    # spy also sees the wrapper's own idempotent close if it were
    # double-called — it must not be.)
    assert close_calls == [True], (
        "an open connection must be closed before the error propagates"
    )
    assert isinstance(excinfo.value.__cause__, sqlite3.DatabaseError)
 # ---------------------------------------------------------------------------
 # Startup through async_setup_entry (the real wiring)
 # ---------------------------------------------------------------------------
 def _make_setup(hass, registry):
    def _setup(entry):
        return async_setup_entry(hass, entry)
    return _setup
 async def _setup_entry(hass, entry, registry) -> object:
    entry = _wire(entry, registry)
    await async_setup_entry(hass, entry)
    return entry
 async def test_setup_with_corrupt_file_raises_not_ready(
    hass, make_entry, monkeypatch, tmp_path, caplog
 ) -> None:
    from custom_components.nestquest.store import async_get_db_path
    db_path = await async_get_db_path(hass)
    await _avalid_database(db_path)
    data = db_path.read_bytes()
    assert len(data) > 64
    db_path.write_bytes(data[: len(data) // 2])
    corrupt_bytes = db_path.read_bytes()
    entry = make_entry()
    with pytest.raises(ConfigEntryNotReady):
        await _setup_entry(hass, entry, hass.registry)
    # File untouched, no runtime data stored.
    assert db_path.read_bytes() == corrupt_bytes
    assert hass.data["nestquest"].get(entry.entry_id) is None
 async def test_setup_with_missing_file_creates_and_loads(
    hass, make_entry
 ) -> None:
    entry = await _setup_entry(hass, make_entry(), hass.registry)
    database = entry.runtime_data.database
    try:
        row = await database.fetch_one(
            f"SELECT version FROM {VERSION_TABLE}"
        )
        assert row == (1,)
    finally:
        await database.close()
    assert await async_unload_entry(hass, entry) is True
 async def test_setup_with_valid_file_preserves_rows(
    hass, make_entry
 ) -> None:
    entry_a = await _setup_entry(hass, make_entry(), hass.registry)
    database_a = entry_a.runtime_data.database
    await database_a.execute(
        "INSERT INTO children (display_name, created_at) VALUES (?, ?)",
        ("Ada", "2026-09-14T12:00:00+00:00"),
    )
    # Unload pops the runtime record AND closes the connection; a
    # fresh setup for the same entry_id then reopens the file.
    assert await async_unload_entry(hass, entry_a) is True
    entry_b = await _setup_entry(hass, make_entry(), hass.registry)
    database_b = entry_b.runtime_data.database
    assert database_b is not database_a
    try:
        row = await database_b.fetch_one("SELECT COUNT(*) FROM children")
        assert row == (1,)
    finally:
        await database_b.close()
    assert await async_unload_entry(hass, entry_b) is True
 # ---------------------------------------------------------------------------
 # No code path deletes or overwrites an existing database file
 # ---------------------------------------------------------------------------
 def test_package_never_unlinks_or_truncates_db_files() -> None:
    """Guardrail: no code path may delete or overwrite an existing
    database file.  Scans the package for unlink/rmtree/write on the
    DB filename constant or *.db patterns.
    """
    import re
    from pathlib import Path
    package = Path(
        __import__(
            "custom_components.nestquest", fromlist=["__file__"]
        ).__file__
    ).parent
    pattern = re.compile(
        r"(unlink|rmtree|write_bytes|write_text|open\s*\(\s*[^)]*['\"]w)"
        r"[^\n]*(nestquest\.db|\.db['\"])|"
        r"(nestquest\.db|\.db['\"])[^\n]*"
        r"(unlink|rmtree|write_bytes|write_text)",
        re.IGNORECASE,
    )
    offenders: list[str] = []
    for py in sorted(package.glob("*.py")):
        if pattern.search(py.read_text()):
            offenders.append(py.name)
    assert offenders == [], (
        f"code paths that could delete/overwrite the database: {offenders}"
    )
@@ -101,20 +101,12 @@ def test_executor_helper_creates_directory(tmp_path) -> None:
 def test_no_sqlite3_import_outside_db_module() -> None:
-    """sqlite3 is banned package-wide except two justified modules.
+    """sqlite3 is banned package-wide except db.py, the connection wrapper."""
    db.py is the connection wrapper (all statement execution lives
    there).  __init__.py imports sqlite3 ONLY for its exception types
    to classify corruption on startup (done-condition: corrupt file ->
    ConfigEntryNotReady); it never opens a connection or executes
    SQL — that is asserted separately below.
    """
    pkg_dir = Path(store.__file__).parent
    py_files = list(pkg_dir.glob("*.py"))
    assert py_files
    allowed = {"db.py", "__init__.py"}
    for py_file in py_files:
-        if py_file.name in allowed:
+        if py_file.name == "db.py":
            continue
        tree = ast.parse(py_file.read_text(encoding="utf-8"), filename=str(py_file))
        for node in ast.walk(tree):
@@ -129,36 +121,6 @@ def test_no_sqlite3_import_outside_db_module() -> None:
                )
 def test_init_module_never_opens_or_executes_sqlite() -> None:
    """The __init__.py sqlite3 usage is limited to (a) exception typing
    for corruption classification and (b) the ONE read-only preflight
    connection (mode=ro URI, SELECT-only) that detects corruption in an
    existing file before the real read-write open.  No read-write
    connect, no cursor writes, no non-SELECT statements may appear.
    """
    import re
    from pathlib import Path
    init_path = (
        Path(store.__file__).parent / "__init__.py"
    ).read_text(encoding="utf-8")
    # The preflight connect must be the URI read-only form, and only
    # SELECT statements and the integrity_check PRAGMA may run there.
    assert "sqlite3.connect(uri, uri=True)" in init_path
    assert "?mode=ro" in init_path
    assert 'PRAGMA integrity_check' in init_path
    forbidden = re.compile(
        r"sqlite3\s*\.\s*connect(?!\(\s*uri\s*,)"
        r"|\bconn(ection)?\s*\.\s*execute\s*\(\s*[\"'](?!SELECT\s|PRAGMA\s+integrity_check)"
        r"|\bexecutemany\b|\bexecutescript\b",
        re.IGNORECASE,
    )
    assert forbidden.search(init_path) is None, (
        "__init__.py may use sqlite3 only for exception typing and the "
        "read-only corruption preflight (SELECT + integrity_check)"
    )
 def test_no_file_creation_calls_in_store() -> None:
    source = Path(store.__file__).read_text(encoding="utf-8")
    tree = ast.parse(source, filename=str(store.__file__))
Author	SHA1	Message	Date
Joshua King	a887df232a	test: derive occurred-at query date from the event stamp; capture generated_at at creation Both comparisons now use the recorded value, never a fresh clock read, so a midnight boundary mid-test cannot flake them.	2026-09-14 08:26:58 -04:00
Joshua King	6e1f22e18a	fix: rollover-safe today reads under lock, INSERT INTO regex, midnight-safe dynamic dates - upsert re-checks 'today' after acquiring the connection lock (call date can go stale while queued across midnight); delete cutoff clamps to the lock-time date too - mutation regex now matches INSERT [OR x] INTO completion_events; test_schema.py's raw-SQL CHECK-constraint helper exempted narrowly via INSERT-strip, guard file self-scanned - NOW replaced by _now_stamp() (today 08:00 UTC, computed at call time); occurred-at test derives its query date from the stamp so it is midnight-safe - new rollover-contention test: a lock holder delays A's upsert across a shifted clock; the in-lock re-check rejects the now-past date	2026-09-14 08:24:17 -04:00
Joshua King	e48123e488	fix: immutable completed instances, no past generation, strict UTC timestamps, hardened append-only guards - upsert refuses due_dates before today (never generate in the past) and refuses regeneration of any instance carrying a completion event (immutability); the ON CONFLICT refresh only ever touches open instances - delete_future_uncompleted clamps the cutoff to today so backdating cannot rewrite past open instances - occurred_at enforces ONE strict UTC shape (YYYY-MM-DDTHH:MM:SS+00:00) at append; occurred-at range scope uses exact half-open bounds - was_on_time: bool \| None with type(was_on_time) is bool validation - 0.5 silently coercing to False is rejected - append-only guards hardened: package-wide mutation regex covers UPDATE OR IGNORE/qualified names/quotes, DAO INSERT exempted narrowly; behavioral call-interception test proves only INSERT/SELECT ever reach the DB for completion_events; compensating self-scan for the instances test file's rules-table exemption	2026-09-14 08:17:48 -04:00
Joshua King	ea6cf73bdc	feat: add typed DAO layer for task_instances and append-only completion_events (#21 )	2026-09-14 08:10:41 -04:00