"""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." )