Propagate asyncio flag from the dialect to selected pool classes

Fixed an issue that presented itself when using the :class:`_pool.NullPool`
or the :class:`_pool.StaticPool` with an async engine. This mostly affected
the aiosqlite dialect.

Fixes: #6575
Change-Id: Ic1e27d99ffcb20ed4de82ea78f430a0f3b629d86

1 files changed, 35 insertions, 1 deletions

diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 5b6dcfa45..70671134f 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -10,7 +10,8 @@ from sqlalchemy import pool
 from sqlalchemy import select
 from sqlalchemy import testing
 from sqlalchemy.engine import default
-from sqlalchemy.pool.impl import _AsyncConnDialect
+from sqlalchemy.pool.base import _AsyncConnDialect
+from sqlalchemy.pool.base import _ConnDialect
 from sqlalchemy.testing import assert_raises
 from sqlalchemy.testing import assert_raises_context_ok
 from sqlalchemy.testing import assert_raises_message
@@ -280,6 +281,39 @@ class PoolTest(PoolTestBase):
         if "use_lifo" in pool_args:
             eq_(p1._pool.use_lifo, p2._pool.use_lifo)
 
+    @testing.combinations(
+        (pool.QueuePool, False),
+        (pool.AsyncAdaptedQueuePool, True),
+        (pool.FallbackAsyncAdaptedQueuePool, True),
+        (pool.NullPool, None),
+        (pool.SingletonThreadPool, False),
+        (pool.StaticPool, None),
+        (pool.AssertionPool, None),
+    )
+    def test_is_asyncio_from_dialect(self, pool_cls, is_async_king):
+        p = pool_cls(creator=object())
+        for is_async in (True, False):
+            if is_async:
+                p._dialect = _AsyncConnDialect()
+            else:
+                p._dialect = _ConnDialect
+            if is_async_king is None:
+                eq_(p._is_asyncio, is_async)
+            else:
+                eq_(p._is_asyncio, is_async_king)
+
+    @testing.combinations(
+        (pool.QueuePool, False),
+        (pool.AsyncAdaptedQueuePool, True),
+        (pool.FallbackAsyncAdaptedQueuePool, True),
+        (pool.NullPool, False),
+        (pool.SingletonThreadPool, False),
+        (pool.StaticPool, False),
+        (pool.AssertionPool, False),
+    )
+    def test_is_asyncio_from_dialect_cls(self, pool_cls, is_async):
+        eq_(pool_cls._is_asyncio, is_async)
+
 
 class PoolDialectTest(PoolTestBase):
     def _dialect(self):