"""Regression: migration 126 must survive a partial-failure retry. Production sequence that motivated this test (2026-05-07): 1. PR #193 introduced migration 126 (sso_apps_catalog) that left the newly-added ``sso_grants.created_app_id`` with the table's default collation (``utf8mb4_bin`` on prod TiDB). 2. ``op.create_foreign_key`` then failed with TiDB error 3780 (CHAR(36) column collation mismatch with ``sso_apps.id``'s ``utf8mb4_unicode_ci``). 3. PR #194's "fix" only addressed ``sso_apps.id`` — missed the ``sso_grants.created_app_id`` half — so the deploy kept failing. 4. TiDB DDL auto-commits CREATE TABLE + ALTER ADD COLUMN, leaving ``sso_apps`` (correct collation) and ``sso_grants.created_app_id`` (wrong collation) on disk while ``alembic_version`` rolled back to 125. The next ``alembic upgrade`` retries from migration 126 and crashes again on CREATE TABLE ("table already exists"). Migration 126 is now idempotent: when it sees ``sso_apps`` already exists it routes through ``_resume_from_table_exists`` which finishes the work the prior run skipped (add missing columns, normalize the ``created_app_id`` collation, create the FK if missing). We don't have TiDB locally so we can't reproduce the collation mismatch directly — but we CAN verify the structural shape of the fix: * The migration imports cleanly. * It defines both an entrypoint ``upgrade`` and the resume helper. * The resume helper makes the three calls the production path needs (add-column-if-missing / normalize-collation / create-FK-if-missing). * The test fixture's alembic-upgrade-to-head succeeds (proven by the fact that the fixture is consumable at all in the rest of the suite). CI on prod TiDB is the canonical end-to-end check — what we pin here is the surface contract the resume path must offer. """ from __future__ import annotations import importlib.util import inspect as py_inspect from pathlib import Path _MIGRATION_PATH = ( Path(__file__).resolve().parents[2] / "alembic" / "versions" / "126_sso_apps_catalog.py" ) def _load_migration(): spec = importlib.util.spec_from_file_location( "migration_126", _MIGRATION_PATH, ) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def test_migration_126_module_imports_cleanly(): """Top-level imports + revision metadata sane. Guards against a refactor accidentally renaming the helpers we depend on.""" m = _load_migration() assert m.revision == "126" assert m.down_revision == "125" assert callable(m.upgrade) assert callable(m.downgrade) assert callable(m._resume_from_table_exists) def test_resume_helper_signature_takes_required_kwargs(): """``_resume_from_table_exists`` is what runs on prod when the earlier deploy half-applied the migration. The kwargs it takes constitute the public contract the upgrade path passes in. Pin the shape so a refactor can't silently drop e.g. ``person_id`` and leave ``add_column`` rendering a NULL type.""" m = _load_migration() sig = py_inspect.signature(m._resume_from_table_exists) assert set(sig.parameters) == {"sso_grants_cols", "sso_grants_fks", "person_id"} def test_upgrade_calls_collation_normalization_helper(): """The whole point of this fix: BOTH the green-field ``upgrade()`` path AND the resume path must call ``normalize_fk_column_collation`` for ``sso_grants.created_app_id`` BEFORE the FK is created. Source-grep the file content rather than mock alembic — alembic's globals are too fiddly for unit tests and the contract we care about is "the call exists in both branches".""" src = _MIGRATION_PATH.read_text() # Both call-sites carry the same 3-arg shape — count both. occurrences = src.count('normalize_fk_column_collation(') assert occurrences >= 2, ( f"Expected >=2 calls to normalize_fk_column_collation " f"(green-field + resume paths); saw {occurrences}. " "Without it on prod TiDB, FK create errors with code 3780 " "because sso_grants.created_app_id inherits utf8mb4_bin " "while sso_apps.id is utf8mb4_unicode_ci." ) # And the fix specifically targets created_app_id, not some # other column that happens to be FK'd elsewhere. assert 'column="created_app_id"' in src assert 'parent_table="sso_apps"' in src def test_upgrade_and_resume_both_create_fk_when_missing(): """Source-level check that the FK ``fk_sso_grants_created_app`` is created on both code paths and that the resume path guards on ``not in sso_grants_fks`` for idempotency.""" src = _MIGRATION_PATH.read_text() # FK name appears in: green-field create_foreign_key + resume's # create_foreign_key + downgrade's drop_constraint = 3 sites. assert src.count('"fk_sso_grants_created_app"') >= 3 # Idempotency guard for resume. assert '"fk_sso_grants_created_app" not in sso_grants_fks' in src def test_resume_path_idempotent_for_columns_already_added(): """The resume path computes ``missing`` from the inspector's view of existing ``sso_grants`` columns. If everything's already there (third run after a successful retry), nothing should be added. Source-level: the loop body is gated on ``if missing:``.""" src = _MIGRATION_PATH.read_text() assert "if missing:" in src # The list of columns to add is explicit; pin the names so a # rename doesn't silently miss one in the resume path. for col in ("acs_url", "entity_id", "x509_cert", "saml_metadata_xml", "created_app_id"): assert f'"{col}"' in src, f"Resume path must list column {col!r}"