Class SocketGetter

test/test_pooling.py:104–132

Utility for TestPooling. Checks out a socket and holds it forever. Used in test_no_wait_queue_timeout.

class SocketGetter(MongoTask):
    """Utility for TestPooling.

    Checks out a socket and holds it forever. Used in
    test_no_wait_queue_timeout.
    """

    def __init__(self, client, pool):
        super().__init__(client)
        self.state = "init"
        self.pool = pool
        self.sock = None

    def run_mongo_thread(self):
        self.state = "get_socket"

        # Call 'pin_cursor' so we can hold the socket.
        with self.pool.checkout() as sock:
            sock.pin_cursor()
            self.sock = sock

        self.state = "connection"

    def release_conn(self):
        if self.sock:
            self.sock.unpin()
            self.sock = None
            return True
        return False


def run_cases(client, cases):