1 files changed, 55 insertions, 5 deletions

diff --git a/lib/sqlalchemy/testing/util.py b/lib/sqlalchemy/testing/util.py
index eb9fcd1cd..01185c284 100644
--- a/lib/sqlalchemy/testing/util.py
+++ b/lib/sqlalchemy/testing/util.py
@@ -14,6 +14,7 @@ import types
 from . import config
 from . import mock
 from .. import inspect
+from ..engine import Connection
 from ..schema import Column
 from ..schema import DropConstraint
 from ..schema import DropTable
@@ -207,11 +208,13 @@ def fail(msg):
 
 @decorator
 def provide_metadata(fn, *args, **kw):
-    """Provide bound MetaData for a single test, dropping afterwards."""
+    """Provide bound MetaData for a single test, dropping afterwards.
 
-    # import cycle that only occurs with py2k's import resolver
-    # in py3k this can be moved top level.
-    from . import engines
+    Legacy; use the "metadata" pytest fixture.
+
+    """
+
+    from . import fixtures
 
     metadata = schema.MetaData()
     self = args[0]
@@ -220,7 +223,31 @@ def provide_metadata(fn, *args, **kw):
     try:
         return fn(*args, **kw)
     finally:
-        engines.drop_all_tables(metadata, config.db)
+        # close out some things that get in the way of dropping tables.
+        # when using the "metadata" fixture, there is a set ordering
+        # of things that makes sure things are cleaned up in order, however
+        # the simple "decorator" nature of this legacy function means
+        # we have to hardcode some of that cleanup ahead of time.
+
+        # close ORM sessions
+        fixtures._close_all_sessions()
+
+        # integrate with the "connection" fixture as there are many
+        # tests where it is used along with provide_metadata
+        if fixtures._connection_fixture_connection:
+            # TODO: this warning can be used to find all the places
+            # this is used with connection fixture
+            # warn("mixing legacy provide metadata with connection fixture")
+            drop_all_tables_from_metadata(
+                metadata, fixtures._connection_fixture_connection
+            )
+            # as the provide_metadata fixture is often used with "testing.db",
+            # when we do the drop we have to commit the transaction so that
+            # the DB is actually updated as the CREATE would have been
+            # committed
+            fixtures._connection_fixture_connection.get_transaction().commit()
+        else:
+            drop_all_tables_from_metadata(metadata, config.db)
         self.metadata = prev_meta
 
 
@@ -359,6 +386,29 @@ class adict(dict):
     get_all = __call__
 
 
+def drop_all_tables_from_metadata(metadata, engine_or_connection):
+    from . import engines
+
+    def go(connection):
+        engines.testing_reaper.prepare_for_drop_tables(connection)
+
+        if not connection.dialect.supports_alter:
+            from . import assertions
+
+            with assertions.expect_warnings(
+                "Can't sort tables", assert_=False
+            ):
+                metadata.drop_all(connection)
+        else:
+            metadata.drop_all(connection)
+
+    if not isinstance(engine_or_connection, Connection):
+        with engine_or_connection.begin() as connection:
+            go(connection)
+    else:
+        go(engine_or_connection)
+
+
 def drop_all_tables(engine, inspector, schema=None, include_names=None):
 
     if include_names is not None: