Name the check constraints as their table definitions do

Revision 0002 passed full check constraint names, such as
ck_simulations_kind, which the naming convention prefixed a second time,
so the database held ck_simulations_ck_simulations_kind while the table
definitions name it ck_simulations_kind. alembic check does not compare
check constraints, so nothing noticed; a later revision dropping one by its
defined name would have failed. The revision now marks the names as final
(op.f), and a test compares every constraint name in the migrated database
with the names the table definitions give.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

Author: anth-volk

libs/policyengine-simulation-persistence/tests/test_migration_cycle.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import re
 import time
 
 import pytest
@@ -18,10 +19,13 @@
     HEAD_REVISION,
     SIMULATION_SCHEMA,
     create_simulation_engine,
+    simulation_metadata,
 )
 from report_support import economy_pair, economy_report, report_store
 from sqlalchemy import create_engine, inspect, text
+from sqlalchemy.dialects import postgresql
 from sqlalchemy.exc import OperationalError, ProgrammingError
+from sqlalchemy.schema import CreateTable
 
 pytestmark = pytest.mark.postgres
 
@@ -229,3 +233,36 @@ def test_the_runtime_role_cannot_change_or_delete_a_definition(
                 connection.execute(text(statement))
     finally:
         engine.dispose()
+
+
+def _declared_constraint_names() -> set[str]:
+    """Return the names the table definitions give their constraints."""
+    names: set[str] = set()
+    for table in simulation_metadata.sorted_tables:
+        ddl = str(CreateTable(table).compile(dialect=postgresql.dialect()))
+        names.update(re.findall(r"CONSTRAINT (\w+)", ddl))
+    return names
+
+
+def test_every_constraint_has_the_name_its_definition_gives_it(
+    migrated_database,
+) -> None:
+    """Compare names here, since alembic check does not compare check constraints."""
+    engine = _superuser_engine(migrated_database)
+    try:
+        with engine.connect() as connection:
+            database_names = set(
+                connection.execute(
+                    text(
+                        "SELECT conname FROM pg_constraint "
+                        "JOIN pg_namespace ON pg_namespace.oid = connamespace "
+                        "WHERE nspname = 'simulation' "
+                        "AND contype IN ('c', 'f', 'p', 'u') "
+                        "AND conrelid <> 'simulation.alembic_version'::regclass"
+                    )
+                ).scalars()
+            )
+    finally:
+        engine.dispose()
+
+    assert database_names == _declared_constraint_names()

migrations/versions/0002_simulations_and_reports.py

@@ -54,27 +54,29 @@ def _create_simulations() -> None:
         ),
         sa.CheckConstraint(
             "kind IN ('economy', 'household')",
-            name="ck_simulations_kind",
+            name=op.f("ck_simulations_kind"),
+        ),
+        sa.CheckConstraint(
+            "country ~ '^[a-z]{2}$'", name=op.f("ck_simulations_country")
         ),
-        sa.CheckConstraint("country ~ '^[a-z]{2}$'", name="ck_simulations_country"),
         sa.CheckConstraint(
             "(kind = 'household' AND household_id IS NOT NULL "
             "AND region_code IS NULL AND dataset_name IS NULL) "
             "OR (kind = 'economy' AND household_id IS NULL "
             "AND region_code IS NOT NULL)",
-            name="ck_simulations_population_matches_kind",
+            name=op.f("ck_simulations_population_matches_kind"),
         ),
         sa.CheckConstraint(
             "region_code IS NULL OR region_code ~ '^\\S+$'",
-            name="ck_simulations_region_code",
+            name=op.f("ck_simulations_region_code"),
         ),
         sa.CheckConstraint(
             "dataset_name IS NULL OR dataset_name ~ '^\\S+$'",
-            name="ck_simulations_dataset_name",
+            name=op.f("ck_simulations_dataset_name"),
         ),
         sa.CheckConstraint(
             "legacy_simulation_id IS NULL OR legacy_simulation_id > 0",
-            name="ck_simulations_legacy_simulation_id",
+            name=op.f("ck_simulations_legacy_simulation_id"),
         ),
         sa.ForeignKeyConstraint(
             ["policy_id", "country"],
@@ -136,19 +138,21 @@ def _create_reports() -> None:
             server_default=sa.text("now()"),
             nullable=False,
         ),
-        sa.CheckConstraint("kind IN ('economy', 'household')", name="ck_reports_kind"),
-        sa.CheckConstraint("country ~ '^[a-z]{2}$'", name="ck_reports_country"),
-        sa.CheckConstraint("year BETWEEN 1900 AND 2200", name="ck_reports_year"),
         sa.CheckConstraint(
-            "jsonb_typeof(options) = 'object'", name="ck_reports_options"
+            "kind IN ('economy', 'household')", name=op.f("ck_reports_kind")
+        ),
+        sa.CheckConstraint("country ~ '^[a-z]{2}$'", name=op.f("ck_reports_country")),
+        sa.CheckConstraint("year BETWEEN 1900 AND 2200", name=op.f("ck_reports_year")),
+        sa.CheckConstraint(
+            "jsonb_typeof(options) = 'object'", name=op.f("ck_reports_options")
         ),
         sa.CheckConstraint(
             "kind = 'household' OR reform_simulation_id IS NOT NULL",
-            name="ck_reports_economy_has_reform",
+            name=op.f("ck_reports_economy_has_reform"),
         ),
         sa.CheckConstraint(
             "legacy_report_id IS NULL OR legacy_report_id > 0",
-            name="ck_reports_legacy_report_id",
+            name=op.f("ck_reports_legacy_report_id"),
         ),
         sa.ForeignKeyConstraint(
             ["baseline_simulation_id", "country", "kind"],