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

5 files changed, 241 insertions, 227 deletions

diff --git a/test/sql/test_constraints.py b/test/sql/test_constraints.py
index 019409ba3..8c1fa5424 100644
--- a/test/sql/test_constraints.py
+++ b/test/sql/test_constraints.py
@@ -59,7 +59,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
         )
         self.assert_sql_execution(
             testing.db,
-            lambda: metadata.create_all(checkfirst=False),
+            lambda: metadata.create_all(testing.db, checkfirst=False),
             CompiledSQL(
                 "CREATE TABLE employees ("
                 "id INTEGER NOT NULL, "
@@ -292,7 +292,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
         assertions.append(AllOf(*fk_assertions))
 
         with self.sql_execution_asserter() as asserter:
-            metadata.create_all(checkfirst=False)
+            metadata.create_all(testing.db, checkfirst=False)
         asserter.assert_(*assertions)
 
         assertions = [
@@ -302,7 +302,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
         ]
 
         with self.sql_execution_asserter() as asserter:
-            metadata.drop_all(checkfirst=False),
+            metadata.drop_all(testing.db, checkfirst=False),
         asserter.assert_(*assertions)
 
     def _assert_cyclic_constraint_no_alter(
@@ -356,7 +356,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
         assertions = [AllOf(*table_assertions)]
 
         with self.sql_execution_asserter() as asserter:
-            metadata.create_all(checkfirst=False)
+            metadata.create_all(testing.db, checkfirst=False)
         asserter.assert_(*assertions)
 
         assertions = [
@@ -366,15 +366,15 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
         if sqlite_warning:
             with expect_warnings("Can't sort tables for DROP; "):
                 with self.sql_execution_asserter() as asserter:
-                    metadata.drop_all(checkfirst=False),
+                    metadata.drop_all(testing.db, checkfirst=False),
         else:
             with self.sql_execution_asserter() as asserter:
-                metadata.drop_all(checkfirst=False),
+                metadata.drop_all(testing.db, checkfirst=False),
         asserter.assert_(*assertions)
 
     @testing.force_drop_names("a", "b")
     def test_cycle_unnamed_fks(self):
-        metadata = MetaData(testing.db)
+        metadata = MetaData()
 
         Table(
             "a",
@@ -417,7 +417,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         ]
         with self.sql_execution_asserter() as asserter:
-            metadata.create_all(checkfirst=False)
+            metadata.create_all(testing.db, checkfirst=False)
 
         if testing.db.dialect.supports_alter:
             asserter.assert_(*assertions)
@@ -431,6 +431,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
                 "cycle have names so that they can be dropped using "
                 "DROP CONSTRAINT.",
                 metadata.drop_all,
+                testing.db,
                 checkfirst=False,
             )
         else:
@@ -439,7 +440,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
                 "foreign key dependency exists between tables"
             ):
                 with self.sql_execution_asserter() as asserter:
-                    metadata.drop_all(checkfirst=False)
+                    metadata.drop_all(testing.db, checkfirst=False)
 
             asserter.assert_(
                 AllOf(CompiledSQL("DROP TABLE b"), CompiledSQL("DROP TABLE a"))
@@ -447,7 +448,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
 
     @testing.force_drop_names("a", "b")
     def test_cycle_named_fks(self):
-        metadata = MetaData(testing.db)
+        metadata = MetaData()
 
         Table(
             "a",
@@ -491,13 +492,13 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
             ),
         ]
         with self.sql_execution_asserter() as asserter:
-            metadata.create_all(checkfirst=False)
+            metadata.create_all(testing.db, checkfirst=False)
 
         if testing.db.dialect.supports_alter:
             asserter.assert_(*assertions)
 
             with self.sql_execution_asserter() as asserter:
-                metadata.drop_all(checkfirst=False)
+                metadata.drop_all(testing.db, checkfirst=False)
 
             asserter.assert_(
                 CompiledSQL("ALTER TABLE b DROP CONSTRAINT aidfk"),
@@ -507,7 +508,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
             )
         else:
             with self.sql_execution_asserter() as asserter:
-                metadata.drop_all(checkfirst=False)
+                metadata.drop_all(testing.db, checkfirst=False)
 
             asserter.assert_(
                 AllOf(CompiledSQL("DROP TABLE b"), CompiledSQL("DROP TABLE a"))
@@ -536,7 +537,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
 
         self.assert_sql_execution(
             testing.db,
-            lambda: metadata.create_all(checkfirst=False),
+            lambda: metadata.create_all(testing.db, checkfirst=False),
             AllOf(
                 CompiledSQL(
                     "CREATE TABLE foo ("
@@ -579,7 +580,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
 
         self.assert_sql_execution(
             testing.db,
-            lambda: metadata.create_all(checkfirst=False),
+            lambda: metadata.create_all(testing.db, checkfirst=False),
             AllOf(
                 CompiledSQL(
                     "CREATE TABLE foo ("
@@ -628,7 +629,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
 
         self.assert_sql_execution(
             testing.db,
-            lambda: metadata.create_all(checkfirst=False),
+            lambda: metadata.create_all(testing.db, checkfirst=False),
             RegexSQL("^CREATE TABLE"),
             AllOf(
                 CompiledSQL(
@@ -665,7 +666,7 @@ class ConstraintGenTest(fixtures.TestBase, AssertsExecutionResults):
 
         self.assert_sql_execution(
             testing.db,
-            lambda: metadata.create_all(checkfirst=False),
+            lambda: metadata.create_all(testing.db, checkfirst=False),
             RegexSQL("^CREATE TABLE"),
             AllOf(
                 CompiledSQL(
diff --git a/test/sql/test_query.py b/test/sql/test_query.py
index 6d26f7975..9af8ef6da 100644
--- a/test/sql/test_query.py
+++ b/test/sql/test_query.py
@@ -27,7 +27,6 @@ from sqlalchemy import union_all
 from sqlalchemy import VARCHAR
 from sqlalchemy.engine import default
 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_
@@ -35,22 +34,13 @@ from sqlalchemy.testing.schema import Column
 from sqlalchemy.testing.schema import Table
 
 
-# ongoing - these are old tests.  those which are of general use
-# to test a dialect are being slowly migrated to
-# sqlalhcemy.testing.suite
-
-users = users2 = addresses = metadata = None
-
-
-class QueryTest(fixtures.TestBase):
+class QueryTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        global users, users2, addresses, metadata
-        metadata = MetaData(testing.db)
-        users = Table(
-            "query_users",
+    def define_tables(cls, metadata):
+        Table(
+            "users",
             metadata,
             Column(
                 "user_id", INT, primary_key=True, test_needs_autoincrement=True
@@ -58,8 +48,8 @@ class QueryTest(fixtures.TestBase):
             Column("user_name", VARCHAR(20)),
             test_needs_acid=True,
         )
-        addresses = Table(
-            "query_addresses",
+        Table(
+            "addresses",
             metadata,
             Column(
                 "address_id",
@@ -67,12 +57,12 @@ class QueryTest(fixtures.TestBase):
                 primary_key=True,
                 test_needs_autoincrement=True,
             ),
-            Column("user_id", Integer, ForeignKey("query_users.user_id")),
+            Column("user_id", Integer, ForeignKey("users.user_id")),
             Column("address", String(30)),
             test_needs_acid=True,
         )
 
-        users2 = Table(
+        Table(
             "u2",
             metadata,
             Column("user_id", INT, primary_key=True),
@@ -80,19 +70,6 @@ class QueryTest(fixtures.TestBase):
             test_needs_acid=True,
         )
 
-        metadata.create_all()
-
-    @engines.close_first
-    def teardown(self):
-        with testing.db.begin() as conn:
-            conn.execute(addresses.delete())
-            conn.execute(users.delete())
-            conn.execute(users2.delete())
-
-    @classmethod
-    def teardown_class(cls):
-        metadata.drop_all()
-
     @testing.fails_on(
         "firebird", "kinterbasdb doesn't send full type information"
     )
@@ -105,6 +82,8 @@ class QueryTest(fixtures.TestBase):
 
         """
 
+        users = self.tables.users
+
         connection.execute(
             users.insert(),
             {"user_id": 7, "user_name": "jack"},
@@ -133,6 +112,7 @@ class QueryTest(fixtures.TestBase):
 
     @testing.requires.order_by_label_with_expression
     def test_order_by_label_compound(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             {"user_id": 7, "user_name": "jack"},
@@ -174,6 +154,7 @@ class QueryTest(fixtures.TestBase):
         assert row.y == False  # noqa
 
     def test_select_tuple(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             {"user_id": 1, "user_name": "apples"},
@@ -187,6 +168,7 @@ class QueryTest(fixtures.TestBase):
         )
 
     def test_like_ops(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             {"user_id": 1, "user_name": "apples"},
@@ -238,6 +220,7 @@ class QueryTest(fixtures.TestBase):
             eq_(connection.scalar(expr), result)
 
     def test_ilike(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             {"user_id": 1, "user_name": "one"},
@@ -279,11 +262,13 @@ class QueryTest(fixtures.TestBase):
             )
 
     def test_compiled_execute(self, connection):
+        users = self.tables.users
         connection.execute(users.insert(), user_id=7, user_name="jack")
         s = select(users).where(users.c.user_id == bindparam("id")).compile()
         eq_(connection.execute(s, id=7).first()._mapping["user_id"], 7)
 
     def test_compiled_insert_execute(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert().compile(), user_id=7, user_name="jack"
         )
@@ -296,6 +281,7 @@ class QueryTest(fixtures.TestBase):
         This should be run for DB-APIs with both positional and named
         paramstyles.
         """
+        users = self.tables.users
 
         connection.execute(users.insert(), user_id=7, user_name="jack")
         connection.execute(users.insert(), user_id=8, user_name="fred")
@@ -369,6 +355,8 @@ class QueryTest(fixtures.TestBase):
         Tests simple, compound, aliased and DESC clauses.
         """
 
+        users = self.tables.users
+
         connection.execute(users.insert(), user_id=1, user_name="c")
         connection.execute(users.insert(), user_id=2, user_name="b")
         connection.execute(users.insert(), user_id=3, user_name="a")
@@ -469,6 +457,8 @@ class QueryTest(fixtures.TestBase):
         Tests simple, compound, aliased and DESC clauses.
         """
 
+        users = self.tables.users
+
         connection.execute(users.insert(), user_id=1)
         connection.execute(users.insert(), user_id=2, user_name="b")
         connection.execute(users.insert(), user_id=3, user_name="a")
@@ -563,6 +553,7 @@ class QueryTest(fixtures.TestBase):
 
     def test_in_filtering(self, connection):
         """test the behavior of the in_() function."""
+        users = self.tables.users
 
         connection.execute(users.insert(), user_id=7, user_name="jack")
         connection.execute(users.insert(), user_id=8, user_name="fred")
@@ -587,6 +578,7 @@ class QueryTest(fixtures.TestBase):
         assert len(r) == 0
 
     def test_expanding_in(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             [
@@ -626,6 +618,7 @@ class QueryTest(fixtures.TestBase):
 
     @testing.requires.no_quoting_special_bind_names
     def test_expanding_in_special_chars(self, connection):
+        users = self.tables.users
         connection.execute(
             users.insert(),
             [
@@ -663,6 +656,8 @@ class QueryTest(fixtures.TestBase):
         )
 
     def test_expanding_in_multiple(self, connection):
+        users = self.tables.users
+
         connection.execute(
             users.insert(),
             [
@@ -687,6 +682,8 @@ class QueryTest(fixtures.TestBase):
         )
 
     def test_expanding_in_repeated(self, connection):
+        users = self.tables.users
+
         connection.execute(
             users.insert(),
             [
@@ -727,6 +724,8 @@ class QueryTest(fixtures.TestBase):
 
     @testing.requires.tuple_in
     def test_expanding_in_composite(self, connection):
+        users = self.tables.users
+
         connection.execute(
             users.insert(),
             [
@@ -768,7 +767,7 @@ class QueryTest(fixtures.TestBase):
                 return value[3:]
 
         users = Table(
-            "query_users",
+            "users",
             MetaData(),
             Column("user_id", Integer, primary_key=True),
             Column("user_name", NameWithProcess()),
@@ -812,6 +811,8 @@ class QueryTest(fixtures.TestBase):
 
         """
 
+        users = self.tables.users
+
         connection.execute(users.insert(), user_id=7, user_name="jack")
         connection.execute(users.insert(), user_id=8, user_name="fred")
         connection.execute(users.insert(), user_id=9, user_name=None)
@@ -827,6 +828,8 @@ class QueryTest(fixtures.TestBase):
     def test_literal_in(self, connection):
         """similar to test_bind_in but use a bind with a value."""
 
+        users = self.tables.users
+
         connection.execute(users.insert(), user_id=7, user_name="jack")
         connection.execute(users.insert(), user_id=8, user_name="fred")
         connection.execute(users.insert(), user_id=9, user_name=None)
@@ -842,6 +845,7 @@ class QueryTest(fixtures.TestBase):
         that a proper boolean value is generated.
 
         """
+        users = self.tables.users
 
         connection.execute(
             users.insert(),
@@ -932,63 +936,60 @@ class RequiredBindTest(fixtures.TablesTest):
         is_(bindparam("foo", callable_=c, required=False).required, False)
 
 
-class LimitTest(fixtures.TestBase):
+class LimitTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        global users, addresses, metadata
-        metadata = MetaData(testing.db)
-        users = Table(
-            "query_users",
+    def define_tables(cls, metadata):
+        Table(
+            "users",
             metadata,
             Column("user_id", INT, primary_key=True),
             Column("user_name", VARCHAR(20)),
         )
-        addresses = Table(
-            "query_addresses",
+        Table(
+            "addresses",
             metadata,
             Column("address_id", Integer, primary_key=True),
-            Column("user_id", Integer, ForeignKey("query_users.user_id")),
+            Column("user_id", Integer, ForeignKey("users.user_id")),
             Column("address", String(30)),
         )
-        metadata.create_all()
-
-        with testing.db.begin() as conn:
-            conn.execute(users.insert(), user_id=1, user_name="john")
-            conn.execute(
-                addresses.insert(), address_id=1, user_id=1, address="addr1"
-            )
-            conn.execute(users.insert(), user_id=2, user_name="jack")
-            conn.execute(
-                addresses.insert(), address_id=2, user_id=2, address="addr1"
-            )
-            conn.execute(users.insert(), user_id=3, user_name="ed")
-            conn.execute(
-                addresses.insert(), address_id=3, user_id=3, address="addr2"
-            )
-            conn.execute(users.insert(), user_id=4, user_name="wendy")
-            conn.execute(
-                addresses.insert(), address_id=4, user_id=4, address="addr3"
-            )
-            conn.execute(users.insert(), user_id=5, user_name="laura")
-            conn.execute(
-                addresses.insert(), address_id=5, user_id=5, address="addr4"
-            )
-            conn.execute(users.insert(), user_id=6, user_name="ralph")
-            conn.execute(
-                addresses.insert(), address_id=6, user_id=6, address="addr5"
-            )
-            conn.execute(users.insert(), user_id=7, user_name="fido")
-            conn.execute(
-                addresses.insert(), address_id=7, user_id=7, address="addr5"
-            )
 
     @classmethod
-    def teardown_class(cls):
-        metadata.drop_all()
+    def insert_data(cls, connection):
+        users, addresses = cls.tables("users", "addresses")
+        conn = connection
+        conn.execute(users.insert(), user_id=1, user_name="john")
+        conn.execute(
+            addresses.insert(), address_id=1, user_id=1, address="addr1"
+        )
+        conn.execute(users.insert(), user_id=2, user_name="jack")
+        conn.execute(
+            addresses.insert(), address_id=2, user_id=2, address="addr1"
+        )
+        conn.execute(users.insert(), user_id=3, user_name="ed")
+        conn.execute(
+            addresses.insert(), address_id=3, user_id=3, address="addr2"
+        )
+        conn.execute(users.insert(), user_id=4, user_name="wendy")
+        conn.execute(
+            addresses.insert(), address_id=4, user_id=4, address="addr3"
+        )
+        conn.execute(users.insert(), user_id=5, user_name="laura")
+        conn.execute(
+            addresses.insert(), address_id=5, user_id=5, address="addr4"
+        )
+        conn.execute(users.insert(), user_id=6, user_name="ralph")
+        conn.execute(
+            addresses.insert(), address_id=6, user_id=6, address="addr5"
+        )
+        conn.execute(users.insert(), user_id=7, user_name="fido")
+        conn.execute(
+            addresses.insert(), address_id=7, user_id=7, address="addr5"
+        )
 
     def test_select_limit(self, connection):
+        users, addresses = self.tables("users", "addresses")
         r = connection.execute(
             users.select(limit=3, order_by=[users.c.user_id])
         ).fetchall()
@@ -998,6 +999,8 @@ class LimitTest(fixtures.TestBase):
     def test_select_limit_offset(self, connection):
         """Test the interaction between limit and offset"""
 
+        users, addresses = self.tables("users", "addresses")
+
         r = connection.execute(
             users.select(limit=3, offset=2, order_by=[users.c.user_id])
         ).fetchall()
@@ -1010,6 +1013,8 @@ class LimitTest(fixtures.TestBase):
     def test_select_distinct_limit(self, connection):
         """Test the interaction between limit and distinct"""
 
+        users, addresses = self.tables("users", "addresses")
+
         r = sorted(
             [
                 x[0]
@@ -1025,6 +1030,8 @@ class LimitTest(fixtures.TestBase):
     def test_select_distinct_offset(self, connection):
         """Test the interaction between distinct and offset"""
 
+        users, addresses = self.tables("users", "addresses")
+
         r = sorted(
             [
                 x[0]
@@ -1043,6 +1050,8 @@ class LimitTest(fixtures.TestBase):
     def test_select_distinct_limit_offset(self, connection):
         """Test the interaction between limit and limit/offset"""
 
+        users, addresses = self.tables("users", "addresses")
+
         r = connection.execute(
             select(addresses.c.address)
             .order_by(addresses.c.address)
@@ -1054,18 +1063,18 @@ class LimitTest(fixtures.TestBase):
         self.assert_(r[0] != r[1] and r[1] != r[2], repr(r))
 
 
-class CompoundTest(fixtures.TestBase):
+class CompoundTest(fixtures.TablesTest):
 
     """test compound statements like UNION, INTERSECT, particularly their
     ability to nest on different databases."""
 
     __backend__ = True
 
+    run_inserts = "each"
+
     @classmethod
-    def setup_class(cls):
-        global metadata, t1, t2, t3
-        metadata = MetaData(testing.db)
-        t1 = Table(
+    def define_tables(cls, metadata):
+        Table(
             "t1",
             metadata,
             Column(
@@ -1078,7 +1087,7 @@ class CompoundTest(fixtures.TestBase):
             Column("col3", String(40)),
             Column("col4", String(30)),
         )
-        t2 = Table(
+        Table(
             "t2",
             metadata,
             Column(
@@ -1091,7 +1100,7 @@ class CompoundTest(fixtures.TestBase):
             Column("col3", String(40)),
             Column("col4", String(30)),
         )
-        t3 = Table(
+        Table(
             "t3",
             metadata,
             Column(
@@ -1104,47 +1113,42 @@ class CompoundTest(fixtures.TestBase):
             Column("col3", String(40)),
             Column("col4", String(30)),
         )
-        metadata.create_all()
-
-        with testing.db.begin() as conn:
-            conn.execute(
-                t1.insert(),
-                [
-                    dict(col2="t1col2r1", col3="aaa", col4="aaa"),
-                    dict(col2="t1col2r2", col3="bbb", col4="bbb"),
-                    dict(col2="t1col2r3", col3="ccc", col4="ccc"),
-                ],
-            )
-            conn.execute(
-                t2.insert(),
-                [
-                    dict(col2="t2col2r1", col3="aaa", col4="bbb"),
-                    dict(col2="t2col2r2", col3="bbb", col4="ccc"),
-                    dict(col2="t2col2r3", col3="ccc", col4="aaa"),
-                ],
-            )
-            conn.execute(
-                t3.insert(),
-                [
-                    dict(col2="t3col2r1", col3="aaa", col4="ccc"),
-                    dict(col2="t3col2r2", col3="bbb", col4="aaa"),
-                    dict(col2="t3col2r3", col3="ccc", col4="bbb"),
-                ],
-            )
-
-    @engines.close_first
-    def teardown(self):
-        pass
 
     @classmethod
-    def teardown_class(cls):
-        metadata.drop_all()
+    def insert_data(cls, connection):
+        t1, t2, t3 = cls.tables("t1", "t2", "t3")
+        conn = connection
+        conn.execute(
+            t1.insert(),
+            [
+                dict(col2="t1col2r1", col3="aaa", col4="aaa"),
+                dict(col2="t1col2r2", col3="bbb", col4="bbb"),
+                dict(col2="t1col2r3", col3="ccc", col4="ccc"),
+            ],
+        )
+        conn.execute(
+            t2.insert(),
+            [
+                dict(col2="t2col2r1", col3="aaa", col4="bbb"),
+                dict(col2="t2col2r2", col3="bbb", col4="ccc"),
+                dict(col2="t2col2r3", col3="ccc", col4="aaa"),
+            ],
+        )
+        conn.execute(
+            t3.insert(),
+            [
+                dict(col2="t3col2r1", col3="aaa", col4="ccc"),
+                dict(col2="t3col2r2", col3="bbb", col4="aaa"),
+                dict(col2="t3col2r3", col3="ccc", col4="bbb"),
+            ],
+        )
 
     def _fetchall_sorted(self, executed):
         return sorted([tuple(row) for row in executed.fetchall()])
 
     @testing.requires.subqueries
     def test_union(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
         (s1, s2) = (
             select(t1.c.col3.label("col3"), t1.c.col4.label("col4")).where(
                 t1.c.col2.in_(["t1col2r1", "t1col2r2"]),
@@ -1171,6 +1175,8 @@ class CompoundTest(fixtures.TestBase):
 
     @testing.fails_on("firebird", "doesn't like ORDER BY with UNIONs")
     def test_union_ordered(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         (s1, s2) = (
             select(t1.c.col3.label("col3"), t1.c.col4.label("col4")).where(
                 t1.c.col2.in_(["t1col2r1", "t1col2r2"]),
@@ -1192,6 +1198,8 @@ class CompoundTest(fixtures.TestBase):
     @testing.fails_on("firebird", "doesn't like ORDER BY with UNIONs")
     @testing.requires.subqueries
     def test_union_ordered_alias(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         (s1, s2) = (
             select(t1.c.col3.label("col3"), t1.c.col4.label("col4")).where(
                 t1.c.col2.in_(["t1col2r1", "t1col2r2"]),
@@ -1220,6 +1228,8 @@ class CompoundTest(fixtures.TestBase):
     )
     @testing.fails_on("sqlite", "FIXME: unknown")
     def test_union_all(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         e = union_all(
             select(t1.c.col3),
             union(select(t1.c.col3), select(t1.c.col3)),
@@ -1241,6 +1251,8 @@ class CompoundTest(fixtures.TestBase):
 
         """
 
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         u = union(select(t1.c.col3), select(t1.c.col3)).alias()
 
         e = union_all(select(t1.c.col3), select(u.c.col3))
@@ -1256,6 +1268,8 @@ class CompoundTest(fixtures.TestBase):
 
     @testing.requires.intersect
     def test_intersect(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         i = intersect(
             select(t2.c.col3, t2.c.col4),
             select(t2.c.col3, t2.c.col4).where(t2.c.col4 == t3.c.col3),
@@ -1274,6 +1288,8 @@ class CompoundTest(fixtures.TestBase):
     @testing.requires.except_
     @testing.fails_on("sqlite", "Can't handle this style of nesting")
     def test_except_style1(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         e = except_(
             union(
                 select(t1.c.col3, t1.c.col4),
@@ -1300,6 +1316,8 @@ class CompoundTest(fixtures.TestBase):
         # same as style1, but add alias().select() to the except_().
         # sqlite can handle it now.
 
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         e = except_(
             union(
                 select(t1.c.col3, t1.c.col4),
@@ -1333,6 +1351,8 @@ class CompoundTest(fixtures.TestBase):
     @testing.requires.except_
     def test_except_style3(self, connection):
         # aaa, bbb, ccc - (aaa, bbb, ccc - (ccc)) = ccc
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         e = except_(
             select(t1.c.col3),  # aaa, bbb, ccc
             except_(
@@ -1346,6 +1366,8 @@ class CompoundTest(fixtures.TestBase):
     @testing.requires.except_
     def test_except_style4(self, connection):
         # aaa, bbb, ccc - (aaa, bbb, ccc - (ccc)) = ccc
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         e = except_(
             select(t1.c.col3),  # aaa, bbb, ccc
             except_(
@@ -1365,6 +1387,8 @@ class CompoundTest(fixtures.TestBase):
         "sqlite can't handle leading parenthesis",
     )
     def test_intersect_unions(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         u = intersect(
             union(select(t1.c.col3, t1.c.col4), select(t3.c.col3, t3.c.col4)),
             union(select(t2.c.col3, t2.c.col4), select(t3.c.col3, t3.c.col4))
@@ -1378,6 +1402,8 @@ class CompoundTest(fixtures.TestBase):
 
     @testing.requires.intersect
     def test_intersect_unions_2(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         u = intersect(
             union(select(t1.c.col3, t1.c.col4), select(t3.c.col3, t3.c.col4))
             .alias()
@@ -1393,6 +1419,8 @@ class CompoundTest(fixtures.TestBase):
 
     @testing.requires.intersect
     def test_intersect_unions_3(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         u = intersect(
             select(t2.c.col3, t2.c.col4),
             union(
@@ -1410,6 +1438,8 @@ class CompoundTest(fixtures.TestBase):
 
     @testing.requires.intersect
     def test_composite_alias(self, connection):
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         ua = intersect(
             select(t2.c.col3, t2.c.col4),
             union(
@@ -1426,10 +1456,7 @@ class CompoundTest(fixtures.TestBase):
         eq_(found, wanted)
 
 
-t1 = t2 = t3 = None
-
-
-class JoinTest(fixtures.TestBase):
+class JoinTest(fixtures.TablesTest):
 
     """Tests join execution.
 
@@ -1443,56 +1470,48 @@ class JoinTest(fixtures.TestBase):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        global metadata
-        global t1, t2, t3
-
-        metadata = MetaData(testing.db)
-        t1 = Table(
+    def define_tables(cls, metadata):
+        Table(
             "t1",
             metadata,
             Column("t1_id", Integer, primary_key=True),
             Column("name", String(32)),
         )
-        t2 = Table(
+        Table(
             "t2",
             metadata,
             Column("t2_id", Integer, primary_key=True),
             Column("t1_id", Integer, ForeignKey("t1.t1_id")),
             Column("name", String(32)),
         )
-        t3 = Table(
+        Table(
             "t3",
             metadata,
             Column("t3_id", Integer, primary_key=True),
             Column("t2_id", Integer, ForeignKey("t2.t2_id")),
             Column("name", String(32)),
         )
-        metadata.drop_all()
-        metadata.create_all()
-
-        with testing.db.begin() as conn:
-            # t1.10 -> t2.20 -> t3.30
-            # t1.11 -> t2.21
-            # t1.12
-            conn.execute(
-                t1.insert(),
-                {"t1_id": 10, "name": "t1 #10"},
-                {"t1_id": 11, "name": "t1 #11"},
-                {"t1_id": 12, "name": "t1 #12"},
-            )
-            conn.execute(
-                t2.insert(),
-                {"t2_id": 20, "t1_id": 10, "name": "t2 #20"},
-                {"t2_id": 21, "t1_id": 11, "name": "t2 #21"},
-            )
-            conn.execute(
-                t3.insert(), {"t3_id": 30, "t2_id": 20, "name": "t3 #30"}
-            )
 
     @classmethod
-    def teardown_class(cls):
-        metadata.drop_all()
+    def insert_data(cls, connection):
+        conn = connection
+        # t1.10 -> t2.20 -> t3.30
+        # t1.11 -> t2.21
+        # t1.12
+        t1, t2, t3 = cls.tables("t1", "t2", "t3")
+
+        conn.execute(
+            t1.insert(),
+            {"t1_id": 10, "name": "t1 #10"},
+            {"t1_id": 11, "name": "t1 #11"},
+            {"t1_id": 12, "name": "t1 #12"},
+        )
+        conn.execute(
+            t2.insert(),
+            {"t2_id": 20, "t1_id": 10, "name": "t2 #20"},
+            {"t2_id": 21, "t1_id": 11, "name": "t2 #21"},
+        )
+        conn.execute(t3.insert(), {"t3_id": 30, "t2_id": 20, "name": "t3 #30"})
 
     def assertRows(self, statement, expected):
         """Execute a statement and assert that rows returned equal expected."""
@@ -1504,6 +1523,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_join_x1(self):
         """Joins t1->t2."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t1.c.t1_id == t2.c.t1_id, t2.c.t1_id == t1.c.t1_id):
             expr = select(t1.c.t1_id, t2.c.t2_id).select_from(
@@ -1513,6 +1533,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_join_x2(self):
         """Joins t1->t2->t3."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t1.c.t1_id == t2.c.t1_id, t2.c.t1_id == t1.c.t1_id):
             expr = select(t1.c.t1_id, t2.c.t2_id).select_from(
@@ -1522,6 +1543,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_x1(self):
         """Outer joins t1->t2."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = select(t1.c.t1_id, t2.c.t2_id).select_from(
@@ -1531,6 +1553,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_x2(self):
         """Outer joins t1->t2,t3."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = select(t1.c.t1_id, t2.c.t2_id, t3.c.t3_id).select_from(
@@ -1544,6 +1567,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_where_x2_t1(self):
         """Outer joins t1->t2,t3, where on t1."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
@@ -1574,6 +1598,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_where_x2_t2(self):
         """Outer joins t1->t2,t3, where on t2."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
@@ -1604,6 +1629,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_where_x2_t3(self):
         """Outer joins t1->t2,t3, where on t3."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
@@ -1635,6 +1661,8 @@ class JoinTest(fixtures.TestBase):
     def test_outerjoin_where_x2_t1t3(self):
         """Outer joins t1->t2,t3, where on t1 and t3."""
 
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
                 select(t1.c.t1_id, t2.c.t2_id, t3.c.t3_id)
@@ -1664,6 +1692,8 @@ class JoinTest(fixtures.TestBase):
     def test_outerjoin_where_x2_t1t2(self):
         """Outer joins t1->t2,t3, where on t1 and t2."""
 
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
+
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
                 select(t1.c.t1_id, t2.c.t2_id, t3.c.t3_id)
@@ -1693,6 +1723,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_outerjoin_where_x2_t1t2t3(self):
         """Outer joins t1->t2,t3, where on t1, t2 and t3."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
@@ -1729,6 +1760,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_mixed(self):
         """Joins t1->t2, outer t2->t3."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = select(t1.c.t1_id, t2.c.t2_id, t3.c.t3_id).select_from(
@@ -1739,6 +1771,7 @@ class JoinTest(fixtures.TestBase):
 
     def test_mixed_where(self):
         """Joins t1->t2, outer t2->t3, plus a where on each table in turn."""
+        t1, t2, t3 = self.tables("t1", "t2", "t3")
 
         for criteria in (t2.c.t2_id == t3.c.t2_id, t3.c.t2_id == t2.c.t2_id):
             expr = (
@@ -1800,17 +1833,12 @@ class JoinTest(fixtures.TestBase):
             self.assertRows(expr, [(10, 20, 30)])
 
 
-metadata = flds = None
-
-
-class OperatorTest(fixtures.TestBase):
+class OperatorTest(fixtures.TablesTest):
     __backend__ = True
 
     @classmethod
-    def setup_class(cls):
-        global metadata, flds
-        metadata = MetaData(testing.db)
-        flds = Table(
+    def define_tables(cls, metadata):
+        Table(
             "flds",
             metadata,
             Column(
@@ -1822,20 +1850,19 @@ class OperatorTest(fixtures.TestBase):
             Column("intcol", Integer),
             Column("strcol", String(50)),
         )
-        metadata.create_all()
-
-        with testing.db.begin() as conn:
-            conn.execute(
-                flds.insert(),
-                [dict(intcol=5, strcol="foo"), dict(intcol=13, strcol="bar")],
-            )
 
     @classmethod
-    def teardown_class(cls):
-        metadata.drop_all()
+    def insert_data(cls, connection):
+        flds = cls.tables.flds
+        connection.execute(
+            flds.insert(),
+            [dict(intcol=5, strcol="foo"), dict(intcol=13, strcol="bar")],
+        )
 
     # TODO: seems like more tests warranted for this setup.
     def test_modulo(self, connection):
+        flds = self.tables.flds
+
         eq_(
             connection.execute(
                 select(flds.c.intcol % 3).order_by(flds.c.idcol)
@@ -1845,6 +1872,8 @@ class OperatorTest(fixtures.TestBase):
 
     @testing.requires.window_functions
     def test_over(self, connection):
+        flds = self.tables.flds
+
         eq_(
             connection.execute(
                 select(
diff --git a/test/sql/test_quote.py b/test/sql/test_quote.py
index a78d6c16b..c743918c8 100644
--- a/test/sql/test_quote.py
+++ b/test/sql/test_quote.py
@@ -187,31 +187,6 @@ class QuoteExecTest(fixtures.TablesTest):
 class QuoteTest(fixtures.TestBase, AssertsCompiledSQL):
     __dialect__ = "default"
 
-    @classmethod
-    def setup_class(cls):
-        # TODO: figure out which databases/which identifiers allow special
-        # characters to be used, such as: spaces, quote characters,
-        # punctuation characters, set up tests for those as well.
-
-        global table1, table2
-        metadata = MetaData(testing.db)
-
-        table1 = Table(
-            "WorstCase1",
-            metadata,
-            Column("lowercase", Integer, primary_key=True),
-            Column("UPPERCASE", Integer),
-            Column("MixedCase", Integer),
-            Column("ASC", Integer, key="a123"),
-        )
-        table2 = Table(
-            "WorstCase2",
-            metadata,
-            Column("desc", Integer, primary_key=True, key="d123"),
-            Column("Union", Integer, key="u123"),
-            Column("MixedCase", Integer),
-        )
-
     @testing.crashes("oracle", "FIXME: unknown, verify not fails_on")
     @testing.requires.subqueries
     def test_labels(self):
@@ -234,6 +209,23 @@ class QuoteTest(fixtures.TestBase, AssertsCompiledSQL):
         where the "UPPERCASE" column of "LaLa" doesn't exist.
         """
 
+        metadata = MetaData()
+        table1 = Table(
+            "WorstCase1",
+            metadata,
+            Column("lowercase", Integer, primary_key=True),
+            Column("UPPERCASE", Integer),
+            Column("MixedCase", Integer),
+            Column("ASC", Integer, key="a123"),
+        )
+        Table(
+            "WorstCase2",
+            metadata,
+            Column("desc", Integer, primary_key=True, key="d123"),
+            Column("Union", Integer, key="u123"),
+            Column("MixedCase", Integer),
+        )
+
         self.assert_compile(
             table1.select(distinct=True).alias("LaLa").select(),
             "SELECT "
diff --git a/test/sql/test_sequences.py b/test/sql/test_sequences.py
index 1809e0cca..b5ba28df7 100644
--- a/test/sql/test_sequences.py
+++ b/test/sql/test_sequences.py
@@ -520,7 +520,7 @@ class SequenceAsServerDefaultTest(
 
     def test_drop_ordering(self):
         with self.sql_execution_asserter(testing.db) as asserter:
-            self.metadata.drop_all(checkfirst=False)
+            self.tables_test_metadata.drop_all(checkfirst=False)
 
         asserter.assert_(
             AllOf(
diff --git a/test/sql/test_types.py b/test/sql/test_types.py
index 919c4b4f9..77aefc190 100644
--- a/test/sql/test_types.py
+++ b/test/sql/test_types.py
@@ -3505,34 +3505,26 @@ class PickleTest(fixtures.TestBase):
             assert p1.compare_values(p1.copy_value(obj), obj)
 
 
-meta = None
-
-
 class CallableTest(fixtures.TestBase):
-    @classmethod
-    def setup_class(cls):
-        global meta
-        meta = MetaData(testing.db)
-
-    @classmethod
-    def teardown_class(cls):
-        meta.drop_all()
-
-    def test_callable_as_arg(self):
+    @testing.provide_metadata
+    def test_callable_as_arg(self, connection):
         ucode = util.partial(Unicode)
 
-        thing_table = Table("thing", meta, Column("name", ucode(20)))
+        thing_table = Table("thing", self.metadata, Column("name", ucode(20)))
         assert isinstance(thing_table.c.name.type, Unicode)
-        thing_table.create()
+        thing_table.create(connection)
 
-    def test_callable_as_kwarg(self):
+    @testing.provide_metadata
+    def test_callable_as_kwarg(self, connection):
         ucode = util.partial(Unicode)
 
         thang_table = Table(
-            "thang", meta, Column("name", type_=ucode(20), primary_key=True)
+            "thang",
+            self.metadata,
+            Column("name", type_=ucode(20), primary_key=True),
         )
         assert isinstance(thang_table.c.name.type, Unicode)
-        thang_table.create()
+        thang_table.create(connection)
 
 
 class LiteralTest(fixtures.TestBase):