Padreug
c50455d5f6
The migration version bump lands in the core dbversions table while the DDL lands in ext_libra — the two writes are not atomic. A failed bump re-runs the whole migration on next boot; bare CREATE/ALTER/INSERT then crashes the extension until manual dbversions surgery. - CREATE TABLE / CREATE INDEX -> IF NOT EXISTS - ALTER TABLE ADD COLUMN -> _alter_add_column_safe (same swallow pattern as the events/withdraw fork migrations) - seed INSERTs (default accounts, virtual parents, default roles) -> ON CONFLICT (name) DO NOTHING Tests run the chain twice against a fresh SQLite DB (full-chain rerun and per-migration rerun); both fail against the previous migrations. Addresses CODE-REVIEW-2026-06 findings #3 and #12. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
109 lines
3.6 KiB
Python
109 lines
3.6 KiB
Python
"""Migration idempotency tests.
|
|
|
|
The migration version bump lands in the core LNbits DB (`dbversions`)
|
|
while the DDL lands in `ext_libra` — the two writes are not atomic. If
|
|
the bump fails after the DDL commits, the whole migration re-runs on
|
|
next boot, so every statement must be a silent no-op on re-run instead
|
|
of crashing with `duplicate column` / `table exists` / UNIQUE
|
|
violations (which bricks the extension until manual dbversions
|
|
surgery).
|
|
|
|
These tests run the full migration chain twice against a fresh SQLite
|
|
database — the second pass simulates the lost-version-bump re-run.
|
|
"""
|
|
import importlib
|
|
import re
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from lnbits.db import Database
|
|
|
|
pytestmark = pytest.mark.anyio
|
|
|
|
|
|
def _module(name: str):
|
|
"""Import a libra submodule under whichever path the active LNbits layout
|
|
uses (default `lnbits.extensions.libra` or bare `libra`)."""
|
|
for prefix in ("lnbits.extensions.libra", "libra"):
|
|
try:
|
|
return importlib.import_module(f"{prefix}.{name}")
|
|
except ModuleNotFoundError:
|
|
continue
|
|
raise ModuleNotFoundError(f"libra.{name}: tried both import paths")
|
|
|
|
|
|
migrations = _module("migrations")
|
|
|
|
# Same discovery as lnbits.core.helpers.run_migration: m### prefix,
|
|
# module definition order.
|
|
_MIGRATION_RE = re.compile(r"^m(\d\d\d)_")
|
|
MIGRATION_FUNCTIONS = [
|
|
fn for name, fn in vars(migrations).items() if _MIGRATION_RE.match(name)
|
|
]
|
|
|
|
# Tables the chain must leave behind — one probe row read per table
|
|
# proves both existence and queryability after a double run.
|
|
EXPECTED_TABLES = [
|
|
"accounts",
|
|
"extension_settings",
|
|
"user_wallet_settings",
|
|
"manual_payment_requests",
|
|
"balance_assertions",
|
|
"user_equity_status",
|
|
"account_permissions",
|
|
"roles",
|
|
"role_permissions",
|
|
"user_roles",
|
|
]
|
|
|
|
|
|
async def _run_all_migrations(db: Database) -> None:
|
|
async with db.connect() as conn:
|
|
for migrate in MIGRATION_FUNCTIONS:
|
|
await migrate(conn)
|
|
|
|
|
|
async def _seed_counts(db: Database) -> dict:
|
|
async with db.connect() as conn:
|
|
accounts = await conn.fetchall("SELECT id, name FROM accounts")
|
|
roles = await conn.fetchall("SELECT id, name FROM roles")
|
|
return {
|
|
"account_names": sorted(r["name"] for r in accounts),
|
|
"account_ids": sorted(r["id"] for r in accounts),
|
|
"role_names": sorted(r["name"] for r in roles),
|
|
"role_ids": sorted(r["id"] for r in roles),
|
|
}
|
|
|
|
|
|
async def test_migrations_rerun_is_noop():
|
|
"""Full chain twice: second run must not raise and not re-seed."""
|
|
db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
|
|
|
|
await _run_all_migrations(db)
|
|
first = await _seed_counts(db)
|
|
|
|
# Simulate the lost dbversions bump: everything runs again.
|
|
await _run_all_migrations(db)
|
|
second = await _seed_counts(db)
|
|
|
|
# Seeds must not duplicate (names) and must not be replaced (ids).
|
|
assert second == first
|
|
assert first["account_names"], "seed accounts missing after migration"
|
|
assert "Employee" in first["role_names"]
|
|
|
|
# Every table exists and is queryable after the double run.
|
|
async with db.connect() as conn:
|
|
for table in EXPECTED_TABLES:
|
|
await conn.fetchall(f"SELECT * FROM {table} LIMIT 1") # noqa: S608
|
|
|
|
|
|
async def test_single_migration_rerun_is_noop():
|
|
"""Each migration individually survives an immediate re-run (the
|
|
version bump fails right after that one migration committed)."""
|
|
db = Database(f"ext_libra_migtest_{uuid4().hex[:8]}")
|
|
|
|
async with db.connect() as conn:
|
|
for migrate in MIGRATION_FUNCTIONS:
|
|
await migrate(conn)
|
|
await migrate(conn) # re-run before "bumping" to the next
|