diff --git a/backend/tests/test_migration_0020.py b/backend/tests/test_migration_0020.py index 11fa2ac..f2be792 100644 --- a/backend/tests/test_migration_0020.py +++ b/backend/tests/test_migration_0020.py @@ -14,6 +14,11 @@ the fresh DB up to v20, insert representative rows, then replay the exact UPDATE statement the migration uses. Replaying the UPDATE proves the SQL works as intended even though the migration itself already ran over an empty table. + +The backfill UPDATE is loaded directly from the migration file (not +a hand-maintained copy) so the test cannot drift from production — +see :func:`_load_migration_0020_backfill_sql` and the regression +locks in ``test_migration_0020_no_drift.py``. """ from __future__ import annotations @@ -27,16 +32,48 @@ import sqlalchemy as sa from cyclone import db_migrate -# The backfill UPDATE the migration executes (extracted so the test can -# replay it against rows that didn't exist when init_db ran). -BACKFILL_SQL = ( - "UPDATE batches " - "SET transaction_set_control_number = " - "json_extract(raw_result_json, '$.envelope.transaction_set_control_number') " - "WHERE raw_result_json IS NOT NULL " - "AND json_extract(raw_result_json, '$.envelope.transaction_set_control_number') " - "IS NOT NULL" -) +def _migration_0020_path() -> Path: + return ( + Path(__file__).parent.parent / "src" / "cyclone" / "migrations" + / "0020_add_batch_txn_set_control_number.sql" + ) + + +def _extract_update_statements(sql: str) -> list[str]: + """Split a migration into statements; return the UPDATE ones. + + Mirrors db_migrate.run()'s splitter (strip ``--`` comments, + split on ``;``) so the test extraction can never disagree with + what the runner actually executes. + """ + lines = [ + line for line in sql.splitlines() + if not line.strip().startswith("--") + ] + cleaned = "\n".join(lines) + return [ + stmt.strip() for stmt in cleaned.split(";") + if stmt.strip() + ] + + +def _load_migration_0020_backfill_sql() -> str: + """Return migration 0020's UPDATE statement (the backfill). + + Reads the migration file at test time and extracts its UPDATE. + Test code that needs to replay the backfill against rows that + didn't exist when init_db ran uses this helper — guarantees the + replayed SQL is byte-identical to what production will run. + """ + sql = _migration_0020_path().read_text() + updates = [ + s for s in _extract_update_statements(sql) + if s.upper().startswith("UPDATE ") + ] + assert len(updates) == 1, ( + f"expected exactly 1 UPDATE in migration 0020, found {len(updates)}: {updates}" + ) + return updates[0] def _fresh_engine(path: Path) -> sa.Engine: @@ -115,7 +152,7 @@ def test_migration_0020_backfills_when_key_present( "VALUES ('B-ST02-1', '837p', 'mig0020-st02.txt', '2026-07-07 00:00:00', ?)", (json.dumps(raw_json),), ) - conn.exec_driver_sql(BACKFILL_SQL) + conn.exec_driver_sql(_load_migration_0020_backfill_sql()) with migrated_engine.connect() as conn: row = conn.exec_driver_sql( @@ -141,7 +178,7 @@ def test_migration_0020_backfill_conditional_on_key_present( "VALUES ('B-NOST-1', '837p', 'mig0020-nost.txt', '2026-07-07 00:00:00', ?)", (json.dumps(raw_json),), ) - conn.exec_driver_sql(BACKFILL_SQL) + conn.exec_driver_sql(_load_migration_0020_backfill_sql()) with migrated_engine.connect() as conn: row = conn.exec_driver_sql( @@ -163,7 +200,7 @@ def test_migration_0020_backfill_handles_null_raw_result_json( "INSERT INTO batches (id, kind, input_filename, parsed_at) " "VALUES ('B-NULL-1', '837p', 'mig0020-null.txt', '2026-07-07 00:00:00')" ) - conn.exec_driver_sql(BACKFILL_SQL) + conn.exec_driver_sql(_load_migration_0020_backfill_sql()) with migrated_engine.connect() as conn: row = conn.exec_driver_sql( diff --git a/backend/tests/test_migration_0020_no_drift.py b/backend/tests/test_migration_0020_no_drift.py new file mode 100644 index 0000000..b32fb7d --- /dev/null +++ b/backend/tests/test_migration_0020_no_drift.py @@ -0,0 +1,66 @@ +"""SP37 follow-up #4: detect drift between test BACKFILL_SQL and migration SQL. + +Followup #1 of the SP37 final-state tracker. The previous test file +maintained a hand-copied ``BACKFILL_SQL`` constant alongside the +migration file. If a future contributor edited one but not the other, +the test would silently replay a different SQL than production — +defeating the regression test. + +The fix: ``test_migration_0020.py`` now reads the migration file at +test time and extracts its UPDATE. This file imports the helper and +pins the invariant — so a future contributor who edits the migration +automatically gets the new SQL replayed in tests, and a contributor +who removes the backfill UPDATE gets a loud extraction failure. +""" +from __future__ import annotations + +from test_migration_0020 import ( + _extract_update_statements, + _load_migration_0020_backfill_sql, + _migration_0020_path, +) + + +def test_migration_0020_backfill_sql_uses_migration_file(): + """The backfill SQL used in tests must come from the migration file. + + Guards against the previous pattern of a hand-maintained + BACKFILL_SQL constant that could drift from the actual migration. + """ + sql = _load_migration_0020_backfill_sql() + # Sanity check: the SQL targets batches.transaction_set_control_number + # via json_extract on raw_result_json. If a contributor changes + # the column name or the JSON path, this assertion catches it. + assert "UPDATE batches" in sql + assert "SET transaction_set_control_number" in sql + assert "json_extract(raw_result_json" in sql + assert "$.envelope.transaction_set_control_number" in sql + + +def test_migration_0020_backfill_sql_is_non_empty_single_statement(): + """The helper returns a non-empty SQL string suitable for direct + execution via exec_driver_sql. The existing + ``test_migration_0020.py`` tests use this helper to replay the + SQL against representative rows — so any successful replay + exercises the migration's actual UPDATE. + """ + sql = _load_migration_0020_backfill_sql() + assert sql # non-empty + assert ";" not in sql # already stripped by the splitter + + +def test_migration_0020_has_exactly_one_update(): + """Guardrail: if a future migration adds a second UPDATE, the + extraction fails loudly so the test author can decide which one + is the backfill. + """ + sql = _migration_0020_path().read_text() + updates = [ + s for s in _extract_update_statements(sql) + if s.upper().startswith("UPDATE ") + ] + assert len(updates) == 1, ( + f"migration 0020 should have exactly 1 UPDATE (the backfill); " + f"found {len(updates)}. If you added a second UPDATE, update " + f"_load_migration_0020_backfill_sql() to pick the right one." + ) \ No newline at end of file