diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index f34c276..d6d97be 100644 --- a/custom_components/nestquest/__init__.py +++ b/custom_components/nestquest/__init__.py @@ -10,6 +10,8 @@ from homeassistant.config_entries import ConfigEntry from homeassistant.core import HomeAssistant from .const import DOMAIN, LOGGER +from .db import NestQuestDatabase +from .store import async_get_db_path _LOGGER = LOGGER @@ -20,6 +22,7 @@ class NestQuestRuntimeData: entry_id: str options: dict[str, Any] + database: NestQuestDatabase remove_update_listener: Callable[[], Any] @@ -31,6 +34,14 @@ async def async_setup(hass: HomeAssistant, config: dict[str, Any]) -> bool: async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: """Set up NestQuest from a config entry.""" hass.data.setdefault(DOMAIN, {}) + existing = hass.data[DOMAIN].get(entry.entry_id) + if existing is not None: + db = getattr(existing, "database", None) + if db is None: + db = await NestQuestDatabase(hass).open(await async_get_db_path(hass)) + existing.database = db + entry.runtime_data = existing + return True async def _async_update_listener( listener_hass: HomeAssistant, listener_entry: ConfigEntry @@ -38,10 +49,17 @@ 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 NestQuestDatabase(hass).open(await async_get_db_path(hass)) + try: + remove_update_listener = entry.add_update_listener(_async_update_listener) + except BaseException: + await database.close() + raise runtime_data = NestQuestRuntimeData( entry_id=entry.entry_id, options=dict(entry.options), - remove_update_listener=entry.add_update_listener(_async_update_listener), + database=database, + remove_update_listener=remove_update_listener, ) entry.runtime_data = runtime_data hass.data[DOMAIN][entry.entry_id] = runtime_data @@ -58,10 +76,22 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: runtime_data = getattr(entry, "runtime_data", None) if runtime_data is not None: remove_update_listener = getattr(runtime_data, "remove_update_listener", None) - if remove_update_listener is not None: - result = remove_update_listener() - if inspect.isawaitable(result): - await result + unload_error: BaseException | None = None + try: + if remove_update_listener is not None: + result = remove_update_listener() + if inspect.isawaitable(result): + await result + except BaseException as err: + unload_error = err + finally: + database = getattr(runtime_data, "database", None) + if database is not None: + await database.close() + if unload_error is not None: + if getattr(entry, "runtime_data", None) is not None: + entry.runtime_data = None + raise unload_error if getattr(entry, "runtime_data", None) is not None: entry.runtime_data = None return True diff --git a/custom_components/nestquest/db.py b/custom_components/nestquest/db.py new file mode 100644 index 0000000..c05af6b --- /dev/null +++ b/custom_components/nestquest/db.py @@ -0,0 +1,422 @@ +"""Async SQLite connection wrapper for the NestQuest integration.""" +from __future__ import annotations + +import asyncio +import sqlite3 +from collections.abc import AsyncIterator, Awaitable, Sequence +from contextlib import asynccontextmanager +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from homeassistant.core import HomeAssistant + + +@dataclass(frozen=True) +class ExecutionResult: + """Plain-data outcome of one statement. + + ``rowcount`` and ``lastrowid`` mirror the corresponding + :class:`sqlite3.Cursor` attributes; both are captured on the executor + thread so no sqlite3 object ever crosses back to the event loop. + """ + + rowcount: int + lastrowid: int | None + + +class NestQuestDatabase: + """Async wrapper around a single SQLite connection. + + Every sqlite3 call is delegated to the Home Assistant executor through + ``hass.async_add_executor_job`` so the event loop is never blocked, and + only plain Python data (row tuples and :class:`ExecutionResult`) is + returned to the event loop. + + Every operation is serialized through one asyncio.Lock per wrapper. A + ``transaction()`` holds that lock for its whole BEGIN..COMMIT span, so + interleaved coroutines cannot corrupt the transaction and concurrent + BEGINs are impossible; statements issued while a transaction is open + queue until it completes. Concurrent transactions on one wrapper are + not supported: a second ``transaction()`` entered while one is active + raises :class:`RuntimeError`. ``close()`` serializes on the same lock + and is idempotent. + + Every connection-affecting executor job (connect, pragmas, statements, + transaction lifecycle and close) is awaited through the settlement + helper :meth:`_await_settled`: if the awaiting task is cancelled while + a job is in flight, the job still settles on the worker thread before + the cancellation propagates, so the wrapper lock is never released + while the executor thread is mid-job. + """ + + def __init__(self, hass: HomeAssistant) -> None: + """Initialize the wrapper around a Home Assistant instance.""" + self._hass = hass + self._conn: sqlite3.Connection | None = None + self._rowcount: int = -1 + self._lastrowid: int | None = None + self._lock: asyncio.Lock | None = None + self._lock_loop: asyncio.AbstractEventLoop | None = None + self._in_transaction = False + self._tx_task: asyncio.Task | None = None + + def _get_lock(self) -> asyncio.Lock: + """Return the per-wrapper lock, (re)bound to the running loop. + + HA drives every wrapper call from one event loop, so the lock is + created once and reused; re-binding only when the running loop + differs keeps the wrapper usable from plain synchronous drivers + that run each call on a fresh loop. + """ + loop = asyncio.get_running_loop() + if self._lock is None or self._lock_loop is not loop: + self._lock = asyncio.Lock() + self._lock_loop = loop + return self._lock + + @property + def connected(self) -> bool: + """Return True while the underlying connection is open.""" + return self._conn is not None + + @property + def rowcount(self) -> int: + """Row count of the most recent statement, or -1 when unset.""" + return self._rowcount + + @property + def lastrowid(self) -> int | None: + """Row id of the most recent insert, or None when unset.""" + return self._lastrowid + + @asynccontextmanager + async def _serialized(self) -> AsyncIterator[None]: + """Hold the wrapper lock unless the caller owns the active transaction. + + A transaction holds the lock for its whole BEGIN..COMMIT span; + statements issued by the transaction's own task are already + serialized by it and must not re-acquire the non-reentrant lock. + Any other task queues on the lock until the transaction ends. + """ + if ( + self._in_transaction + and self._tx_task is not None + and asyncio.current_task() is self._tx_task + ): + yield + return + async with self._get_lock(): + yield + + async def open(self, path: Path | str) -> NestQuestDatabase: + """Open the SQLite file with WAL journal mode and foreign keys on. + + Cancellation-safe: if this call is cancelled while the connect job + is in flight, the job still settles (see :meth:`_await_settled`). + A connection the settled job created is never adopted by the + cancelled call; it is closed through a settled executor job instead + of being abandoned to garbage collection, and the cancellation + propagates afterwards, so a follow-up ``open()`` starts from a + clean state. + """ + if self._conn is not None: + return self + async with self._get_lock(): + if self._conn is not None: + return self + + def _open() -> sqlite3.Connection: + conn = sqlite3.connect( + path, check_same_thread=False, isolation_level=None + ) + try: + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + except BaseException: + conn.close() + raise + return conn + + settled, cancelled = await self._await_settled_state( + self._hass.async_add_executor_job(_open) + ) + if cancelled: + if isinstance(settled, sqlite3.Connection): + + def _close_abandoned() -> None: + settled.close() + + await self._await_settled( + self._hass.async_add_executor_job(_close_abandoned) + ) + raise asyncio.CancelledError() + self._conn = settled + return self + + async def execute( + self, sql: str, parameters: Sequence[Any] = () + ) -> ExecutionResult: + """Run one statement in the executor and return its plain result.""" + async with self._serialized(): + conn = self._require_conn() + + def _execute() -> ExecutionResult: + cursor = conn.execute(sql, parameters) + return ExecutionResult(cursor.rowcount, cursor.lastrowid) + + result = await self._await_settled( + self._hass.async_add_executor_job(_execute) + ) + self._rowcount = result.rowcount + self._lastrowid = result.lastrowid + return result + + async def execute_many( + self, sql: str, parameters: Sequence[Sequence[Any]] + ) -> ExecutionResult: + """Run executemany in the executor and return its plain result.""" + async with self._serialized(): + conn = self._require_conn() + + def _execute_many() -> ExecutionResult: + cursor = conn.executemany(sql, parameters) + return ExecutionResult(cursor.rowcount, cursor.lastrowid) + + result = await self._await_settled( + self._hass.async_add_executor_job(_execute_many) + ) + self._rowcount = result.rowcount + self._lastrowid = result.lastrowid + return result + + async def fetch_one( + self, sql: str, parameters: Sequence[Any] = () + ) -> tuple | None: + """Run a query in the executor and return its first row, or None.""" + async with self._serialized(): + conn = self._require_conn() + + def _fetch_one() -> tuple[tuple | None, int, int | None]: + cursor = conn.execute(sql, parameters) + return cursor.fetchone(), cursor.rowcount, cursor.lastrowid + + row, rowcount, lastrowid = await self._await_settled( + self._hass.async_add_executor_job(_fetch_one) + ) + self._rowcount = rowcount + self._lastrowid = lastrowid + return row + + async def fetch_all( + self, sql: str, parameters: Sequence[Any] = () + ) -> list[tuple]: + """Run a query in the executor and return all of its rows.""" + async with self._serialized(): + conn = self._require_conn() + + def _fetch_all() -> tuple[list[tuple], int, int | None]: + cursor = conn.execute(sql, parameters) + return cursor.fetchall(), cursor.rowcount, cursor.lastrowid + + rows, rowcount, lastrowid = await self._await_settled( + self._hass.async_add_executor_job(_fetch_all) + ) + self._rowcount = rowcount + self._lastrowid = lastrowid + return rows + + @asynccontextmanager + async def transaction(self) -> AsyncIterator[None]: + """Run a block inside BEGIN/COMMIT, rolling back on any exception. + + The wrapper lock is held for the entire BEGIN..COMMIT span, so no + other statement can interleave inside the transaction. Entering a + second transaction while one is active raises RuntimeError. A + failing COMMIT (e.g. a deferred foreign-key violation) also triggers + a ROLLBACK before the commit error propagates, so the real + connection is never left inside a stray transaction. + + The lifecycle is cancellation-safe. Transaction ownership and the + wrapper lock are only released after the executor work has fully + settled, even when the owning task is cancelled mid-flight: if a + cancelled BEGIN or COMMIT still completes on the worker thread, + the corresponding ROLLBACK is shielded and awaited before any state + is cleared. A lifecycle statement that fails plus a failing + follow-up cleanup surface both errors in an ExceptionGroup. + """ + if self._in_transaction: + raise RuntimeError( + "A NestQuest transaction is already active on this " + "connection; concurrent transactions are not supported" + ) + async with self._get_lock(): + conn = self._require_conn() + self._in_transaction = True + self._tx_task = asyncio.current_task() + try: + + def _begin() -> None: + conn.execute("BEGIN") + + await self._settle_lifecycle( + conn, + self._hass.async_add_executor_job(_begin), + "BEGIN did not settle cleanly and the follow-up " + "cleanup also raised; the transaction may need manual " + "cleanup", + ) + try: + yield + except BaseException: + + def _rollback() -> None: + conn.execute("ROLLBACK") + + await self._settle_lifecycle( + conn, + self._hass.async_add_executor_job(_rollback), + "ROLLBACK failed and the follow-up cleanup also " + "raised; the transaction may need manual cleanup", + ) + raise + + def _commit() -> None: + conn.execute("COMMIT") + + await self._settle_lifecycle( + conn, + self._hass.async_add_executor_job(_commit), + "COMMIT failed and the follow-up cleanup also raised; " + "the transaction may need manual cleanup", + ) + finally: + self._in_transaction = False + self._tx_task = None + + async def _settle_lifecycle( + self, conn: sqlite3.Connection, job: Awaitable[Any], message: str + ) -> None: + """Run one BEGIN/ROLLBACK/COMMIT job, then force the connection clean. + + The job always settles first (see :meth:`_await_settled`), so a + cancelled statement that the worker thread completed anyway is + never abandoned mid-flight. If the job raised, a follow-up + ROLLBACK — itself settled — guarantees the connection ends outside + any SQLite transaction before the error propagates. When both the + job and the cleanup raise, both errors are preserved in an + ExceptionGroup instead of the original being lost. + """ + try: + await self._await_settled(job) + except BaseException as settle_error: + errors: list[BaseException] = [settle_error] + try: + await self._ensure_no_stray_transaction(conn) + except BaseException as cleanup_error: + errors.append(cleanup_error) + if len(errors) == 2: + raise BaseExceptionGroup(message, errors) from settle_error + raise + + async def _ensure_no_stray_transaction( + self, conn: sqlite3.Connection + ) -> None: + """Force the connection out of any transaction via ROLLBACK. + + Runs after a failed or cancelled lifecycle statement. The worker + thread may still be finishing (or may already have completed) the + very statement whose await was cancelled, so both the check and + the ROLLBACK settle in the executor before the caller releases + transaction ownership. + """ + + def _cleanup() -> bool: + if conn.in_transaction: + conn.execute("ROLLBACK") + return conn.in_transaction + + stray = await self._await_settled( + self._hass.async_add_executor_job(_cleanup) + ) + if stray: + raise RuntimeError( + "The SQLite connection is still inside a transaction after " + "cleanup; the transaction may need manual cleanup" + ) + + async def _await_settled(self, job: Awaitable[Any]) -> Any: + """Await ``job``, letting it settle even if this task is cancelled. + + Cancellation is observed, not swallowed: it is re-raised only once + the job has fully settled, so the caller can finalize against a + known connection state instead of abandoning in-flight work. The + job is wrapped in a task because a bare coroutine dies un-awaited + once its await is cancelled. A second cancellation during + settlement is absorbed the same way; if the job itself fails after + a cancellation, both errors surface in an ExceptionGroup. + """ + result, cancelled = await self._await_settled_state(job) + if cancelled: + raise asyncio.CancelledError() + return result + + async def _await_settled_state(self, job: Awaitable[Any]) -> tuple[Any, bool]: + """Await ``job`` to settlement and report ``(result, cancelled)``. + + Same settlement guarantees as :meth:`_await_settled`, but returns + the settled outcome together with whether the awaiting task was + cancelled, so callers that must act on a settled result before + re-raising cancellation (e.g. closing an abandoned connection) + can do so without losing the result. + """ + cancelled = False + inner = asyncio.ensure_future(job) + while not inner.done(): + try: + await asyncio.shield(inner) + except asyncio.CancelledError: + cancelled = True + try: + result = inner.result() + except BaseException as job_error: + if cancelled: + raise BaseExceptionGroup( + "executor job failed while settling after cancellation", + [asyncio.CancelledError(), job_error], + ) from job_error + raise + return result, cancelled + + async def close(self) -> None: + """Close the connection in the executor; safe to call repeatedly. + + Serialized on the wrapper lock, so an in-flight operation or open + transaction always completes before the connection is closed. + """ + if ( + self._in_transaction + and self._tx_task is not None + and asyncio.current_task() is self._tx_task + ): + raise RuntimeError( + "Cannot close the NestQuest database from inside its own " + "transaction; exit the transaction first" + ) + async with self._get_lock(): + conn, self._conn = self._conn, None + self._rowcount, self._lastrowid = -1, None + if conn is None: + return + + def _close() -> None: + conn.close() + + await self._await_settled(self._hass.async_add_executor_job(_close)) + + def _require_conn(self) -> sqlite3.Connection: + if self._conn is None: + raise RuntimeError( + "The NestQuest database connection is closed; call open() first" + ) + return self._conn \ No newline at end of file diff --git a/tests/conftest.py b/tests/conftest.py index e702c10..8a9e501 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -25,6 +25,9 @@ import asyncio import importlib.util import inspect import sys +import tempfile +import weakref +from pathlib import Path from types import SimpleNamespace from unittest.mock import MagicMock @@ -218,13 +221,35 @@ def wire_entry_to_registry(entry, registry: ListenerRegistry): return entry +class _HassNamespace(SimpleNamespace): + """SimpleNamespace subclass so the hass stub can hold weak references.""" + + def make_hass() -> tuple: """Create a hass stub plus its listener registry, mirroring HA's shape.""" - hass = SimpleNamespace( + config_dir = Path(tempfile.mkdtemp(prefix="nestquest-hass-")) + hass = _HassNamespace( data={}, config=MagicMock(), config_entries=SimpleNamespace(async_reload=None), ) + + def _cleanup_config_dir(): + import shutil + + shutil.rmtree(config_dir, ignore_errors=True) + + weakref.finalize(hass, _cleanup_config_dir) + hass.config.path = MagicMock( + side_effect=lambda name: str(config_dir / (name or "config")) + ) + + def _run_executor_job(fn, *args, **kwargs): + return asyncio.get_running_loop().run_in_executor( + None, lambda: fn(*args, **kwargs) + ) + + hass.async_add_executor_job = _run_executor_job registry = ListenerRegistry(hass) async def _async_reload(entry_id: str) -> None: diff --git a/tests/test_db.py b/tests/test_db.py new file mode 100644 index 0000000..7640c57 --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,998 @@ +"""Tests for db.py, the async SQLite connection wrapper of NestQuest.""" +from __future__ import annotations + +import ast +import asyncio +import sqlite3 +import threading +from pathlib import Path +from typing import Any +from unittest.mock import AsyncMock, MagicMock + +import pytest + +from conftest import make_hass + +from custom_components.nestquest import db as db_module +from custom_components.nestquest.db import NestQuestDatabase + + +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) -> tuple[NestQuestDatabase, MagicMock]: + hass = _make_hass_mock() + database = NestQuestDatabase(hass) + _run(database.open(path)) + return database, hass + + +# --------------------------------------------------------------------------- +# open(): pragmas, executor delegation, idempotence +# --------------------------------------------------------------------------- + + +def test_open_enables_wal_journal_mode(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "wal.db") + row = _run(database.fetch_one("PRAGMA journal_mode")) + assert row is not None and str(row[0]).lower() == "wal" + _run(database.close()) + + +def test_open_enables_foreign_key_enforcement(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "fk.db") + row = _run(database.fetch_one("PRAGMA foreign_keys")) + assert row is not None and int(row[0]) == 1 + _run(database.close()) + + +def test_open_delegates_connection_to_executor(tmp_path) -> None: + hass = _make_hass_mock() + database = NestQuestDatabase(hass) + _run(database.open(tmp_path / "exec.db")) + first_call = hass.async_add_executor_job.call_args_list[0] + assert callable(first_call.args[0]) + _run(database.close()) + + +def test_open_twice_returns_same_wrapper_without_second_connect(tmp_path) -> None: + path = tmp_path / "once.db" + hass = _make_hass_mock() + database = NestQuestDatabase(hass) + first = _run(database.open(path)) + second = _run(database.open(path)) + assert first is database + assert second is database + assert database.connected is True + opens = [ + c + for c in hass.async_add_executor_job.call_args_list + if getattr(c.args[0], "__name__", "") == "_open" + ] + assert len(opens) == 1 + _run(database.close()) + + +def test_open_accepts_str_path(tmp_path) -> None: + database, _hass = _open_db(str(tmp_path / "str.db")) + assert database.connected is True + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Executor delegation: every sqlite3 call happens off the event loop +# --------------------------------------------------------------------------- + + +def test_every_method_routes_sqlite3_through_executor(tmp_path) -> None: + """Each wrapper call must go via hass.async_add_executor_job.""" + database, hass = _open_db(tmp_path / "route.db") + calls_before = hass.async_add_executor_job.await_count + + _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")) + _run(database.execute("INSERT INTO t (name) VALUES (?)", ("a",))) + _run(database.execute_many("INSERT INTO t (name) VALUES (?)", [("b",), ("c",)])) + row = _run(database.fetch_one("SELECT name FROM t WHERE id = 1")) + rows = _run(database.fetch_all("SELECT name FROM t ORDER BY id")) + + assert row == ("a",) + assert rows == [("a",), ("b",), ("c",)] + assert hass.async_add_executor_job.await_count == calls_before + 5 + _run(database.close()) + + +async def test_every_wrapper_call_awaits_executor_job(tmp_path) -> None: + """The executor surface is awaited per wrapper call on a live loop.""" + hass = _make_hass_mock() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "loop.db") + + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)") + await database.fetch_one("SELECT 1") + await database.fetch_all("SELECT 1") + await database.execute_many("INSERT INTO t (id) VALUES (?)", [(1,)]) + async with database.transaction(): + pass + + assert hass.async_add_executor_job.await_count == 7 + assert all(callable(c.args[0]) for c in hass.async_add_executor_job.call_args_list) + await database.close() + + +async def test_public_methods_return_only_plain_data(tmp_path) -> None: + """execute/execute_many return plain dataclasses, never sqlite3 objects.""" + hass = _make_hass_mock() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "plain.db") + result = await database.execute( + "CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)" + ) + assert not isinstance(result, (sqlite3.Connection, sqlite3.Cursor)) + insert = await database.execute("INSERT INTO t (name) VALUES (?)", ("a",)) + assert not isinstance(insert, (sqlite3.Connection, sqlite3.Cursor)) + assert insert.rowcount == 1 + assert insert.lastrowid == 1 + many = await database.execute_many( + "INSERT INTO t (name) VALUES (?)", [("b",), ("c",)] + ) + assert not isinstance(many, (sqlite3.Connection, sqlite3.Cursor)) + assert many.rowcount == 2 + row = await database.fetch_one("SELECT name FROM t WHERE id = 1") + assert row == ("a",) + rows = await database.fetch_all("SELECT name FROM t ORDER BY id") + assert rows == [("a",), ("b",), ("c",)] + for value in (*rows, row): + assert not isinstance(value, sqlite3.Cursor) + await database.close() + + +async def test_no_sqlite3_call_on_event_loop_with_real_executor(tmp_path) -> None: + """With the conftest hass (real run_in_executor), queries still succeed.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "real.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)") + assert await database.fetch_one("SELECT 1") == (1,) + await database.close() + + +async def test_real_executor_never_touches_loop_thread(tmp_path, monkeypatch) -> None: + """Every real sqlite3 call must run on a non-event-loop thread. + + sqlite3.connect is instrumented so a proxy wraps the connection and + every cursor it hands out; each proxied call records the thread it ran + on. A recorded thread id equal to the event-loop thread would mean a + sqlite3 call crossed the executor boundary. + """ + loop_thread = threading.get_ident() + recorded: list[int] = [] + guard = threading.Lock() + + def _note() -> None: + with guard: + recorded.append(threading.get_ident()) + + class _Proxy: + """Generic proxy recording the thread of every attribute call.""" + + def __init__(self, target: Any) -> None: + object.__setattr__(self, "_target", target) + + def __getattr__(self, name: str): + attr = getattr(object.__getattribute__(self, "_target"), name) + if not callable(attr): + return attr + + def _wrapped(*args, **kwargs): + _note() + result = attr(*args, **kwargs) + if isinstance(result, sqlite3.Cursor): + return _Proxy(result) + return result + + return _wrapped + + real_connect = sqlite3.connect + connect_threads: list[int] = [] + + def _traced_connect(*args, **kwargs): + _note() + connect_threads.append(threading.get_ident()) + return _Proxy(real_connect(*args, **kwargs)) + + monkeypatch.setattr(db_module.sqlite3, "connect", _traced_connect) + + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "threads.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)") + await database.execute("INSERT INTO t (id) VALUES (?)", (1,)) + await database.execute_many("INSERT INTO t (id) VALUES (?)", [(2,), (3,)]) + assert await database.fetch_one("SELECT id FROM t WHERE id = 1") == (1,) + assert await database.fetch_all("SELECT id FROM t") == [(1,), (2,), (3,)] + async with database.transaction(): + await database.execute("INSERT INTO t (id) VALUES (?)", (4,)) + await database.close() + + assert recorded, "expected at least one recorded sqlite3 call" + assert len(connect_threads) == 1, ( + f"expected exactly one traced sqlite3.connect, got {len(connect_threads)}" + ) + assert loop_thread not in recorded, ( + f"sqlite3 call ran on the event-loop thread {loop_thread}; " + f"recorded threads: {sorted(set(recorded))}" + ) + + +# --------------------------------------------------------------------------- +# Source scan: sqlite3.* only inside executor-delegated functions +# --------------------------------------------------------------------------- + + +def _iter_with_scope(node, chain): + """Yield (node, enclosing-function chain), depth-first.""" + for child in ast.iter_child_nodes(node): + yield child, chain + next_chain = chain + (child,) if isinstance( + child, (ast.FunctionDef, ast.AsyncFunctionDef) + ) else chain + yield from _iter_with_scope(child, next_chain) + + +def test_db_sqlite3_calls_only_inside_executor_delegated_functions() -> None: + """Every sqlite3.* call in db.py sits inside an executor-delegating function. + + A function is executor-delegating when it (or, for nested closures, any of + its enclosing functions) passes work to ``hass.async_add_executor_job``. + """ + source = Path(db_module.__file__).read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(db_module.__file__)) + + delegated = { + node.name + for node in ast.walk(tree) + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + and any( + isinstance(sub, ast.Call) + and isinstance(sub.func, ast.Attribute) + and sub.func.attr == "async_add_executor_job" + for sub in ast.walk(node) + ) + } + assert delegated, "expected executor-delegated functions in db.py" + + violations = [] + for node, chain in _iter_with_scope(tree, ()): + if ( + isinstance(node, ast.Call) + and isinstance(node.func, ast.Attribute) + and isinstance(node.func.value, ast.Name) + and node.func.value.id == "sqlite3" + and not any(f.name in delegated for f in chain) + ): + violations.append((node.lineno, node.func.attr)) + assert violations == [], ( + f"Direct sqlite3 calls outside executor-delegated functions: {violations}" + ) + + +def test_package_wide_sqlite3_scan_excludes_only_db_module() -> None: + """The package-wide store scan exempts exactly the db module.""" + pkg_dir = Path(db_module.__file__).parent + scanned = [p.name for p in pkg_dir.glob("*.py") if p.name != "db.py"] + assert scanned, "expected other package modules to be scanned" + assert "db.py" not in scanned + + +# --------------------------------------------------------------------------- +# close(): idempotence and post-close errors +# --------------------------------------------------------------------------- + + +def test_close_is_idempotent(tmp_path) -> None: + database, hass = _open_db(tmp_path / "idem.db") + _run(database.close()) + assert database.connected is False + _run(database.close()) + assert database.connected is False + assert hass.async_add_executor_job.await_count == 2 + assert database.rowcount == -1 + assert database.lastrowid is None + + +def test_operations_after_close_raise_clear_error(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "closed.db") + _run(database.close()) + + async def _txn(): + async with database.transaction(): + pass + + with pytest.raises(RuntimeError, match="closed"): + _run(database.execute("SELECT 1")) + with pytest.raises(RuntimeError, match="closed"): + _run(database.fetch_one("SELECT 1")) + with pytest.raises(RuntimeError, match="closed"): + _run(database.fetch_all("SELECT 1")) + with pytest.raises(RuntimeError, match="closed"): + _run(database.execute_many("SELECT 1", [])) + with pytest.raises(RuntimeError, match="closed"): + _run(_txn()) + + +# --------------------------------------------------------------------------- +# Transactions: commit on success, rollback on exception +# --------------------------------------------------------------------------- + + +def test_transaction_commits_on_success(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "commit.db") + _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")) + + async def _block(): + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + + _run(_block()) + assert _run(database.fetch_all("SELECT name FROM t")) == [("kept",)] + _run(database.close()) + + +def test_transaction_rolls_back_on_exception(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "rollback.db") + _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")) + _run(database.execute("INSERT INTO t (name) VALUES (?)", ("before",))) + + async def _block(): + with pytest.raises(ValueError, match="boom"): + async with database.transaction(): + await database.execute( + "INSERT INTO t (name) VALUES (?)", ("dropped",) + ) + raise ValueError("boom") + + _run(_block()) + assert _run(database.fetch_all("SELECT name FROM t")) == [("before",)] + _run(database.close()) + + +async def test_commit_failure_rolls_back_and_next_transaction_works( + tmp_path, +) -> None: + """A COMMIT failure rolls back and leaves the wrapper reusable. + + With PRAGMA defer_foreign_keys=ON a violating insert succeeds inside + the transaction and the foreign-key violation only surfaces at COMMIT. + The commit error must propagate, the real connection must end up + outside any transaction, and the next transaction() plus statement + must run cleanly. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "deferred.db") + await database.execute("CREATE TABLE parent (id INTEGER PRIMARY KEY)") + await database.execute( + "CREATE TABLE child (" + "id INTEGER PRIMARY KEY, " + "parent_id INTEGER NOT NULL REFERENCES parent(id))" + ) + + with pytest.raises(sqlite3.IntegrityError, match="FOREIGN KEY"): + async with database.transaction(): + await database.execute("PRAGMA defer_foreign_keys=ON") + await database.execute( + "INSERT INTO child (parent_id) VALUES (?)", (999,) + ) + + assert database._in_transaction is False + assert database._tx_task is None + assert database._conn is not None + assert database._conn.in_transaction is False + assert await database.fetch_all("SELECT COUNT(*) FROM child") == [(0,)] + + async with database.transaction(): + await database.execute("INSERT INTO parent (id) VALUES (?)", (1,)) + await database.execute( + "INSERT INTO child (parent_id) VALUES (?)", (1,) + ) + assert await database.fetch_all("SELECT COUNT(*) FROM child") == [(1,)] + await database.close() + + +async def test_cancel_during_open_closes_abandoned_connection(tmp_path) -> None: + """NQ-PR12-010: cancel during open() must close the settled connection. + + The _open executor job is gated: the open task is cancelled while the + connect is still in flight, then the worker is allowed to complete it + anyway. The settled connection must never be adopted, must be closed + through a settled executor-backed close before the cancellation + propagates, and a follow-up open() must succeed cleanly. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + + open_started = asyncio.Event() + release_open = asyncio.Event() + jobs: list[str] = [] + captured: dict[str, Any] = {} + real_add = hass.async_add_executor_job + + async def _gated_add_executor_job(fn, *args, **kwargs): + name = getattr(fn, "__name__", "") + jobs.append(name) + if name == "_open": + open_started.set() + await release_open.wait() + result = await real_add(fn, *args, **kwargs) + if name == "_open": + captured["conn"] = result + return result + + hass.async_add_executor_job = _gated_add_executor_job + + open_task = asyncio.create_task(database.open(tmp_path / "cancel-open.db")) + await open_started.wait() + open_task.cancel() + await asyncio.sleep(0.05) + assert database._conn is None, ( + "the cancelled open adopted a connection that was still settling" + ) + release_open.set() + with pytest.raises(asyncio.CancelledError): + await open_task + + assert database.connected is False + assert "_close_abandoned" in jobs, ( + "the abandoned connection was not closed via an executor job" + ) + conn = captured.get("conn") + assert conn is not None + + def _probe() -> str: + try: + conn.execute("SELECT 1") + except sqlite3.ProgrammingError: + return "closed" + return "open" + + assert await real_add(_probe) == "closed" + + hass.async_add_executor_job = real_add + await database.open(tmp_path / "cancel-open.db") + assert database.connected is True + assert await database.fetch_one("SELECT 1") == (1,) + await database.close() + + +async def test_cancel_during_begin_settles_then_rolls_back(tmp_path) -> None: + """NQ-PR12-007: cancel during BEGIN must not leak a stray transaction. + + The BEGIN executor job is gated: the transaction task is cancelled + while BEGIN is still in flight, then the worker is allowed to + complete it anyway. The wrapper must keep transaction ownership + until the submitted work settles, roll the completed BEGIN back, + clear its state, and stay usable for the next transaction. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "cancel-begin.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + await database.execute("INSERT INTO t (name) VALUES (?)", ("before",)) + + begin_started = asyncio.Event() + release_begin = asyncio.Event() + real_add = hass.async_add_executor_job + + async def _gated_add_executor_job(fn, *args, **kwargs): + if getattr(fn, "__name__", "") == "_begin": + begin_started.set() + await release_begin.wait() + return await real_add(fn, *args, **kwargs) + + hass.async_add_executor_job = _gated_add_executor_job + + async def _txn() -> None: + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + + txn_task = asyncio.create_task(_txn()) + await begin_started.wait() + txn_task.cancel() + await asyncio.sleep(0.05) + assert database._in_transaction is True, ( + "ownership was released while the cancelled BEGIN was still in flight" + ) + release_begin.set() + with pytest.raises(asyncio.CancelledError): + await txn_task + + assert database._in_transaction is False + assert database._tx_task is None + assert database._conn is not None + assert database._conn.in_transaction is False + + hass.async_add_executor_job = real_add + assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [ + ("before",) + ] + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [ + ("before",), + ("kept",), + ] + await database.close() + + +async def test_cancel_during_commit_settles_then_propagates_cancellation( + tmp_path, +) -> None: + """NQ-PR12-007: cancel during COMMIT must not mask the cancellation. + + COMMIT is gated; the task is cancelled while it is in flight and the + worker then completes it anyway. The cancellation must propagate as + CancelledError (not be converted by a doomed ROLLBACK of an already + committed transaction), the connection must end outside any + transaction, and the wrapper must stay usable. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "cancel-commit.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + + commit_started = asyncio.Event() + release_commit = asyncio.Event() + real_add = hass.async_add_executor_job + + async def _gated_add_executor_job(fn, *args, **kwargs): + if getattr(fn, "__name__", "") == "_commit": + commit_started.set() + await release_commit.wait() + return await real_add(fn, *args, **kwargs) + + hass.async_add_executor_job = _gated_add_executor_job + + async def _txn() -> None: + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + + txn_task = asyncio.create_task(_txn()) + await commit_started.wait() + txn_task.cancel() + await asyncio.sleep(0.05) + assert database._in_transaction is True, ( + "ownership was released while the cancelled COMMIT was still in flight" + ) + release_commit.set() + with pytest.raises(asyncio.CancelledError): + await txn_task + + assert database._in_transaction is False + assert database._tx_task is None + assert database._conn is not None + assert database._conn.in_transaction is False + + hass.async_add_executor_job = real_add + assert await database.fetch_all("SELECT name FROM t") == [("kept",)] + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("again",)) + assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [ + ("kept",), + ("again",), + ] + await database.close() + + +async def test_cancel_during_in_transaction_execute_settles_then_rolls_back( + tmp_path, +) -> None: + """NQ-PR12-009: cancel during an in-transaction statement must not race ROLLBACK. + + The in-transaction INSERT executor job is gated: the transaction task + is cancelled while the statement is still in flight, then the worker + is allowed to complete it anyway. The ROLLBACK must only run after + the statement has settled — while the statement is in flight the + transaction is still open — and the cancelled statement's write must + not persist. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "cancel-execute.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + await database.execute("INSERT INTO t (name) VALUES (?)", ("before",)) + + insert_started = asyncio.Event() + release_insert = asyncio.Event() + real_add = hass.async_add_executor_job + + async def _gated_add_executor_job(fn, *args, **kwargs): + if getattr(fn, "__name__", "") == "_execute": + insert_started.set() + await release_insert.wait() + return await real_add(fn, *args, **kwargs) + + hass.async_add_executor_job = _gated_add_executor_job + + async def _txn() -> None: + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("lost",)) + + txn_task = asyncio.create_task(_txn()) + await insert_started.wait() + txn_task.cancel() + await asyncio.sleep(0.05) + assert database._in_transaction is True, ( + "ownership was released while the cancelled statement was in flight" + ) + assert database._conn is not None + assert database._conn.in_transaction is True, ( + "ROLLBACK ran before the cancelled statement settled" + ) + release_insert.set() + with pytest.raises(asyncio.CancelledError): + await txn_task + + assert database._in_transaction is False + assert database._tx_task is None + assert database._conn is not None + assert database._conn.in_transaction is False + + hass.async_add_executor_job = real_add + assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [ + ("before",) + ] + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + assert await database.fetch_all("SELECT name FROM t ORDER BY id") == [ + ("before",), + ("kept",), + ] + await database.close() + + +async def test_commit_and_rollback_failures_are_both_preserved(tmp_path) -> None: + """NQ-PR12-008: both errors survive when COMMIT and ROLLBACK both fail. + + The COMMIT executor call is forced to fail, and the follow-up + cleanup ROLLBACK is forced to fail as well. The raised error must + keep both original exceptions inspectable instead of discarding the + commit failure behind the rollback failure. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "both-fail.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY)") + + commit_failure = sqlite3.OperationalError("commit exploded") + rollback_failure = sqlite3.OperationalError("rollback exploded") + real_add = hass.async_add_executor_job + + async def _failing_add_executor_job(fn, *args, **kwargs): + name = getattr(fn, "__name__", "") + if name == "_commit": + raise commit_failure + if name == "_cleanup": + raise rollback_failure + return await real_add(fn, *args, **kwargs) + + hass.async_add_executor_job = _failing_add_executor_job + + with pytest.raises(BaseExceptionGroup) as exc_info: + async with database.transaction(): + pass + + assert commit_failure in exc_info.value.exceptions + assert rollback_failure in exc_info.value.exceptions + assert database._in_transaction is False + assert database._tx_task is None + await database.close() + + +async def test_transaction_persists_across_connections(tmp_path) -> None: + """A committed transaction is visible to a brand-new connection.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "persist.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + await database.close() + + other = NestQuestDatabase(hass) + await other.open(tmp_path / "persist.db") + assert await other.fetch_all("SELECT name FROM t") == [("kept",)] + await other.close() + + +# --------------------------------------------------------------------------- +# Serialization: one lock per wrapper, no cross-commit interleaving +# --------------------------------------------------------------------------- + + +async def test_concurrent_writers_do_not_interleave_commits(tmp_path) -> None: + """Real-executor proof: nothing commits inside another transaction's span. + + A transaction runner repeatedly counts rows, inserts, and counts again + inside BEGIN..COMMIT; the count must grow by exactly one, proving no + other writer committed in between even though several writer tasks fire + statements concurrently on real executor threads. Statements issued + while a transaction is open queue behind it on the wrapper lock. + """ + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "serial.db") + await database.execute("CREATE TABLE t (kind TEXT, seq INTEGER)") + tx_rounds, writers, writer_rounds = 6, 5, 8 + total = tx_rounds + writers * writer_rounds + + async def _tx_runner() -> None: + for rnd in range(tx_rounds): + async with database.transaction(): + rows = await database.fetch_all("SELECT COUNT(*) FROM t") + seen = rows[0][0] if rows else -1 + await database.execute( + "INSERT INTO t (kind, seq) VALUES (?, ?)", ("tx", rnd) + ) + rows = await database.fetch_all("SELECT COUNT(*) FROM t") + grown = rows[0][0] if rows else -1 + if grown != seen + 1: + raise AssertionError( + f"interleaved commit: count {seen} -> {grown} " + f"in tx round {rnd}" + ) + + async def _writer(worker: int) -> None: + for rnd in range(writer_rounds): + await database.execute( + "INSERT INTO t (kind, seq) VALUES (?, ?)", ("w", worker) + ) + + await asyncio.gather( + _tx_runner(), *(_writer(w) for w in range(writers)) + ) + + rows = await database.fetch_all("SELECT COUNT(*) FROM t") + assert rows[0][0] == total, f"expected {total} rows, got {rows[0][0]}" + await database.close() + + +async def test_transaction_span_excludes_concurrent_transaction(tmp_path) -> None: + """A second transaction caller is rejected with a clear error.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "reject.db") + await database.execute("CREATE TABLE t (id INTEGER)") + + started = asyncio.Event() + + async def _first() -> None: + async with database.transaction(): + started.set() + await asyncio.sleep(0.05) + + first = asyncio.create_task(_first()) + await started.wait() + with pytest.raises(RuntimeError, match="transaction"): + async with database.transaction(): + pass + await first + await database.execute("INSERT INTO t (id) VALUES (?)", (1,)) + assert await database.fetch_one("SELECT id FROM t") == (1,) + await database.close() + + +async def test_transaction_blocks_other_statements_until_commit(tmp_path) -> None: + """A statement issued during an open transaction queues behind it.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "block.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + + order: list[str] = [] + started = asyncio.Event() + release = asyncio.Event() + + async def _txn() -> None: + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("in_tx",)) + order.append("tx-start") + started.set() + await release.wait() + order.append("tx-end") + + async def _outsider() -> None: + await started.wait() + await database.execute("INSERT INTO t (name) VALUES (?)", ("after",)) + order.append("outside") + + txn_task = asyncio.create_task(_txn()) + await started.wait() + outsider_task = asyncio.create_task(_outsider()) + await asyncio.sleep(0.05) + assert order == ["tx-start"], f"outsider ran inside transaction: {order}" + release.set() + await asyncio.gather(txn_task, outsider_task) + assert order == ["tx-start", "tx-end", "outside"] + rows = await database.fetch_all("SELECT name FROM t ORDER BY id") + assert rows == [("in_tx",), ("after",)] + await database.close() + + +async def test_concurrent_transaction_is_rejected_with_clear_error( + tmp_path, +) -> None: + """A second concurrent transaction() gets a clear RuntimeError.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "concurrent-tx.db") + await database.execute("CREATE TABLE t (id INTEGER)") + + started = asyncio.Event() + + async def _first() -> None: + async with database.transaction(): + started.set() + await asyncio.sleep(0.05) + + first = asyncio.create_task(_first()) + await started.wait() + with pytest.raises(RuntimeError, match="transaction"): + async with database.transaction(): + pass + await first + await database.close() + + +async def test_close_waits_for_in_flight_transaction(tmp_path) -> None: + """close() serializes: an open transaction finishes before shutdown.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "closerace.db") + await database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)") + + started = asyncio.Event() + release = asyncio.Event() + + async def _txn() -> None: + async with database.transaction(): + await database.execute("INSERT INTO t (name) VALUES (?)", ("kept",)) + started.set() + await release.wait() + + txn_task = asyncio.create_task(_txn()) + await started.wait() + close_task = asyncio.create_task(database.close()) + await asyncio.sleep(0.05) + assert database.connected is True, "close() raced an open transaction" + release.set() + await asyncio.gather(txn_task, close_task) + assert database.connected is False + + witness = NestQuestDatabase(hass) + await witness.open(tmp_path / "closerace.db") + assert await witness.fetch_all("SELECT name FROM t") == [("kept",)] + await witness.close() + + +async def test_close_during_transaction_from_same_task_is_rejected( + tmp_path, +) -> None: + """close() inside its own transaction body raises instead of deadlocking.""" + hass, _registry = make_hass() + database = NestQuestDatabase(hass) + await database.open(tmp_path / "selfclose.db") + await database.execute("CREATE TABLE t (id INTEGER)") + with pytest.raises(RuntimeError, match="transaction"): + async with database.transaction(): + await database.close() + assert database.connected is True + await database.close() + + +# --------------------------------------------------------------------------- +# rowcount / lastrowid surface +# --------------------------------------------------------------------------- + + +def test_rowcount_and_lastrowid_before_any_operation(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "unset.db") + assert database.rowcount == -1 + assert database.lastrowid is None + _run(database.close()) + + +def test_execute_exposes_rowcount_and_lastrowid(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "counts.db") + _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")) + _run(database.execute("INSERT INTO t (name) VALUES (?)", ("x",))) + assert database.rowcount == 1 + assert database.lastrowid == 1 + _run(database.execute("INSERT INTO t (name) VALUES (?)", ("y",))) + assert database.rowcount == 1 + assert database.lastrowid == 2 + _run(database.close()) + + +def test_execute_many_exposes_rowcount(tmp_path) -> None: + database, _hass = _open_db(tmp_path / "many.db") + _run(database.execute("CREATE TABLE t (id INTEGER PRIMARY KEY, name TEXT)")) + _run(database.execute_many("INSERT INTO t (name) VALUES (?)", [("a",), ("b",)])) + assert database.rowcount == 2 + _run(database.execute("INSERT INTO t (name) VALUES (?)", ("c",))) + assert database.lastrowid == 3 + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Lifecycle wiring: opened once per entry, closed on unload +# --------------------------------------------------------------------------- + + +async def test_setup_entry_opens_database_in_runtime_data(hass, make_entry) -> None: + from conftest import wire_entry_to_registry + + from custom_components.nestquest import async_setup_entry, async_unload_entry + + entry = wire_entry_to_registry(make_entry(), hass.registry) + assert await async_setup_entry(hass, entry) is True + runtime = entry.runtime_data + assert runtime.database is not None + assert runtime.database.connected is True + await async_unload_entry(hass, entry) + + +async def test_unload_entry_closes_database(hass, make_entry) -> None: + from conftest import wire_entry_to_registry + + from custom_components.nestquest import async_setup_entry, async_unload_entry + + entry = wire_entry_to_registry(make_entry(), hass.registry) + await async_setup_entry(hass, entry) + database = entry.runtime_data.database + assert database.connected is True + assert await async_unload_entry(hass, entry) is True + assert database.connected is False + with pytest.raises(RuntimeError, match="closed"): + await database.fetch_one("SELECT 1") + + +async def test_second_setup_does_not_open_second_connection(hass, make_entry) -> None: + from conftest import wire_entry_to_registry + + from custom_components.nestquest import async_setup_entry, async_unload_entry + + entry = wire_entry_to_registry(make_entry(), hass.registry) + await async_setup_entry(hass, entry) + first = entry.runtime_data.database + + await async_setup_entry(hass, entry) + second = entry.runtime_data.database + + assert first is second + assert first.connected is True + await async_unload_entry(hass, entry) + + +async def test_setup_unload_cycles_reopen_fresh_connection(hass, make_entry) -> None: + from conftest import wire_entry_to_registry + + from custom_components.nestquest import async_setup_entry, async_unload_entry + + entry = wire_entry_to_registry(make_entry(), hass.registry) + seen: list[NestQuestDatabase] = [] + for _ in range(3): + await async_setup_entry(hass, entry) + runtime = entry.runtime_data + assert runtime.database.connected is True + seen.append(runtime.database) + await async_unload_entry(hass, entry) + assert runtime.database.connected is False + assert len(set(seen)) == 3 + assert hass.data == {} \ No newline at end of file diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index b12ed36..b145aad 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -3,6 +3,8 @@ from __future__ import annotations import asyncio +import pytest + from conftest import make_config_entry, make_hass, wire_entry_to_registry from custom_components.nestquest import DOMAIN, async_setup_entry, async_unload_entry @@ -196,4 +198,42 @@ async def test_setup_awaitable_listener_removal_supported() -> None: assert result is True assert removed == [True] assert hass.data == {} - assert entry.runtime_data is None \ No newline at end of file + assert entry.runtime_data is None + + +async def test_setup_closes_database_when_listener_registration_fails() -> None: + """A listener-registration failure must not leak the opened database.""" + from custom_components.nestquest.db import NestQuestDatabase + + class _Boom(Exception): + pass + + entry = make_config_entry(entry_id="listener-boom") + entry.add_update_listener = lambda listener: (_ for _ in ()).throw(_Boom()) + hass, _registry = make_hass() + with pytest.raises(_Boom): + await async_setup_entry(hass, entry) + assert hass.data.get(DOMAIN, {}) == {} + assert not hasattr(entry, "runtime_data") + + +async def test_unload_closes_database_even_when_listener_removal_raises() -> None: + """Unload removes the listener, closes the DB in a finally, then raises.""" + from custom_components.nestquest.db import NestQuestDatabase + + class _RemoveBoom(Exception): + pass + + def _remover(): + raise _RemoveBoom() + + entry = make_config_entry(entry_id="remove-boom") + entry.add_update_listener = lambda listener: _remover + hass, _registry = make_hass() + await async_setup_entry(hass, entry) + database = entry.runtime_data.database + with pytest.raises(_RemoveBoom): + await async_unload_entry(hass, entry) + assert database.connected is False, "database leaked despite remover error" + assert entry.runtime_data is None + assert DOMAIN not in hass.data \ No newline at end of file diff --git a/tests/test_store.py b/tests/test_store.py index 95866b4..3b20923 100644 --- a/tests/test_store.py +++ b/tests/test_store.py @@ -100,11 +100,14 @@ def test_executor_helper_creates_directory(tmp_path) -> None: assert nested.is_dir() -def test_no_sqlite3_import_in_package() -> None: +def test_no_sqlite3_import_outside_db_module() -> None: + """sqlite3 is banned package-wide except db.py, the connection wrapper.""" pkg_dir = Path(store.__file__).parent py_files = list(pkg_dir.glob("*.py")) assert py_files for py_file in py_files: + 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): if isinstance(node, ast.Import):