Merge "Deprecate create_engine.implicit_returning" into main

1 files changed, 53 insertions, 52 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 92e59d9a8..2011744fb 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -19,7 +19,6 @@ from sqlalchemy.sql import select
 from sqlalchemy.sql import text
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import expect_warnings
 from sqlalchemy.testing import fixtures
@@ -918,68 +917,70 @@ class CTEDefaultTest(fixtures.TablesTest):
         eq_(conn.execute(select(q.c.x, q.c.y)).first(), expected)
 
 
-class PKDefaultTest(fixtures.TablesTest):
+class PKDefaultTest(fixtures.TestBase):
     __requires__ = ("subqueries",)
     __backend__ = True
 
-    @classmethod
-    def define_tables(cls, metadata):
-        t2 = Table("t2", metadata, Column("nextid", Integer))
+    @testing.fixture
+    def table_fixture(self, metadata, connection):
+        def go(implicit_returning):
+            t2 = Table(
+                "t2",
+                metadata,
+                Column("nextid", Integer),
+                implicit_returning=implicit_returning,
+            )
 
-        Table(
-            "t1",
-            metadata,
-            Column(
-                "id",
-                Integer,
-                primary_key=True,
-                default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
-            ),
-            Column("data", String(30)),
-        )
+            t1 = Table(
+                "t1",
+                metadata,
+                Column(
+                    "id",
+                    Integer,
+                    primary_key=True,
+                    default=sa.select(func.max(t2.c.nextid)).scalar_subquery(),
+                ),
+                Column("data", String(30)),
+                implicit_returning=implicit_returning,
+            )
 
-        Table(
-            "date_table",
-            metadata,
-            Column(
-                "date_id",
-                DateTime(timezone=True),
-                default=text("current_timestamp"),
-                primary_key=True,
-            ),
-        )
+            date_table = Table(
+                "date_table",
+                metadata,
+                Column(
+                    "date_id",
+                    DateTime(timezone=True),
+                    default=text("current_timestamp"),
+                    primary_key=True,
+                ),
+                implicit_returning=implicit_returning,
+            )
 
-    @testing.requires.returning
-    def test_with_implicit_returning(self):
-        self._test(True)
+            metadata.create_all(connection)
+            return t1, t2, date_table
 
-    def test_regular(self):
-        self._test(False)
+        return go
 
-    def _test(self, returning):
-        t2, t1, date_table = (
-            self.tables.t2,
-            self.tables.t1,
-            self.tables.date_table,
-        )
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_pk_default(self, connection, table_fixture, implicit_returning):
+        t1, t2, date_table = table_fixture(implicit_returning)
 
-        if not returning and not testing.db.dialect.implicit_returning:
-            engine = testing.db
-        else:
-            engine = engines.testing_engine(
-                options={"implicit_returning": returning}
-            )
-        with engine.begin() as conn:
-            conn.execute(t2.insert(), dict(nextid=1))
-            r = conn.execute(t1.insert(), dict(data="hi"))
-            eq_((1,), r.inserted_primary_key)
+        conn = connection
+
+        conn.execute(t2.insert(), dict(nextid=1))
+        r = conn.execute(t1.insert(), dict(data="hi"))
+        eq_((1,), r.inserted_primary_key)
 
-            conn.execute(t2.insert(), dict(nextid=2))
-            r = conn.execute(t1.insert(), dict(data="there"))
-            eq_((2,), r.inserted_primary_key)
+        conn.execute(t2.insert(), dict(nextid=2))
+        r = conn.execute(t1.insert(), dict(data="there"))
+        eq_((2,), r.inserted_primary_key)
 
-            r = conn.execute(date_table.insert())
-            assert isinstance(r.inserted_primary_key[0], datetime.datetime)
+        r = conn.execute(date_table.insert())
+        assert isinstance(r.inserted_primary_key[0], datetime.datetime)
 
 
 class PKIncrementTest(fixtures.TablesTest):