From ae7a064d63bde751a2c3eadbc48f7456ea993e65 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 20:20:31 -0400 Subject: [PATCH 1/6] 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): -- 2.55.0 From a56a7e3d3710166992f2cbe58b9cd1e292920528 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:07:42 -0400 Subject: [PATCH 2/6] fix: make db wrapper executor-safe, serialized, and leak-proof --- custom_components/nestquest/__init__.py | 30 ++- custom_components/nestquest/db.py | 237 +++++++++++++++----- tests/test_db.py | 283 +++++++++++++++++++++++- tests/test_lifecycle.py | 42 +++- 4 files changed, 525 insertions(+), 67 deletions(-) diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index 679311b..d6d97be 100644 --- a/custom_components/nestquest/__init__.py +++ b/custom_components/nestquest/__init__.py @@ -50,11 +50,16 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: 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), database=database, - remove_update_listener=entry.add_update_listener(_async_update_listener), + remove_update_listener=remove_update_listener, ) entry.runtime_data = runtime_data hass.data[DOMAIN][entry.entry_id] = runtime_data @@ -71,13 +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 - database = getattr(runtime_data, "database", None) - if database is not None: - await database.close() + 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 index c3f68b1..3fcf6bc 100644 --- a/custom_components/nestquest/db.py +++ b/custom_components/nestquest/db.py @@ -1,27 +1,72 @@ """Async SQLite connection wrapper for the NestQuest integration.""" from __future__ import annotations +import asyncio import sqlite3 from collections.abc import AsyncIterator, 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. + ``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. """ 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 + 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: @@ -31,109 +76,189 @@ class NestQuestDatabase: @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 + return self._rowcount @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 + 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.""" 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 - ) - conn.execute("PRAGMA journal_mode=WAL") - conn.execute("PRAGMA foreign_keys=ON") - return conn + 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 - self._conn = await self._hass.async_add_executor_job(_open) + 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() + ) -> ExecutionResult: + """Run one statement in the executor and return its plain result.""" + async with self._serialized(): + conn = self._require_conn() - def _execute() -> sqlite3.Cursor: - return conn.execute(sql, parameters) + def _execute() -> ExecutionResult: + cursor = conn.execute(sql, parameters) + return ExecutionResult(cursor.rowcount, cursor.lastrowid) - self._cursor = await self._hass.async_add_executor_job(_execute) - return self._cursor + result = await 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]] - ) -> sqlite3.Cursor: - """Run executemany in the executor and return its cursor.""" - conn = self._require_conn() + ) -> ExecutionResult: + """Run executemany in the executor and return its plain result.""" + async with self._serialized(): + conn = self._require_conn() - def _execute_many() -> sqlite3.Cursor: - return conn.executemany(sql, parameters) + def _execute_many() -> ExecutionResult: + cursor = conn.executemany(sql, parameters) + return ExecutionResult(cursor.rowcount, cursor.lastrowid) - self._cursor = await self._hass.async_add_executor_job(_execute_many) - return self._cursor + result = await 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.""" - cursor = await self.execute(sql, parameters) + async with self._serialized(): + conn = self._require_conn() - def _fetch_one() -> tuple | None: - return cursor.fetchone() + def _fetch_one() -> tuple[tuple | None, int, int | None]: + cursor = conn.execute(sql, parameters) + return cursor.fetchone(), cursor.rowcount, cursor.lastrowid - return await self._hass.async_add_executor_job(_fetch_one) + row, rowcount, lastrowid = await 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.""" - cursor = await self.execute(sql, parameters) + async with self._serialized(): + conn = self._require_conn() - def _fetch_all() -> list[tuple]: - return cursor.fetchall() + def _fetch_all() -> tuple[list[tuple], int, int | None]: + cursor = conn.execute(sql, parameters) + return cursor.fetchall(), cursor.rowcount, cursor.lastrowid - return await self._hass.async_add_executor_job(_fetch_all) + rows, rowcount, lastrowid = await 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.""" - conn = self._require_conn() + """Run a block inside BEGIN/COMMIT, rolling back on any exception. - def _begin() -> None: - conn.execute("BEGIN") + 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. + """ + 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: - await self._hass.async_add_executor_job(_begin) - try: - yield - except BaseException: + def _begin() -> None: + conn.execute("BEGIN") - def _rollback() -> None: - conn.execute("ROLLBACK") + await self._hass.async_add_executor_job(_begin) + try: + yield + except BaseException: - await self._hass.async_add_executor_job(_rollback) - raise + def _rollback() -> None: + conn.execute("ROLLBACK") - def _commit() -> None: - conn.execute("COMMIT") + await self._hass.async_add_executor_job(_rollback) + raise - await self._hass.async_add_executor_job(_commit) + def _commit() -> None: + conn.execute("COMMIT") + + await self._hass.async_add_executor_job(_commit) + finally: + self._in_transaction = False + self._tx_task = None 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 + """Close the connection in the executor; safe to call repeatedly. - def _close() -> None: - conn.close() + 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 - await self._hass.async_add_executor_job(_close) + def _close() -> None: + conn.close() + + await self._hass.async_add_executor_job(_close) def _require_conn(self) -> sqlite3.Connection: if self._conn is None: diff --git a/tests/test_db.py b/tests/test_db.py index 1380f38..0ef0ea8 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -3,7 +3,10 @@ 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 @@ -101,7 +104,7 @@ def test_every_method_routes_sqlite3_through_executor(tmp_path) -> None: assert row == ("a",) assert rows == [("a",), ("b",), ("c",)] - assert hass.async_add_executor_job.await_count == calls_before + 7 + assert hass.async_add_executor_job.await_count == calls_before + 5 _run(database.close()) @@ -118,11 +121,38 @@ async def test_every_wrapper_call_awaits_executor_job(tmp_path) -> None: async with database.transaction(): pass - assert hass.async_add_executor_job.await_count == 9 + 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() @@ -133,6 +163,68 @@ async def test_no_sqlite3_call_on_event_loop_with_real_executor(tmp_path) -> Non 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 + + def _traced_connect(*args, **kwargs): + 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 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 # --------------------------------------------------------------------------- @@ -205,6 +297,8 @@ def test_close_is_idempotent(tmp_path) -> None: _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: @@ -279,6 +373,191 @@ async def test_transaction_persists_across_connections(tmp_path) -> None: 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 # --------------------------------------------------------------------------- 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 -- 2.55.0 From d527d13570a2ecf3a7530fb670777c8d6a4347c6 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:12:33 -0400 Subject: [PATCH 3/6] fix: roll back on commit failure and record connect thread in loop-safety test --- custom_components/nestquest/db.py | 21 +++++++++++-- tests/test_db.py | 49 +++++++++++++++++++++++++++++++ 2 files changed, 68 insertions(+), 2 deletions(-) diff --git a/custom_components/nestquest/db.py b/custom_components/nestquest/db.py index 3fcf6bc..65163e8 100644 --- a/custom_components/nestquest/db.py +++ b/custom_components/nestquest/db.py @@ -199,7 +199,10 @@ class NestQuestDatabase: 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. + 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. """ if self._in_transaction: raise RuntimeError( @@ -229,7 +232,21 @@ class NestQuestDatabase: def _commit() -> None: conn.execute("COMMIT") - await self._hass.async_add_executor_job(_commit) + try: + await self._hass.async_add_executor_job(_commit) + except BaseException as commit_error: + + def _rollback() -> None: + conn.execute("ROLLBACK") + + try: + await self._hass.async_add_executor_job(_rollback) + except BaseException as rollback_error: + raise RuntimeError( + "COMMIT failed and the follow-up ROLLBACK also " + "raised; the transaction may need manual cleanup" + ) from rollback_error + raise finally: self._in_transaction = False self._tx_task = None diff --git a/tests/test_db.py b/tests/test_db.py index 0ef0ea8..fbd75a1 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -200,8 +200,11 @@ async def test_real_executor_never_touches_loop_thread(tmp_path, monkeypatch) -> 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) @@ -219,6 +222,9 @@ async def test_real_executor_never_touches_loop_thread(tmp_path, monkeypatch) -> 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))}" @@ -357,6 +363,49 @@ def test_transaction_rolls_back_on_exception(tmp_path) -> None: _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_transaction_persists_across_connections(tmp_path) -> None: """A committed transaction is visible to a brand-new connection.""" hass, _registry = make_hass() -- 2.55.0 From 863a144e934e2e051e7ac08edcf17f1193285ea0 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:24:20 -0400 Subject: [PATCH 4/6] fix: make transaction lifecycle cancellation-safe and preserve commit+rollback errors --- custom_components/nestquest/db.py | 128 ++++++++++++++++++++---- tests/test_db.py | 157 ++++++++++++++++++++++++++++++ 2 files changed, 267 insertions(+), 18 deletions(-) diff --git a/custom_components/nestquest/db.py b/custom_components/nestquest/db.py index 65163e8..cd1c93b 100644 --- a/custom_components/nestquest/db.py +++ b/custom_components/nestquest/db.py @@ -3,7 +3,7 @@ from __future__ import annotations import asyncio import sqlite3 -from collections.abc import AsyncIterator, Sequence +from collections.abc import AsyncIterator, Awaitable, Sequence from contextlib import asynccontextmanager from dataclasses import dataclass from pathlib import Path @@ -203,6 +203,14 @@ class NestQuestDatabase: 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( @@ -218,7 +226,13 @@ class NestQuestDatabase: def _begin() -> None: conn.execute("BEGIN") - await self._hass.async_add_executor_job(_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: @@ -226,31 +240,109 @@ class NestQuestDatabase: def _rollback() -> None: conn.execute("ROLLBACK") - await self._hass.async_add_executor_job(_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") - try: - await self._hass.async_add_executor_job(_commit) - except BaseException as commit_error: - - def _rollback() -> None: - conn.execute("ROLLBACK") - - try: - await self._hass.async_add_executor_job(_rollback) - except BaseException as rollback_error: - raise RuntimeError( - "COMMIT failed and the follow-up ROLLBACK also " - "raised; the transaction may need manual cleanup" - ) from rollback_error - raise + 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. + """ + 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 + if cancelled: + raise asyncio.CancelledError() + return result + async def close(self) -> None: """Close the connection in the executor; safe to call repeatedly. diff --git a/tests/test_db.py b/tests/test_db.py index fbd75a1..61d4977 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -406,6 +406,163 @@ async def test_commit_failure_rolls_back_and_next_transaction_works( 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_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() -- 2.55.0 From 97c8040e057fa17c3ed43f72f82ba1ccc4072aeb Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:28:57 -0400 Subject: [PATCH 5/6] fix: settle all executor jobs before releasing the wrapper lock --- custom_components/nestquest/db.py | 29 +++++++++---- tests/test_db.py | 67 +++++++++++++++++++++++++++++++ 2 files changed, 88 insertions(+), 8 deletions(-) diff --git a/custom_components/nestquest/db.py b/custom_components/nestquest/db.py index cd1c93b..bd785a4 100644 --- a/custom_components/nestquest/db.py +++ b/custom_components/nestquest/db.py @@ -41,6 +41,13 @@ class NestQuestDatabase: 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: @@ -122,7 +129,9 @@ class NestQuestDatabase: raise return conn - self._conn = await self._hass.async_add_executor_job(_open) + self._conn = await self._await_settled( + self._hass.async_add_executor_job(_open) + ) return self async def execute( @@ -136,7 +145,9 @@ class NestQuestDatabase: cursor = conn.execute(sql, parameters) return ExecutionResult(cursor.rowcount, cursor.lastrowid) - result = await self._hass.async_add_executor_job(_execute) + result = await self._await_settled( + self._hass.async_add_executor_job(_execute) + ) self._rowcount = result.rowcount self._lastrowid = result.lastrowid return result @@ -152,7 +163,9 @@ class NestQuestDatabase: cursor = conn.executemany(sql, parameters) return ExecutionResult(cursor.rowcount, cursor.lastrowid) - result = await self._hass.async_add_executor_job(_execute_many) + result = await self._await_settled( + self._hass.async_add_executor_job(_execute_many) + ) self._rowcount = result.rowcount self._lastrowid = result.lastrowid return result @@ -168,8 +181,8 @@ class NestQuestDatabase: cursor = conn.execute(sql, parameters) return cursor.fetchone(), cursor.rowcount, cursor.lastrowid - row, rowcount, lastrowid = await self._hass.async_add_executor_job( - _fetch_one + row, rowcount, lastrowid = await self._await_settled( + self._hass.async_add_executor_job(_fetch_one) ) self._rowcount = rowcount self._lastrowid = lastrowid @@ -186,8 +199,8 @@ class NestQuestDatabase: cursor = conn.execute(sql, parameters) return cursor.fetchall(), cursor.rowcount, cursor.lastrowid - rows, rowcount, lastrowid = await self._hass.async_add_executor_job( - _fetch_all + rows, rowcount, lastrowid = await self._await_settled( + self._hass.async_add_executor_job(_fetch_all) ) self._rowcount = rowcount self._lastrowid = lastrowid @@ -367,7 +380,7 @@ class NestQuestDatabase: def _close() -> None: conn.close() - await self._hass.async_add_executor_job(_close) + await self._await_settled(self._hass.async_add_executor_job(_close)) def _require_conn(self) -> sqlite3.Connection: if self._conn is None: diff --git a/tests/test_db.py b/tests/test_db.py index 61d4977..8aceb8c 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -525,6 +525,73 @@ async def test_cancel_during_commit_settles_then_propagates_cancellation( 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. -- 2.55.0 From c646446f33d89c97bbf3af2d6a69e88d3ca06c68 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:33:43 -0400 Subject: [PATCH 6/6] fix: close abandoned connection when open is cancelled after settle --- custom_components/nestquest/db.py | 42 +++++++++++++++++--- tests/test_db.py | 65 +++++++++++++++++++++++++++++++ 2 files changed, 102 insertions(+), 5 deletions(-) diff --git a/custom_components/nestquest/db.py b/custom_components/nestquest/db.py index bd785a4..c05af6b 100644 --- a/custom_components/nestquest/db.py +++ b/custom_components/nestquest/db.py @@ -110,7 +110,16 @@ class NestQuestDatabase: yield async def open(self, path: Path | str) -> NestQuestDatabase: - """Open the SQLite file with WAL journal mode and foreign keys on.""" + """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(): @@ -129,9 +138,20 @@ class NestQuestDatabase: raise return conn - self._conn = await self._await_settled( + 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( @@ -336,6 +356,20 @@ class NestQuestDatabase: 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(): @@ -352,9 +386,7 @@ class NestQuestDatabase: [asyncio.CancelledError(), job_error], ) from job_error raise - if cancelled: - raise asyncio.CancelledError() - return result + return result, cancelled async def close(self) -> None: """Close the connection in the executor; safe to call repeatedly. diff --git a/tests/test_db.py b/tests/test_db.py index 8aceb8c..7640c57 100644 --- a/tests/test_db.py +++ b/tests/test_db.py @@ -406,6 +406,71 @@ async def test_commit_failure_rolls_back_and_next_transaction_works( 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. -- 2.55.0