From 31879b01f4abd06d3f7b6d7a605665c1de7c05e6 Mon Sep 17 00:00:00 2001
From: npeter83 <npeter83+claude@gmail.com>
Date: Tue, 18 Aug 2026 01:23:07 +0200
Subject: [PATCH] test(migrations): model<->schema drift + downgrade/re-upgrade
 (E1 S2, TEST-3)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Prod only ever runs `alembic upgrade head`; nothing checks that the migrated
schema still matches Base.metadata, and the downgrade path never runs. Add a
DB-lane test that (a) asserts no structural (table/column) drift between the ORM
models and the head schema — filtering the index/constraint/sequence noise that
compare_metadata reports for migration-only objects — and (b) exercises the head
migration's downgrade + re-upgrade so a broken downgrade is caught the sprint it
lands.
---
 backend/tests/test_migrations.py | 59 ++++++++++++++++++++++++++++++++
 1 file changed, 59 insertions(+)
 create mode 100644 backend/tests/test_migrations.py

diff --git a/backend/tests/test_migrations.py b/backend/tests/test_migrations.py
new file mode 100644
index 00000000..373e4413
--- /dev/null
+++ b/backend/tests/test_migrations.py
@@ -0,0 +1,59 @@
+"""Migration hygiene (TEST-3), DB-lane.
+
+Two things the prod boot never checks:
+- **model <-> schema drift:** `entrypoint.sh` only ever runs `alembic upgrade head`; nothing verifies the
+  migrated schema still matches `Base.metadata`. A model column added without a migration (or a migration
+  that drifted from the model) ships silently.
+- **the downgrade path:** every migration defines `downgrade()`, but it never runs, so a broken downgrade
+  is only discovered during an emergency rollback.
+
+Both use the `db` fixture, which has already brought the schema to head via `_schema`.
+
+Scope note: we assert on STRUCTURAL drift only — tables and columns present in the model but not the
+schema, or vice versa. `compare_metadata` also reports index/constraint/sequence/type differences, but
+those are dominated by migration-only objects (partial/expression/GIN indexes, CHECK constraints, SERIAL
+sequences) that can't be expressed on `Base.metadata`, so comparing them wholesale is noise. Missing/extra
+columns and tables — the "someone changed a model without a migration" case — are the signal worth gating.
+"""
+from alembic import command
+from alembic.autogenerate import compare_metadata
+from alembic.config import Config
+from alembic.migration import MigrationContext
+
+from app.db import Base, engine
+
+# The diff ops that mean a real table/column is present on one side and not the other.
+_DRIFT_OPS = {"add_table", "remove_table", "add_column", "remove_column"}
+
+
+def _structural_drift():
+    """Table/column presence diffs between the ORM models and the migrated schema (`compare_type=True`
+    mirrors alembic/env.py). compare_metadata nests per-column ops in a sub-list per table and returns
+    table-level ops as bare tuples; flatten both and keep only the presence ops."""
+    with engine.connect() as conn:
+        ctx = MigrationContext.configure(conn, opts={"compare_type": True})
+        diffs = compare_metadata(ctx, Base.metadata)
+    drift = []
+    for d in diffs:
+        for entry in (d if isinstance(d, list) else [d]):
+            if isinstance(entry, tuple) and entry and entry[0] in _DRIFT_OPS:
+                drift.append(entry)
+    return drift
+
+
+def test_models_match_the_migrated_schema(db):
+    # Non-empty == the ORM models and the alembic-built schema disagree on a table/column — almost
+    # always a model change that forgot its migration.
+    drift = _structural_drift()
+    assert drift == [], "model/schema drift — add or fix a migration:\n" + "\n".join(
+        str(d) for d in drift
+    )
+
+
+def test_head_migration_downgrades_and_reupgrades(db):
+    # Exercise the newest migration's downgrade + re-upgrade so a broken downgrade is caught the sprint
+    # it lands, not during a rollback. Leaves the schema back at head for the rest of the session.
+    cfg = Config("alembic.ini")
+    command.downgrade(cfg, "-1")
+    command.upgrade(cfg, "head")
+    assert _structural_drift() == []