def test_over_max_no_timeout_blocks(self):
        class Thread(threading.Thread):
            def __init__(self, pool):
                super(Thread, self).__init__()
                self.state = 'init'
                self.pool = pool
                self.sock = None

            def run(self):
                self.state = 'get_socket'
                self.sock = self.pool.get_socket()
                self.state = 'sock'

        pool = self.get_pool(None, 2, None)
        s1 = pool.get_socket()
        self.assertTrue(None is not s1)
        s2 = pool.get_socket()
        self.assertTrue(None is not s2)
        self.assertNotEqual(s1, s2)
        t = Thread(pool)
        t.start()
        while t.state != 'get_socket':
            time.sleep(0.1)
        self.assertEqual(t.state, 'get_socket')
        time.sleep(5)
        self.assertEqual(t.state, 'get_socket')
        pool.maybe_return_socket(s1)
        while t.state != 'sock':
            time.sleep(0.1)
        self.assertEqual(t.state, 'sock')
        self.assertEqual(t.sock, s1)

 def test_wait_queue_timeout(self):
     wait_queue_timeout = 2  # Seconds
     pool = self.get_pool_with_wait_queue_timeout(wait_queue_timeout)
     pool.get_socket()
     start = time.time()
     self.assertRaises(ConnectionFailure, pool.get_socket)
     duration = time.time() - start
     self.assertTrue(
         abs(wait_queue_timeout - duration) < 1,
         "Waited %.2f seconds for a socket, expected %f" % (
             duration, wait_queue_timeout))

 def test_over_max_times_out(self):
     conn_timeout = 2
     pool = self.get_pool(conn_timeout, conn_timeout + 5, conn_timeout)
     s1 = pool.get_socket()
     self.assertTrue(None is not s1)
     s2 = pool.get_socket()
     self.assertTrue(None is not s2)
     self.assertNotEqual(s1, s2)
     start = time.time()
     self.assertRaises(socket.timeout, pool.get_socket)
     end = time.time()
     self.assertTrue(end - start > conn_timeout)
     self.assertTrue(end - start < conn_timeout + 5)

    def test_wait_queue_multiple(self):
        pool = self.get_pool_with_wait_queue_multiple(3)

        # Reach max_size sockets.
        pool.get_socket()
        pool.get_socket()

        # Reach max_size * wait_queue_multiple waiters.
        threads = []
        for _ in xrange(6):
            t = SocketGetter(self, pool)
            t.start()
            threads.append(t)

        self.sleep(1)
        for t in threads:
            self.assertEqual(t.state, 'get_socket')

        self.assertRaises(ExceededMaxWaiters, pool.get_socket)

    def __socket(self, mongo):
        """Get a SocketInfo from the pool.
        """
        pool = mongo['pool']
        if self.__auto_start_request:
            # No effect if a request already started
            self.start_request()

        sock_info = pool.get_socket()

        if self.__auth_credentials:
            self.__check_auth(sock_info)
        return sock_info

    def __is_master(self, host):
        """Directly call ismaster.
        """
        pool = self.pool_class(host, self.__max_pool_size,
                               self.__net_timeout, self.__conn_timeout,
                               self.__use_ssl)
        sock_info = pool.get_socket()
        response = self.__simple_command(
            sock_info, 'admin', {'ismaster': 1}
        )

        pool.return_socket(sock_info)
        return response, pool

 def test_wait_queue_multiple_unset(self):
     pool = self.get_pool_with_wait_queue_multiple(None)
     socks = []
     for _ in xrange(2):
         sock = pool.get_socket()
         socks.append(sock)
     threads = []
     for _ in xrange(30):
         t = SocketGetter(self, pool)
         t.start()
         threads.append(t)
     self.sleep(1)
     for t in threads:
         self.assertEqual(t.state, 'get_socket')

    def __is_master(self, host):
        """Directly call ismaster.
        """
        pool = self.pool_class(host, self.__max_pool_size,
                               self.__net_timeout, self.__conn_timeout,
                               self.__use_ssl)
        sock_info = pool.get_socket()
        try:
            response = self.__simple_command(
                sock_info, 'admin', {'ismaster': 1}
            )

            pool.maybe_return_socket(sock_info)
            return response, pool
        except (ConnectionFailure, socket.error):
            pool.discard_socket(sock_info)
            raise

    def test_blocking(self):
        # Verify get_socket() with no wait_queue_timeout blocks forever.
        pool = self.get_pool_with_wait_queue_timeout(None)

        # Reach max_size.
        s1 = pool.get_socket()
        t = SocketGetter(self, pool)
        t.start()
        while t.state != 'get_socket':
            self.sleep(0.1)

        self.sleep(1)
        self.assertEqual(t.state, 'get_socket')
        pool.maybe_return_socket(s1)
        while t.state != 'sock':
            self.sleep(0.1)

        self.assertEqual(t.state, 'sock')
        self.assertEqual(t.sock, s1)

    def test_wait_queue_multiple(self):
        class Thread(threading.Thread):
            def __init__(self, pool):
                super(Thread, self).__init__()
                self.state = 'init'
                self.pool = pool
                self.sock = None

            def run(self):
                self.state = 'get_socket'
                self.sock = self.pool.get_socket()
                self.state = 'sock'

        pool = self.get_pool(None, None, None, 3)
        socks = []
        for _ in xrange(2):
            sock = pool.get_socket()
            self.assertTrue(sock is not None)
            socks.append(sock)
        threads = []
        for _ in xrange(6):
            t = Thread(pool)
            t.start()
            threads.append(t)
        time.sleep(1)
        for t in threads:
            self.assertEqual(t.state, 'get_socket')
        self.assertRaises(ExceededMaxWaiters, pool.get_socket)
        while threads:
            for sock in socks:
                pool.maybe_return_socket(sock)
            socks = []
            for t in list(threads):
                if t.sock is not None:
                    socks.append(t.sock)
                    t.join()
                    threads.remove(t)