Method test_wait_queue_timeout

test/test_pooling.py:322–338

(self)

            pass

    def test_wait_queue_timeout(self):
        wait_queue_timeout = 2  # Seconds
        pool = self.create_pool(max_pool_size=1, wait_queue_timeout=wait_queue_timeout)
        self.addCleanup(pool.close)

        with pool.checkout():
            start = time.time()
            with self.assertRaises(ConnectionFailure):
                with pool.checkout():
                    pass

        duration = time.time() - start
        self.assertLess(
            abs(wait_queue_timeout - duration),
            1,
            f"Waited {duration:.2f} seconds for a socket, expected {wait_queue_timeout:f}",
        )

    def test_no_wait_queue_timeout(self):
        # Verify get_socket() with no wait_queue_timeout blocks forever.