Add tests for pool when connection opener fails

Author: robsdedude

neo4j/_async/io/_pool.py

@@ -248,24 +248,14 @@ async def deactivate(self, address):
                         await conn.close()
                     except OSError:
                         pass
-            if not connections:
-                await self.remove(address)
+            if not self.connections[address]:
+                del self.connections[address]
 
     def on_write_failure(self, address):
         raise WriteServiceUnavailable(
             "No write service available for pool {}".format(self)
         )
 
-    async def remove(self, address):
-        """ Remove an address from the connection pool, if present, closing
-        all connections to that address.
-        """
-        async with self.lock:
-            for connection in self.connections.pop(address, ()):
-                try:
-                    await connection.close()
-                except OSError:
-                    pass
 
     async def close(self):
         """ Close all connections and empty the pool.
@@ -274,7 +264,8 @@ async def close(self):
         try:
             async with self.lock:
                 for address in list(self.connections):
-                    await self.remove(address)
+                    for connection in self.connections.pop(address, ()):
+                        await connection.close()
         except TypeError:
             pass
 

neo4j/_sync/io/_pool.py

@@ -248,24 +248,14 @@ def deactivate(self, address):
                         conn.close()
                     except OSError:
                         pass
-            if not connections:
-                self.remove(address)
+            if not self.connections[address]:
+                del self.connections[address]
 
     def on_write_failure(self, address):
         raise WriteServiceUnavailable(
             "No write service available for pool {}".format(self)
         )
 
-    def remove(self, address):
-        """ Remove an address from the connection pool, if present, closing
-        all connections to that address.
-        """
-        with self.lock:
-            for connection in self.connections.pop(address, ()):
-                try:
-                    connection.close()
-                except OSError:
-                    pass
 
     def close(self):
         """ Close all connections and empty the pool.
@@ -274,7 +264,8 @@ def close(self):
         try:
             with self.lock:
                 for address in list(self.connections):
-                    self.remove(address)
+                    for connection in self.connections.pop(address, ()):
+                        connection.close()
         except TypeError:
             pass
 

tests/unit/async_/io/test_neo4j_pool.py

@@ -35,6 +35,10 @@
 )
 
 from ...._async_compat import mark_async_test
+from ..._async_compat import (
+    AsyncMock,
+    mark_async_test,
+)
 from ..work import async_fake_connection_generator
 
 
@@ -384,3 +388,17 @@ def liveness_side_effect(*args, **kwargs):
     cx3.reset.assert_awaited_once()
     assert cx1 not in pool.connections[cx1.addr]
     assert cx3 in pool.connections[cx1.addr]
+
+
+
+@mark_async_test
+async def test_failing_opener_leaves_connections_in_use_alone(opener):
+    pool = AsyncNeo4jPool(
+        opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    cx1 = await pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    opener.side_effect = ServiceUnavailable("Server overloaded")
+    with pytest.raises((ServiceUnavailable, SessionExpired)):
+        await pool.acquire(READ_ACCESS, 30, "test_db", None)
+    assert not cx1.closed()

tests/unit/sync/io/test_neo4j_pool.py

@@ -35,6 +35,10 @@
 )
 
 from ...._async_compat import mark_sync_test
+from ..._async_compat import (
+    mark_sync_test,
+    Mock,
+)
 from ..work import fake_connection_generator
 
 
@@ -384,3 +388,17 @@ def liveness_side_effect(*args, **kwargs):
     cx3.reset.assert_called_once()
     assert cx1 not in pool.connections[cx1.addr]
     assert cx3 in pool.connections[cx1.addr]
+
+
+
+@mark_sync_test
+def test_failing_opener_leaves_connections_in_use_alone(opener):
+    pool = Neo4jPool(
+        opener, PoolConfig(), WorkspaceConfig(), ROUTER_ADDRESS
+    )
+    cx1 = pool.acquire(READ_ACCESS, 30, "test_db", None)
+
+    opener.side_effect = ServiceUnavailable("Server overloaded")
+    with pytest.raises((ServiceUnavailable, SessionExpired)):
+        pool.acquire(READ_ACCESS, 30, "test_db", None)
+    assert not cx1.closed()