integrate connection.terminate() for supporting dialects

Integrated support for asyncpg's ``terminate()`` method call for cases
where the connection pool is recycling a possibly timed-out connection,
where a connection is being garbage collected that wasn't gracefully
closed, as well as when the connection has been invalidated. This allows
asyncpg to abandon the connection without waiting for a response that may
incur long timeouts.

Fixes: #8419
Change-Id: Ia575af779d5733b483a72dff3690b8bbbad2bb05

1 files changed, 15 insertions, 5 deletions

diff --git a/test/engine/test_pool.py b/test/engine/test_pool.py
index 2bbb976a8..473d462a3 100644
--- a/test/engine/test_pool.py
+++ b/test/engine/test_pool.py
@@ -87,10 +87,13 @@ class PoolTestBase(fixtures.TestBase):
     def _queuepool_dbapi_fixture(self, **kw):
         dbapi = MockDBAPI()
         _is_asyncio = kw.pop("_is_asyncio", False)
+        _has_terminate = kw.pop("_has_terminate", False)
         p = pool.QueuePool(creator=lambda: dbapi.connect("foo.db"), **kw)
         if _is_asyncio:
             p._is_asyncio = True
             p._dialect = _AsyncConnDialect()
+        if _has_terminate:
+            p._dialect.has_terminate = True
         return dbapi, p
 
 
@@ -445,8 +448,10 @@ class PoolEventsTest(PoolTestBase):
 
         return p, canary
 
-    def _checkin_event_fixture(self, _is_asyncio=False):
-        p = self._queuepool_fixture(_is_asyncio=_is_asyncio)
+    def _checkin_event_fixture(self, _is_asyncio=False, _has_terminate=False):
+        p = self._queuepool_fixture(
+            _is_asyncio=_is_asyncio, _has_terminate=_has_terminate
+        )
         canary = []
 
         @event.listens_for(p, "checkin")
@@ -721,9 +726,12 @@ class PoolEventsTest(PoolTestBase):
         assert canary.call_args_list[0][0][0] is dbapi_con
         assert canary.call_args_list[0][0][2] is exc
 
-    @testing.combinations((True,), (False,))
-    def test_checkin_event_gc(self, detach_gced):
-        p, canary = self._checkin_event_fixture(_is_asyncio=detach_gced)
+    @testing.combinations((True,), (False,), argnames="is_asyncio")
+    @testing.combinations((True,), (False,), argnames="has_terminate")
+    def test_checkin_event_gc(self, is_asyncio, has_terminate):
+        p, canary = self._checkin_event_fixture(
+            _is_asyncio=is_asyncio, _has_terminate=has_terminate
+        )
 
         c1 = p.connect()
 
@@ -733,6 +741,8 @@ class PoolEventsTest(PoolTestBase):
         del c1
         lazy_gc()
 
+        detach_gced = is_asyncio and not has_terminate
+
         if detach_gced:
             # "close_detached" is not called because for asyncio the
             # connection is just lost.