Class MockClient

test/pymongo_mocks.py:95–251

class MockClient(MongoClient):
    def __init__(
        self,
        standalones,
        members,
        mongoses,
        hello_hosts=None,
        arbiters=None,
        down_hosts=None,
        *args,
        **kwargs,
    ):
        """A MongoClient connected to the default server, with a mock topology.

        standalones, members, mongoses, arbiters, and down_hosts determine the
        configuration of the topology. They are formatted like ['a:1', 'b:2'].
        hello_hosts provides an alternative host list for the server's
        mocked hello response; see test_connect_with_internal_ips.
        """
        self.mock_standalones = standalones[:]
        self.mock_members = members[:]

        if self.mock_members:
            self.mock_primary = self.mock_members[0]
        else:
            self.mock_primary = None

        # Hosts that should be considered an arbiter.
        self.mock_arbiters = arbiters[:] if arbiters else []

        if hello_hosts is not None:
            self.mock_hello_hosts = hello_hosts
        else:
            self.mock_hello_hosts = members[:]

        self.mock_mongoses = mongoses[:]

        # Hosts that should raise socket errors.
        self.mock_down_hosts = down_hosts[:] if down_hosts else []

        # Hostname -> (min wire version, max wire version)
        self.mock_wire_versions = {}

        # Hostname -> max write batch size
        self.mock_max_write_batch_sizes = {}

        # Hostname -> round trip time
        self.mock_rtts = {}

        kwargs["_pool_class"] = partial(MockPool, self)
        kwargs["_monitor_class"] = partial(SyncMockMonitor, self)

        client_options = client_context.default_client_options.copy()
        client_options.update(kwargs)

        super().__init__(*args, **client_options)

    @classmethod