Deprecate create_engine.implicit_returning

The :paramref:`_sa.create_engine.implicit_returning` parameter is
deprecated on the :func:`_sa.create_engine` function only; the parameter
remains available on the :class:`_schema.Table` object. This parameter was
originally intended to enable the "implicit returning" feature of
SQLAlchemy when it was first developed and was not enabled by default.
Under modern use, there's no reason this parameter should be disabled, and
it has been observed to cause confusion as it degrades performance and
makes it more difficult for the ORM to retrieve recently inserted server
defaults. The parameter remains available on :class:`_schema.Table` to
specifically suit database-level edge cases which make RETURNING
infeasible, the sole example currently being SQL Server's limitation that
INSERT RETURNING may not be used on a table that has INSERT triggers on it.

Also removed from the Oracle dialect some logic that would upgrade
an Oracle 8/8i server version to use implicit returning if the
parameter were explictly passed; these versions of Oracle
still support RETURNING so the feature is now enabled for all
Oracle versions.

Fixes: #6962
Change-Id: Ib338e300cd7c8026c3083043f645084a8211aed8

4 files changed, 328 insertions, 322 deletions

diff --git a/test/sql/test_defaults.py b/test/sql/test_defaults.py
index 31cc15155..aa3728a98 100644
--- a/test/sql/test_defaults.py
+++ b/test/sql/test_defaults.py
@@ -20,7 +20,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
@@ -921,68 +920,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):
diff --git a/test/sql/test_insert_exec.py b/test/sql/test_insert_exec.py
index e0e3b60a9..ef1a3be09 100644
--- a/test/sql/test_insert_exec.py
+++ b/test/sql/test_insert_exec.py
@@ -5,17 +5,16 @@ from sqlalchemy import func
 from sqlalchemy import INT
 from sqlalchemy import Integer
 from sqlalchemy import literal
-from sqlalchemy import MetaData
 from sqlalchemy import Sequence
 from sqlalchemy import sql
 from sqlalchemy import String
 from sqlalchemy import testing
 from sqlalchemy import VARCHAR
 from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_
+from sqlalchemy.testing import mock
 from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
 
@@ -89,10 +88,10 @@ class InsertExecTest(fixtures.TablesTest):
             ],
         )
 
-    def _test_lastrow_accessor(self, table_, values, assertvalues):
+    def _test_lastrow_accessor(self, connection, table_, values, assertvalues):
         """Tests the inserted_primary_key and lastrow_has_id() functions."""
 
-        def insert_values(engine, table_, values):
+        def insert_values(table_, values):
             """
             Inserts a row into a table, returns the full list of values
             INSERTed including defaults that fired off on the DB side and
@@ -100,71 +99,62 @@ class InsertExecTest(fixtures.TablesTest):
             """
 
             # verify implicit_returning is working
-            if engine.dialect.implicit_returning:
+            if (
+                connection.dialect.implicit_returning
+                and table_.implicit_returning
+            ):
                 ins = table_.insert()
-                comp = ins.compile(engine, column_keys=list(values))
+                comp = ins.compile(connection, column_keys=list(values))
                 if not set(values).issuperset(
                     c.key for c in table_.primary_key
                 ):
                     is_(bool(comp.returning), True)
 
-            with engine.begin() as connection:
-                result = connection.execute(table_.insert(), values)
-                ret = values.copy()
-
-                ipk = result.inserted_primary_key
-                for col, id_ in zip(table_.primary_key, ipk):
-                    ret[col.key] = id_
-
-                if result.lastrow_has_defaults():
-                    criterion = and_(
-                        *[
-                            col == id_
-                            for col, id_ in zip(
-                                table_.primary_key, result.inserted_primary_key
-                            )
-                        ]
-                    )
-                    row = connection.execute(
-                        table_.select().where(criterion)
-                    ).first()
-                    for c in table_.c:
-                        ret[c.key] = row._mapping[c]
+            result = connection.execute(table_.insert(), values)
+            ret = values.copy()
+
+            ipk = result.inserted_primary_key
+            for col, id_ in zip(table_.primary_key, ipk):
+                ret[col.key] = id_
+
+            if result.lastrow_has_defaults():
+                criterion = and_(
+                    *[
+                        col == id_
+                        for col, id_ in zip(
+                            table_.primary_key, result.inserted_primary_key
+                        )
+                    ]
+                )
+                row = connection.execute(
+                    table_.select().where(criterion)
+                ).first()
+                for c in table_.c:
+                    ret[c.key] = row._mapping[c]
             return ret, ipk
 
-        if testing.against("postgresql", "oracle", "mssql"):
-            assert testing.db.dialect.implicit_returning
-
-        if testing.db.dialect.implicit_returning:
-            test_engines = [
-                engines.testing_engine(options={"implicit_returning": False}),
-                engines.testing_engine(options={"implicit_returning": True}),
-            ]
-        else:
-            test_engines = [testing.db]
-
-        for engine in test_engines:
-            try:
-                table_.create(bind=engine, checkfirst=True)
-                i, ipk = insert_values(engine, table_, values)
-                eq_(i, assertvalues)
+        table_.create(connection, checkfirst=True)
+        i, ipk = insert_values(table_, values)
+        eq_(i, assertvalues)
 
-                # named tuple tests
-                for col in table_.primary_key:
-                    eq_(getattr(ipk, col.key), assertvalues[col.key])
-                    eq_(ipk._mapping[col.key], assertvalues[col.key])
+        # named tuple tests
+        for col in table_.primary_key:
+            eq_(getattr(ipk, col.key), assertvalues[col.key])
+            eq_(ipk._mapping[col.key], assertvalues[col.key])
 
-                eq_(
-                    ipk._fields, tuple([col.key for col in table_.primary_key])
-                )
-
-            finally:
-                table_.drop(bind=engine)
+        eq_(ipk._fields, tuple([col.key for col in table_.primary_key]))
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_one(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_one(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t1",
                 metadata,
@@ -175,15 +165,23 @@ class InsertExecTest(fixtures.TablesTest):
                     test_needs_autoincrement=True,
                 ),
                 Column("foo", String(30), primary_key=True),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi"},
         )
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_two(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_two(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t2",
                 metadata,
@@ -195,29 +193,45 @@ class InsertExecTest(fixtures.TablesTest):
                 ),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi", "bar": "hi"},
         )
 
-    def test_lastrow_accessor_three(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_three(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t3",
                 metadata,
                 Column("id", String(40), primary_key=True),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30)),
+                implicit_returning=implicit_returning,
             ),
             {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
             {"id": "hi", "foo": "thisisfoo", "bar": "thisisbar"},
         )
 
     @testing.requires.sequences
-    def test_lastrow_accessor_four(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_four(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t4",
                 metadata,
@@ -229,15 +243,23 @@ class InsertExecTest(fixtures.TablesTest):
                 ),
                 Column("foo", String(30), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi", "id": 1},
             {"id": 1, "foo": "hi", "bar": "hi"},
         )
 
     @testing.requires.sequences
-    def test_lastrow_accessor_four_a(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_four_a(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t4",
                 metadata,
@@ -248,28 +270,44 @@ class InsertExecTest(fixtures.TablesTest):
                     primary_key=True,
                 ),
                 Column("foo", String(30)),
+                implicit_returning=implicit_returning,
             ),
             {"foo": "hi"},
             {"id": 1, "foo": "hi"},
         )
 
-    def test_lastrow_accessor_five(self):
-        metadata = MetaData()
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_five(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t5",
                 metadata,
                 Column("id", String(10), primary_key=True),
                 Column("bar", String(30), server_default="hi"),
+                implicit_returning=implicit_returning,
             ),
             {"id": "id1"},
             {"id": "id1", "bar": "hi"},
         )
 
-    @testing.skip_if("sqlite")
-    def test_lastrow_accessor_six(self):
-        metadata = MetaData()
+    @testing.requires.supports_autoincrement_w_composite_pk
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_lastrow_accessor_six(
+        self, metadata, connection, implicit_returning
+    ):
         self._test_lastrow_accessor(
+            connection,
             Table(
                 "t6",
                 metadata,
@@ -280,6 +318,7 @@ class InsertExecTest(fixtures.TablesTest):
                     test_needs_autoincrement=True,
                 ),
                 Column("bar", Integer, primary_key=True),
+                implicit_returning=implicit_returning,
             ),
             {"bar": 0},
             {"id": 1, "bar": 0},
@@ -287,35 +326,28 @@ class InsertExecTest(fixtures.TablesTest):
 
     # TODO: why not in the sqlite suite?
     @testing.only_on("sqlite+pysqlite")
-    @testing.provide_metadata
-    def test_lastrowid_zero(self):
+    def test_lastrowid_zero(self, metadata, connection):
         from sqlalchemy.dialects import sqlite
 
-        eng = engines.testing_engine()
-
         class ExcCtx(sqlite.base.SQLiteExecutionContext):
             def get_lastrowid(self):
                 return 0
 
-        eng.dialect.execution_ctx_cls = ExcCtx
         t = Table(
             "t",
             self.metadata,
             Column("x", Integer, primary_key=True),
             Column("y", Integer),
         )
-        with eng.begin() as conn:
-            t.create(conn)
-            r = conn.execute(t.insert().values(y=5))
+        t.create(connection)
+        with mock.patch.object(
+            connection.dialect, "execution_ctx_cls", ExcCtx
+        ):
+            r = connection.execute(t.insert().values(y=5))
             eq_(r.inserted_primary_key, (0,))
 
-    @testing.fails_on(
-        "sqlite", "sqlite autoincrement doesn't work with composite pks"
-    )
-    @testing.provide_metadata
-    def test_misordered_lastrow(self, connection):
-        metadata = self.metadata
-
+    @testing.requires.supports_autoincrement_w_composite_pk
+    def test_misordered_lastrow(self, connection, metadata):
         related = Table(
             "related",
             metadata,
diff --git a/test/sql/test_returning.py b/test/sql/test_returning.py
index 4069416d4..138e7a4c6 100644
--- a/test/sql/test_returning.py
+++ b/test/sql/test_returning.py
@@ -16,7 +16,6 @@ from sqlalchemy import update
 from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import AssertsExecutionResults
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing.schema import Column
@@ -767,35 +766,3 @@ class ReturnDefaultsTest(fixtures.TablesTest):
             result.inserted_primary_key_rows,
             [(10,), (11,), (12,), (13,), (14,), (15,)],
         )
-
-
-class ImplicitReturningFlag(fixtures.TestBase):
-    __backend__ = True
-
-    def test_flag_turned_off(self):
-        e = engines.testing_engine(options={"implicit_returning": False})
-        assert e.dialect.implicit_returning is False
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is False
-
-    def test_flag_turned_on(self):
-        e = engines.testing_engine(options={"implicit_returning": True})
-        assert e.dialect.implicit_returning is True
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is True
-
-    def test_flag_turned_default(self):
-        supports = [False]
-
-        def go():
-            supports[0] = True
-
-        testing.requires.returning(go)()
-        e = engines.testing_engine()
-
-        # version detection on connect sets it
-        c = e.connect()
-        c.close()
-        assert e.dialect.implicit_returning is supports[0]
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index d6906f9e6..c5c76ad08 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -11,7 +11,6 @@ from sqlalchemy.schema import CreateSequence
 from sqlalchemy.schema import DropSequence
 from sqlalchemy.sql import select
 from sqlalchemy.testing import assert_raises_message
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import fixtures
 from sqlalchemy.testing import is_false
@@ -180,29 +179,30 @@ class SequenceExecTest(fixtures.TestBase):
         connection.execute(t1.insert().values(x=s.next_value()))
         self._assert_seq_result(connection.scalar(t1.select()))
 
-    @testing.provide_metadata
-    def test_inserted_pk_no_returning(self):
+    def test_inserted_pk_no_returning(self, metadata, connection):
         """test inserted_primary_key contains [None] when
         pk_col=next_value(), implicit returning is not used."""
 
         # I'm not really sure what this test wants to accomlish.
 
-        metadata = self.metadata
-        t1 = Table("t", metadata, Column("x", Integer, primary_key=True))
+        t1 = Table(
+            "t",
+            metadata,
+            Column("x", Integer, primary_key=True),
+            implicit_returning=False,
+        )
         s = Sequence("my_sequence_here", metadata=metadata)
 
-        e = engines.testing_engine(options={"implicit_returning": False})
-        with e.begin() as conn:
-
-            t1.create(conn)
-            s.create(conn)
+        conn = connection
+        t1.create(conn)
+        s.create(conn)
 
-            r = conn.execute(t1.insert().values(x=s.next_value()))
+        r = conn.execute(t1.insert().values(x=s.next_value()))
 
-            if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
-                eq_(r.inserted_primary_key, (1,))
-            else:
-                eq_(r.inserted_primary_key, (None,))
+        if testing.requires.emulated_lastrowid_even_with_sequences.enabled:
+            eq_(r.inserted_primary_key, (1,))
+        else:
+            eq_(r.inserted_primary_key, (None,))
 
     @testing.combinations(
         ("implicit_returning",),
@@ -213,43 +213,40 @@ class SequenceExecTest(fixtures.TestBase):
         argnames="returning",
     )
     @testing.requires.multivalues_inserts
-    def test_seq_multivalues_inline(self, metadata, testing_engine, returning):
+    def test_seq_multivalues_inline(self, metadata, connection, returning):
+        _implicit_returning = "no_implicit_returning" not in returning
         t1 = Table(
             "t",
             metadata,
             Column("x", Integer, Sequence("my_seq"), primary_key=True),
             Column("data", String(50)),
+            implicit_returning=_implicit_returning,
         )
 
-        e = engines.testing_engine(
-            options={
-                "implicit_returning": "no_implicit_returning" not in returning
-            }
+        metadata.create_all(connection)
+        conn = connection
+
+        stmt = t1.insert().values(
+            [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
         )
-        metadata.create_all(e)
-        with e.begin() as conn:
+        if returning == "explicit_returning":
+            stmt = stmt.returning(t1.c.x)
+        elif "return_defaults" in returning:
+            stmt = stmt.return_defaults()
 
-            stmt = t1.insert().values(
-                [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
-            )
-            if returning == "explicit_returning":
-                stmt = stmt.returning(t1.c.x)
-            elif "return_defaults" in returning:
-                stmt = stmt.return_defaults()
-
-            r = conn.execute(stmt)
-            if returning == "explicit_returning":
-                eq_(r.all(), [(1,), (2,), (3,)])
-            elif "return_defaults" in returning:
-                eq_(r.returned_defaults_rows, None)
+        r = conn.execute(stmt)
+        if returning == "explicit_returning":
+            eq_(r.all(), [(1,), (2,), (3,)])
+        elif "return_defaults" in returning:
+            eq_(r.returned_defaults_rows, None)
 
-                # TODO: not sure what this is
-                eq_(r.inserted_primary_key_rows, [(None,)])
+            # TODO: not sure what this is
+            eq_(r.inserted_primary_key_rows, [(None,)])
 
-            eq_(
-                conn.execute(t1.select().order_by(t1.c.x)).all(),
-                [(1, "d1"), (2, "d2"), (3, "d3")],
-            )
+        eq_(
+            conn.execute(t1.select().order_by(t1.c.x)).all(),
+            [(1, "d1"), (2, "d2"), (3, "d3")],
+        )
 
     @testing.combinations(
         ("implicit_returning",),
@@ -272,54 +269,49 @@ class SequenceExecTest(fixtures.TestBase):
         argnames="returning",
     )
     def test_seq_multivalues_executemany(
-        self, metadata, testing_engine, returning
+        self, connection, metadata, returning
     ):
+        _implicit_returning = "no_implicit_returning" not in returning
         t1 = Table(
             "t",
             metadata,
             Column("x", Integer, Sequence("my_seq"), primary_key=True),
             Column("data", String(50)),
+            implicit_returning=_implicit_returning,
         )
 
-        e = engines.testing_engine(
-            options={
-                "implicit_returning": "no_implicit_returning" not in returning
-            }
-        )
-        metadata.create_all(e)
-        with e.begin() as conn:
+        metadata.create_all(connection)
+        conn = connection
 
-            stmt = t1.insert()
-            if returning == "explicit_returning":
-                stmt = stmt.returning(t1.c.x)
-            elif "return_defaults" in returning:
-                stmt = stmt.return_defaults()
+        stmt = t1.insert()
+        if returning == "explicit_returning":
+            stmt = stmt.returning(t1.c.x)
+        elif "return_defaults" in returning:
+            stmt = stmt.return_defaults()
 
-            r = conn.execute(
-                stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
-            )
-            if returning == "explicit_returning":
-                eq_(r.all(), [(1,), (2,), (3,)])
-            elif "return_defaults" in returning:
-                if "no_implicit_returning" in returning:
-                    eq_(r.returned_defaults_rows, None)
-                    eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
-                else:
-                    eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
-                    eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
-
-            eq_(
-                conn.execute(t1.select().order_by(t1.c.x)).all(),
-                [(1, "d1"), (2, "d2"), (3, "d3")],
-            )
+        r = conn.execute(
+            stmt, [{"data": "d1"}, {"data": "d2"}, {"data": "d3"}]
+        )
+        if returning == "explicit_returning":
+            eq_(r.all(), [(1,), (2,), (3,)])
+        elif "return_defaults" in returning:
+            if "no_implicit_returning" in returning:
+                eq_(r.returned_defaults_rows, None)
+                eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+            else:
+                eq_(r.returned_defaults_rows, [(1,), (2,), (3,)])
+                eq_(r.inserted_primary_key_rows, [(1,), (2,), (3,)])
+
+        eq_(
+            conn.execute(t1.select().order_by(t1.c.x)).all(),
+            [(1, "d1"), (2, "d2"), (3, "d3")],
+        )
 
     @testing.requires.returning
-    @testing.provide_metadata
-    def test_inserted_pk_implicit_returning(self):
+    def test_inserted_pk_implicit_returning(self, connection, metadata):
         """test inserted_primary_key contains the result when
         pk_col=next_value(), when implicit returning is used."""
 
-        metadata = self.metadata
         s = Sequence("my_sequence")
         t1 = Table(
             "t",
@@ -329,13 +321,12 @@ class SequenceExecTest(fixtures.TestBase):
                 Integer,
                 primary_key=True,
             ),
+            implicit_returning=True,
         )
-        t1.create(testing.db)
+        t1.create(connection)
 
-        e = engines.testing_engine(options={"implicit_returning": True})
-        with e.begin() as conn:
-            r = conn.execute(t1.insert().values(x=s.next_value()))
-            self._assert_seq_result(r.inserted_primary_key[0])
+        r = connection.execute(t1.insert().values(x=s.next_value()))
+        self._assert_seq_result(r.inserted_primary_key[0])
 
 
 class SequenceTest(fixtures.TestBase, testing.AssertsCompiledSQL):
@@ -501,42 +492,56 @@ class TableBoundSequenceTest(fixtures.TablesTest):
     __requires__ = ("sequences",)
     __backend__ = True
 
-    @classmethod
-    def define_tables(cls, metadata):
-        Table(
-            "cartitems",
-            metadata,
-            Column(
-                "cart_id",
-                Integer,
-                Sequence("cart_id_seq"),
-                primary_key=True,
-                autoincrement=False,
-            ),
-            Column("description", String(40)),
-            Column("createdate", sa.DateTime()),
-        )
+    @testing.fixture
+    def table_fixture(self, metadata, connection, implicit_returning):
+        def go(implicit_returning):
+            cartitems = Table(
+                "cartitems",
+                metadata,
+                Column(
+                    "cart_id",
+                    Integer,
+                    Sequence("cart_id_seq"),
+                    primary_key=True,
+                    autoincrement=False,
+                ),
+                Column("description", String(40)),
+                Column("createdate", sa.DateTime()),
+                implicit_returning=implicit_returning,
+            )
 
-        # a little bit of implicit case sensitive naming test going on here
-        Table(
-            "Manager",
-            metadata,
-            Column(
-                "obj_id",
-                Integer,
-                Sequence("obj_id_seq"),
-            ),
-            Column("name", String(128)),
-            Column(
-                "id",
-                Integer,
-                Sequence("Manager_id_seq", optional=True),
-                primary_key=True,
-            ),
-        )
+            # a little bit of implicit case sensitive naming test going on here
+            Manager = Table(
+                "Manager",
+                metadata,
+                Column(
+                    "obj_id",
+                    Integer,
+                    Sequence("obj_id_seq"),
+                ),
+                Column("name", String(128)),
+                Column(
+                    "id",
+                    Integer,
+                    Sequence("Manager_id_seq", optional=True),
+                    primary_key=True,
+                ),
+                implicit_returning=implicit_returning,
+            )
+            metadata.create_all(connection)
+            return Manager, cartitems
 
-    def test_insert_via_seq(self, connection):
-        cartitems = self.tables.cartitems
+        return go
+
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_insert_via_seq(
+        self, table_fixture, connection, implicit_returning
+    ):
+        Manager, cartitems = table_fixture(implicit_returning)
 
         connection.execute(cartitems.insert(), dict(description="hi"))
         connection.execute(cartitems.insert(), dict(description="there"))
@@ -554,52 +559,53 @@ class TableBoundSequenceTest(fixtures.TablesTest):
             expected,
         )
 
-    def test_seq_nonpk(self):
+    @testing.combinations(
+        (True, testing.requires.returning),
+        (False,),
+        argnames="implicit_returning",
+    )
+    def test_seq_nonpk(self, connection, table_fixture, implicit_returning):
         """test sequences fire off as defaults on non-pk columns"""
 
-        sometable = self.tables.Manager
-
-        engine = engines.testing_engine(options={"implicit_returning": False})
+        sometable, cartitems = table_fixture(implicit_returning)
 
-        with engine.begin() as conn:
-            result = conn.execute(sometable.insert(), dict(name="somename"))
+        conn = connection
+        result = conn.execute(sometable.insert(), dict(name="somename"))
 
-            eq_(result.postfetch_cols(), [sometable.c.obj_id])
+        eq_(result.postfetch_cols(), [sometable.c.obj_id])
 
-            result = conn.execute(sometable.insert(), dict(name="someother"))
+        result = conn.execute(sometable.insert(), dict(name="someother"))
 
-            conn.execute(
-                sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
-            )
+        conn.execute(
+            sometable.insert(), [{"name": "name3"}, {"name": "name4"}]
+        )
 
-            dsb = testing.db.dialect.default_sequence_base
-            eq_(
-                list(
-                    conn.execute(sometable.select().order_by(sometable.c.id))
+        dsb = testing.db.dialect.default_sequence_base
+        eq_(
+            list(conn.execute(sometable.select().order_by(sometable.c.id))),
+            [
+                (
+                    dsb,
+                    "somename",
+                    dsb,
                 ),
-                [
-                    (
-                        dsb,
-                        "somename",
-                        dsb,
-                    ),
-                    (
-                        dsb + 1,
-                        "someother",
-                        dsb + 1,
-                    ),
-                    (
-                        dsb + 2,
-                        "name3",
-                        dsb + 2,
-                    ),
-                    (
-                        dsb + 3,
-                        "name4",
-                        dsb + 3,
-                    ),
-                ],
-            )
+                (
+                    dsb + 1,
+                    "someother",
+                    dsb + 1,
+                ),
+                (
+                    dsb + 2,
+                    "name3",
+                    dsb + 2,
+                ),
+                (
+                    dsb + 3,
+                    "name4",
+                    dsb + 3,
+                ),
+            ],
+        )
 
 
 class SequenceAsServerDefaultTest(