remove metadata.bind use from test suite

importantly this means we can remove bound metadata from
the fixtures that are used by Alembic's test suite.

hopefully this is the last one that has to happen to allow
Alembic to be fully 1.4/2.0.

Start moving from @testing.provide_metadata to a pytest
metadata fixture.  This does not seem to have any negative
effects even though TablesTest uses a "self.metadata" attribute.

Change-Id: Iae6ab95938a7e92b6d42086aec534af27b5577d3

4 files changed, 498 insertions, 621 deletions

diff --git a/test/engine/test_ddlevents.py b/test/engine/test_ddlevents.py
index 5cbb47854..396b48aa4 100644
--- a/test/engine/test_ddlevents.py
+++ b/test/engine/test_ddlevents.py
@@ -376,7 +376,7 @@ class DDLEventTest(fixtures.TestBase):
 class DDLExecutionTest(fixtures.TestBase):
     def setup(self):
         self.engine = engines.mock_engine()
-        self.metadata = MetaData(self.engine)
+        self.metadata = MetaData()
         self.users = Table(
             "users",
             self.metadata,
@@ -391,14 +391,14 @@ class DDLExecutionTest(fixtures.TestBase):
         event.listen(users, "before_drop", DDL("xyzzy"))
         event.listen(users, "after_drop", DDL("fnord"))
 
-        users.create()
+        users.create(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" in strings
         assert "klptzyxm" in strings
         assert "xyzzy" not in strings
         assert "fnord" not in strings
         del engine.mock[:]
-        users.drop()
+        users.drop(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" not in strings
         assert "klptzyxm" not in strings
@@ -413,14 +413,14 @@ class DDLExecutionTest(fixtures.TestBase):
         event.listen(users, "before_drop", DDL("xyzzy"))
         event.listen(users, "after_drop", DDL("fnord"))
 
-        metadata.create_all()
+        metadata.create_all(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" in strings
         assert "klptzyxm" in strings
         assert "xyzzy" not in strings
         assert "fnord" not in strings
         del engine.mock[:]
-        metadata.drop_all()
+        metadata.drop_all(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" not in strings
         assert "klptzyxm" not in strings
@@ -435,14 +435,14 @@ class DDLExecutionTest(fixtures.TestBase):
         event.listen(metadata, "before_drop", DDL("xyzzy"))
         event.listen(metadata, "after_drop", DDL("fnord"))
 
-        metadata.create_all()
+        metadata.create_all(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" in strings
         assert "klptzyxm" in strings
         assert "xyzzy" not in strings
         assert "fnord" not in strings
         del engine.mock[:]
-        metadata.drop_all()
+        metadata.drop_all(self.engine)
         strings = [str(x) for x in engine.mock]
         assert "mxyzptlk" not in strings
         assert "klptzyxm" not in strings
diff --git a/test/engine/test_deprecations.py b/test/engine/test_deprecations.py
index 4ca081be2..286231614 100644
--- a/test/engine/test_deprecations.py
+++ b/test/engine/test_deprecations.py
@@ -158,7 +158,10 @@ class ConnectionlessDeprecationTest(fixtures.TestBase):
                 bind.begin()
             try:
                 for args in (([bind], {}), ([], {"bind": bind})):
-                    metadata = MetaData(*args[0], **args[1])
+                    with testing.expect_deprecated_20(
+                        "The MetaData.bind argument is deprecated "
+                    ):
+                        metadata = MetaData(*args[0], **args[1])
                     table = Table(
                         "test_table", metadata, Column("foo", Integer)
                     )
@@ -315,11 +318,11 @@ class ConnectionlessDeprecationTest(fixtures.TestBase):
         ):
             eq_(testing.db.execute(stmt).fetchall(), [(1,)])
 
-    @testing.provide_metadata
-    def test_implicit_execute(self):
-        table = Table("t", self.metadata, Column("a", Integer))
+    def test_implicit_execute(self, metadata):
+        table = Table("t", metadata, Column("a", Integer))
         table.create(testing.db)
 
+        metadata.bind = testing.db
         stmt = table.insert().values(a=1)
         with testing.expect_deprecated_20(
             r"The Executable.execute\(\) method is considered legacy",
@@ -1225,7 +1228,7 @@ class DeprecatedReflectionTest(fixtures.TablesTest):
             is_true(testing.db.has_table("user"))
 
     def test_engine_table_names(self):
-        metadata = self.metadata
+        metadata = self.tables_test_metadata
 
         with testing.expect_deprecated(
             r"The Engine.table_names\(\) method is deprecated"
@@ -1235,7 +1238,8 @@ class DeprecatedReflectionTest(fixtures.TablesTest):
 
     def test_reflecttable(self):
         inspector = inspect(testing.db)
-        metadata = self.metadata
+        metadata = MetaData()
+
         table = Table("user", metadata)
         with testing.expect_deprecated_20(
             r"The Inspector.reflecttable\(\) method is considered "
@@ -1632,7 +1636,7 @@ class EngineEventsTest(fixtures.TestBase):
 class DDLExecutionTest(fixtures.TestBase):
     def setup(self):
         self.engine = engines.mock_engine()
-        self.metadata = MetaData(self.engine)
+        self.metadata = MetaData()
         self.users = Table(
             "users",
             self.metadata,
@@ -1742,7 +1746,7 @@ class AutocommitTextTest(AutocommitKeywordFixture, fixtures.TestBase):
         self._test_keyword("SELECT foo FROM table", False)
 
 
-class ExplicitAutoCommitTest(fixtures.TestBase):
+class ExplicitAutoCommitTest(fixtures.TablesTest):
 
     """test the 'autocommit' flag on select() and text() objects.
 
@@ -1752,36 +1756,31 @@ class ExplicitAutoCommitTest(fixtures.TestBase):
     __only_on__ = "postgresql"
 
     @classmethod
-    def setup_class(cls):
-        global metadata, foo
-        metadata = MetaData(testing.db)
-        foo = Table(
+    def define_tables(cls, metadata):
+        Table(
             "foo",
             metadata,
             Column("id", Integer, primary_key=True),
             Column("data", String(100)),
         )
-        with testing.db.begin() as conn:
-            metadata.create_all(conn)
-            conn.exec_driver_sql(
+
+        event.listen(
+            metadata,
+            "after_create",
+            DDL(
                 "create function insert_foo(varchar) "
                 "returns integer as 'insert into foo(data) "
                 "values ($1);select 1;' language sql"
-            )
-
-    def teardown(self):
-        with testing.db.begin() as conn:
-            conn.execute(foo.delete())
-
-    @classmethod
-    def teardown_class(cls):
-        with testing.db.begin() as conn:
-            conn.exec_driver_sql("drop function insert_foo(varchar)")
-            metadata.drop_all(conn)
+            ),
+        )
+        event.listen(
+            metadata, "before_drop", DDL("drop function insert_foo(varchar)")
+        )
 
     def test_control(self):
 
         # test that not using autocommit does not commit
+        foo = self.tables.foo
 
         conn1 = testing.db.connect()
         conn2 = testing.db.connect()
@@ -1799,6 +1798,8 @@ class ExplicitAutoCommitTest(fixtures.TestBase):
         conn2.close()
 
     def test_explicit_compiled(self):
+        foo = self.tables.foo
+
         conn1 = testing.db.connect()
         conn2 = testing.db.connect()
 
@@ -1816,6 +1817,8 @@ class ExplicitAutoCommitTest(fixtures.TestBase):
         conn2.close()
 
     def test_explicit_connection(self):
+        foo = self.tables.foo
+
         conn1 = testing.db.connect()
         conn2 = testing.db.connect()
         with testing.expect_deprecated_20(
@@ -1853,6 +1856,8 @@ class ExplicitAutoCommitTest(fixtures.TestBase):
         conn2.close()
 
     def test_explicit_text(self):
+        foo = self.tables.foo
+
         conn1 = testing.db.connect()
         conn2 = testing.db.connect()
         with testing.expect_deprecated_20(
@@ -1869,6 +1874,8 @@ class ExplicitAutoCommitTest(fixtures.TestBase):
         conn2.close()
 
     def test_implicit_text(self):
+        foo = self.tables.foo
+
         conn1 = testing.db.connect()
         conn2 = testing.db.connect()
         with testing.expect_deprecated_20(
diff --git a/test/engine/test_execute.py b/test/engine/test_execute.py
index 6239d1f18..21d4e06e0 100644
--- a/test/engine/test_execute.py
+++ b/test/engine/test_execute.py
@@ -830,12 +830,11 @@ class ConvenienceExecuteTest(fixtures.TablesTest):
             self._assert_no_data()
 
 
-class CompiledCacheTest(fixtures.TablesTest):
+class CompiledCacheTest(fixtures.TestBase):
     __backend__ = True
 
-    @classmethod
-    def define_tables(cls, metadata):
-        Table(
+    def test_cache(self, connection, metadata):
+        users = Table(
             "users",
             metadata,
             Column(
@@ -844,9 +843,7 @@ class CompiledCacheTest(fixtures.TablesTest):
             Column("user_name", VARCHAR(20)),
             Column("extra_data", VARCHAR(20)),
         )
-
-    def test_cache(self, connection):
-        users = self.tables.users
+        users.create(connection)
 
         conn = connection
         cache = {}
@@ -912,8 +909,17 @@ class CompiledCacheTest(fixtures.TablesTest):
         # the statement values (only the keys).
         eq_(ref_blob(), None)
 
-    def test_keys_independent_of_ordering(self, connection):
-        users = self.tables.users
+    def test_keys_independent_of_ordering(self, connection, metadata):
+        users = Table(
+            "users",
+            metadata,
+            Column(
+                "user_id", INT, primary_key=True, test_needs_autoincrement=True
+            ),
+            Column("user_name", VARCHAR(20)),
+            Column("extra_data", VARCHAR(20)),
+        )
+        users.create(connection)
 
         connection.execute(
             users.insert(),
@@ -961,13 +967,10 @@ class CompiledCacheTest(fixtures.TablesTest):
         eq_(len(cache), 1)
 
     @testing.requires.schemas
-    @testing.provide_metadata
-    def test_schema_translate_in_key(self):
-        Table("x", self.metadata, Column("q", Integer))
-        Table(
-            "x", self.metadata, Column("q", Integer), schema=config.test_schema
-        )
-        self.metadata.create_all()
+    def test_schema_translate_in_key(self, metadata, connection):
+        Table("x", metadata, Column("q", Integer))
+        Table("x", metadata, Column("q", Integer), schema=config.test_schema)
+        metadata.create_all(connection)
 
         m = MetaData()
         t1 = Table("x", m, Column("q", Integer))
@@ -975,33 +978,30 @@ class CompiledCacheTest(fixtures.TablesTest):
         stmt = select(t1.c.q)
 
         cache = {}
-        with config.db.begin() as conn:
-            conn = conn.execution_options(compiled_cache=cache)
-            conn.execute(ins, {"q": 1})
-            eq_(conn.scalar(stmt), 1)
 
-        with config.db.begin() as conn:
-            conn = conn.execution_options(
-                compiled_cache=cache,
-                schema_translate_map={None: config.test_schema},
-            )
-            conn.execute(ins, {"q": 2})
-            eq_(conn.scalar(stmt), 2)
+        conn = connection.execution_options(compiled_cache=cache)
+        conn.execute(ins, {"q": 1})
+        eq_(conn.scalar(stmt), 1)
 
-        with config.db.begin() as conn:
-            conn = conn.execution_options(
-                compiled_cache=cache,
-                schema_translate_map={None: None},
-            )
-            # should use default schema again even though statement
-            # was compiled with test_schema in the map
-            eq_(conn.scalar(stmt), 1)
+        conn = connection.execution_options(
+            compiled_cache=cache,
+            schema_translate_map={None: config.test_schema},
+        )
+        conn.execute(ins, {"q": 2})
+        eq_(conn.scalar(stmt), 2)
 
-        with config.db.begin() as conn:
-            conn = conn.execution_options(
-                compiled_cache=cache,
-            )
-            eq_(conn.scalar(stmt), 1)
+        conn = connection.execution_options(
+            compiled_cache=cache,
+            schema_translate_map={None: None},
+        )
+        # should use default schema again even though statement
+        # was compiled with test_schema in the map
+        eq_(conn.scalar(stmt), 1)
+
+        conn = connection.execution_options(
+            compiled_cache=cache,
+        )
+        eq_(conn.scalar(stmt), 1)
 
 
 class MockStrategyTest(fixtures.TestBase):
@@ -1079,7 +1079,7 @@ class SchemaTranslateTest(fixtures.TestBase, testing.AssertsExecutionResults):
         Table("t1", metadata, Column("x", Integer), schema=config.test_schema)
         Table("t2", metadata, Column("x", Integer), schema=config.test_schema)
         Table("t3", metadata, Column("x", Integer), schema=None)
-        metadata.create_all()
+        metadata.create_all(testing.db)
 
     def test_ddl_hastable(self):
 
@@ -1772,7 +1772,7 @@ class EngineEventsTest(fixtures.TestBase):
         ]:
             event.listen(engine, "before_execute", execute)
             event.listen(engine, "before_cursor_execute", cursor_execute)
-            m = MetaData(engine)
+            m = MetaData()
             t1 = Table(
                 "t1",
                 m,
diff --git a/test/engine/test_reflection.py b/test/engine/test_reflection.py
index 48b6c40d7..658cdd79f 100644
--- a/test/engine/test_reflection.py
+++ b/test/engine/test_reflection.py
@@ -21,7 +21,6 @@ from sqlalchemy.testing import assert_raises_message
 from sqlalchemy.testing import AssertsCompiledSQL
 from sqlalchemy.testing import ComparesTables
 from sqlalchemy.testing import config
-from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import eq_regex
 from sqlalchemy.testing import expect_warnings
@@ -43,13 +42,8 @@ from sqlalchemy.util import ue
 class ReflectionTest(fixtures.TestBase, ComparesTables):
     __backend__ = True
 
-    @testing.exclude(
-        "mssql", "<", (10, 0, 0), "Date is only supported on MSSQL 2008+"
-    )
-    @testing.exclude("mysql", "<", (4, 1, 1), "early types are squirrely")
-    @testing.provide_metadata
-    def test_basic_reflection(self):
-        meta = self.metadata
+    def test_basic_reflection(self, connection, metadata):
+        meta = metadata
 
         users = Table(
             "engine_users",
@@ -85,25 +79,22 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("email_address", sa.String(20)),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
 
         meta2 = MetaData()
         reflected_users = Table(
-            "engine_users", meta2, autoload_with=testing.db
+            "engine_users", meta2, autoload_with=connection
         )
         reflected_addresses = Table(
             "engine_email_addresses",
             meta2,
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         self.assert_tables_equal(users, reflected_users)
         self.assert_tables_equal(addresses, reflected_addresses)
 
-    @testing.provide_metadata
-    def test_autoload_with_imply_autoload(
-        self,
-    ):
-        meta = self.metadata
+    def test_autoload_with_imply_autoload(self, metadata, connection):
+        meta = metadata
         t = Table(
             "t",
             meta,
@@ -111,15 +102,14 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("x", sa.String(20)),
             Column("y", sa.Integer),
         )
-        meta.create_all()
+        meta.create_all(connection)
 
         meta2 = MetaData()
-        reflected_t = Table("t", meta2, autoload_with=testing.db)
+        reflected_t = Table("t", meta2, autoload_with=connection)
         self.assert_tables_equal(t, reflected_t)
 
-    @testing.provide_metadata
-    def test_two_foreign_keys(self):
-        meta = self.metadata
+    def test_two_foreign_keys(self, metadata, connection):
+        meta = metadata
         Table(
             "t1",
             meta,
@@ -140,18 +130,17 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("id", sa.Integer, primary_key=True),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
         t1r, t2r, t3r = [
-            Table(x, meta2, autoload_with=testing.db)
+            Table(x, meta2, autoload_with=connection)
             for x in ("t1", "t2", "t3")
         ]
         assert t1r.c.t2id.references(t2r.c.id)
         assert t1r.c.t3id.references(t3r.c.id)
 
-    @testing.provide_metadata
-    def test_resolve_fks_false_table(self):
-        meta = self.metadata
+    def test_resolve_fks_false_table(self, connection, metadata):
+        meta = metadata
         Table(
             "t1",
             meta,
@@ -165,9 +154,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("id", sa.Integer, primary_key=True),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
-        t1 = Table("t1", meta2, resolve_fks=False, autoload_with=testing.db)
+        t1 = Table("t1", meta2, resolve_fks=False, autoload_with=connection)
         in_("t1", meta2.tables)
         not_in("t2", meta2.tables)
 
@@ -176,14 +165,13 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             lambda: list(t1.c.t2id.foreign_keys)[0].column,
         )
 
-        t2 = Table("t2", meta2, autoload_with=testing.db)
+        t2 = Table("t2", meta2, autoload_with=connection)
 
         # now it resolves
         is_true(t1.c.t2id.references(t2.c.id))
 
-    @testing.provide_metadata
-    def test_resolve_fks_false_extend_existing(self):
-        meta = self.metadata
+    def test_resolve_fks_false_extend_existing(self, connection, metadata):
+        meta = metadata
         Table(
             "t1",
             meta,
@@ -197,7 +185,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("id", sa.Integer, primary_key=True),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
         Table("t1", meta2)
         in_("t1", meta2.tables)
@@ -206,7 +194,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             "t1",
             meta2,
             resolve_fks=False,
-            autoload_with=testing.db,
+            autoload_with=connection,
             extend_existing=True,
         )
         not_in("t2", meta2.tables)
@@ -216,14 +204,13 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             lambda: list(t1.c.t2id.foreign_keys)[0].column,
         )
 
-        t2 = Table("t2", meta2, autoload_with=testing.db)
+        t2 = Table("t2", meta2, autoload_with=connection)
 
         # now it resolves
         is_true(t1.c.t2id.references(t2.c.id))
 
-    @testing.provide_metadata
-    def test_resolve_fks_false_metadata(self):
-        meta = self.metadata
+    def test_resolve_fks_false_metadata(self, connection, metadata):
+        meta = metadata
         Table(
             "t1",
             meta,
@@ -237,9 +224,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("id", sa.Integer, primary_key=True),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
-        meta2.reflect(testing.db, resolve_fks=False, only=["t1"])
+        meta2.reflect(connection, resolve_fks=False, only=["t1"])
         in_("t1", meta2.tables)
         not_in("t2", meta2.tables)
 
@@ -250,36 +237,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             lambda: list(t1.c.t2id.foreign_keys)[0].column,
         )
 
-        meta2.reflect(testing.db, resolve_fks=False)
+        meta2.reflect(connection, resolve_fks=False)
 
         t2 = meta2.tables["t2"]
         is_true(t1.c.t2id.references(t2.c.id))
 
-    def test_nonexistent(self):
+    def test_nonexistent(self, connection):
         meta = MetaData()
         assert_raises(
             sa.exc.NoSuchTableError,
             Table,
             "nonexistent",
             meta,
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         assert "nonexistent" not in meta.tables
 
-    @testing.provide_metadata
-    def test_include_columns(self):
-        meta = self.metadata
+    def test_include_columns(self, connection, metadata):
+        meta = metadata
         foo = Table(
             "foo",
             meta,
             *[Column(n, sa.String(30)) for n in ["a", "b", "c", "d", "e", "f"]]
         )
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
         foo = Table(
             "foo",
             meta2,
-            autoload_with=testing.db,
+            autoload_with=connection,
             include_columns=["b", "f", "e"],
         )
         # test that cols come back in original order
@@ -291,7 +277,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
 
         # test against a table which is already reflected
         meta3 = MetaData()
-        foo = Table("foo", meta3, autoload_with=testing.db)
+        foo = Table("foo", meta3, autoload_with=connection)
 
         foo = Table(
             "foo", meta3, include_columns=["b", "f", "e"], extend_existing=True
@@ -302,9 +288,8 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         for c in ("a", "c", "d"):
             assert c not in foo.c
 
-    @testing.provide_metadata
-    def test_extend_existing(self):
-        meta = self.metadata
+    def test_extend_existing(self, connection, metadata):
+        meta = metadata
 
         Table(
             "t",
@@ -314,7 +299,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("y", Integer),
             Column("z", Integer, server_default="5"),
         )
-        meta.create_all()
+        meta.create_all(connection)
 
         m2 = MetaData()
         old_z = Column("z", String, primary_key=True)
@@ -327,7 +312,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             m2,
             old_y,
             extend_existing=True,
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         eq_(set(t2.columns.keys()), set(["x", "y", "z", "q", "id"]))
 
@@ -346,7 +331,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             "t",
             m3,
             extend_existing=False,
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         eq_(set(t3.columns.keys()), set(["z"]))
 
@@ -362,7 +347,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             old_y,
             extend_existing=True,
             autoload_replace=False,
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         eq_(set(t4.columns.keys()), set(["x", "y", "z", "q", "id"]))
         eq_(list(t4.primary_key.columns), [t4.c.z, t4.c.id])
@@ -371,9 +356,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert t4.c.z.type._type_affinity is String
         assert t4.c.q is old_q
 
-    @testing.provide_metadata
-    def test_extend_existing_reflect_all_dont_dupe_index(self):
-        m = self.metadata
+    def test_extend_existing_reflect_all_dont_dupe_index(
+        self, connection, metadata
+    ):
+        m = metadata
         d = Table(
             "d",
             m,
@@ -389,10 +375,10 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("id", Integer, primary_key=True),
             Column("aid", ForeignKey("d.id")),
         )
-        m.create_all()
+        m.create_all(connection)
 
         m2 = MetaData()
-        m2.reflect(testing.db, extend_existing=True)
+        m2.reflect(connection, extend_existing=True)
 
         eq_(
             len(
@@ -422,51 +408,51 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             )
 
     @testing.emits_warning(r".*omitted columns")
-    @testing.provide_metadata
-    def test_include_columns_indexes(self):
-        m = self.metadata
+    def test_include_columns_indexes(self, connection, metadata):
+        m = metadata
 
         t1 = Table("t1", m, Column("a", sa.Integer), Column("b", sa.Integer))
         sa.Index("foobar", t1.c.a, t1.c.b)
         sa.Index("bat", t1.c.a)
-        m.create_all()
+        m.create_all(connection)
         m2 = MetaData()
-        t2 = Table("t1", m2, autoload_with=testing.db)
+        t2 = Table("t1", m2, autoload_with=connection)
         assert len(t2.indexes) == 2
 
         m2 = MetaData()
-        t2 = Table("t1", m2, autoload_with=testing.db, include_columns=["a"])
+        t2 = Table("t1", m2, autoload_with=connection, include_columns=["a"])
         assert len(t2.indexes) == 1
 
         m2 = MetaData()
         t2 = Table(
-            "t1", m2, autoload_with=testing.db, include_columns=["a", "b"]
+            "t1", m2, autoload_with=connection, include_columns=["a", "b"]
         )
         assert len(t2.indexes) == 2
 
-    @testing.provide_metadata
-    def test_autoload_replace_foreign_key_nonpresent(self):
+    def test_autoload_replace_foreign_key_nonpresent(
+        self, connection, metadata
+    ):
         """test autoload_replace=False with col plus FK
         establishes the FK not present in the DB.
 
         """
-        Table("a", self.metadata, Column("id", Integer, primary_key=True))
+        Table("a", metadata, Column("id", Integer, primary_key=True))
         Table(
             "b",
-            self.metadata,
+            metadata,
             Column("id", Integer, primary_key=True),
             Column("a_id", Integer),
         )
-        self.metadata.create_all()
+        metadata.create_all(connection)
 
         m2 = MetaData()
         b2 = Table("b", m2, Column("a_id", Integer, sa.ForeignKey("a.id")))
-        a2 = Table("a", m2, autoload_with=testing.db)
+        a2 = Table("a", m2, autoload_with=connection)
         b2 = Table(
             "b",
             m2,
             extend_existing=True,
-            autoload_with=testing.db,
+            autoload_with=connection,
             autoload_replace=False,
         )
 
@@ -474,30 +460,31 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert b2.c.a_id.references(a2.c.id)
         eq_(len(b2.constraints), 2)
 
-    @testing.provide_metadata
-    def test_autoload_replace_foreign_key_ispresent(self):
+    def test_autoload_replace_foreign_key_ispresent(
+        self, connection, metadata
+    ):
         """test autoload_replace=False with col plus FK mirroring
         DB-reflected FK skips the reflected FK and installs
         the in-python one only.
 
         """
-        Table("a", self.metadata, Column("id", Integer, primary_key=True))
+        Table("a", metadata, Column("id", Integer, primary_key=True))
         Table(
             "b",
-            self.metadata,
+            metadata,
             Column("id", Integer, primary_key=True),
             Column("a_id", Integer, sa.ForeignKey("a.id")),
         )
-        self.metadata.create_all()
+        metadata.create_all(connection)
 
         m2 = MetaData()
         b2 = Table("b", m2, Column("a_id", Integer, sa.ForeignKey("a.id")))
-        a2 = Table("a", m2, autoload_with=testing.db)
+        a2 = Table("a", m2, autoload_with=connection)
         b2 = Table(
             "b",
             m2,
             extend_existing=True,
-            autoload_with=testing.db,
+            autoload_with=connection,
             autoload_replace=False,
         )
 
@@ -505,29 +492,28 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert b2.c.a_id.references(a2.c.id)
         eq_(len(b2.constraints), 2)
 
-    @testing.provide_metadata
-    def test_autoload_replace_foreign_key_removed(self):
+    def test_autoload_replace_foreign_key_removed(self, connection, metadata):
         """test autoload_replace=False with col minus FK that's in the
         DB means the FK is skipped and doesn't get installed at all.
 
         """
-        Table("a", self.metadata, Column("id", Integer, primary_key=True))
+        Table("a", metadata, Column("id", Integer, primary_key=True))
         Table(
             "b",
-            self.metadata,
+            metadata,
             Column("id", Integer, primary_key=True),
             Column("a_id", Integer, sa.ForeignKey("a.id")),
         )
-        self.metadata.create_all()
+        metadata.create_all(connection)
 
         m2 = MetaData()
         b2 = Table("b", m2, Column("a_id", Integer))
-        a2 = Table("a", m2, autoload_with=testing.db)
+        a2 = Table("a", m2, autoload_with=connection)
         b2 = Table(
             "b",
             m2,
             extend_existing=True,
-            autoload_with=testing.db,
+            autoload_with=connection,
             autoload_replace=False,
         )
 
@@ -535,10 +521,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert not b2.c.a_id.references(a2.c.id)
         eq_(len(b2.constraints), 1)
 
-    @testing.provide_metadata
-    def test_autoload_replace_primary_key(self):
-        Table("a", self.metadata, Column("id", Integer))
-        self.metadata.create_all()
+    def test_autoload_replace_primary_key(self, connection, metadata):
+        Table("a", metadata, Column("id", Integer))
+        metadata.create_all(connection)
 
         m2 = MetaData()
         a2 = Table("a", m2, Column("id", Integer, primary_key=True))
@@ -546,7 +531,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         Table(
             "a",
             m2,
-            autoload_with=testing.db,
+            autoload_with=connection,
             autoload_replace=False,
             extend_existing=True,
         )
@@ -555,15 +540,14 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
     def test_autoload_replace_arg(self):
         Table("t", MetaData(), autoload_replace=False)
 
-    @testing.provide_metadata
-    def test_autoincrement_col(self):
+    def test_autoincrement_col(self, connection, metadata):
         """test that 'autoincrement' is reflected according to sqla's policy.
 
         Don't mark this test as unsupported for any backend !
 
         """
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "test",
             meta,
@@ -581,41 +565,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("data", sa.String(50)),
             mysql_engine="InnoDB",
         )
-        meta.create_all()
+        meta.create_all(connection)
         m2 = MetaData()
-        t1a = Table("test", m2, autoload_with=testing.db)
+        t1a = Table("test", m2, autoload_with=connection)
         assert t1a._autoincrement_column is t1a.c.id
 
-        t2a = Table("test2", m2, autoload_with=testing.db)
+        t2a = Table("test2", m2, autoload_with=connection)
         assert t2a._autoincrement_column is None
 
     @skip("sqlite")
-    @testing.provide_metadata
-    def test_unknown_types(self):
+    def test_unknown_types(self, connection, metadata):
         """Test the handling of unknown types for the given dialect.
 
         sqlite is skipped because it has special rules for unknown types using
         'affinity types' - this feature is tested in that dialect's test spec.
         """
-        meta = self.metadata
+        meta = metadata
         t = Table("test", meta, Column("foo", sa.DateTime))
 
-        ischema_names = testing.db.dialect.ischema_names
-        t.create()
-        testing.db.dialect.ischema_names = {}
-        try:
-            m2 = MetaData(testing.db)
+        t.create(connection)
+
+        with mock.patch.object(connection.dialect, "ischema_names", {}):
+            m2 = MetaData()
 
             with testing.expect_warnings("Did not recognize type"):
-                t3 = Table("test", m2, autoload_with=testing.db)
+                t3 = Table("test", m2, autoload_with=connection)
                 is_(t3.c.foo.type.__class__, sa.types.NullType)
 
-        finally:
-            testing.db.dialect.ischema_names = ischema_names
-
-    @testing.provide_metadata
-    def test_basic_override(self):
-        meta = self.metadata
+    def test_basic_override(self, connection, metadata):
+        meta = metadata
         table = Table(
             "override_test",
             meta,
@@ -623,7 +601,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("col2", sa.String(20)),
             Column("col3", sa.Numeric),
         )
-        table.create()
+        table.create(connection)
 
         meta2 = MetaData()
         table = Table(
@@ -631,16 +609,15 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             meta2,
             Column("col2", sa.Unicode()),
             Column("col4", sa.String(30)),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
 
         self.assert_(isinstance(table.c.col1.type, sa.Integer))
         self.assert_(isinstance(table.c.col2.type, sa.Unicode))
         self.assert_(isinstance(table.c.col4.type, sa.String))
 
-    @testing.provide_metadata
-    def test_override_upgrade_pk_flag(self):
-        meta = self.metadata
+    def test_override_upgrade_pk_flag(self, connection, metadata):
+        meta = metadata
         table = Table(
             "override_test",
             meta,
@@ -648,26 +625,25 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("col2", sa.String(20)),
             Column("col3", sa.Numeric),
         )
-        table.create()
+        table.create(connection)
 
         meta2 = MetaData()
         table = Table(
             "override_test",
             meta2,
             Column("col1", sa.Integer, primary_key=True),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
 
         eq_(list(table.primary_key), [table.c.col1])
         eq_(table.c.col1.primary_key, True)
 
-    @testing.provide_metadata
-    def test_override_pkfk(self):
+    def test_override_pkfk(self, connection, metadata):
         """test that you can override columns which contain foreign keys
         to other reflected tables, where the foreign key column is also
         a primary key column"""
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "users",
             meta,
@@ -681,7 +657,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("street", sa.String(30)),
         )
 
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
         a2 = Table(
             "addresses",
@@ -689,36 +665,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column(
                 "id", sa.Integer, sa.ForeignKey("users.id"), primary_key=True
             ),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
-        u2 = Table("users", meta2, autoload_with=testing.db)
+        u2 = Table("users", meta2, autoload_with=connection)
 
         assert list(a2.primary_key) == [a2.c.id]
         assert list(u2.primary_key) == [u2.c.id]
         assert u2.join(a2).onclause.compare(u2.c.id == a2.c.id)
 
         meta3 = MetaData()
-        u3 = Table("users", meta3, autoload_with=testing.db)
+        u3 = Table("users", meta3, autoload_with=connection)
         a3 = Table(
             "addresses",
             meta3,
             Column(
                 "id", sa.Integer, sa.ForeignKey("users.id"), primary_key=True
             ),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
 
         assert list(a3.primary_key) == [a3.c.id]
         assert list(u3.primary_key) == [u3.c.id]
         assert u3.join(a3).onclause.compare(u3.c.id == a3.c.id)
 
-    @testing.provide_metadata
-    def test_override_nonexistent_fk(self):
+    def test_override_nonexistent_fk(self, connection, metadata):
         """test that you can override columns and create new foreign
         keys to other reflected tables which have no foreign keys.  this
         is common with MySQL MyISAM tables."""
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "users",
             meta,
@@ -733,15 +708,15 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("user_id", sa.Integer),
         )
 
-        meta.create_all()
+        meta.create_all(connection)
         meta2 = MetaData()
         a2 = Table(
             "addresses",
             meta2,
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
-        u2 = Table("users", meta2, autoload_with=testing.db)
+        u2 = Table("users", meta2, autoload_with=connection)
         assert len(a2.c.user_id.foreign_keys) == 1
         assert len(a2.foreign_keys) == 1
         assert [c.parent for c in a2.foreign_keys] == [a2.c.user_id]
@@ -750,13 +725,13 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert u2.join(a2).onclause.compare(u2.c.id == a2.c.user_id)
         meta3 = MetaData()
 
-        u3 = Table("users", meta3, autoload_with=testing.db)
+        u3 = Table("users", meta3, autoload_with=connection)
 
         a3 = Table(
             "addresses",
             meta3,
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         assert u3.join(a3).onclause.compare(u3.c.id == a3.c.user_id)
 
@@ -766,7 +741,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             "users",
             meta4,
             Column("id", sa.Integer, key="u_id", primary_key=True),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
 
         a4 = Table(
@@ -777,7 +752,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column(
                 "user_id", sa.Integer, sa.ForeignKey("users.u_id"), key="id"
             ),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
 
         # for the thing happening here with the column collection,
@@ -789,12 +764,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert len(a4.columns) == 3
         assert len(a4.constraints) == 2
 
-    @testing.provide_metadata
-    def test_override_composite_fk(self):
+    def test_override_composite_fk(self, connection, metadata):
         """Test double-remove of composite foreign key, when replaced."""
 
-        metadata = self.metadata
-
         Table(
             "a",
             metadata,
@@ -810,26 +782,25 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             sa.ForeignKeyConstraint(["x", "y"], ["a.x", "a.y"]),
         )
 
-        metadata.create_all()
+        metadata.create_all(connection)
 
         meta2 = MetaData()
 
         c1 = Column("x", sa.Integer, primary_key=True)
         c2 = Column("y", sa.Integer, primary_key=True)
         f1 = sa.ForeignKeyConstraint(["x", "y"], ["a.x", "a.y"])
-        b1 = Table("b", meta2, c1, c2, f1, autoload_with=testing.db)
+        b1 = Table("b", meta2, c1, c2, f1, autoload_with=connection)
 
         assert b1.c.x is c1
         assert b1.c.y is c2
         assert f1 in b1.constraints
         assert len(b1.constraints) == 2
 
-    @testing.provide_metadata
-    def test_override_keys(self):
+    def test_override_keys(self, connection, metadata):
         """test that columns can be overridden with a 'key',
         and that ForeignKey targeting during reflection still works."""
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "a",
             meta,
@@ -843,27 +814,26 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("y", sa.Integer, sa.ForeignKey("a.x")),
             test_needs_fk=True,
         )
-        meta.create_all(testing.db)
+        meta.create_all(connection)
         m2 = MetaData()
         a2 = Table(
             "a",
             m2,
             Column("x", sa.Integer, primary_key=True, key="x1"),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
-        b2 = Table("b", m2, autoload_with=testing.db)
+        b2 = Table("b", m2, autoload_with=connection)
         assert a2.join(b2).onclause.compare(a2.c.x1 == b2.c.y)
         assert b2.c.y.references(a2.c.x1)
 
-    @testing.provide_metadata
-    def test_nonreflected_fk_raises(self):
+    def test_nonreflected_fk_raises(self, connection, metadata):
         """test that a NoReferencedColumnError is raised when reflecting
         a table with an FK to another table which has not included the target
         column in its reflection.
 
         """
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "a",
             meta,
@@ -877,21 +847,19 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("y", sa.Integer, sa.ForeignKey("a.x")),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
         m2 = MetaData()
-        a2 = Table("a", m2, include_columns=["z"], autoload_with=testing.db)
-        b2 = Table("b", m2, autoload_with=testing.db)
+        a2 = Table("a", m2, include_columns=["z"], autoload_with=connection)
+        b2 = Table("b", m2, autoload_with=connection)
 
         assert_raises(sa.exc.NoReferencedColumnError, a2.join, b2)
 
-    @testing.exclude("mysql", "<", (4, 1, 1), "innodb funkiness")
-    @testing.provide_metadata
-    def test_override_existing_fk(self):
+    def test_override_existing_fk(self, connection, metadata):
         """test that you can override columns and specify new foreign
         keys to other reflected tables, on columns which *do* already
         have that foreign key, and that the FK is not duped."""
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "users",
             meta,
@@ -907,15 +875,15 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             test_needs_fk=True,
         )
 
-        meta.create_all(testing.db)
+        meta.create_all(connection)
         meta2 = MetaData()
         a2 = Table(
             "addresses",
             meta2,
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
-        u2 = Table("users", meta2, autoload_with=testing.db)
+        u2 = Table("users", meta2, autoload_with=connection)
         s = sa.select(a2).subquery()
 
         assert s.c.user_id is not None
@@ -932,14 +900,14 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             "users",
             meta2,
             Column("id", sa.Integer, primary_key=True),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         a2 = Table(
             "addresses",
             meta2,
             Column("id", sa.Integer, primary_key=True),
             Column("user_id", sa.Integer, sa.ForeignKey("users.id")),
-            autoload_with=testing.db,
+            autoload_with=connection,
         )
         s = sa.select(a2).subquery()
 
@@ -953,8 +921,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert u2.join(a2).onclause.compare(u2.c.id == a2.c.user_id)
 
     @testing.only_on(["postgresql", "mysql"])
-    @testing.provide_metadata
-    def test_fk_options(self):
+    def test_fk_options(self, connection, metadata):
         """test that foreign key reflection includes options (on
         backends with {dialect}.get_foreign_keys() support)"""
 
@@ -989,7 +956,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             )
             test_attrs = ("onupdate", "ondelete")
 
-        meta = self.metadata
+        meta = metadata
         Table(
             "users",
             meta,
@@ -1004,40 +971,38 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             Column("user_id", sa.Integer, addresses_user_id_fkey),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
 
         meta2 = MetaData()
-        meta2.reflect(testing.db)
+        meta2.reflect(connection)
         for fk in meta2.tables["addresses"].foreign_keys:
             ref = addresses_user_id_fkey
             for attr in test_attrs:
                 eq_(getattr(fk, attr), getattr(ref, attr))
 
-    @testing.provide_metadata
-    def test_pks_not_uniques(self):
+    def test_pks_not_uniques(self, connection, metadata):
         """test that primary key reflection not tripped up by unique
         indexes"""
 
-        with testing.db.begin() as conn:
-            conn.exec_driver_sql(
-                """
-                CREATE TABLE book (
-                    id INTEGER NOT NULL,
-                    title VARCHAR(100) NOT NULL,
-                    series INTEGER,
-                    series_id INTEGER,
-                    UNIQUE(series, series_id),
-                    PRIMARY KEY(id)
-                )"""
-            )
+        conn = connection
+        conn.exec_driver_sql(
+            """
+            CREATE TABLE book (
+                id INTEGER NOT NULL,
+                title VARCHAR(100) NOT NULL,
+                series INTEGER,
+                series_id INTEGER,
+                UNIQUE(series, series_id),
+                PRIMARY KEY(id)
+            )"""
+        )
 
-        book = Table("book", self.metadata, autoload_with=testing.db)
+        book = Table("book", metadata, autoload_with=connection)
         assert book.primary_key.contains_column(book.c.id)
         assert not book.primary_key.contains_column(book.c.series)
         eq_(len(book.primary_key), 1)
 
-    def test_fk_error(self):
-        metadata = MetaData(testing.db)
+    def test_fk_error(self, connection, metadata):
         Table(
             "slots",
             metadata,
@@ -1052,37 +1017,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             "could not find table 'pkgs' with which to generate "
             "a foreign key to target column 'pkg_id'",
             metadata.create_all,
+            connection,
         )
 
-    @testing.provide_metadata
-    def test_composite_pks(self):
+    def test_composite_pks(self, connection, metadata):
         """test reflection of a composite primary key"""
 
-        with testing.db.begin() as conn:
-            conn.exec_driver_sql(
-                """
-                CREATE TABLE book (
-                    id INTEGER NOT NULL,
-                    isbn VARCHAR(50) NOT NULL,
-                    title VARCHAR(100) NOT NULL,
-                    series INTEGER NOT NULL,
-                    series_id INTEGER NOT NULL,
-                    UNIQUE(series, series_id),
-                    PRIMARY KEY(id, isbn)
-                )"""
-            )
-        book = Table("book", self.metadata, autoload_with=testing.db)
+        conn = connection
+        conn.exec_driver_sql(
+            """
+            CREATE TABLE book (
+                id INTEGER NOT NULL,
+                isbn VARCHAR(50) NOT NULL,
+                title VARCHAR(100) NOT NULL,
+                series INTEGER NOT NULL,
+                series_id INTEGER NOT NULL,
+                UNIQUE(series, series_id),
+                PRIMARY KEY(id, isbn)
+            )"""
+        )
+        book = Table("book", metadata, autoload_with=connection)
         assert book.primary_key.contains_column(book.c.id)
         assert book.primary_key.contains_column(book.c.isbn)
         assert not book.primary_key.contains_column(book.c.series)
         eq_(len(book.primary_key), 2)
 
-    @testing.exclude("mysql", "<", (4, 1, 1), "innodb funkiness")
-    @testing.provide_metadata
-    def test_composite_fk(self):
+    def test_composite_fk(self, connection, metadata):
         """test reflection of composite foreign keys"""
 
-        meta = self.metadata
+        meta = metadata
         multi = Table(
             "multi",
             meta,
@@ -1107,11 +1070,11 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             ),
             test_needs_fk=True,
         )
-        meta.create_all()
+        meta.create_all(connection)
 
         meta2 = MetaData()
-        table = Table("multi", meta2, autoload_with=testing.db)
-        table2 = Table("multi2", meta2, autoload_with=testing.db)
+        table = Table("multi", meta2, autoload_with=connection)
+        table2 = Table("multi2", meta2, autoload_with=connection)
         self.assert_tables_equal(multi, table)
         self.assert_tables_equal(multi2, table2)
         j = sa.join(table, table2)
@@ -1126,13 +1089,12 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
 
     @testing.crashes("oracle", "FIXME: unknown, confirm not fails_on")
     @testing.requires.check_constraints
-    @testing.provide_metadata
-    def test_reserved(self):
+    def test_reserved(self, connection, metadata):
 
         # check a table that uses a SQL reserved name doesn't cause an
         # error
 
-        meta = self.metadata
+        meta = metadata
         table_a = Table(
             "select",
             meta,
@@ -1142,11 +1104,11 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         )
         sa.Index("where", table_a.c["from"])
 
-        if meta.bind.dialect.requires_name_normalize:
+        if connection.dialect.requires_name_normalize:
             check_col = "TRUE"
         else:
             check_col = "true"
-        quoter = meta.bind.dialect.identifier_preparer.quote_identifier
+        quoter = connection.dialect.identifier_preparer.quote_identifier
 
         Table(
             "false",
@@ -1164,120 +1126,81 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
             sa.PrimaryKeyConstraint("or", "join", name="to"),
         )
         index_c = sa.Index("else", table_c.c.join)
-        meta.create_all()
-        index_c.drop()
+        meta.create_all(connection)
+        index_c.drop(connection)
         meta2 = MetaData()
-        Table("select", meta2, autoload_with=testing.db)
-        Table("false", meta2, autoload_with=testing.db)
-        Table("is", meta2, autoload_with=testing.db)
+        Table("select", meta2, autoload_with=connection)
+        Table("false", meta2, autoload_with=connection)
+        Table("is", meta2, autoload_with=connection)
 
-    @testing.provide_metadata
-    def _test_reflect_uses_bind(self, fn):
-        from sqlalchemy.pool import AssertionPool
-
-        e = engines.testing_engine(options={"poolclass": AssertionPool})
-        fn(e)
-
-    def test_reflect_uses_bind_constructor_conn_reflect(self):
-        self._test_reflect_uses_bind(lambda e: MetaData(e.connect()).reflect())
-
-    def test_reflect_uses_bind_constructor_engine_reflect(self):
-        self._test_reflect_uses_bind(lambda e: MetaData(e).reflect())
-
-    def test_reflect_uses_bind_conn_reflect(self):
-        self._test_reflect_uses_bind(lambda e: MetaData().reflect(e.connect()))
-
-    def test_reflect_uses_bind_engine_reflect(self):
-        self._test_reflect_uses_bind(lambda e: MetaData().reflect(e))
-
-    def test_reflect_uses_bind_option_engine_reflect(self):
-        self._test_reflect_uses_bind(
-            lambda e: MetaData().reflect(e.execution_options(foo="bar"))
-        )
-
-    @testing.provide_metadata
-    def test_reflect_all(self):
-        existing = inspect(testing.db).get_table_names()
+    def test_reflect_all(self, connection, metadata):
         names = ["rt_%s" % name for name in ("a", "b", "c", "d", "e")]
         nameset = set(names)
-        for name in names:
-            # be sure our starting environment is sane
-            self.assert_(name not in existing)
-        self.assert_("rt_f" not in existing)
 
-        baseline = self.metadata
+        baseline = metadata
         for name in names:
             Table(name, baseline, Column("id", sa.Integer, primary_key=True))
-        baseline.create_all()
+        baseline.create_all(connection)
 
-        m1 = MetaData(testing.db)
-        self.assert_(not m1.tables)
-        m1.reflect()
-        self.assert_(nameset.issubset(set(m1.tables.keys())))
+        m1 = MetaData()
+        is_false(m1.tables)
+        m1.reflect(connection)
+        is_true(nameset.issubset(set(m1.tables.keys())))
 
         m2 = MetaData()
-        m2.reflect(testing.db, only=["rt_a", "rt_b"])
-        self.assert_(set(m2.tables.keys()) == set(["rt_a", "rt_b"]))
+        m2.reflect(connection, only=["rt_a", "rt_b"])
+        eq_(set(m2.tables.keys()), set(["rt_a", "rt_b"]))
 
         m3 = MetaData()
-        c = testing.db.connect()
-        m3.reflect(bind=c, only=lambda name, meta: name == "rt_c")
-        self.assert_(set(m3.tables.keys()) == set(["rt_c"]))
+        m3.reflect(connection, only=lambda name, meta: name == "rt_c")
+        eq_(set(m3.tables.keys()), set(["rt_c"]))
 
-        m4 = MetaData(testing.db)
+        m4 = MetaData()
 
         assert_raises_message(
             sa.exc.InvalidRequestError,
             r"Could not reflect: requested table\(s\) not available in "
             r"Engine\(.*?\): \(rt_f\)",
             m4.reflect,
+            connection,
             only=["rt_a", "rt_f"],
         )
 
-        m5 = MetaData(testing.db)
-        m5.reflect(only=[])
-        self.assert_(not m5.tables)
+        m5 = MetaData()
+        m5.reflect(connection, only=[])
+        is_false(m5.tables)
 
-        m6 = MetaData(testing.db)
-        m6.reflect(only=lambda n, m: False)
-        self.assert_(not m6.tables)
+        m6 = MetaData()
+        m6.reflect(connection, only=lambda n, m: False)
+        is_false(m6.tables)
 
-        m7 = MetaData(testing.db)
-        m7.reflect()
-        self.assert_(nameset.issubset(set(m7.tables.keys())))
+        m7 = MetaData()
+        m7.reflect(connection)
+        is_true(nameset.issubset(set(m7.tables.keys())))
 
-        m8 = MetaData()
-        assert_raises(sa.exc.UnboundExecutionError, m8.reflect)
-
-        m8_e1 = MetaData(testing.db)
+        m8_e1 = MetaData()
         rt_c = Table("rt_c", m8_e1)
-        m8_e1.reflect(extend_existing=True)
+        m8_e1.reflect(connection, extend_existing=True)
         eq_(set(m8_e1.tables.keys()), set(names))
         eq_(rt_c.c.keys(), ["id"])
 
-        m8_e2 = MetaData(testing.db)
+        m8_e2 = MetaData()
         rt_c = Table("rt_c", m8_e2)
-        m8_e2.reflect(extend_existing=True, only=["rt_a", "rt_c"])
+        m8_e2.reflect(connection, extend_existing=True, only=["rt_a", "rt_c"])
         eq_(set(m8_e2.tables.keys()), set(["rt_a", "rt_c"]))
         eq_(rt_c.c.keys(), ["id"])
 
-        if existing:
-            print("Other tables present in database, skipping some checks.")
-        else:
-            baseline.drop_all()
-            m9 = MetaData(testing.db)
-            m9.reflect()
-            self.assert_(not m9.tables)
+        baseline.drop_all(connection)
+        m9 = MetaData()
+        m9.reflect(connection)
+        is_false(m9.tables)
 
-    @testing.provide_metadata
-    def test_reflect_all_unreflectable_table(self):
+    def test_reflect_all_unreflectable_table(self, connection, metadata):
         names = ["rt_%s" % name for name in ("a", "b", "c", "d", "e")]
 
         for name in names:
-            Table(
-                name, self.metadata, Column("id", sa.Integer, primary_key=True)
-            )
-        self.metadata.create_all()
+            Table(name, metadata, Column("id", sa.Integer, primary_key=True))
+        metadata.create_all(connection)
 
         m = MetaData()
 
@@ -1292,7 +1215,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
 
         with mock.patch.object(inspector, "reflect_table", patched):
             with expect_warnings("Skipping table rt_c: Can't reflect rt_c"):
-                m.reflect(bind=testing.db)
+                m.reflect(connection)
 
             assert_raises_message(
                 sa.exc.UnreflectableTableError,
@@ -1300,23 +1223,11 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
                 Table,
                 "rt_c",
                 m,
-                autoload_with=testing.db,
+                autoload_with=connection,
             )
 
-    def test_reflect_all_conn_closing(self):
-        m1 = MetaData()
-        c = testing.db.connect()
-        m1.reflect(bind=c)
-        assert not c.closed
-
-    def test_inspector_conn_closing(self):
-        c = testing.db.connect()
-        inspect(c)
-        assert not c.closed
-
-    @testing.provide_metadata
-    def test_index_reflection(self):
-        m1 = self.metadata
+    def test_index_reflection(self, connection, metadata):
+        m1 = metadata
         t1 = Table(
             "party",
             m1,
@@ -1325,9 +1236,9 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         )
         sa.Index("idx1", t1.c.id, unique=True)
         sa.Index("idx2", t1.c.name, t1.c.id, unique=False)
-        m1.create_all()
+        m1.create_all(connection)
         m2 = MetaData()
-        t2 = Table("party", m2, autoload_with=testing.db)
+        t2 = Table("party", m2, autoload_with=connection)
 
         assert len(t2.indexes) == 3
         # Make sure indexes are in the order we expect them in
@@ -1345,18 +1256,17 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         assert set([t2.c.name]) == set(r3.columns)
 
     @testing.requires.comment_reflection
-    @testing.provide_metadata
-    def test_comment_reflection(self):
-        m1 = self.metadata
+    def test_comment_reflection(self, connection, metadata):
+        m1 = metadata
         Table(
             "sometable",
             m1,
             Column("id", sa.Integer, comment="c1 comment"),
             comment="t1 comment",
         )
-        m1.create_all()
+        m1.create_all(connection)
         m2 = MetaData()
-        t2 = Table("sometable", m2, autoload_with=testing.db)
+        t2 = Table("sometable", m2, autoload_with=connection)
 
         eq_(t2.comment, "t1 comment")
         eq_(t2.c.id.comment, "c1 comment")
@@ -1366,18 +1276,17 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         eq_(t3.c.id.comment, "c1 comment")
 
     @testing.requires.check_constraint_reflection
-    @testing.provide_metadata
-    def test_check_constraint_reflection(self):
-        m1 = self.metadata
+    def test_check_constraint_reflection(self, connection, metadata):
+        m1 = metadata
         Table(
             "x",
             m1,
             Column("q", Integer),
             sa.CheckConstraint("q > 10", name="ck1"),
         )
-        m1.create_all()
+        m1.create_all(connection)
         m2 = MetaData()
-        t2 = Table("x", m2, autoload_with=testing.db)
+        t2 = Table("x", m2, autoload_with=connection)
 
         ck = [
             const
@@ -1388,40 +1297,35 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
         eq_regex(ck.sqltext.text, r"[\(`]*q[\)`]* > 10")
         eq_(ck.name, "ck1")
 
-    @testing.provide_metadata
-    def test_index_reflection_cols_busted(self):
-        t = Table(
-            "x", self.metadata, Column("a", Integer), Column("b", Integer)
-        )
+    def test_index_reflection_cols_busted(self, connection, metadata):
+        t = Table("x", metadata, Column("a", Integer), Column("b", Integer))
         sa.Index("x_ix", t.c.a, t.c.b)
-        self.metadata.create_all()
+        metadata.create_all(connection)
 
         def mock_get_columns(self, connection, table_name, **kw):
             return [{"name": "b", "type": Integer, "primary_key": False}]
 
         with testing.mock.patch.object(
-            testing.db.dialect, "get_columns", mock_get_columns
+            connection.dialect, "get_columns", mock_get_columns
         ):
             m = MetaData()
             with testing.expect_warnings(
                 "index key 'a' was not located in columns"
             ):
-                t = Table("x", m, autoload_with=testing.db)
+                t = Table("x", m, autoload_with=connection)
 
         eq_(list(t.indexes)[0].columns, [t.c.b])
 
     @testing.requires.views
-    @testing.provide_metadata
-    def test_views(self):
-        metadata = self.metadata
+    def test_views(self, connection, metadata):
         users, addresses, dingalings = createTables(metadata)
         try:
-            metadata.create_all()
-            _create_views(metadata.bind, None)
+            metadata.create_all(connection)
+            _create_views(connection, None)
             m2 = MetaData()
-            users_v = Table("users_v", m2, autoload_with=testing.db)
+            users_v = Table("users_v", m2, autoload_with=connection)
             addresses_v = Table(
-                "email_addresses_v", m2, autoload_with=testing.db
+                "email_addresses_v", m2, autoload_with=connection
             )
 
             for c1, c2 in zip(users_v.c, users.c):
@@ -1432,25 +1336,23 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
                 eq_(c1.name, c2.name)
                 self.assert_types_base(c1, c2)
         finally:
-            _drop_views(metadata.bind)
+            _drop_views(connection)
 
     @testing.requires.views
-    @testing.provide_metadata
-    def test_reflect_all_with_views(self):
-        metadata = self.metadata
+    def test_reflect_all_with_views(self, connection, metadata):
         users, addresses, dingalings = createTables(metadata, None)
         try:
-            metadata.create_all()
-            _create_views(metadata.bind, None)
-            m2 = MetaData(testing.db)
+            metadata.create_all(connection)
+            _create_views(connection, None)
+            m2 = MetaData()
 
-            m2.reflect(views=False)
+            m2.reflect(connection, views=False)
             eq_(
                 set(m2.tables), set(["users", "email_addresses", "dingalings"])
             )
 
-            m2 = MetaData(testing.db)
-            m2.reflect(views=True)
+            m2 = MetaData()
+            m2.reflect(connection, views=True)
             eq_(
                 set(m2.tables),
                 set(
@@ -1464,7 +1366,7 @@ class ReflectionTest(fixtures.TestBase, ComparesTables):
                 ),
             )
         finally:
-            _drop_views(metadata.bind)
+            _drop_views(connection)
 
 
 class CreateDropTest(fixtures.TablesTest):
@@ -1473,6 +1375,13 @@ class CreateDropTest(fixtures.TablesTest):
     run_create_tables = None
 
     @classmethod
+    def teardown_class(cls):
+        # TablesTest is used here without
+        # run_create_tables, so add an explicit drop of whatever is in
+        # metadata
+        cls._tables_metadata.drop_all(testing.db)
+
+    @classmethod
     def define_tables(cls, metadata):
         Table(
             "users",
@@ -1525,11 +1434,8 @@ class CreateDropTest(fixtures.TablesTest):
             Column("item_name", sa.VARCHAR(50)),
         )
 
-    def teardown(self):
-        self.metadata.drop_all(testing.db)
-
     def test_sorter(self):
-        tables = self.metadata.sorted_tables
+        tables = self.tables_test_metadata.sorted_tables
         table_names = [t.name for t in tables]
         ua = [n for n in table_names if n in ("users", "email_addresses")]
         oi = [n for n in table_names if n in ("orders", "items")]
@@ -1537,39 +1443,41 @@ class CreateDropTest(fixtures.TablesTest):
         eq_(ua, ["users", "email_addresses"])
         eq_(oi, ["orders", "items"])
 
-    def test_checkfirst(self):
-        insp = inspect(testing.db)
+    def test_checkfirst(self, connection):
+        insp = inspect(connection)
         users = self.tables.users
 
         is_false(insp.has_table("users"))
-        users.create(bind=testing.db)
+        users.create(connection)
         is_true(insp.has_table("users"))
-        users.create(bind=testing.db, checkfirst=True)
-        users.drop(bind=testing.db)
-        users.drop(bind=testing.db, checkfirst=True)
+        users.create(connection, checkfirst=True)
+        users.drop(connection)
+        users.drop(connection, checkfirst=True)
         is_false(insp.has_table("users"))
-        users.create(bind=testing.db, checkfirst=True)
-        users.drop(bind=testing.db)
+        users.create(connection, checkfirst=True)
+        users.drop(connection)
 
-    def test_createdrop(self):
-        insp = inspect(testing.db)
-        metadata = self.metadata
-        metadata.create_all(bind=testing.db)
+    def test_createdrop(self, connection):
+        insp = inspect(connection)
+
+        metadata = self.tables_test_metadata
+
+        metadata.create_all(connection)
         is_true(insp.has_table("items"))
         is_true(insp.has_table("email_addresses"))
-        metadata.create_all(bind=testing.db)
+        metadata.create_all(connection)
         is_true(insp.has_table("items"))
 
-        metadata.drop_all(bind=testing.db)
+        metadata.drop_all(connection)
         is_false(insp.has_table("items"))
         is_false(insp.has_table("email_addresses"))
-        metadata.drop_all(bind=testing.db)
+        metadata.drop_all(connection)
         is_false(insp.has_table("items"))
 
-    def test_tablenames(self):
-        metadata = self.metadata
-        metadata.create_all(bind=testing.db)
-        insp = inspect(testing.db)
+    def test_tablenames(self, connection):
+        metadata = self.tables_test_metadata
+        metadata.create_all(bind=connection)
+        insp = inspect(connection)
 
         # ensure all tables we created are in the list.
         is_true(set(insp.get_table_names()).issuperset(metadata.tables))
@@ -1597,12 +1505,11 @@ class SchemaManipulationTest(fixtures.TestBase):
         assert addresses.constraints == set([addresses.primary_key, fk])
 
 
-class UnicodeReflectionTest(fixtures.TestBase):
+class UnicodeReflectionTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        cls.metadata = metadata = MetaData()
+    def define_tables(cls, metadata):
 
         no_multibyte_period = set([("plain", "col_plain", "ix_plain")])
         no_has_table = [
@@ -1671,32 +1578,24 @@ class UnicodeReflectionTest(fixtures.TestBase):
             )
             schema.Index(ixname, t.c[cname])
 
-        metadata.create_all(testing.db)
         cls.names = names
 
-    @classmethod
-    def teardown_class(cls):
-        cls.metadata.drop_all(testing.db, checkfirst=False)
-
     @testing.requires.unicode_connections
-    def test_has_table(self):
-        insp = inspect(testing.db)
+    def test_has_table(self, connection):
+        insp = inspect(connection)
         for tname, cname, ixname in self.names:
             assert insp.has_table(tname), "Can't detect name %s" % tname
 
     @testing.requires.unicode_connections
-    def test_basic(self):
+    def test_basic(self, connection):
         # the 'convert_unicode' should not get in the way of the
         # reflection process.  reflect_table for oracle, postgresql
         # (others?) expect non-unicode strings in result sets/bind
         # params
 
-        bind = testing.db
         names = set([rec[0] for rec in self.names])
 
-        reflected = set(inspect(bind).get_table_names())
-
-        # Jython 2.5 on Java 5 lacks unicodedata.normalize
+        reflected = set(inspect(connection).get_table_names())
 
         if not names.issubset(reflected) and hasattr(unicodedata, "normalize"):
 
@@ -1711,14 +1610,14 @@ class UnicodeReflectionTest(fixtures.TestBase):
             # Yep.  But still ensure that bulk reflection and
             # create/drop work with either normalization.
 
-        r = MetaData(bind)
-        r.reflect()
-        r.drop_all(checkfirst=False)
-        r.create_all(checkfirst=False)
+        r = MetaData()
+        r.reflect(connection)
+        r.drop_all(connection, checkfirst=False)
+        r.create_all(connection, checkfirst=False)
 
     @testing.requires.unicode_connections
-    def test_get_names(self):
-        inspector = inspect(testing.db)
+    def test_get_names(self, connection):
+        inspector = inspect(connection)
         names = dict(
             (tname, (cname, ixname)) for tname, cname, ixname in self.names
         )
@@ -1760,8 +1659,7 @@ class SchemaTest(fixtures.TestBase):
     @testing.requires.cross_schema_fk_reflection
     @testing.requires.implicit_default_schema
     @testing.provide_metadata
-    def test_blank_schema_arg(self):
-        metadata = self.metadata
+    def test_blank_schema_arg(self, connection, metadata):
 
         Table(
             "some_table",
@@ -1778,37 +1676,27 @@ class SchemaTest(fixtures.TestBase):
             schema=None,
             test_needs_fk=True,
         )
-        metadata.create_all()
-        with testing.db.connect() as conn:
-            meta2 = MetaData(conn, schema=testing.config.test_schema)
-            meta2.reflect()
+        metadata.create_all(connection)
+        meta2 = MetaData(schema=testing.config.test_schema)
+        meta2.reflect(connection)
 
-            eq_(
-                set(meta2.tables),
-                set(
-                    [
-                        "some_other_table",
-                        "%s.some_table" % testing.config.test_schema,
-                    ]
-                ),
-            )
+        eq_(
+            set(meta2.tables),
+            set(
+                [
+                    "some_other_table",
+                    "%s.some_table" % testing.config.test_schema,
+                ]
+            ),
+        )
 
     @testing.requires.schemas
-    def test_explicit_default_schema(self):
-        engine = testing.db
-        engine.connect().close()
-
-        if testing.against("sqlite"):
-            # Works for CREATE TABLE main.foo, SELECT FROM main.foo, etc.,
-            # but fails on:
-            #   FOREIGN KEY(col2) REFERENCES main.table1 (col1)
-            schema = "main"
-        else:
-            schema = engine.dialect.default_schema_name
+    def test_explicit_default_schema(self, connection, metadata):
+
+        schema = connection.dialect.default_schema_name
 
         assert bool(schema)
 
-        metadata = MetaData()
         Table(
             "table1",
             metadata,
@@ -1826,54 +1714,41 @@ class SchemaTest(fixtures.TestBase):
             test_needs_fk=True,
             schema=schema,
         )
-        try:
-            metadata.create_all(engine)
-            metadata.create_all(engine, checkfirst=True)
-            assert len(metadata.tables) == 2
-            metadata.clear()
-
-            Table("table1", metadata, autoload_with=engine, schema=schema)
-            Table("table2", metadata, autoload_with=engine, schema=schema)
-            assert len(metadata.tables) == 2
-        finally:
-            metadata.drop_all(engine)
+        metadata.create_all(connection)
+        metadata.create_all(connection, checkfirst=True)
+        eq_(len(metadata.tables), 2)
+
+        m1 = MetaData()
+        Table("table1", m1, autoload_with=connection, schema=schema)
+        Table("table2", m1, autoload_with=connection, schema=schema)
+        eq_(len(m1.tables), 2)
 
     @testing.requires.schemas
-    @testing.provide_metadata
-    def test_schema_translation(self):
+    def test_schema_translation(self, connection, metadata):
         Table(
             "foob",
-            self.metadata,
+            metadata,
             Column("q", Integer),
             schema=config.test_schema,
         )
-        self.metadata.create_all()
+        metadata.create_all(connection)
 
         m = MetaData()
         map_ = {"foob": config.test_schema}
-        with config.db.connect().execution_options(
-            schema_translate_map=map_
-        ) as conn:
-            t = Table("foob", m, schema="foob", autoload_with=conn)
-            eq_(t.schema, "foob")
-            eq_(t.c.keys(), ["q"])
+
+        c2 = connection.execution_options(schema_translate_map=map_)
+        t = Table("foob", m, schema="foob", autoload_with=c2)
+        eq_(t.schema, "foob")
+        eq_(t.c.keys(), ["q"])
 
     @testing.requires.schemas
     @testing.fails_on("sybase", "FIXME: unknown")
-    def test_explicit_default_schema_metadata(self):
-        engine = testing.db
-
-        if testing.against("sqlite"):
-            # Works for CREATE TABLE main.foo, SELECT FROM main.foo, etc.,
-            # but fails on:
-            #   FOREIGN KEY(col2) REFERENCES main.table1 (col1)
-            schema = "main"
-        else:
-            schema = engine.dialect.default_schema_name
+    def test_explicit_default_schema_metadata(self, connection, metadata):
+        schema = connection.dialect.default_schema_name
 
-        assert bool(schema)
+        is_true(schema)
 
-        metadata = MetaData(schema=schema)
+        metadata.schema = schema
         Table(
             "table1",
             metadata,
@@ -1887,26 +1762,21 @@ class SchemaTest(fixtures.TestBase):
             Column("col2", sa.Integer, sa.ForeignKey("table1.col1")),
             test_needs_fk=True,
         )
-        try:
-            metadata.create_all(engine)
-            metadata.create_all(engine, checkfirst=True)
-            assert len(metadata.tables) == 2
-            metadata.clear()
-
-            Table("table1", metadata, autoload_with=engine)
-            Table("table2", metadata, autoload_with=engine)
-            assert len(metadata.tables) == 2
-        finally:
-            metadata.drop_all(engine)
+        metadata.create_all(connection)
+        metadata.create_all(connection, checkfirst=True)
+
+        m1 = MetaData(schema=schema)
+
+        Table("table1", m1, autoload_with=connection)
+        Table("table2", m1, autoload_with=connection)
+        eq_(len(m1.tables), 2)
 
     @testing.requires.schemas
-    @testing.provide_metadata
-    def test_metadata_reflect_schema(self):
-        metadata = self.metadata
+    def test_metadata_reflect_schema(self, connection, metadata):
         createTables(metadata, testing.config.test_schema)
-        metadata.create_all()
-        m2 = MetaData(schema=testing.config.test_schema, bind=testing.db)
-        m2.reflect()
+        metadata.create_all(connection)
+        m2 = MetaData(schema=testing.config.test_schema)
+        m2.reflect(connection)
         eq_(
             set(m2.tables),
             set(
@@ -1921,24 +1791,23 @@ class SchemaTest(fixtures.TestBase):
     @testing.requires.schemas
     @testing.requires.cross_schema_fk_reflection
     @testing.requires.implicit_default_schema
-    @testing.provide_metadata
-    def test_reflect_all_schemas_default_overlap(self):
-        Table("t", self.metadata, Column("id", Integer, primary_key=True))
+    def test_reflect_all_schemas_default_overlap(self, connection, metadata):
+        Table("t", metadata, Column("id", Integer, primary_key=True))
 
         Table(
             "t",
-            self.metadata,
+            metadata,
             Column("id1", sa.ForeignKey("t.id")),
             schema=testing.config.test_schema,
         )
 
-        self.metadata.create_all()
+        metadata.create_all(connection)
         m2 = MetaData()
-        m2.reflect(testing.db, schema=testing.config.test_schema)
+        m2.reflect(connection, schema=testing.config.test_schema)
 
         m3 = MetaData()
-        m3.reflect(testing.db)
-        m3.reflect(testing.db, schema=testing.config.test_schema)
+        m3.reflect(connection)
+        m3.reflect(connection, schema=testing.config.test_schema)
 
         eq_(
             set((t.name, t.schema) for t in m2.tables.values()),
@@ -2015,30 +1884,28 @@ def createIndexes(con, schema=None):
 
 
 @testing.requires.views
-def _create_views(con, schema=None):
-    with testing.db.begin() as conn:
-        for table_name in ("users", "email_addresses"):
-            fullname = table_name
-            if schema:
-                fullname = "%s.%s" % (schema, table_name)
-            view_name = fullname + "_v"
-            query = "CREATE VIEW %s AS SELECT * FROM %s" % (
-                view_name,
-                fullname,
-            )
-            conn.execute(sa.sql.text(query))
+def _create_views(conn, schema=None):
+    for table_name in ("users", "email_addresses"):
+        fullname = table_name
+        if schema:
+            fullname = "%s.%s" % (schema, table_name)
+        view_name = fullname + "_v"
+        query = "CREATE VIEW %s AS SELECT * FROM %s" % (
+            view_name,
+            fullname,
+        )
+        conn.execute(sa.sql.text(query))
 
 
 @testing.requires.views
-def _drop_views(con, schema=None):
-    with testing.db.begin() as conn:
-        for table_name in ("email_addresses", "users"):
-            fullname = table_name
-            if schema:
-                fullname = "%s.%s" % (schema, table_name)
-            view_name = fullname + "_v"
-            query = "DROP VIEW %s" % view_name
-            conn.execute(sa.sql.text(query))
+def _drop_views(conn, schema=None):
+    for table_name in ("email_addresses", "users"):
+        fullname = table_name
+        if schema:
+            fullname = "%s.%s" % (schema, table_name)
+        view_name = fullname + "_v"
+        query = "DROP VIEW %s" % view_name
+        conn.execute(sa.sql.text(query))
 
 
 class ReverseCasingReflectTest(fixtures.TestBase, AssertsCompiledSQL):
@@ -2064,9 +1931,9 @@ class ReverseCasingReflectTest(fixtures.TestBase, AssertsCompiledSQL):
             conn.exec_driver_sql("drop table weird_casing")
 
     @testing.requires.denormalized_names
-    def test_direct_quoting(self):
+    def test_direct_quoting(self, connection):
         m = MetaData()
-        t = Table("weird_casing", m, autoload_with=testing.db)
+        t = Table("weird_casing", m, autoload_with=connection)
         self.assert_compile(
             t.select(),
             "SELECT weird_casing.col1, "
@@ -2097,13 +1964,13 @@ class CaseSensitiveTest(fixtures.TablesTest):
         )
 
     @testing.fails_if(testing.requires._has_mysql_on_windows)
-    def test_table_names(self):
-        x = inspect(testing.db).get_table_names()
+    def test_table_names(self, connection):
+        x = inspect(connection).get_table_names()
         assert set(["SomeTable", "SomeOtherTable"]).issubset(x)
 
-    def test_reflect_exact_name(self):
+    def test_reflect_exact_name(self, connection):
         m = MetaData()
-        t1 = Table("SomeTable", m, autoload_with=testing.db)
+        t1 = Table("SomeTable", m, autoload_with=connection)
         eq_(t1.name, "SomeTable")
         assert t1.c.x is not None
 
@@ -2111,47 +1978,43 @@ class CaseSensitiveTest(fixtures.TablesTest):
         lambda: testing.against(("mysql", "<", (5, 5)))
         and not testing.requires._has_mysql_fully_case_sensitive()
     )
-    def test_reflect_via_fk(self):
+    def test_reflect_via_fk(self, connection):
         m = MetaData()
-        t2 = Table("SomeOtherTable", m, autoload_with=testing.db)
+        t2 = Table("SomeOtherTable", m, autoload_with=connection)
         eq_(t2.name, "SomeOtherTable")
         assert "SomeTable" in m.tables
 
     @testing.fails_if(testing.requires._has_mysql_fully_case_sensitive)
     @testing.fails_on_everything_except("sqlite", "mysql", "mssql")
-    def test_reflect_case_insensitive(self):
+    def test_reflect_case_insensitive(self, connection):
         m = MetaData()
-        t2 = Table("sOmEtAbLe", m, autoload_with=testing.db)
+        t2 = Table("sOmEtAbLe", m, autoload_with=connection)
         eq_(t2.name, "sOmEtAbLe")
 
 
-class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
+class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        cls.metadata = MetaData()
-        cls.to_reflect = Table(
+    def define_tables(cls, metadata):
+        to_reflect = Table(
             "to_reflect",
-            cls.metadata,
+            metadata,
             Column("x", sa.Integer, primary_key=True, autoincrement=False),
             Column("y", sa.Integer),
             test_needs_fk=True,
         )
-        cls.related = Table(
+        Table(
             "related",
-            cls.metadata,
+            metadata,
             Column("q", sa.Integer, sa.ForeignKey("to_reflect.x")),
             test_needs_fk=True,
         )
-        sa.Index("some_index", cls.to_reflect.c.y)
-        cls.metadata.create_all(testing.db)
+        sa.Index("some_index", to_reflect.c.y)
 
-    @classmethod
-    def teardown_class(cls):
-        cls.metadata.drop_all(testing.db)
-
-    def _do_test(self, col, update, assert_, tablename="to_reflect"):
+    def _do_test(
+        self, connection, col, update, assert_, tablename="to_reflect"
+    ):
         # load the actual Table class, not the test
         # wrapper
         from sqlalchemy.schema import Table
@@ -2165,31 +2028,31 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
         t = Table(
             tablename,
             m,
-            autoload_with=testing.db,
+            autoload_with=connection,
             listeners=[("column_reflect", column_reflect)],
         )
         assert_(t)
 
         m = MetaData()
         self.event_listen(Table, "column_reflect", column_reflect)
-        t2 = Table(tablename, m, autoload_with=testing.db)
+        t2 = Table(tablename, m, autoload_with=connection)
         assert_(t2)
 
-    def test_override_key(self):
+    def test_override_key(self, connection):
         def assertions(table):
             eq_(table.c.YXZ.name, "x")
             eq_(set(table.primary_key), set([table.c.YXZ]))
 
-        self._do_test("x", {"key": "YXZ"}, assertions)
+        self._do_test(connection, "x", {"key": "YXZ"}, assertions)
 
-    def test_override_index(self):
+    def test_override_index(self, connection):
         def assertions(table):
             idx = list(table.indexes)[0]
             eq_(idx.columns, [table.c.YXZ])
 
-        self._do_test("y", {"key": "YXZ"}, assertions)
+        self._do_test(connection, "y", {"key": "YXZ"}, assertions)
 
-    def test_override_key_fk(self):
+    def test_override_key_fk(self, connection):
         m = MetaData()
 
         def column_reflect(insp, table, column_info):
@@ -2202,48 +2065,51 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
         to_reflect = Table(
             "to_reflect",
             m,
-            autoload_with=testing.db,
+            autoload_with=connection,
             listeners=[("column_reflect", column_reflect)],
         )
         related = Table(
             "related",
             m,
-            autoload_with=testing.db,
+            autoload_with=connection,
             listeners=[("column_reflect", column_reflect)],
         )
 
         assert related.c.qyz.references(to_reflect.c.xyz)
 
-    def test_override_type(self):
+    def test_override_type(self, connection):
         def assert_(table):
             assert isinstance(table.c.x.type, sa.String)
 
-        self._do_test("x", {"type": sa.String}, assert_)
+        self._do_test(connection, "x", {"type": sa.String}, assert_)
 
-    def test_override_info(self):
+    def test_override_info(self, connection):
         self._do_test(
+            connection,
             "x",
             {"info": {"a": "b"}},
             lambda table: eq_(table.c.x.info, {"a": "b"}),
         )
 
-    def test_override_server_default_fetchedvalue(self):
+    def test_override_server_default_fetchedvalue(self, connection):
         my_default = FetchedValue()
         self._do_test(
+            connection,
             "x",
             {"default": my_default},
             lambda table: eq_(table.c.x.server_default, my_default),
         )
 
-    def test_override_server_default_default_clause(self):
+    def test_override_server_default_default_clause(self, connection):
         my_default = DefaultClause("1")
         self._do_test(
+            connection,
             "x",
             {"default": my_default},
             lambda table: eq_(table.c.x.server_default, my_default),
         )
 
-    def test_override_server_default_plain_text(self):
+    def test_override_server_default_plain_text(self, connection):
         my_default = "1"
 
         def assert_text_of_one(table):
@@ -2254,9 +2120,11 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
             )
             eq_(str(table.c.x.server_default.arg), "1")
 
-        self._do_test("x", {"default": my_default}, assert_text_of_one)
+        self._do_test(
+            connection, "x", {"default": my_default}, assert_text_of_one
+        )
 
-    def test_override_server_default_textclause(self):
+    def test_override_server_default_textclause(self, connection):
         my_default = sa.text("1")
 
         def assert_text_of_one(table):
@@ -2267,9 +2135,11 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
             )
             eq_(str(table.c.x.server_default.arg), "1")
 
-        self._do_test("x", {"default": my_default}, assert_text_of_one)
+        self._do_test(
+            connection, "x", {"default": my_default}, assert_text_of_one
+        )
 
-    def test_listen_metadata_obj(self):
+    def test_listen_metadata_obj(self, connection):
         m1 = MetaData()
 
         m2 = MetaData()
@@ -2280,13 +2150,13 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
         def go(insp, table, info):
             canary.append(info["name"])
 
-        Table("related", m1, autoload_with=testing.db)
+        Table("related", m1, autoload_with=connection)
 
-        Table("related", m2, autoload_with=testing.db)
+        Table("related", m2, autoload_with=connection)
 
         eq_(canary, ["q", "x", "y"])
 
-    def test_listen_metadata_cls(self):
+    def test_listen_metadata_cls(self, connection):
         m1 = MetaData()
 
         m2 = MetaData()
@@ -2298,9 +2168,9 @@ class ColumnEventsTest(fixtures.RemovesEvents, fixtures.TestBase):
 
         self.event_listen(MetaData, "column_reflect", go)
 
-        Table("related", m1, autoload_with=testing.db)
+        Table("related", m1, autoload_with=connection)
 
-        Table("related", m2, autoload_with=testing.db)
+        Table("related", m2, autoload_with=connection)
 
         eq_(canary, ["q", "x", "y", "q", "x", "y"])