add prose tests

Author: blink1073

test/asynchronous/test_pooling.py

@@ -513,6 +513,67 @@ async def test_connection_timeout_message(self):
             str(error.exception),
         )
 
+    @async_client_context.require_failCommand_appName
+    async def test_pool_backoff_preserves_existing_collections(self):
+        client = await self.async_rs_or_single_client()
+        coll = self.db.t
+        pool = await async_get_pool(client)
+        await coll.insert_many([{"x": 1} for _ in range(10)])
+        t = SocketGetter(self.c, pool)
+        await t.start()
+        while t.state != "connection":
+            await asyncio.sleep(0.1)
+
+        assert not t.sock.conn_closed()
+
+        # Mock a session establishment overload.
+        mock_connection_fail = {
+            "configureFailPoint": "failCommand",
+            "mode": {"times": 1},
+            "data": {
+                "closeConnection": True,
+            },
+        }
+
+        async with self.fail_point(mock_connection_fail):
+            await coll.find_one({})
+
+        # Make sure the pool is out of backoff state.
+        assert pool._backoff == 0
+
+        # Make sure the existing socket was not affected.
+        assert not t.sock.conn_closed()
+
+        # Cleanup
+        await t.release_conn()
+        await t.join()
+        await pool.close()
+
+    async def test_pool_check_backoff(self):
+        # Test that Pool recovers from two connection failures in a row.
+        # This exercises code at the end of Pool._check().
+        cx_pool = await self.create_pool(max_pool_size=1, connect_timeout=1, wait_queue_timeout=1)
+        self.addAsyncCleanup(cx_pool.close)
+
+        async with cx_pool.checkout() as conn:
+            # Simulate a closed socket without telling the Connection it's
+            # closed.
+            await conn.conn.close()
+
+        # Enable backoff.
+        cx_pool._backoff = 1
+
+        # Swap pool's address with a bad one.
+        address, cx_pool.address = cx_pool.address, ("foo.com", 1234)
+        with self.assertRaises(AutoReconnect):
+            async with cx_pool.checkout():
+                pass
+
+        # Back to normal, semaphore was correctly released.
+        cx_pool.address = address
+        async with cx_pool.checkout():
+            pass
+
 
 class TestPoolMaxSize(_TestPoolingBase):
     async def test_max_pool_size(self):

test/test_pooling.py

@@ -511,6 +511,67 @@ def test_connection_timeout_message(self):
             str(error.exception),
         )
 
+    @client_context.require_failCommand_appName
+    def test_pool_backoff_preserves_existing_collections(self):
+        client = self.rs_or_single_client()
+        coll = self.db.t
+        pool = get_pool(client)
+        coll.insert_many([{"x": 1} for _ in range(10)])
+        t = SocketGetter(self.c, pool)
+        t.start()
+        while t.state != "connection":
+            time.sleep(0.1)
+
+        assert not t.sock.conn_closed()
+
+        # Mock a session establishment overload.
+        mock_connection_fail = {
+            "configureFailPoint": "failCommand",
+            "mode": {"times": 1},
+            "data": {
+                "closeConnection": True,
+            },
+        }
+
+        with self.fail_point(mock_connection_fail):
+            coll.find_one({})
+
+        # Make sure the pool is out of backoff state.
+        assert pool._backoff == 0
+
+        # Make sure the existing socket was not affected.
+        assert not t.sock.conn_closed()
+
+        # Cleanup
+        t.release_conn()
+        t.join()
+        pool.close()
+
+    def test_pool_check_backoff(self):
+        # Test that Pool recovers from two connection failures in a row.
+        # This exercises code at the end of Pool._check().
+        cx_pool = self.create_pool(max_pool_size=1, connect_timeout=1, wait_queue_timeout=1)
+        self.addCleanup(cx_pool.close)
+
+        with cx_pool.checkout() as conn:
+            # Simulate a closed socket without telling the Connection it's
+            # closed.
+            conn.conn.close()
+
+        # Enable backoff.
+        cx_pool._backoff = 1
+
+        # Swap pool's address with a bad one.
+        address, cx_pool.address = cx_pool.address, ("foo.com", 1234)
+        with self.assertRaises(AutoReconnect):
+            with cx_pool.checkout():
+                pass
+
+        # Back to normal, semaphore was correctly released.
+        cx_pool.address = address
+        with cx_pool.checkout():
+            pass
+
 
 class TestPoolMaxSize(_TestPoolingBase):
     def test_max_pool_size(self):