rollback connection before teardown

this test needs to clear out the transaction
if running on SQLAlchemy 2.0.

Change-Id: Ic3f76433ba9b9ea5485a81caae745d3c7f2a4ee8

2 files changed, 9 insertions, 0 deletions

diff --git a/alembic/testing/fixtures.py b/alembic/testing/fixtures.py
index c273665..5937d48 100644
--- a/alembic/testing/fixtures.py
+++ b/alembic/testing/fixtures.py
@@ -245,6 +245,7 @@ class AlterColRoundTripFixture:
         ), "server defaults %r and %r didn't compare as equivalent" % (s1, s2)
 
     def tearDown(self):
+        sqla_compat._safe_rollback_connection_transaction(self.conn)
         with self.conn.begin():
             self.metadata.drop_all(self.conn)
         self.conn.close()
diff --git a/alembic/util/sqla_compat.py b/alembic/util/sqla_compat.py
index 21dee89..6f98af4 100644
--- a/alembic/util/sqla_compat.py
+++ b/alembic/util/sqla_compat.py
@@ -122,6 +122,14 @@ def _safe_begin_connection_transaction(
         return connection.begin()
 
 
+def _safe_rollback_connection_transaction(
+    connection: "Connection",
+) -> "Transaction":
+    transaction = _get_connection_transaction(connection)
+    if transaction:
+        transaction.rollback()
+
+
 def _get_connection_in_transaction(connection: Optional["Connection"]) -> bool:
     try:
         in_transaction = connection.in_transaction  # type: ignore