From c6d42e8277e006490d97879b87fa297d7424291f Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 06:54:45 -0400 Subject: [PATCH 1/5] feat: add versioned schema migration runner (#17) --- custom_components/nestquest/migrations.py | 161 +++++++++ tests/test_migrations.py | 388 ++++++++++++++++++++++ 2 files changed, 549 insertions(+) create mode 100644 custom_components/nestquest/migrations.py create mode 100644 tests/test_migrations.py diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py new file mode 100644 index 0000000..9866421 --- /dev/null +++ b/custom_components/nestquest/migrations.py @@ -0,0 +1,161 @@ +"""Versioned schema migration runner for the NestQuest SQLite database. + +The runner owns one metadata table, ``nestquest_schema_version``, holding +a single row with the schema version the database file currently +reflects. Migrations are declared as an ordered list: entry ``n`` +upgrades version ``n-1`` to version ``n``. Applying a migration means +running its DDL inside one transaction and stamping the new version in +that same transaction, so a database never persists a half-applied +migration: any failure rolls the statements and the version stamp back +together, leaving the file at its prior version. + +Version 0 means "the file exists but carries no NestQuest schema yet" +— a fresh file, or a file this integration has never written. Starting +the runner against version 0 applies migration 1, which applies the +full v1 DDL from :mod:`.schema` and stamps version 1. + +Each migration's statements and its version stamp run inside ONE +transaction, and the stamp is written as DELETE + INSERT rather than +UPDATE because a version-0 database has no version row to update yet. +An UPDATE form would silently match zero rows there and leave the +database migrated but unstamped — reporting version 0 forever, and +re-applying migration 1 on every startup. The DELETE+INSERT shape is +also what makes the stamp a replace, so the row never duplicates. + +Idempotence across restarts: a database already at the latest version +has no pending migrations, so the runner opens a read-only check, finds +nothing to do, and leaves the file byte-untouched (no write transaction +is opened, so no WAL churn). + +Migration list policy: entries are append-only. Once shipped, a +migration's SQL must never be edited — an old database applying it for +the first time and a fresh database would then diverge. Fix-forward +with a new migration instead. The version stamp lives in the same +SQLite file as the data it describes, so restoring a backup restores +its matching schema version with it. + +Corrupt or missing files are not this module's concern: startup +handling of those cases is task d5bca74b, which decides whether to +open fresh or raise ConfigEntryNotReady. This runner assumes it is +handed an open, working connection. +""" +from __future__ import annotations + +import logging +from collections.abc import Sequence + +from .db import NestQuestDatabase +from .schema import SCHEMA_V1_STATEMENTS + +LOGGER = logging.getLogger(__name__) + +#: Metadata table holding the applied schema version. Prefixed with the +#: integration domain so a config directory shared with other tools can +#: never collide with a foreign ``schema_version`` table. +VERSION_TABLE = "nestquest_schema_version" + +#: One-based index into :data:`MIGRATIONS`: entry ``n`` upgrades +#: version ``n-1`` to version ``n``. +MIGRATION_1_V1_DDL: list[str] = list(SCHEMA_V1_STATEMENTS) + +#: The ordered migration list. Append-only: never edit an applied +#: entry, add the next one instead. +MIGRATIONS: Sequence[Sequence[str]] = [ + MIGRATION_1_V1_DDL, +] + + +async def _create_version_table(database: NestQuestDatabase) -> None: + """Create the version metadata table if it does not exist. + + The table is created outside the per-migration transaction: it is + the runner's own bookkeeping, needed before any migration can stamp + a version, and creating it is itself idempotent. + """ + await database.execute( + f""" + CREATE TABLE IF NOT EXISTS {VERSION_TABLE} ( + version INTEGER NOT NULL + ) + """ + ) + + +async def read_schema_version(database: NestQuestDatabase) -> int: + """Return the schema version recorded in the database. + + Returns 0 when the version table or its row is missing — the + "fresh file" case. A stored version higher than the runner knows + is surfaced as-is; the caller decides how to treat a future version. + + Reading is deliberately passive: a missing table raises, it is not + swallowed into 0, because :func:`apply_migrations` always creates + the table before reading and a direct reader hitting "no such + table" is a caller bug worth surfacing, not a fresh file. + """ + row = await database.fetch_one( + f"SELECT version FROM {VERSION_TABLE} LIMIT 1" + ) + if row is None: + return 0 + version = row[0] + if not isinstance(version, int): + raise RuntimeError( + f"Schema version metadata is not an integer: {version!r}" + ) + return version + + +async def apply_migrations( + database: NestQuestDatabase, + migrations: Sequence[Sequence[str]] = MIGRATIONS, +) -> int: + """Bring the database schema up to the latest version. + + Reads the current version, applies every pending migration in order + — each inside one transaction that also stamps the new version — + and returns the resulting version. Applying zero pending + migrations (database already current) is a no-op that logs at debug + level and opens no write transaction. + + Raises whatever the failing migration raised, with the database + rolled back to its prior version and prior rows untouched. + """ + await _create_version_table(database) + current = await read_schema_version(database) + latest = len(migrations) + + if current > latest: + raise RuntimeError( + f"Database schema version {current} is newer than this " + f"integration understands (latest known: {latest}); " + "downgrades are not supported" + ) + + if current == latest: + LOGGER.debug( + "NestQuest schema already at version %d; no migrations to apply", + current, + ) + return current + + for target in range(current + 1, latest + 1): + statements = migrations[target - 1] + async with database.transaction(): + for sql in statements: + await database.execute(sql) + # DELETE + INSERT, not UPDATE: a fresh file has no version + # row yet, and UPDATE would silently match zero rows. + await database.execute(f"DELETE FROM {VERSION_TABLE}") + await database.execute( + f"INSERT INTO {VERSION_TABLE} (version) VALUES (?)", + (target,), + ) + LOGGER.info( + "NestQuest schema migrated to version %d (%d statement%s)", + target, + len(statements), + "s" if len(statements) != 1 else "", + ) + + return latest \ No newline at end of file diff --git a/tests/test_migrations.py b/tests/test_migrations.py new file mode 100644 index 0000000..fa08504 --- /dev/null +++ b/tests/test_migrations.py @@ -0,0 +1,388 @@ +"""Tests for migrations.py: the versioned schema migration runner.""" +from __future__ import annotations + +import asyncio +import logging +import sqlite3 +from unittest.mock import AsyncMock + +import pytest + +from custom_components.nestquest.db import NestQuestDatabase +from custom_components.nestquest.migrations import ( + MIGRATIONS, + VERSION_TABLE, + apply_migrations, + read_schema_version, +) +from custom_components.nestquest.schema import SCHEMA_V1_STATEMENTS + + +def _run(coro): + return asyncio.new_event_loop().run_until_complete(coro) + + +def _make_hass_mock(): + from unittest.mock import AsyncMock, MagicMock + + hass = MagicMock() + hass.async_add_executor_job = AsyncMock(side_effect=(lambda fn, *a: fn(*a))) + return hass + + +def _open_db(path) -> NestQuestDatabase: + database = NestQuestDatabase(_make_hass_mock()) + _run(database.open(path)) + return database + + +def _migrate(database, migrations=MIGRATIONS): + return _run(apply_migrations(database, migrations)) + + +def _version(database) -> int: + return _run(read_schema_version(database)) + + +def _tables(database) -> set[str]: + rows = _run( + database.fetch_all( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name NOT LIKE 'sqlite_%'" + ) + ) + return {row[0] for row in rows} + + +def _counts(database) -> dict[str, int]: + """Row counts for every v1 data table plus the version table.""" + tables = ( + "children", + "admin_users", + "schedule_rules", + "task_definitions", + "presence_schedules", + "presence_overrides", + "task_instances", + "completion_events", + ) + counts = {} + for table in tables: + row = _run(database.fetch_one(f"SELECT COUNT(*) FROM {table}")) + counts[table] = row[0] + return counts + + +# --------------------------------------------------------------------------- +# Fresh file: version 0 -> version 1 +# --------------------------------------------------------------------------- + + +def test_fresh_file_migrates_to_version_1(tmp_path) -> None: + database = _open_db(tmp_path / "fresh.db") + try: + final = _migrate(database) + assert final == 1 + assert _version(database) == 1 + finally: + _run(database.close()) + + +def test_fresh_file_migration_creates_all_v1_tables(tmp_path) -> None: + database = _open_db(tmp_path / "fresh-tables.db") + try: + _migrate(database) + expected = { + "children", + "admin_users", + "schedule_rules", + "task_definitions", + "presence_schedules", + "presence_overrides", + "task_instances", + "completion_events", + VERSION_TABLE, + } + assert _tables(database) == expected + finally: + _run(database.close()) + + +def test_missing_version_table_raises_on_direct_read(tmp_path) -> None: + database = _open_db(tmp_path / "no-table.db") + try: + with pytest.raises(sqlite3.OperationalError, match="no such table"): + _run(read_schema_version(database)) + row = _run( + database.fetch_one( + "SELECT name FROM sqlite_master WHERE type = 'table' " + f"AND name = '{VERSION_TABLE}'" + ) + ) + assert row is None, "read_schema_version must not create the table" + finally: + _run(database.close()) + + +def test_empty_version_table_reads_as_version_0(tmp_path) -> None: + database = _open_db(tmp_path / "empty-table.db") + try: + _run( + database.execute( + f"CREATE TABLE {VERSION_TABLE} (version INTEGER NOT NULL)" + ) + ) + assert _version(database) == 0 + finally: + _run(database.close()) + + +def test_migration_1_is_the_full_v1_ddl() -> None: + assert MIGRATIONS[0] == SCHEMA_V1_STATEMENTS + assert len(MIGRATIONS) == 1 + + +# --------------------------------------------------------------------------- +# Idempotence: second start is a no-op +# --------------------------------------------------------------------------- + + +def test_second_run_applies_nothing_and_keeps_rows(tmp_path) -> None: + database = _open_db(tmp_path / "idempotent.db") + try: + _migrate(database) + _run( + database.execute( + "INSERT INTO children (display_name, created_at) VALUES (?, ?)", + ("Ada", "2026-09-13T00:00:00+00:00"), + ) + ) + _run( + database.execute( + "INSERT INTO admin_users (ha_user_id, added_at) VALUES (?, ?)", + ("user-1", "2026-09-13T00:00:00+00:00"), + ) + ) + final = _migrate(database) + assert final == 1 + assert _version(database) == 1 + assert _counts(database)["children"] == 1 + assert _counts(database)["admin_users"] == 1 + finally: + _run(database.close()) + + +def test_second_run_writes_nothing_to_the_file(tmp_path, caplog) -> None: + database = _open_db(tmp_path / "noop.db") + try: + _migrate(database) + db_file = tmp_path / "noop.db" + size_before = db_file.stat().st_size + mtime_before = db_file.stat().st_mtime_ns + + _migrate(database) + + # A no-op run must not open a write transaction: the main + # database file is byte-identical and untouched (mtime too), and + # no WAL checkpoint traffic dirtied it. + assert db_file.stat().st_size == size_before + assert db_file.stat().st_mtime_ns == mtime_before + finally: + _run(database.close()) + + +def test_noop_run_logs_no_migration(caplog, tmp_path) -> None: + database = _open_db(tmp_path / "noop-log.db") + try: + _migrate(database) + caplog.clear() + with caplog.at_level(logging.INFO, logger="custom_components.nestquest.migrations"): + _migrate(database) + migrated = [ + r for r in caplog.records if "migrated" in r.getMessage().lower() + ] + assert migrated == [] + finally: + _run(database.close()) + + +def test_noop_run_is_logged_at_debug_level(caplog, tmp_path) -> None: + database = _open_db(tmp_path / "noop-debug.db") + try: + _migrate(database) + caplog.clear() + with caplog.at_level(logging.DEBUG, logger="custom_components.nestquest.migrations"): + _migrate(database) + assert any( + "already at version 1" in r.getMessage() + for r in caplog.records + ) + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Failure atomicity: a failing migration leaves the prior version intact +# --------------------------------------------------------------------------- + + +def _failing_migrations(prior_statements: list[str]) -> list[list[str]]: + """Two-migration list where migration 2 fails mid-application.""" + return [ + prior_statements, + [ + "CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)", + "CREATE TABLE broken (", + ], + ] + + +def test_failing_migration_rolls_back_to_prior_version(tmp_path) -> None: + database = _open_db(tmp_path / "rollback.db") + try: + # Manually simulate a database at version 1: apply migration 1's + # statements directly and stamp version 1 through the runner. + _migrate(database) + version_row = _run( + database.fetch_one(f"SELECT version FROM {VERSION_TABLE}") + ) + assert version_row == (1,) + + with pytest.raises(sqlite3.OperationalError): + _migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS)) + + # Version stamp unchanged, statements from the failed migration + # absent, and the v1 tables untouched. + assert _version(database) == 1 + assert "later_table" not in _tables(database) + assert _counts(database)["children"] == 0 + assert _counts(database)["completion_events"] == 0 + finally: + _run(database.close()) + + +def test_failing_migration_on_fresh_file_leaves_version_0(tmp_path) -> None: + database = _open_db(tmp_path / "fresh-rollback.db") + try: + # The fresh file still reaches migration 2 (the failing one): + # migration 1 commits, stamping version 1, before migration 2 + # fails and rolls back. The run aborts at version 1 — the last + # fully-applied migration — not at version 0. + with pytest.raises(sqlite3.OperationalError): + _migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS)) + assert _version(database) == 1 + # Migration 2's tables must not survive the failed attempt. + assert "later_table" not in _tables(database) + # Migration 1's tables do: it committed before the failure. + assert "children" in _tables(database) + assert "task_instances" in _tables(database) + finally: + _run(database.close()) + + +def test_failed_migration_can_be_retried_after_fix(tmp_path) -> None: + database = _open_db(tmp_path / "retry.db") + try: + _migrate(database) + broken = _failing_migrations(SCHEMA_V1_STATEMENTS) + with pytest.raises(sqlite3.OperationalError): + _migrate(database, broken) + # "Fix" the migration and retry: it now applies cleanly. + broken[1] = [ + "CREATE TABLE IF NOT EXISTS later_table (id INTEGER PRIMARY KEY)" + ] + final = _migrate(database, broken) + assert final == 2 + assert _version(database) == 2 + assert "later_table" in _tables(database) + finally: + _run(database.close()) + + +def test_failing_migration_error_is_preserved(tmp_path) -> None: + database = _open_db(tmp_path / "error-preserved.db") + try: + with pytest.raises(sqlite3.OperationalError, match="incomplete input"): + _migrate(database, _failing_migrations(SCHEMA_V1_STATEMENTS)) + finally: + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Version guard rails +# --------------------------------------------------------------------------- + + +def test_future_version_is_rejected(tmp_path) -> None: + database = _open_db(tmp_path / "future.db") + try: + _migrate(database) + _run( + database.execute( + f"UPDATE {VERSION_TABLE} SET version = ?", (99,) + ) + ) + with pytest.raises(RuntimeError, match="newer than"): + _migrate(database) + # Nothing was modified by the rejected run. + assert _version(database) == 99 + finally: + _run(database.close()) + + +def test_non_integer_version_raises_runtime_error(tmp_path) -> None: + database = _open_db(tmp_path / "string-version.db") + try: + _run( + database.execute( + f"CREATE TABLE {VERSION_TABLE} (version TEXT NOT NULL)" + ) + ) + _run( + database.execute( + f"INSERT INTO {VERSION_TABLE} (version) VALUES (?)", + ("one",), + ) + ) + with pytest.raises(RuntimeError, match="not an integer"): + _run(read_schema_version(database)) + finally: + _run(database.close()) + + +def test_multiple_pending_migrations_apply_in_order(tmp_path) -> None: + database = _open_db(tmp_path / "multi-step.db") + try: + migrations = [ + ["CREATE TABLE IF NOT EXISTS step_a (id INTEGER PRIMARY KEY)"], + ["CREATE TABLE IF NOT EXISTS step_b (id INTEGER PRIMARY KEY)"], + ["CREATE TABLE IF NOT EXISTS step_c (id INTEGER PRIMARY KEY)"], + ] + final = _run(apply_migrations(database, migrations)) + assert final == 3 + assert _version(database) == 3 + assert {"step_a", "step_b", "step_c"} <= _tables(database) + finally: + _run(database.close()) + + +def test_partial_pending_migrations_resume_from_current_version( + tmp_path, +) -> None: + database = _open_db(tmp_path / "resume.db") + try: + migrations = [ + ["CREATE TABLE IF NOT EXISTS step_a (id INTEGER PRIMARY KEY)"], + ["CREATE TABLE IF NOT EXISTS step_b (id INTEGER PRIMARY KEY)"], + ] + # Advance to version 1 only, by applying a truncated list of the + # same migrations: patch apply_migrations' latest by passing a + # single-migration list first, then the full list. + _run(apply_migrations(database, migrations[:1])) + assert _version(database) == 1 + final = _run(apply_migrations(database, migrations)) + assert final == 2 + assert _version(database) == 2 + assert "step_b" in _tables(database) + finally: + _run(database.close()) \ No newline at end of file -- 2.55.0 From 492c1149f04f68795f0635ac5e0fdb98904f0eda Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 06:59:30 -0400 Subject: [PATCH 2/5] fix: address review findings - startup wiring, singleton version table, version guards - __init__.py now calls apply_migrations() on setup instead of raw DDL (HIGH finding: the runner was never invoked at startup); lifecycle wiring tests in test_migrations.py cover fresh migrate, reload reuse, and close-on-failure - VERSION_TABLE DDL enforces singleton row (id=1 PRIMARY KEY + CHECK) and version >= 1; stamp is INSERT .. ON CONFLICT DO UPDATE keyed on the singleton row - read_schema_version rejects non-integer and sub-1 versions with explicit guards (bool excluded) - no-op test asserts content hash + mtime, not just size --- custom_components/nestquest/__init__.py | 4 +- custom_components/nestquest/migrations.py | 46 +++-- tests/test_migrations.py | 222 ++++++++++++++++++++-- 3 files changed, 239 insertions(+), 33 deletions(-) diff --git a/custom_components/nestquest/__init__.py b/custom_components/nestquest/__init__.py index d4c16f5..8b8454d 100644 --- a/custom_components/nestquest/__init__.py +++ b/custom_components/nestquest/__init__.py @@ -11,7 +11,7 @@ from homeassistant.core import HomeAssistant from .const import DOMAIN, LOGGER from .db import NestQuestDatabase -from .schema import SCHEMA_V1_STATEMENTS, async_apply_ddl +from .migrations import apply_migrations from .store import async_get_db_path _LOGGER = LOGGER @@ -52,7 +52,7 @@ async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool: database = await NestQuestDatabase(hass).open(await async_get_db_path(hass)) try: - await async_apply_ddl(database, SCHEMA_V1_STATEMENTS) + await apply_migrations(database) remove_update_listener = entry.add_update_listener(_async_update_listener) except BaseException: await database.close() diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py index 9866421..8f35717 100644 --- a/custom_components/nestquest/migrations.py +++ b/custom_components/nestquest/migrations.py @@ -54,6 +54,20 @@ LOGGER = logging.getLogger(__name__) #: never collide with a foreign ``schema_version`` table. VERSION_TABLE = "nestquest_schema_version" +#: DDL for the version metadata table. The singleton-row invariant is +#: enforced by the schema itself: ``id`` is fixed at 1 by CHECK, keyed +#: PRIMARY KEY, so the table can never hold a second row, and the +#: stamp write is an UPDATE (no DELETE+INSERT needed). ``version`` is +#: a positive integer: version 0 means "no row" and is never stored, +#: so a negative or zero stored value is corruption, not a state the +#: runner can act on. +VERSION_TABLE_DDL = f""" + CREATE TABLE IF NOT EXISTS {VERSION_TABLE} ( + id INTEGER NOT NULL PRIMARY KEY CHECK (id = 1), + version INTEGER NOT NULL CHECK (version >= 1) + ) +""" + #: One-based index into :data:`MIGRATIONS`: entry ``n`` upgrades #: version ``n-1`` to version ``n``. MIGRATION_1_V1_DDL: list[str] = list(SCHEMA_V1_STATEMENTS) @@ -68,17 +82,12 @@ MIGRATIONS: Sequence[Sequence[str]] = [ async def _create_version_table(database: NestQuestDatabase) -> None: """Create the version metadata table if it does not exist. - The table is created outside the per-migration transaction: it is - the runner's own bookkeeping, needed before any migration can stamp - a version, and creating it is itself idempotent. + The table is created before any version check: it is the runner's + own bookkeeping, and creating it is idempotent. The DDL enforces + the singleton-row invariant (``id`` fixed at 1) and a positive + integer version at the database level. """ - await database.execute( - f""" - CREATE TABLE IF NOT EXISTS {VERSION_TABLE} ( - version INTEGER NOT NULL - ) - """ - ) + await database.execute(VERSION_TABLE_DDL) async def read_schema_version(database: NestQuestDatabase) -> int: @@ -99,10 +108,14 @@ async def read_schema_version(database: NestQuestDatabase) -> int: if row is None: return 0 version = row[0] - if not isinstance(version, int): + if not isinstance(version, int) or isinstance(version, bool): raise RuntimeError( f"Schema version metadata is not an integer: {version!r}" ) + if version < 1: + raise RuntimeError( + f"Schema version metadata is invalid (must be >= 1): {version!r}" + ) return version @@ -144,11 +157,14 @@ async def apply_migrations( async with database.transaction(): for sql in statements: await database.execute(sql) - # DELETE + INSERT, not UPDATE: a fresh file has no version - # row yet, and UPDATE would silently match zero rows. - await database.execute(f"DELETE FROM {VERSION_TABLE}") + # UPDATE keyed on the singleton row (id = 1): a fresh file + # has no row yet, but migration 1 is the only migration a + # version-0 database can run, so the row is seeded by the + # same statement batch when needed. UPDATE alone would + # silently match zero rows there, so seed-if-absent first. await database.execute( - f"INSERT INTO {VERSION_TABLE} (version) VALUES (?)", + f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, ?) " + "ON CONFLICT (id) DO UPDATE SET version = excluded.version", (target,), ) LOGGER.info( diff --git a/tests/test_migrations.py b/tests/test_migrations.py index fa08504..7a89951 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -2,6 +2,7 @@ from __future__ import annotations import asyncio +import hashlib import logging import sqlite3 from unittest.mock import AsyncMock @@ -12,6 +13,7 @@ from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.migrations import ( MIGRATIONS, VERSION_TABLE, + VERSION_TABLE_DDL, apply_migrations, read_schema_version, ) @@ -127,11 +129,7 @@ def test_missing_version_table_raises_on_direct_read(tmp_path) -> None: def test_empty_version_table_reads_as_version_0(tmp_path) -> None: database = _open_db(tmp_path / "empty-table.db") try: - _run( - database.execute( - f"CREATE TABLE {VERSION_TABLE} (version INTEGER NOT NULL)" - ) - ) + _run(database.execute(VERSION_TABLE_DDL)) assert _version(database) == 0 finally: _run(database.close()) @@ -177,16 +175,18 @@ def test_second_run_writes_nothing_to_the_file(tmp_path, caplog) -> None: try: _migrate(database) db_file = tmp_path / "noop.db" - size_before = db_file.stat().st_size + digest_before = hashlib.sha256(db_file.read_bytes()).hexdigest() mtime_before = db_file.stat().st_mtime_ns _migrate(database) # A no-op run must not open a write transaction: the main - # database file is byte-identical and untouched (mtime too), and - # no WAL checkpoint traffic dirtied it. - assert db_file.stat().st_size == size_before + # database file is byte-identical (content hash, not just size) + # and untouched (mtime too). assert db_file.stat().st_mtime_ns == mtime_before + assert ( + hashlib.sha256(db_file.read_bytes()).hexdigest() == digest_before + ) finally: _run(database.close()) @@ -333,14 +333,13 @@ def test_future_version_is_rejected(tmp_path) -> None: def test_non_integer_version_raises_runtime_error(tmp_path) -> None: database = _open_db(tmp_path / "string-version.db") try: + _run(database.execute(VERSION_TABLE_DDL)) + # The CHECK enforces version >= 1 but SQLite's loose typing + # stores TEXT under an INTEGER-affinity column when it cannot + # be losslessly coerced; guard the read path against it. _run( database.execute( - f"CREATE TABLE {VERSION_TABLE} (version TEXT NOT NULL)" - ) - ) - _run( - database.execute( - f"INSERT INTO {VERSION_TABLE} (version) VALUES (?)", + f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, ?)", ("one",), ) ) @@ -350,6 +349,95 @@ def test_non_integer_version_raises_runtime_error(tmp_path) -> None: _run(database.close()) +def test_negative_version_read_guard_rejects_below_one() -> None: + """The read path rejects sub-1 versions even if storage missed them. + + The VERSION_TABLE CHECK already makes negative/zero unstorable; this + exercises the defense-in-depth guard in read_schema_version itself + against a database stub returning such a row (e.g. a future schema + relaxing the CHECK, or a hand-edited file). + """ + from unittest.mock import AsyncMock, MagicMock + + for bad in (-1, 0): + database = MagicMock() + database.fetch_one = AsyncMock(return_value=(bad,)) + with pytest.raises(RuntimeError, match="must be >= 1"): + _run(read_schema_version(database)) + + +def test_non_integer_version_read_guard_rejects_text() -> None: + from unittest.mock import AsyncMock, MagicMock + + for bad in ("one", 1.5, True): + database = MagicMock() + database.fetch_one = AsyncMock(return_value=(bad,)) + with pytest.raises(RuntimeError, match="not an integer"): + _run(read_schema_version(database)) + + +def test_zero_version_row_is_unstorable(tmp_path) -> None: + """The CHECK makes sub-1 versions impossible through SQL. + + PRAGMA writable_schema does not bypass CHECKs on ordinary tables, + so the negative/zero read guard above is pure defense-in-depth; + here we prove the storage-level rule the normal path relies on. + """ + database = _open_db(tmp_path / "zero-version.db") + try: + _run(database.execute(VERSION_TABLE_DDL)) + for bad in (0, -1): + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _run( + database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) " + "VALUES (1, ?)", + (bad,), + ) + ) + finally: + _run(database.close()) + + +def test_version_table_enforces_singleton_row(tmp_path) -> None: + database = _open_db(tmp_path / "singleton.db") + try: + _run(database.execute(VERSION_TABLE_DDL)) + _run( + database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, 1)" + ) + ) + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _run( + database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) " + "VALUES (2, 1)" + ) + ) + finally: + _run(database.close()) + + +def test_version_table_check_rejects_version_zero_and_negative( + tmp_path, +) -> None: + database = _open_db(tmp_path / "version-check.db") + try: + _run(database.execute(VERSION_TABLE_DDL)) + for bad in (0, -1): + with pytest.raises(sqlite3.IntegrityError, match="CHECK"): + _run( + database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) " + "VALUES (1, ?)", + (bad,), + ) + ) + finally: + _run(database.close()) + + def test_multiple_pending_migrations_apply_in_order(tmp_path) -> None: database = _open_db(tmp_path / "multi-step.db") try: @@ -385,4 +473,106 @@ def test_partial_pending_migrations_resume_from_current_version( assert _version(database) == 2 assert "step_b" in _tables(database) finally: - _run(database.close()) \ No newline at end of file + _run(database.close()) + + +# --------------------------------------------------------------------------- +# Startup wiring: setup runs the migration runner, not raw DDL +# --------------------------------------------------------------------------- + + +async def _setup_entry(hass, entry, registry) -> object: + from tests.test_lifecycle import _wire # reuse the established wiring + + entry = _wire(entry, registry) + from custom_components.nestquest import async_setup_entry + + await async_setup_entry(hass, entry) + return entry + + +async def test_setup_entry_migrates_fresh_database(hass, make_entry) -> None: + import sqlite3 as sqlite3_mod + + from custom_components.nestquest.migrations import read_schema_version + from custom_components.nestquest.store import async_get_db_path + + entry = await _setup_entry(hass, make_entry(), hass.registry) + database = entry.runtime_data.database + assert await read_schema_version(database) == 1 + # The v1 tables exist because the runner applied migration 1. + names = { + row[0] + for row in await database.fetch_all( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name NOT LIKE 'sqlite_%'" + ) + } + assert "children" in names and "completion_events" in names + # The version table exists with its singleton row stamped. + columns = await database.fetch_all(f"PRAGMA table_info({VERSION_TABLE})") + assert [(row[1], row[3]) for row in columns] == [("id", 1), ("version", 1)] + db_path = await async_get_db_path(hass) + conn = sqlite3_mod.connect(db_path) + try: + conn.execute(f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (2, 1)") + violated = False + except sqlite3_mod.IntegrityError: + violated = True + finally: + conn.close() + assert violated, "singleton-row CHECK must be live in the setup-created DB" + await database.close() + + +async def test_setup_entry_twice_keeps_version_and_rows( + hass, make_entry +) -> None: + from custom_components.nestquest.migrations import read_schema_version + + entry_a = await _setup_entry(hass, make_entry(), hass.registry) + database_a = entry_a.runtime_data.database + await database_a.execute( + "INSERT INTO children (display_name, created_at) VALUES (?, ?)", + ("Ada", "2026-09-13T00:00:00+00:00"), + ) + + # A reload re-runs setup for the same entry_id against the same + # file: setup reuses the domain-data record and its still-open + # connection rather than reopening, and nothing is re-applied. + entry_b = await _setup_entry(hass, make_entry(), hass.registry) + record_a = hass.data["nestquest"][entry_a.entry_id] + record_b = hass.data["nestquest"][entry_b.entry_id] + assert record_a is record_b + assert record_a.database is database_a + assert await read_schema_version(database_a) == 1 + row = await database_a.fetch_one("SELECT COUNT(*) FROM children") + assert row == (1,) + await database_a.close() + + +async def test_setup_entry_failure_closes_database( + hass, make_entry, monkeypatch +) -> None: + import pytest as pytest_module + + import custom_components.nestquest as nestquest_module + from custom_components.nestquest.migrations import ( + MIGRATIONS as _MIGRATIONS, + ) + + async def _failing(database, migrations=_MIGRATIONS): + raise RuntimeError("boom") + + monkeypatch.setattr( + nestquest_module, "apply_migrations", _failing, raising=True + ) + entry = make_entry() + with pytest_module.raises(RuntimeError, match="boom"): + await _setup_entry(hass, entry, hass.registry) + # Setup closed the connection before re-raising: no leaked handle. + # The record only lands in hass.data after the listener is added, so + # on failure the connection must be gone from the wrapper itself. + record = hass.data["nestquest"].get(entry.entry_id) + if record is not None: + assert not record.database.connected \ No newline at end of file -- 2.55.0 From a679004720cbef580d2a994010c70f03a273fadc Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 07:01:49 -0400 Subject: [PATCH 3/5] test: hash WAL alongside main db in no-op test; exercise unload+setup reload lifecycle --- tests/test_migrations.py | 52 +++++++++++++++++++++++++++------------- 1 file changed, 36 insertions(+), 16 deletions(-) diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 7a89951..5e7e844 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -174,19 +174,33 @@ def test_second_run_writes_nothing_to_the_file(tmp_path, caplog) -> None: database = _open_db(tmp_path / "noop.db") try: _migrate(database) + _run(database.execute("PRAGMA wal_checkpoint(TRUNCATE)")) db_file = tmp_path / "noop.db" + wal_file = tmp_path / "noop.db-wal" digest_before = hashlib.sha256(db_file.read_bytes()).hexdigest() mtime_before = db_file.stat().st_mtime_ns + wal_digest_before = ( + hashlib.sha256(wal_file.read_bytes()).hexdigest() + if wal_file.exists() + else None + ) _migrate(database) # A no-op run must not open a write transaction: the main # database file is byte-identical (content hash, not just size) - # and untouched (mtime too). + # and untouched (mtime too), and the WAL gained no frames. assert db_file.stat().st_mtime_ns == mtime_before assert ( hashlib.sha256(db_file.read_bytes()).hexdigest() == digest_before ) + if wal_digest_before is None: + assert not wal_file.exists() + else: + assert ( + hashlib.sha256(wal_file.read_bytes()).hexdigest() + == wal_digest_before + ) finally: _run(database.close()) @@ -525,30 +539,36 @@ async def test_setup_entry_migrates_fresh_database(hass, make_entry) -> None: await database.close() -async def test_setup_entry_twice_keeps_version_and_rows( +async def test_options_reload_reopens_and_rechecks_version( hass, make_entry ) -> None: + """A listener-triggered reload re-runs setup against the same file. + + Unload then setup is the reload lifecycle: the old connection is + closed, apply_migrations runs again on the persisted file, keeps + version and rows intact, and the reopened connection is live. + """ + from custom_components.nestquest import async_unload_entry from custom_components.nestquest.migrations import read_schema_version - entry_a = await _setup_entry(hass, make_entry(), hass.registry) - database_a = entry_a.runtime_data.database + entry = await _setup_entry(hass, make_entry(), hass.registry) + database_a = entry.runtime_data.database await database_a.execute( "INSERT INTO children (display_name, created_at) VALUES (?, ?)", ("Ada", "2026-09-13T00:00:00+00:00"), ) + assert await async_unload_entry(hass, entry) is True + assert not database_a.connected - # A reload re-runs setup for the same entry_id against the same - # file: setup reuses the domain-data record and its still-open - # connection rather than reopening, and nothing is re-applied. - entry_b = await _setup_entry(hass, make_entry(), hass.registry) - record_a = hass.data["nestquest"][entry_a.entry_id] - record_b = hass.data["nestquest"][entry_b.entry_id] - assert record_a is record_b - assert record_a.database is database_a - assert await read_schema_version(database_a) == 1 - row = await database_a.fetch_one("SELECT COUNT(*) FROM children") - assert row == (1,) - await database_a.close() + entry_reloaded = await _setup_entry(hass, make_entry(), hass.registry) + database_b = entry_reloaded.runtime_data.database + assert database_b is not database_a + try: + assert await read_schema_version(database_b) == 1 + row = await database_b.fetch_one("SELECT COUNT(*) FROM children") + assert row == (1,) + finally: + await database_b.close() async def test_setup_entry_failure_closes_database( -- 2.55.0 From c0c540c1940b57281915a4990ef4797e9201e402 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 07:05:13 -0400 Subject: [PATCH 4/5] fix: re-read schema version inside each migration transaction; spy on close in setup-failure test - apply_migrations re-reads the version under the transaction lock before applying, so two interleaved runners on one connection cannot both apply a non-idempotent migration from a stale read - setup-failure test now spies on NestQuestDatabase.close and asserts it ran and no runtime record was stored (was vacuous) --- custom_components/nestquest/migrations.py | 64 ++++++++++++++++------- tests/test_migrations.py | 27 ++++++++-- 2 files changed, 68 insertions(+), 23 deletions(-) diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py index 8f35717..00526b9 100644 --- a/custom_components/nestquest/migrations.py +++ b/custom_components/nestquest/migrations.py @@ -131,6 +131,13 @@ async def apply_migrations( migrations (database already current) is a no-op that logs at debug level and opens no write transaction. + The version is re-read inside each migration transaction before + deciding to apply it: the outer read is advisory only (loop bounds, + no-op fast path), and on one shared connection two interleaved + runners would otherwise both read the same version and both apply + a non-idempotent migration. Under the transaction's lock the + second runner sees the first runner's stamp and skips. + Raises whatever the failing migration raised, with the database rolled back to its prior version and prior rows untouched. """ @@ -153,25 +160,46 @@ async def apply_migrations( return current for target in range(current + 1, latest + 1): - statements = migrations[target - 1] + applied = False async with database.transaction(): - for sql in statements: - await database.execute(sql) - # UPDATE keyed on the singleton row (id = 1): a fresh file - # has no row yet, but migration 1 is the only migration a - # version-0 database can run, so the row is seeded by the - # same statement batch when needed. UPDATE alone would - # silently match zero rows there, so seed-if-absent first. - await database.execute( - f"INSERT INTO {VERSION_TABLE} (id, version) VALUES (1, ?) " - "ON CONFLICT (id) DO UPDATE SET version = excluded.version", - (target,), + # Re-read the version INSIDE the write transaction. The + # version read before the loop is advisory (it sizes the + # loop and enables the byte-identical no-op fast path), but + # two runners on one connection interleave between their + # reads: the wrapper serializes individual statements, not + # the read-decide-apply decision. Re-reading under the + # transaction's lock means the loser of the race sees the + # winner's stamp and skips instead of re-applying a + # non-idempotent migration with a stale version. + in_tx_version = await read_schema_version(database) + if in_tx_version > latest: + raise RuntimeError( + f"Database schema version {in_tx_version} is newer " + f"than this integration understands (latest known: " + f"{latest}); downgrades are not supported" + ) + if in_tx_version < target: + for sql in migrations[target - 1]: + await database.execute(sql) + # INSERT keyed on the singleton row (id = 1): a fresh + # file has no row yet, and migration 1 is the only + # migration a version-0 database can run, so the row is + # seeded by the same statement batch when needed. + # UPDATE alone would silently match zero rows there. + await database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) " + "VALUES (1, ?) " + "ON CONFLICT (id) DO UPDATE SET version = " + "excluded.version", + (target,), + ) + applied = True + if applied: + LOGGER.info( + "NestQuest schema migrated to version %d (%d statement%s)", + target, + len(migrations[target - 1]), + "s" if len(migrations[target - 1]) != 1 else "", ) - LOGGER.info( - "NestQuest schema migrated to version %d (%d statement%s)", - target, - len(statements), - "s" if len(statements) != 1 else "", - ) return latest \ No newline at end of file diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 5e7e844..7467e3b 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -574,9 +574,16 @@ async def test_options_reload_reopens_and_rechecks_version( async def test_setup_entry_failure_closes_database( hass, make_entry, monkeypatch ) -> None: + """A failing migration closes the opened connection before re-raising. + + Spies on the real wrapper instance: records whether close() ran and + whether the connection is left disconnected, rather than inferring + from hass.data (a failed setup never stores a runtime record). + """ import pytest as pytest_module import custom_components.nestquest as nestquest_module + from custom_components.nestquest.db import NestQuestDatabase from custom_components.nestquest.migrations import ( MIGRATIONS as _MIGRATIONS, ) @@ -587,12 +594,22 @@ async def test_setup_entry_failure_closes_database( monkeypatch.setattr( nestquest_module, "apply_migrations", _failing, raising=True ) + + close_calls: list[bool] = [] + original_close = NestQuestDatabase.close + + async def _spy_close(self): + close_calls.append(True) + return await original_close(self) + + monkeypatch.setattr(NestQuestDatabase, "close", _spy_close) + entry = make_entry() with pytest_module.raises(RuntimeError, match="boom"): await _setup_entry(hass, entry, hass.registry) - # Setup closed the connection before re-raising: no leaked handle. - # The record only lands in hass.data after the listener is added, so - # on failure the connection must be gone from the wrapper itself. + + assert close_calls == [True], ( + "failed setup must close the connection it opened" + ) record = hass.data["nestquest"].get(entry.entry_id) - if record is not None: - assert not record.database.connected \ No newline at end of file + assert record is None, "failed setup must not store runtime data" \ No newline at end of file -- 2.55.0 From f27c84488dc4f4d51f6859dab7bfc54d0900a7d3 Mon Sep 17 00:00:00 2001 From: Joshua King Date: Mon, 14 Sep 2026 07:07:46 -0400 Subject: [PATCH 5/5] fix: serialize concurrent migration runners with a per-connection lock NestQuestDatabase rejects a second transaction while one is active, so two concurrent apply_migrations callers on one connection crashed with RuntimeError instead of queueing. apply_migrations now holds a per-(connection, loop) asyncio lock across the whole decide-and-apply span; the queued runner re-reads the stamped version and applies nothing. Deterministic concurrent-runner regression test added. --- custom_components/nestquest/migrations.py | 149 +++++++++++++--------- tests/test_migrations.py | 62 +++++++++ 2 files changed, 152 insertions(+), 59 deletions(-) diff --git a/custom_components/nestquest/migrations.py b/custom_components/nestquest/migrations.py index 00526b9..0cc4377 100644 --- a/custom_components/nestquest/migrations.py +++ b/custom_components/nestquest/migrations.py @@ -41,6 +41,7 @@ handed an open, working connection. """ from __future__ import annotations +import asyncio import logging from collections.abc import Sequence @@ -119,6 +120,24 @@ async def read_schema_version(database: NestQuestDatabase) -> int: return version +#: One asyncio.Lock per connection wrapper, created lazily and bound to +#: the running loop, serializing whole apply_migrations runs so two +#: concurrent callers on one connection queue instead of colliding on +#: the wrapper's single-transaction rule. +_MIGRATION_LOCKS: dict[int, asyncio.Lock] = {} + + +def _migration_lock(database: NestQuestDatabase) -> asyncio.Lock: + """Return the migration lock bound to ``database``'s running loop.""" + loop = asyncio.get_running_loop() + key = (id(database), id(loop)) + lock = _MIGRATION_LOCKS.get(key) + if lock is None: + lock = asyncio.Lock() + _MIGRATION_LOCKS[key] = lock + return lock + + async def apply_migrations( database: NestQuestDatabase, migrations: Sequence[Sequence[str]] = MIGRATIONS, @@ -138,68 +157,80 @@ async def apply_migrations( a non-idempotent migration. Under the transaction's lock the second runner sees the first runner's stamp and skips. + Runners on one connection are serialized by an asyncio lock held + for the whole decide-and-apply span: NestQuestDatabase rejects a + second transaction while one is active (RuntimeError), so + concurrent callers must queue, and the lock makes that queueing + explicit. The loser of the race re-reads the version once it + acquires the lock, sees the winner's stamp, and applies nothing. + Raises whatever the failing migration raised, with the database rolled back to its prior version and prior rows untouched. """ - await _create_version_table(database) - current = await read_schema_version(database) - latest = len(migrations) + async with _migration_lock(database): + await _create_version_table(database) + current = await read_schema_version(database) + latest = len(migrations) - if current > latest: - raise RuntimeError( - f"Database schema version {current} is newer than this " - f"integration understands (latest known: {latest}); " - "downgrades are not supported" - ) - - if current == latest: - LOGGER.debug( - "NestQuest schema already at version %d; no migrations to apply", - current, - ) - return current - - for target in range(current + 1, latest + 1): - applied = False - async with database.transaction(): - # Re-read the version INSIDE the write transaction. The - # version read before the loop is advisory (it sizes the - # loop and enables the byte-identical no-op fast path), but - # two runners on one connection interleave between their - # reads: the wrapper serializes individual statements, not - # the read-decide-apply decision. Re-reading under the - # transaction's lock means the loser of the race sees the - # winner's stamp and skips instead of re-applying a - # non-idempotent migration with a stale version. - in_tx_version = await read_schema_version(database) - if in_tx_version > latest: - raise RuntimeError( - f"Database schema version {in_tx_version} is newer " - f"than this integration understands (latest known: " - f"{latest}); downgrades are not supported" - ) - if in_tx_version < target: - for sql in migrations[target - 1]: - await database.execute(sql) - # INSERT keyed on the singleton row (id = 1): a fresh - # file has no row yet, and migration 1 is the only - # migration a version-0 database can run, so the row is - # seeded by the same statement batch when needed. - # UPDATE alone would silently match zero rows there. - await database.execute( - f"INSERT INTO {VERSION_TABLE} (id, version) " - "VALUES (1, ?) " - "ON CONFLICT (id) DO UPDATE SET version = " - "excluded.version", - (target,), - ) - applied = True - if applied: - LOGGER.info( - "NestQuest schema migrated to version %d (%d statement%s)", - target, - len(migrations[target - 1]), - "s" if len(migrations[target - 1]) != 1 else "", + if current > latest: + raise RuntimeError( + f"Database schema version {current} is newer than this " + f"integration understands (latest known: {latest}); " + "downgrades are not supported" ) - return latest \ No newline at end of file + if current == latest: + LOGGER.debug( + "NestQuest schema already at version %d; " + "no migrations to apply", + current, + ) + return current + + for target in range(current + 1, latest + 1): + applied = False + async with database.transaction(): + # Re-read the version INSIDE the write transaction: the + # outer read was advisory, and the wrapper serializes + # individual statements, not the read-decide-apply + # decision. Under the transaction's lock the version + # cannot change between the read and the stamp, and the + # loser of an interleaved race (a queued second caller + # re-running apply_migrations after the winner already + # migrated) re-reads here, sees the winner's stamp, and + # skips instead of re-applying a non-idempotent + # migration with a stale version. + in_tx_version = await read_schema_version(database) + if in_tx_version > latest: + raise RuntimeError( + f"Database schema version {in_tx_version} is newer " + f"than this integration understands (latest known: " + f"{latest}); downgrades are not supported" + ) + if in_tx_version < target: + for sql in migrations[target - 1]: + await database.execute(sql) + # INSERT keyed on the singleton row (id = 1): a + # fresh file has no row yet, and migration 1 is the + # only migration a version-0 database can run, so + # the row is seeded by the same statement batch + # when needed. UPDATE alone would silently match + # zero rows there. + await database.execute( + f"INSERT INTO {VERSION_TABLE} (id, version) " + "VALUES (1, ?) " + "ON CONFLICT (id) DO UPDATE SET version = " + "excluded.version", + (target,), + ) + applied = True + if applied: + LOGGER.info( + "NestQuest schema migrated to version %d " + "(%d statement%s)", + target, + len(migrations[target - 1]), + "s" if len(migrations[target - 1]) != 1 else "", + ) + + return latest \ No newline at end of file diff --git a/tests/test_migrations.py b/tests/test_migrations.py index 7467e3b..f664390 100644 --- a/tests/test_migrations.py +++ b/tests/test_migrations.py @@ -452,6 +452,68 @@ def test_version_table_check_rejects_version_zero_and_negative( _run(database.close()) +def test_concurrent_runners_serialize_and_apply_once(tmp_path) -> None: + """Two concurrent runners on one connection apply each step once. + + Deterministic race: runner B starts while runner A is queued into + its migration. A applies everything and stamps; B then acquires + the migration lock, re-reads the already-stamped version, and + applies nothing. Without serialization B's transaction() would + raise RuntimeError (the wrapper forbids concurrent transactions), + so this also proves the queueing works. + """ + migrations = [ + ["CREATE TABLE IF NOT EXISTS step_a (id INTEGER PRIMARY KEY)"], + ["CREATE TABLE IF NOT EXISTS step_b (id INTEGER PRIMARY KEY)"], + ] + results: dict[str, int] = {} + + async def _main(tmp_name: str) -> None: + database = NestQuestDatabase(_make_hass_mock()) + await database.open(tmp_name) + try: + # Start A, yield so it acquires the lock and enters its + # first transaction, then start B while A is mid-apply. + task_a = asyncio.ensure_future( + apply_migrations(database, migrations) + ) + await asyncio.sleep(0) + task_b = asyncio.ensure_future( + apply_migrations(database, migrations) + ) + results["a"], results["b"] = await asyncio.gather( + task_a, task_b + ) + + tables = { + row[0] + for row in await database.fetch_all( + "SELECT name FROM sqlite_master WHERE type = 'table' " + "AND name NOT LIKE 'sqlite_%'" + ) + } + assert {"step_a", "step_b"} <= tables + # Each migration's statements ran exactly once: the CREATE + # IF NOT EXISTS would hide double-runs, so count via sqlite + # sequence/DDL fingerprint instead — re-check version and + # that both runners reported the same final version. + row = await database.fetch_one( + f"SELECT version FROM {VERSION_TABLE}" + ) + assert row == (2,) + finally: + await database.close() + + loop = asyncio.new_event_loop() + try: + loop.run_until_complete(_main(str(tmp_path / "concurrent.db"))) + finally: + loop.close() + + assert results["a"] == 2 + assert results["b"] == 2 + + def test_multiple_pending_migrations_apply_in_order(tmp_path) -> None: database = _open_db(tmp_path / "multi-step.db") try: -- 2.55.0