Method test_no_wait_queue_timeout

test/test_pooling.py:340–363

(self)

        )

    def test_no_wait_queue_timeout(self):
        # Verify get_socket() with no wait_queue_timeout blocks forever.
        pool = self.create_pool(max_pool_size=1)
        self.addCleanup(pool.close)

        # Reach max_size.
        with pool.checkout() as s1:
            t = SocketGetter(self.c, pool)
            t.start()
            while t.state != "get_socket":
                time.sleep(0.1)

            time.sleep(1)
            self.assertEqual(t.state, "get_socket")

        while t.state != "connection":
            time.sleep(0.1)

        self.assertEqual(t.state, "connection")
        self.assertEqual(t.sock, s1)
        # Cleanup
        t.release_conn()
        t.join()
        pool.close()

    def test_checkout_more_than_max_pool_size(self):
        pool = self.create_pool(max_pool_size=2)