ensure exception raised for all stream w/ sync result

Fixed issue where the :meth:`_asyncio.AsyncSession.execute` method failed
to raise an informative exception if the ``stream_results`` execution
option were used, which is incompatible with a sync-style
:class:`_result.Result` object. An exception is now raised in this scenario
in the same way one is already raised when using ``stream_results`` in
conjunction with the :meth:`_asyncio.AsyncConnection.execute` method.
Additionally, for improved stability with state-sensitive dialects such as
asyncmy, the cursor is now closed when this error condition is raised;
previously with the asyncmy dialect, the connection would go into an
invalid state with unconsumed server side results remaining.

Fixes: #7667
Change-Id: I6eb7affe08584889b57423a90258295f8b7085dc

1 files changed, 27 insertions, 0 deletions

diff --git a/test/ext/asyncio/test_engine_py3k.py b/test/ext/asyncio/test_engine_py3k.py
index 0fdbc28df..1f40cbdec 100644
--- a/test/ext/asyncio/test_engine_py3k.py
+++ b/test/ext/asyncio/test_engine_py3k.py
@@ -18,6 +18,7 @@ from sqlalchemy import union_all
 from sqlalchemy.ext.asyncio import async_engine_from_config
 from sqlalchemy.ext.asyncio import create_async_engine
 from sqlalchemy.ext.asyncio import engine as _async_engine
+from sqlalchemy.ext.asyncio import exc as async_exc
 from sqlalchemy.ext.asyncio import exc as asyncio_exc
 from sqlalchemy.ext.asyncio.base import ReversibleProxy
 from sqlalchemy.ext.asyncio.engine import AsyncConnection
@@ -724,6 +725,32 @@ class AsyncInspection(EngineFixture):
 
 
 class AsyncResultTest(EngineFixture):
+    @async_test
+    async def test_no_ss_cursor_w_execute(self, async_engine):
+        users = self.tables.users
+        async with async_engine.connect() as conn:
+            conn = await conn.execution_options(stream_results=True)
+            with expect_raises_message(
+                async_exc.AsyncMethodRequired,
+                r"Can't use the AsyncConnection.execute\(\) method with a "
+                r"server-side cursor. Use the AsyncConnection.stream\(\) "
+                r"method for an async streaming result set.",
+            ):
+                await conn.execute(select(users))
+
+    @async_test
+    async def test_no_ss_cursor_w_exec_driver_sql(self, async_engine):
+        async with async_engine.connect() as conn:
+            conn = await conn.execution_options(stream_results=True)
+            with expect_raises_message(
+                async_exc.AsyncMethodRequired,
+                r"Can't use the AsyncConnection.exec_driver_sql\(\) "
+                r"method with a "
+                r"server-side cursor. Use the AsyncConnection.stream\(\) "
+                r"method for an async streaming result set.",
+            ):
+                await conn.exec_driver_sql("SELECT * FROM users")
+
     @testing.combinations(
         (None,), ("scalars",), ("mappings",), argnames="filter_"
     )