def test_acceptable_latency(self):
        client = MockClient(
            standalones=[],
            members=[],
            mongoses=['a:1', 'b:2', 'c:3'],
            host='a:1,b:2,c:3',
            secondaryAcceptableLatencyMS=7)

        self.assertEqual(7, client.secondary_acceptable_latency_ms)
        # No error
        client.db.collection.find_one()

        client = MockClient(
            standalones=[],
            members=[],
            mongoses=['a:1', 'b:2', 'c:3'],
            host='a:1,b:2,c:3',
            secondaryAcceptableLatencyMS=0)

        self.assertEqual(0, client.secondary_acceptable_latency_ms)
        # No error
        client.db.collection.find_one()
        # Our chosen mongos goes down.
        client.kill_host('%s:%s' % (client.host, client.port))
        try:
            client.db.collection.find_one()
        except:
            pass
        # No error
        client.db.collection.find_one()

    def test_client(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2'],
            mongoses=[],
            host='a:1',
            replicaSet='rs')

        wait_until(lambda: len(c.nodes) == 2, 'discover both nodes')

        # MongoClient connects to primary by default.
        self.assertEqual(c.address, ('a', 1))
        self.assertEqual(set([('a', 1), ('b', 2)]), c.nodes)

        # C is added.
        c.mock_members.append('c:3')
        c.mock_ismaster_hosts.append('c:3')

        c.close()
        c.db.command('ismaster')

        self.assertEqual(c.address, ('a', 1))

        wait_until(lambda: set([('a', 1), ('b', 2), ('c', 3)]) == c.nodes,
                   'reconnect to both secondaries')

    def test_client(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='a:1,b:2,c:3',
            replicaSet='rs')

        # MongoClient connects to primary by default.
        self.assertEqual('a', c.host)
        self.assertEqual(1, c.port)

        # C is brought up as a standalone.
        c.mock_members.remove('c:3')
        c.mock_standalones.append('c:3')

        # Fail over.
        c.kill_host('a:1')
        c.kill_host('b:2')

        # Force reconnect.
        c.disconnect()

        try:
            c.db.collection.find_one()
        except ConfigurationError, e:
            self.assertTrue('not a member of replica set' in str(e))

    def test_wire_version(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='a:1',
            replicaSet='rs',
            connect=False)

        c.set_wire_version_range('a:1', 1, 5)
        c.set_wire_version_range('b:2', 0, 1)
        c.set_wire_version_range('c:3', 1, 2)
        c.db.command('ismaster')  # Connect.

        c.set_wire_version_range('a:1', 2, 2)

        # A secondary doesn't overlap with us.
        c.set_wire_version_range('b:2', 5, 6)

        def raises_configuration_error():
            try:
                c.db.collection.find_one()
                return False
            except ConfigurationError:
                return True

        wait_until(raises_configuration_error,
                   'notice we are incompatible with server')

        self.assertRaises(ConfigurationError, c.db.collection.insert_one, {})

    def _test_network_error(self, operation_callback):
        # Verify only the disconnected server is reset by a network failure.

        # Disable background refresh.
        with client_knobs(heartbeat_frequency=999999):
            c = MockClient(
                standalones=[], members=["a:1", "b:2"], mongoses=[], host="a:1", replicaSet="rs", connect=False
            )

            # Set host-specific information so we can test whether it is reset.
            c.set_wire_version_range("a:1", 0, 1)
            c.set_wire_version_range("b:2", 0, 2)
            c._get_topology().select_servers(writable_server_selector)
            wait_until(lambda: len(c.nodes) == 2, "connect")

            c.kill_host("a:1")

            # MongoClient is disconnected from the primary.
            self.assertRaises(AutoReconnect, operation_callback, c)

            # The primary's description is reset.
            server_a = c._get_topology().get_server_by_address(("a", 1))
            sd_a = server_a.description
            self.assertEqual(SERVER_TYPE.Unknown, sd_a.server_type)
            self.assertEqual(0, sd_a.min_wire_version)
            self.assertEqual(0, sd_a.max_wire_version)

            # ...but not the secondary's.
            server_b = c._get_topology().get_server_by_address(("b", 2))
            sd_b = server_b.description
            self.assertEqual(SERVER_TYPE.RSSecondary, sd_b.server_type)
            self.assertEqual(0, sd_b.min_wire_version)
            self.assertEqual(2, sd_b.max_wire_version)

    def mock_client(self, **kwargs):
        mock_client = MockClient(
            standalones=[],
            members=[],
            mongoses=['a:1', 'b:2', 'c:3'],
            host='a:1,b:2,c:3',
            connect=False,
            **kwargs)

        # Latencies in seconds.
        mock_client.mock_rtts['a:1'] = 0.020
        mock_client.mock_rtts['b:2'] = 0.025
        mock_client.mock_rtts['c:3'] = 0.040
        return mock_client

    def test_discover_primary(self):
        # Disable background refresh.
        with client_knobs(heartbeat_frequency=999999):
            c = MockClient(
                standalones=[],
                members=["a:1", "b:2", "c:3"],
                mongoses=[],
                host="b:2",  # Pass a secondary.
                replicaSet="rs",
            )

            wait_until(lambda: len(c.nodes) == 3, "connect")
            self.assertEqual(c.address, ("a", 1))

            # Fail over.
            c.kill_host("a:1")
            c.mock_primary = "b:2"

            c.close()
            self.assertEqual(0, len(c.nodes))

            t = c._get_topology()
            t.select_servers(writable_server_selector)  # Reconnect.
            self.assertEqual(c.address, ("b", 2))

            # a:1 not longer in nodes.
            self.assertLess(len(c.nodes), 3)

            # c:3 is rediscovered.
            t.select_server_by_address(("c", 3))

    def test_discover_primary(self):
        # Disable background refresh.
        with client_knobs(heartbeat_frequency=999999):
            c = MockClient(
                standalones=[],
                members=['a:1', 'b:2', 'c:3'],
                mongoses=[],
                host='b:2',  # Pass a secondary.
                replicaSet='rs')

            wait_until(lambda: len(c.nodes) == 3, 'connect')
            self.assertEqual(c.address, ('a', 1))

            # Fail over.
            c.kill_host('a:1')
            c.mock_primary = 'b:2'

            c.close()
            self.assertEqual(0, len(c.nodes))

            t = c._get_topology()
            t.select_servers(writable_server_selector)  # Reconnect.
            self.assertEqual(c.address, ('b', 2))

            # a:1 not longer in nodes.
            self.assertLess(len(c.nodes), 3)

            # c:3 is rediscovered.
            t.select_server_by_address(('c', 3))

    def test_failover(self):
        nthreads = 1

        # ['1:1', '2:2', '3:3', ...]
        mock_hosts = ['%d:%d' % (i, i) for i in range(50)]
        client = MockClient(
            standalones=[],
            members=[],
            mongoses=mock_hosts,
            host=','.join(mock_hosts))

        self.assertEqual(len(mock_hosts), len(client.nodes))

        # Our chosen mongos goes down.
        client.kill_host('%s:%s' % (client.host, client.port))

        # Trigger failover. AutoReconnect should be raised exactly once.
        errors = []
        passed = []

        def f():
            try:
                client.db.collection.find_one()
            except AutoReconnect:
                errors.append(True)

                # Second attempt succeeds.
                client.db.collection.find_one()

            passed.append(True)

        threads = [threading.Thread(target=f) for _ in range(nthreads)]
        for t in threads:
            t.start()

        for t in threads:
            t.join()

        self.assertEqual(1, len(errors))
        self.assertEqual(nthreads, len(passed))

        # Down host is still in list.
        self.assertEqual(len(mock_hosts), len(client.nodes))

    def test_max_write_batch_size(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2'],
            mongoses=[],
            host='a:1',
            replicaSet='rs',
            connect=False)

        c.set_max_write_batch_size('a:1', 1)
        c.set_max_write_batch_size('b:2', 2)

        # Uses primary's max batch size.
        self.assertEqual(c.max_write_batch_size, 1)

        # b becomes primary.
        c.mock_primary = 'b:2'
        wait_until(lambda: c.max_write_batch_size == 2,
                   'update max_write_batch_size')

    def test_backport_localthresholdms_kwarg(self):
        # Test that localThresholdMS takes precedence over
        # secondaryAcceptableLatencyMS.
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            client = MockClient(
                standalones=[],
                members=[],
                mongoses=['a:1', 'b:2', 'c:3'],
                host='a:1,b:2,c:3',
                localThresholdMS=7,
                secondaryAcceptableLatencyMS=0)

            self.assertEqual(7, client.secondary_acceptable_latency_ms)
            self.assertEqual(7, client.local_threshold_ms)
            # No error
            client.db.collection.find_one()

            client = MockClient(
                standalones=[],
                members=[],
                mongoses=['a:1', 'b:2', 'c:3'],
                host='a:1,b:2,c:3',
                localThresholdMS=0,
                secondaryAcceptableLatencyMS=15)

            self.assertEqual(0, client.secondary_acceptable_latency_ms)
            self.assertEqual(0, client.local_threshold_ms)

            # Test that using localThresholdMS works in the same way as using
            # secondaryAcceptableLatencyMS.
            client.db.collection.find_one()
            # Our chosen mongos goes down.
            client.kill_host('%s:%s' % client.address)
            try:
                client.db.collection.find_one()
            except:
                pass
            # No error
            client.db.collection.find_one()
        finally:
            ctx.exit()

    def test_client(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2'],
            mongoses=[],
            host='a:1',
            replicaSet='rs')

        # MongoClient connects to primary by default.
        self.assertEqual(('a', 1), c.address)
        self.assertEqual(set([('a', 1), ('b', 2)]), c.nodes)

        # C is added.
        c.mock_members.append('c:3')
        c.mock_ismaster_hosts.append('c:3')

        c.close()
        c.db.collection.find_one()

        self.assertEqual(('a', 1), c.address)
        self.assertEqual(set([('a', 1), ('b', 2), ('c', 3)]), c.nodes)

    def test_reconnect(self):
        # Verify the node list isn't forgotten during a network failure.
        c = MockClient(
            standalones=[], members=["a:1", "b:2", "c:3"], mongoses=[], host="b:2", replicaSet="rs"  # Pass a secondary.
        )

        wait_until(lambda: len(c.nodes) == 3, "connect")

        # Total failure.
        c.kill_host("a:1")
        c.kill_host("b:2")
        c.kill_host("c:3")

        # MongoClient discovers it's alone.
        self.assertRaises(AutoReconnect, c.db.collection.find_one)

        # But it can reconnect.
        c.revive_host("a:1")
        c._get_topology().select_servers(writable_server_selector)
        self.assertEqual(c.address, ("a", 1))

    def test_reconnect(self):
        # Verify the node list isn't forgotten during a network failure.
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='b:2',  # Pass a secondary.
            replicaSet='rs')

        wait_until(lambda: len(c.nodes) == 3, 'connect')

        # Total failure.
        c.kill_host('a:1')
        c.kill_host('b:2')
        c.kill_host('c:3')

        # MongoClient discovers it's alone.
        self.assertRaises(AutoReconnect, c.db.collection.find_one)

        # But it can reconnect.
        c.revive_host('a:1')
        c._get_topology().select_servers(writable_server_selector)
        self.assertEqual(c.address, ('a', 1))

    def test_wire_version(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='b:2',  # Pass a secondary.
            replicaSet='rs',
            _connect=False)

        c.set_wire_version_range('a:1', 1, 5)
        c.db.collection.find_one()  # Connect.
        self.assertEqual(c.min_wire_version, 1)
        self.assertEqual(c.max_wire_version, 5)

        c.set_wire_version_range('a:1', 10, 11)
        c.disconnect()
        self.assertRaises(ConfigurationError, c.db.collection.find_one)

    def test_reconnect(self):
        # Verify the node list isn't forgotten during a network failure.
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='b:2',  # Pass a secondary.
            replicaSet='rs')

        # Total failure.
        c.kill_host('a:1')
        c.kill_host('b:2')
        c.kill_host('c:3')

        # MongoClient discovers it's alone.
        self.assertRaises(AutoReconnect, c.db.collection.find_one)

        # But it remembers its node list.
        self.assertEqual(3, len(c.nodes))

        # So it can reconnect.
        c.revive_host('a:1')
        c.db.collection.find_one()

    def test_discover_primary(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='b:2',  # Pass a secondary.
            replicaSet='rs')

        self.assertEqual(('a', 1), c.address)
        self.assertEqual(3, len(c.nodes))

        # Fail over.
        c.kill_host('a:1')
        c.mock_primary = 'b:2'

        # Force reconnect.
        c.close()
        c.db.collection.find_one()
        self.assertEqual(('b', 2), c.address)

        # a:1 is still in nodes.
        self.assertEqual(3, len(c.nodes))

    def test_max_wire_version(self):
        c = MockClient(
            standalones=[],
            members=['a:1', 'b:2', 'c:3'],
            mongoses=[],
            host='b:2',  # Pass a secondary.
            replicaSet='rs',
            _connect=False)

        c.set_max_write_batch_size('a:1', 1)
        c.set_max_write_batch_size('b:2', 2)

        # Starts with default max batch size.
        self.assertEqual(1000, c.max_write_batch_size)
        c.db.collection.find_one()  # Connect.
        # Uses primary's max batch size.
        self.assertEqual(c.max_write_batch_size, 1)

        # b becomes primary.
        c.mock_primary = 'b:2'
        c.close()
        self.assertEqual(1000, c.max_write_batch_size)
        c.db.collection.find_one()  # Connect.
        self.assertEqual(c.max_write_batch_size, 2)

    def test_wire_version_mongos_ha(self):
        # TODO: Reimplement Mongos HA with PyMongo 3's MongoClient.
        raise SkipTest('Mongos HA must be reimplemented in PyMongo 3')

        c = MockClient(
            standalones=[],
            members=[],
            mongoses=['a:1', 'b:2', 'c:3'],
            host='a:1,b:2,c:3',
            connect=False)

        c.set_wire_version_range('a:1', 2, 5)
        c.set_wire_version_range('b:2', 2, 2)
        c.set_wire_version_range('c:3', 1, 1)
        c.db.command('ismaster')  # Connect.

        # Which member did we use?
        used_host = '%s:%s' % c.address
        expected_min, expected_max = c.mock_wire_versions[used_host]
        self.assertEqual(expected_min, c.min_wire_version)
        self.assertEqual(expected_max, c.max_wire_version)

        c.set_wire_version_range('a:1', 0, 0)
        c.set_wire_version_range('b:2', 0, 0)
        c.set_wire_version_range('c:3', 0, 0)
        c.close()
        c.db.command('ismaster')
        used_host = '%s:%s' % c.address
        expected_min, expected_max = c.mock_wire_versions[used_host]
        self.assertEqual(expected_min, c.min_wire_version)
        self.assertEqual(expected_max, c.max_wire_version)

    def test_wire_version_mongos_ha(self):
        c = MockClient(
            standalones=[],
            members=[],
            mongoses=['a:1', 'b:2', 'c:3'],
            host='a:1,b:2,c:3',
            _connect=False)

        c.set_wire_version_range('a:1', 2, 5)
        c.set_wire_version_range('b:2', 2, 2)
        c.set_wire_version_range('c:3', 1, 1)
        c.db.collection.find_one()  # Connect.

        # Which member did we use?
        used_host = '%s:%s' % (c.host, c.port)
        expected_min, expected_max = c.mock_wire_versions[used_host]
        self.assertEqual(expected_min, c.min_wire_version)
        self.assertEqual(expected_max, c.max_wire_version)

        c.set_wire_version_range('a:1', 0, 0)
        c.set_wire_version_range('b:2', 0, 0)
        c.set_wire_version_range('c:3', 0, 0)
        c.disconnect()
        c.db.collection.find_one()
        used_host = '%s:%s' % (c.host, c.port)
        expected_min, expected_max = c.mock_wire_versions[used_host]
        self.assertEqual(expected_min, c.min_wire_version)
        self.assertEqual(expected_max, c.max_wire_version)