From a56a7e3d3710166992f2cbe58b9cd1e292920528 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Sun, 13 Sep 2026 21:07:42 -0400 Subject: [PATCH] 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