From ae7a064d63bde751a2c3eadbc48f7456ea993e65 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 20:20:31 -0400 Subject: [PATCH] feat: add async sqlite connection wrapper with executor-backed queries --- custom_components/nestquest/__init__.py | 16 + custom_components/nestquest/db.py | 143 +++++++++ tests/conftest.py | 27 +- tests/test_db.py | 381 ++++++++++++++++++++++++ tests/test_store.py | 5 +- 5 files changed, 570 insertions(+), 2 deletions(-) create mode 100644 custom_components/nestquest/db.py create mode 100644 tests/test_db.py diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index f34c276..679311b 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,9 +49,11 @@ 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)) runtime_data = NestQuestRuntimeData( entry_id=entry.entry_id, options=dict(entry.options), + database=database, remove_update_listener=entry.add_update_listener(_async_update_listener), ) entry.runtime_data = runtime_data @@ -62,6 +75,9 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: result = remove_update_listener() if inspect.isawaitable(result): await result + database = getattr(runtime_data, "database", None) + if database is not None: + await database.close() 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..c3f68b1 --- /dev/null +++ b/custom_components/nestquest/db.py @@ -0,0 +1,143 @@ +"""Async SQLite connection wrapper for the NestQuest integration.""" +from __future__ import annotations + +import sqlite3 +from collections.abc import AsyncIterator, Sequence +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any + +from homeassistant.core import HomeAssistant + + +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. + """ + + def __init__(self, hass: HomeAssistant) -> None: + """Initialize the wrapper around a Home Assistant instance.""" + self._hass = hass + self._conn: sqlite3.Connection | None = None + self._cursor: sqlite3.Cursor | None = None + + @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._cursor.rowcount if self._cursor is not None else -1 + + @property + def lastrowid(self) -> int | None: + """Row id of the most recent insert, or None when unset.""" + return self._cursor.lastrowid if self._cursor is not None else None + + async def open(self, path: Path | str) -> NestQuestDatabase: + """Open the SQLite file with WAL journal mode and foreign keys on.""" + if self._conn is not None: + return self + + def _open() -> sqlite3.Connection: + conn = sqlite3.connect( + path, check_same_thread=False, isolation_level=None + ) + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA foreign_keys=ON") + return conn + + self._conn = await self._hass.async_add_executor_job(_open) + return self + + async def execute( + self, sql: str, parameters: Sequence[Any] = () + ) -> sqlite3.Cursor: + """Run one statement in the executor and return its cursor.""" + conn = self._require_conn() + + def _execute() -> sqlite3.Cursor: + return conn.execute(sql, parameters) + + self._cursor = await self._hass.async_add_executor_job(_execute) + return self._cursor + + async def execute_many( + self, sql: str, parameters: Sequence[Sequence[Any]] + ) -> sqlite3.Cursor: + """Run executemany in the executor and return its cursor.""" + conn = self._require_conn() + + def _execute_many() -> sqlite3.Cursor: + return conn.executemany(sql, parameters) + + self._cursor = await self._hass.async_add_executor_job(_execute_many) + return self._cursor + + 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.""" + cursor = await self.execute(sql, parameters) + + def _fetch_one() -> tuple | None: + return cursor.fetchone() + + return await self._hass.async_add_executor_job(_fetch_one) + + async def fetch_all( + self, sql: str, parameters: Sequence[Any] = () + ) -> list[tuple]: + """Run a query in the executor and return all of its rows.""" + cursor = await self.execute(sql, parameters) + + def _fetch_all() -> list[tuple]: + return cursor.fetchall() + + return await self._hass.async_add_executor_job(_fetch_all) + + @asynccontextmanager + async def transaction(self) -> AsyncIterator[None]: + """Run a block inside BEGIN/COMMIT, rolling back on any exception.""" + conn = self._require_conn() + + def _begin() -> None: + conn.execute("BEGIN") + + await self._hass.async_add_executor_job(_begin) + try: + yield + except BaseException: + + def _rollback() -> None: + conn.execute("ROLLBACK") + + await self._hass.async_add_executor_job(_rollback) + raise + + def _commit() -> None: + conn.execute("COMMIT") + + await self._hass.async_add_executor_job(_commit) + + async def close(self) -> None: + """Close the connection in the executor; safe to call repeatedly.""" + conn, self._conn, self._cursor = self._conn, None, None + if conn is None: + return + + def _close() -> None: + conn.close() + + await 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..1380f38 --- /dev/null +++ b/tests/test_db.py @@ -0,0 +1,381 @@ +"""Tests for db.py, the async SQLite connection wrapper of NestQuest.""" +from __future__ import annotations + +import ast +import asyncio +from pathlib import Path +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 + 7 + _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 == 9 + assert all(callable(c.args[0]) for c in hass.async_add_executor_job.call_args_list) + 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() + + +# --------------------------------------------------------------------------- +# 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 + + +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_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() + + +# --------------------------------------------------------------------------- +# 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_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):