feat: catch server-default, CHECK-constraint and enum drift the diff test misses

README.md

@@ -47,7 +47,7 @@ from myapp.db import Base
 
 @pytest.mark.integration
 class TestMyMigrations(MigrationTestBase):
-    """All five tests inherited automatically."""
+    """All seven tests inherited automatically."""
 
     @pytest.fixture
     def orm_metadata(self) -> MetaData:
@@ -62,6 +62,8 @@ That's it! You now have:
 
 - `test_stairway_upgrade_downgrade` — each migration forward and back
 - `test_migrations_up_to_date` — schema matches ORM models
+- `test_check_constraints_match` — CHECK constraints match ORM models, by name
+- `test_enum_values_match` — enum values match ORM models, in order
 - `test_single_head_revision` — no unmerged branches
 - `test_downgrade_all_the_way` — full downgrade to base
 - `test_naming_conventions` — indexes and FKs follow conventions

alembic_gauntlet/testing/base.py

@@ -16,7 +16,7 @@ class MigrationTestBase(
 ):
     """Base class for database migration tests.
 
-    Inherit from this class and provide the following fixtures to get all five
+    Inherit from this class and provide the following fixtures to get all seven
     migration tests for free:
 
     Required fixtures:
@@ -30,7 +30,10 @@ class MigrationTestBase(
 
     Optional class attributes:
         - ``migration_diff_ignore_tables: list[str]`` — table names to exclude from
-          schema diff and naming checks (e.g. auto-generated partition tables).
+          the schema diff, check constraint, enum and naming checks (e.g. auto-generated
+          partition tables).
+        - ``migration_diff_compare_server_default: bool`` — also compare server defaults
+          in the schema diff. Off by default; see ``MigrationConsistencyMixin``.
         - ``allowed_index_prefixes``, ``allowed_index_suffixes``, ``allowed_fk_prefixes``,
           ``allowed_fk_suffixes``, ``allowed_check_prefixes``, ``allowed_check_suffixes``,
           ``allowed_uq_prefixes``, ``allowed_uq_suffixes``, ``allowed_pk_prefixes``,

alembic_gauntlet/testing/consistency_mixin.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, ClassVar
 
 import pytest
 from alembic.autogenerate import compare_metadata
@@ -11,7 +11,12 @@
 from alembic.script import ScriptDirectory
 from sqlalchemy import text
 
-from alembic_gauntlet.utils.diff import DEFAULT_IGNORE_TABLES, is_ignored_diff_item
+from alembic_gauntlet.utils.diff import (
+    DEFAULT_IGNORE_TABLES,
+    compare_check_constraints,
+    compare_enums,
+    is_ignored_diff_item,
+)
 from alembic_gauntlet.utils.migrations import (
     get_all_revisions,
     get_current_revision,
@@ -31,7 +36,16 @@
 
 
 class MigrationConsistencyMixin:
-    """Core migration correctness tests."""
+    """Core migration correctness tests.
+
+    Optional class attributes:
+        - ``migration_diff_compare_server_default: bool`` — pass ``compare_server_default``
+          to Alembic in ``test_migrations_up_to_date``. Off by default, because a default
+          present on one side only — a Python-side ``default=`` in the model, a
+          ``server_default`` in the migration — is reported as drift.
+    """
+
+    migration_diff_compare_server_default: ClassVar[bool] = False
 
     async def test_stairway_upgrade_downgrade(
         self,
@@ -96,7 +110,10 @@ def _run_check(sync_conn: Connection) -> MigrationDiff:
             sync_conn.execute(text("SELECT set_config('search_path', :s, true)"), {"s": quoted})
             ctx = MigrationContext.configure(
                 sync_conn,
-                opts={"version_table_schema": isolated_migration_schema},
+                opts={
+                    "version_table_schema": isolated_migration_schema,
+                    "compare_server_default": self.migration_diff_compare_server_default,
+                },
             )
             diff = compare_metadata(ctx, orm_metadata)
             assert isinstance(diff, list)
@@ -110,6 +127,58 @@ def _run_check(sync_conn: Connection) -> MigrationDiff:
             "Run: alembic revision --autogenerate"
         )
 
+    async def test_check_constraints_match(
+        self,
+        alembic_config: Config,
+        migration_engine: AsyncEngine,
+        isolated_migration_schema: str,
+        orm_metadata: MetaData,
+    ) -> None:
+        """Verify the CHECK constraints after a full upgrade match the SQLAlchemy ORM metadata, by name.
+
+        ``compare_metadata()`` never compares CHECK constraints, so a constraint a
+        migration forgot is invisible to ``test_migrations_up_to_date``.
+        """
+        await run_alembic_upgrade(
+            migration_engine,
+            alembic_config,
+            target_schema=isolated_migration_schema,
+        )
+
+        ignore_tables = frozenset(getattr(self, "migration_diff_ignore_tables", ()))
+        async with migration_engine.connect() as conn:
+            differences = await conn.run_sync(
+                lambda sc: compare_check_constraints(sc, orm_metadata, isolated_migration_schema, ignore_tables)
+            )
+
+        assert not differences, "CHECK constraints are out of sync with ORM models:\n" + "\n".join(differences)
+
+    async def test_enum_values_match(
+        self,
+        alembic_config: Config,
+        migration_engine: AsyncEngine,
+        isolated_migration_schema: str,
+        orm_metadata: MetaData,
+    ) -> None:
+        """Verify every native ``Enum`` column's values after a full upgrade match the type in the database, in order.
+
+        ``compare_metadata()`` never compares enum members, so a value a migration
+        forgot to add is invisible to ``test_migrations_up_to_date``.
+        """
+        await run_alembic_upgrade(
+            migration_engine,
+            alembic_config,
+            target_schema=isolated_migration_schema,
+        )
+
+        ignore_tables = frozenset(getattr(self, "migration_diff_ignore_tables", ()))
+        async with migration_engine.connect() as conn:
+            differences = await conn.run_sync(
+                lambda sc: compare_enums(sc, orm_metadata, isolated_migration_schema, ignore_tables)
+            )
+
+        assert not differences, "Enum values are out of sync with ORM models:\n" + "\n".join(differences)
+
     async def test_single_head_revision(self, alembic_config: Config) -> None:
         """Verify there is exactly one head revision (no unmerged branches)."""
         script = ScriptDirectory.from_config(alembic_config)

alembic_gauntlet/utils/diff.py

@@ -1,7 +1,17 @@
-"""Schema diff filtering for migration consistency tests."""
+"""Schema diff helpers for migration consistency tests."""
 
 from __future__ import annotations
 
+from typing import TYPE_CHECKING
+
+from alembic.util.sqla_compat import _get_constraint_final_name
+from sqlalchemy import CheckConstraint, Enum, inspect
+from sqlalchemy.dialects.postgresql.base import PGInspector
+
+if TYPE_CHECKING:
+    from sqlalchemy import MetaData
+    from sqlalchemy.engine import Connection
+
 DEFAULT_IGNORE_TABLES: frozenset[str] = frozenset({"alembic_version"})
 
 
@@ -26,3 +36,102 @@ def is_ignored_diff_item(diff_item: tuple, ignore_tables: frozenset[str]) -> boo
         name = getattr(table, "name", None) if table is not None else None
         return name in ignore_tables if name else False
     return False
+
+
+def compare_check_constraints(
+    sync_conn: Connection,
+    metadata: MetaData,
+    schema: str,
+    ignore_tables: frozenset[str] = frozenset(),
+) -> list[str]:
+    """Compare the CHECK constraints ``metadata`` declares with the ones in ``schema``, by name.
+
+    Alembic's ``compare_metadata()`` never looks at CHECK constraints. Names are resolved
+    the way the DDL would render them — the metadata's naming convention applied, a
+    deferred name such as ``Boolean(create_constraint=True)`` filled in, anything over 63
+    characters truncated — through the same Alembic helper autogenerate uses for index
+    and unique constraint names.
+
+    A named constraint the models declare and the database lacks is always reported.
+    One the database has and the models do not is reported unless the table also
+    declares an unnamed check constraint, which carries whatever name PostgreSQL gave
+    it. Unnamed constraints are never compared.
+
+    Args:
+        sync_conn: Synchronous SQLAlchemy connection.
+        metadata: SQLAlchemy ``MetaData`` of the ORM models.
+        schema: PostgreSQL schema the migrations ran in.
+        ignore_tables: Table names to skip.
+
+    Returns:
+        One line per difference; empty when the constraints match.
+    """
+    inspector = inspect(sync_conn)
+    differences: list[str] = []
+    for table in metadata.sorted_tables:
+        if table.name in ignore_tables:
+            continue
+        names = {
+            _get_constraint_final_name(constraint, sync_conn.dialect)
+            for constraint in table.constraints
+            if isinstance(constraint, CheckConstraint)
+        }
+        expected = {name for name in names if name}
+        actual = {name for c in inspector.get_check_constraints(table.name, schema=schema) if (name := c["name"])}
+        for name in sorted(expected - actual):
+            differences.append(
+                f"Check constraint '{name}' on table '{table.name}' is in the models but not in the database."
+            )
+        if None not in names:
+            for name in sorted(actual - expected):
+                differences.append(
+                    f"Check constraint '{name}' on table '{table.name}' is in the database but not in the models."
+                )
+    return differences
+
+
+def compare_enums(
+    sync_conn: Connection,
+    metadata: MetaData,
+    schema: str,
+    ignore_tables: frozenset[str] = frozenset(),
+) -> list[str]:
+    """Compare the values of every native ``Enum`` column in ``metadata`` with its type in the database.
+
+    Alembic's ``compare_metadata()`` never looks at enum members. Values are compared as
+    ordered lists, so a value added in the wrong position counts as a difference. The type
+    is looked up in ``Enum.schema`` when set, otherwise in ``schema``. Non-native and
+    unnamed enums have no type in the database and are skipped; a type in the database
+    that no column uses is not reported.
+
+    Args:
+        sync_conn: Synchronous SQLAlchemy connection.
+        metadata: SQLAlchemy ``MetaData`` of the ORM models.
+        schema: PostgreSQL schema the migrations ran in.
+        ignore_tables: Table names to skip.
+
+    Returns:
+        One line per difference; empty when the enums match.
+    """
+    inspector = inspect(sync_conn)
+    assert isinstance(inspector, PGInspector)
+    expected: dict[tuple[str, str], list[str]] = {}
+    for table in metadata.sorted_tables:
+        if table.name in ignore_tables:
+            continue
+        for column in table.columns:
+            enum = column.type
+            if isinstance(enum, Enum) and enum.native_enum and enum.name:
+                expected[(enum.schema or schema, enum.name)] = list(enum.enums)
+    actual = {(e["schema"], e["name"]): e["labels"] for e in inspector.get_enums(schema="*")}
+
+    differences: list[str] = []
+    for (type_schema, name), values in sorted(expected.items()):
+        labels = actual.get((type_schema, name))
+        if labels is None:
+            differences.append(
+                f"Enum type '{name}' in schema '{type_schema}' is in the models but not in the database."
+            )
+        elif labels != values:
+            differences.append(f"Enum type '{name}' has values {labels} in the database and {values} in the models.")
+    return differences