unify transactional context managers

Applied consistent behavior to the use case of
calling ``.commit()`` or ``.rollback()`` inside of an existing
``.begin()`` context manager, with the addition of potentially
emitting SQL within the block subsequent to the commit or rollback.
This change continues upon the change first added in
:ticket:`6155` where the use case of calling "rollback" inside of
a ``.begin()`` contextmanager block was proposed:

* calling ``.commit()`` or ``.rollback()`` will now be allowed
without error or warning within all scopes, including
that of legacy and future :class:`_engine.Engine`, ORM
:class:`_orm.Session`, asyncio :class:`.AsyncEngine`.  Previously,
the :class:`_orm.Session` disallowed this.

* The remaining scope of the context manager is then closed;
when the block ends, a check is emitted to see if the transaction
was already ended, and if so the block returns without action.

* It will now raise **an error** if subsequent SQL of any kind
is emitted within the block, **after** ``.commit()`` or
``.rollback()`` is called.   The block should be closed as
the state of the executable object would otherwise be undefined
in this state.

Fixes: #6288
Change-Id: I8b21766ae430f0fa1ac5ef689f4c0fb19fc84336

1 files changed, 10 insertions, 1 deletions

diff --git a/test/ext/asyncio/test_session_py3k.py b/test/ext/asyncio/test_session_py3k.py
index feb557471..e97e2563a 100644
--- a/test/ext/asyncio/test_session_py3k.py
+++ b/test/ext/asyncio/test_session_py3k.py
@@ -14,10 +14,11 @@ from sqlalchemy.testing import engines
 from sqlalchemy.testing import eq_
 from sqlalchemy.testing import is_
 from sqlalchemy.testing import mock
+from .test_engine_py3k import AsyncFixture as _AsyncFixture
 from ...orm import _fixtures
 
 
-class AsyncFixture(_fixtures.FixtureTest):
+class AsyncFixture(_AsyncFixture, _fixtures.FixtureTest):
     __requires__ = ("async_dialect",)
 
     @classmethod
@@ -124,6 +125,14 @@ class AsyncSessionTransactionTest(AsyncFixture):
     run_inserts = None
 
     @async_test
+    async def test_interrupt_ctxmanager_connection(
+        self, async_trans_ctx_manager_fixture, async_session
+    ):
+        fn = async_trans_ctx_manager_fixture
+
+        await fn(async_session, trans_on_subject=True, execute_on_subject=True)
+
+    @async_test
     async def test_sessionmaker_block_one(self, async_engine):
 
         User = self.classes.User