From: Justin Malin <justin@joincandidhealth.com>
Date: Mon, 9 Jun 2025 20:01:33 +0000 (+0000)
Subject: Add additional test
X-Git-Url: http://git.ipfire.org/?a=commitdiff_plain;h=196f08b87ca78a8c07d0fcd3b95009997daf086c;p=thirdparty%2Fsqlalchemy%2Falembic.git

Add additional test
---

diff --git a/alembic/autogenerate/render.py b/alembic/autogenerate/render.py
index d5ca2f4f..04e20fac 100644
--- a/alembic/autogenerate/render.py
+++ b/alembic/autogenerate/render.py
@@ -390,6 +390,7 @@ def _add_unique_constraint(
 def _add_fk_constraint(
     autogen_context: AutogenContext, op: ops.CreateForeignKeyOp
 ) -> str:
+    constraint = op.to_constraint()
     args = [repr(_render_gen_name(autogen_context, op.constraint_name))]
     if not autogen_context._has_batch:
         args.append(repr(_ident(op.source_table)))
@@ -419,10 +420,7 @@ def _add_fk_constraint(
             if value is not None:
                 args.append("%s=%r" % (k, value))
 
-
-    # op.to_constraint() may fail with `multiple values for argument 'name'`.
-    dialect_kwarg_view = getattr(op.kw, "dialect_kwargs", {})
-    dialect_kwargs = _render_dialect_kwargs_items(autogen_context, dialect_kwarg_view)
+    dialect_kwargs = _render_dialect_kwargs_items(autogen_context, constraint.dialect_kwargs)
 
     return "%(prefix)screate_foreign_key(%(args)s%(dialect_kwargs)s)" % {
         "prefix": _alembic_autogenerate_prefix(autogen_context),
diff --git a/tests/test_autogen_render.py b/tests/test_autogen_render.py
index 55e80230..38fe5798 100644
--- a/tests/test_autogen_render.py
+++ b/tests/test_autogen_render.py
@@ -41,7 +41,7 @@ from alembic import op  # noqa
 from alembic import testing
 from alembic.autogenerate import api
 from alembic.migration import MigrationContext
-from alembic.operations import ops
+from alembic.operations import Operations, ops
 from alembic.testing import assert_raises
 from alembic.testing import assertions
 from alembic.testing import config
@@ -314,6 +314,18 @@ class AutogenRenderTest(TestBase):
             "somedialect_foobar='option')",
         )
 
+    def test_add_fk_constraint__dialect_kwargs(self):
+        t1 = self.table()
+        t2 = self.table()
+        item = ForeignKeyConstraint([t1.c.id], [t2.c.id], name="fk", postgresql_not_valid=True)
+        fk_obj = ops.CreateForeignKeyOp.from_constraint(item)
+
+        eq_ignore_whitespace(
+            re.sub( r"u'", "'", autogenerate.render_op_text(self.autogen_context, fk_obj)),
+            "op.create_foreign_key('fk', 'test', 'test', ['id'], ['id'], postgresql_not_valid=True)",
+        )
+
+
     def test_drop_index_batch(self):
         """
         autogenerate.render._drop_index
@@ -2466,6 +2478,52 @@ class AutogenRenderTest(TestBase):
             "server_default=None)",
         )
 
+# class PostgresDialectKwargsTest(TestBase):
+#     def setUp(self):
+#         ctx_opts = {
+#             "sqlalchemy_module_prefix": "sa.",
+#             "alembic_module_prefix": "op.",
+#             "target_metadata": MetaData(),
+#         }
+#         context = MigrationContext.configure(
+#             dialect_name="postgresql", opts=ctx_opts
+#         )
+#         self.autogen_context = api.AutogenContext(context)
+#
+#     def test_add_fk_constraint__dialect_kwargs(self):
+#         t1 = Table(
+#             "t1",
+#             MetaData(),
+#             Column("c1", Integer, primary_key=True),
+#         )
+#         t2 = Table(
+#             "t2",
+#             MetaData(),
+#             Column("c2", Integer, primary_key=True),
+#         )
+#         item = ForeignKeyConstraint([t1.c.c1], [t2.c.c2], name="fk", postgresql_not_valid=True)
+#
+#         # fk_obj = ops.CreateForeignKeyOp(
+#         #     constraint_name="fk",
+#         #     source_table="t1",
+#         #     referent_table="t2",
+#         #     local_cols=["c1"],
+#         #     remote_cols=["c2"],
+#         #     **item.__dict__,
+#         #     postgresql_not_valid=True
+#         # )
+#         fk_obj = ops.CreateForeignKeyOp.from_constraint(item)
+#         print(fk_obj.kw)
+#         fk_obj.kw["postgresql_not_valid"] = True
+#
+#         print(fk_obj.kw)
+#         print(fk_obj.kw.get("dialect_kwargs", {}))
+#
+#         eq_ignore_whitespace(
+#             re.sub( r"u'", "'", autogenerate.render_op_text(self.autogen_context, fk_obj)),
+#             "op.create_foreign_key('fk', 't1', 't2', ['c1'], ['c2'], postgresql_not_valid=True)",
+#         )
+
 
 class RenderNamingConventionTest(TestBase):
     def setUp(self):