contextmanager skips rollback if trans says to skip it

Fixed issue where if an exception occurred when the :class:`_orm.Session`
were to close the connection within the :meth:`_orm.Session.commit` method,
when using a context manager for :meth:`_orm.Session.begin` , it would
attempt a rollback which would not be possible as the :class:`_orm.Session`
was in between where the transaction is committed and the connection is
then to be returned to the pool, raising the exception "this
sessiontransaction is in the committed state". This exception can occur
mostly in an asyncio context where CancelledError can be raised.

Fixes: #7388
Change-Id: I1a85a3a7eae79f3553ddf1e3d245a0d90b0a2f40
(cherry picked from commit a845da8b0fc5bb172e278c399a1de9a2e49d62af)

1 files changed, 30 insertions, 0 deletions

diff --git a/test/engine/test_transaction.py b/test/engine/test_transaction.py
index 9e6142022..43b42647e 100644
--- a/test/engine/test_transaction.py
+++ b/test/engine/test_transaction.py
@@ -487,6 +487,36 @@ class TransactionTest(fixtures.TablesTest):
         result = connection.exec_driver_sql("select * from users")
         assert len(result.fetchall()) == 0
 
+    @testing.requires.independent_connections
+    def test_no_rollback_in_deactive(self, local_connection):
+        """test #7388"""
+
+        def fail(*arg, **kw):
+            raise BaseException("some base exception")
+
+        with mock.patch.object(testing.db.dialect, "do_commit", fail):
+            with expect_raises_message(BaseException, "some base exception"):
+                with local_connection.begin():
+                    pass
+
+    @testing.requires.independent_connections
+    @testing.requires.savepoints
+    def test_no_rollback_in_deactive_savepoint(self, local_connection):
+        """test #7388"""
+
+        def fail(*arg, **kw):
+            raise BaseException("some base exception")
+
+        with mock.patch.object(
+            testing.db.dialect, "do_release_savepoint", fail
+        ):
+            with local_connection.begin():
+                with expect_raises_message(
+                    BaseException, "some base exception"
+                ):
+                    with local_connection.begin_nested():
+                        pass
+
     @testing.requires.savepoints
     def test_nested_subtransaction_rollback(self, local_connection):
         connection = local_connection