tools/verify_migration.py
Rama main · réplica de lectura del repositorio original.
"""Verify a migration against a POPULATED predecessor, on either backend.
./tools/verify_migration.py 0002 # sqlite
PG=1 ./tools/verify_migration.py 0002 # postgres
Every migration in this project gets this before it ships, and the reason is
that a green test suite does not cover it: the suite builds every schema from
nothing at head, so it passes even for a migration that would destroy existing
rows. This walks the path a real deployment takes instead — stop at the
predecessor, put data in, upgrade, check the data survived and the new columns
behave, downgrade, re-upgrade.
The generic checks below catch the usual failures. **Anything specific to the
migration under test still needs adding by hand** — a backfill, for
instance, had to be checked for giving every row a *distinct* UUID, because the
tempting one-line version gives them all the same one and that would make every
account the same person to any app federating with this one. Generic checks
would have called that a pass.
Add those in `extra_checks()`.
"""
from __future__ import annotations
import os
import sys
import tempfile
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(APP_ROOT))
from alembic import command # noqa: E402
from alembic.config import Config # noqa: E402
from alembic.script import ScriptDirectory # noqa: E402
from sqlalchemy import create_engine, inspect, text # noqa: E402
SCRIPTS = str(APP_ROOT / "app" / "migrations")
# A user row that satisfies every NOT NULL in the current schema. Extend it when
# a migration adds one — that is the usual reason this script suddenly fails on
# the INSERT rather than on a check.
USER_INSERT = (
'insert into "user" (id, email, name, password_hash, is_admin,'
" must_change_password, created_at, theme, accent, nav_style,"
" charts_seeded, is_active, public_id)"
" values (:id, :email, :name, 'x', :admin, false,"
" '2026-01-01 00:00:00', 'auto', 'blue', 'both', false, true, :public_id)"
)
def cfg(url: str) -> Config:
c = Config()
c.set_main_option("script_location", SCRIPTS)
c.set_main_option("sqlalchemy.url", url.replace("%", "%%"))
return c
def check(label: str, ok: bool) -> None:
print(f" {'PASS' if ok else 'FAIL'} {label}")
if not ok:
raise SystemExit(f"FAILED: {label}")
def columns(engine, table: str) -> set[str]:
return {c["name"] for c in inspect(engine).get_columns(table)}
def version_of(engine) -> str:
with engine.begin() as conn:
return conn.execute(text("select version_num from alembic_version")).scalar()
def predecessor(target: str) -> str:
script = ScriptDirectory.from_config(cfg("sqlite://"))
rev = script.get_revision(target)
if not rev.down_revision:
raise SystemExit(f"{target} has no predecessor — it is the base revision.")
return rev.down_revision
def new_columns(engine_before: set[str], engine_after: set[str]) -> set[str]:
return engine_after - engine_before
def extra_checks(engine, target: str) -> None:
"""Per-migration assertions, keyed on the revision under test.
Add to this rather than trusting the generic ones — they see that a column
arrived, never what the migration was *for*. A migration that backfills a
value is the case they cannot cover: the column is present and non-null
either way, so only a check written against that migration's intent can
tell a correct backfill from one that wrote the same value to every row.
if target == "0002":
with engine.begin() as conn:
ids = [r[0] for r in conn.execute(text('select ... from ...'))]
check("every value is distinct", len(set(ids)) == len(ids))
"""
def run(url: str, dialect: str, target: str) -> None:
previous = predecessor(target)
print(f"\n=== {dialect}: {previous} -> {target} ===")
c = cfg(url)
engine = create_engine(url)
# 1. A database at the predecessor, as a live deployment would be.
command.upgrade(c, previous)
check(f"stopped at {previous}", version_of(engine) == previous)
before_user = columns(engine, "user")
before_session = columns(engine, "user_session")
# 2. Populate it. Several rows, because "gave every row the same value" is
# a failure one row cannot show.
with engine.begin() as conn:
import uuid as _u
for i in range(1, 4):
conn.execute(text(USER_INSERT), {
"id": i, "email": f"u{i}@example.test", "name": f"u{i}",
"admin": i == 1, "public_id": str(_u.uuid4())})
with engine.begin() as conn:
check("rows inserted at the predecessor",
conn.execute(text('select count(*) from "user"')).scalar() == 3)
# 3. Upgrade.
command.upgrade(c, target)
check(f"at {target}", version_of(engine) == target)
with engine.begin() as conn:
emails = [r[0] for r in conn.execute(text('select email from "user" order by id'))]
check("every pre-existing row survived",
emails == ["u1@example.test", "u2@example.test", "u3@example.test"])
added = (new_columns(before_user, columns(engine, "user"))
| new_columns(before_session, columns(engine, "user_session")))
print(f" ... columns added: {sorted(added) or 'none'}")
extra_checks(engine, target)
# 4. Downgrade. The change goes; the rows must not.
command.downgrade(c, previous)
check(f"back at {previous}", version_of(engine) == previous)
with engine.begin() as conn:
after = [r[0] for r in conn.execute(text('select email from "user" order by id'))]
check("rows survived the downgrade too", after == emails)
# SQLite has no DROP COLUMN before 3.35, so alembic rebuilds the table —
# the neighbours are worth checking, not just the row count.
with engine.begin() as conn:
row = conn.execute(text(
'select name, is_admin, theme from "user" where id = 1')).one()
check("neighbouring columns intact after the table rebuild",
row[0] == "u1" and bool(row[1]) is True and row[2] == "auto")
# 5. Repeatable, not one-way.
command.upgrade(c, target)
check(f"re-upgraded to {target}", version_of(engine) == target)
command.upgrade(c, "head")
print(f" ... head is {version_of(engine)}")
engine.dispose()
def main() -> int:
if len(sys.argv) < 2:
print(__doc__)
return 2
target = sys.argv[1]
if os.environ.get("PG"):
base = "postgresql+psycopg://postgres:postgres@localhost:55432/"
admin = create_engine(base + "postgres", isolation_level="AUTOCOMMIT")
name = f"verify_{target}"
with admin.begin() as conn:
conn.execute(text(f"drop database if exists {name}"))
conn.execute(text(f"create database {name}"))
admin.dispose()
run(base + name, "postgres", target)
else:
with tempfile.TemporaryDirectory() as tmp:
run(f"sqlite:///{tmp}/verify.db", "sqlite", target)
print("\nOK")
return 0
if __name__ == "__main__":
raise SystemExit(main())