Method mock_hello

test/asynchronous/pymongo_mocks.py:186–247

Return mock hello response (a dict) and round trip time.

(self, host)

        self.mock_max_write_batch_sizes[host] = size

    def mock_hello(self, host):
        """Return mock hello response (a dict) and round trip time."""
        if host in self.mock_wire_versions:
            min_wire_version, max_wire_version = self.mock_wire_versions[host]
        else:
            min_wire_version = common.MIN_SUPPORTED_WIRE_VERSION
            max_wire_version = common.MAX_SUPPORTED_WIRE_VERSION

        max_write_batch_size = self.mock_max_write_batch_sizes.get(
            host, common.MAX_WRITE_BATCH_SIZE
        )

        rtt = self.mock_rtts.get(host, 0)

        # host is like 'a:1'.
        if host in self.mock_down_hosts:
            raise NetworkTimeout("mock timeout")

        elif host in self.mock_standalones:
            response = {
                "ok": 1,
                HelloCompat.LEGACY_CMD: True,
                "minWireVersion": min_wire_version,
                "maxWireVersion": max_wire_version,
                "maxWriteBatchSize": max_write_batch_size,
            }
        elif host in self.mock_members:
            primary = host == self.mock_primary

            # Simulate a replica set member.
            response = {
                "ok": 1,
                HelloCompat.LEGACY_CMD: primary,
                "secondary": not primary,
                "setName": "rs",
                "hosts": self.mock_hello_hosts,
                "minWireVersion": min_wire_version,
                "maxWireVersion": max_wire_version,
                "maxWriteBatchSize": max_write_batch_size,
            }

            if self.mock_primary:
                response["primary"] = self.mock_primary

            if host in self.mock_arbiters:
                response["arbiterOnly"] = True
                response["secondary"] = False
        elif host in self.mock_mongoses:
            response = {
                "ok": 1,
                HelloCompat.LEGACY_CMD: True,
                "minWireVersion": min_wire_version,
                "maxWireVersion": max_wire_version,
                "msg": "isdbgrid",
                "maxWriteBatchSize": max_write_batch_size,
            }
        else:
            # In test_internal_ips(), we try to connect to a host listed