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

        # MongoClient connects to primary by default.
        wait_until(lambda: c.address is not None, 'connect to primary')
        self.assertEqual(c.address, ('a', 1))

        # 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.close()

        with self.assertRaises(AutoReconnect):
            c.db.command('ismaster')

        self.assertEqual(c.address, None)

    def test_ipv6(self):
        if client_context.ssl:
            # http://bugs.python.org/issue13034
            if sys.version_info[:2] == (2, 6):
                raise SkipTest("Python 2.6 can't parse SANs")
            if not HAVE_IPADDRESS:
                raise SkipTest("Need the ipaddress module to test with SSL")

        port = client_context.port
        c = rs_client("mongodb://[::1]:%d" % (port,))

        # Client switches to IPv4 once it has first ismaster response.
        msg = 'discovered primary with IPv4 address "%r"' % (self.primary,)
        wait_until(lambda: c.primary == self.primary, msg)

        # Same outcome with both IPv4 and IPv6 seeds.
        c = rs_client("mongodb://[::1]:%d,localhost:%d" % (port, port))

        wait_until(lambda: c.primary == self.primary, msg)

        if client_context.auth_enabled:
            auth_str = "%s:%[email protected]" % (db_user, db_pwd)
        else:
            auth_str = ""

        uri = "mongodb://%slocalhost:%d,[::1]:%d" % (auth_str, port, port)
        client = rs_client(uri)
        client.pymongo_test.test.insert_one({"dummy": u"object"})
        client.pymongo_test_bernie.test.insert_one({"dummy": u"object"})

        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)
        self.assertTrue("pymongo_test_bernie" in dbs)
        client.close()

    def test_kill_cursors_with_tuple(self):
        if (client_context.is_mongos
                and not client_context.version.at_least(2, 4, 7)):
            # Old mongos sends incorrectly formatted error response when
            # cursor isn't found, see SERVER-9738.
            raise SkipTest("Can't test kill_cursors against old mongos")

        self.collection = self.client.pymongo_test.test
        self.collection.drop()

        self.collection.insert_many([{'_id': i} for i in range(200)])
        cursor = self.collection.find().batch_size(1)
        next(cursor)
        self.client.kill_cursors(
            [cursor.cursor_id],
            self.client.address)

        # Prevent killcursors from reaching the server while a getmore is in
        # progress -- the server logs "Assertion: 16089:Cannot kill active
        # cursor."
        time.sleep(2)

        def raises_cursor_not_found():
            try:
                next(cursor)
                return False
            except CursorNotFound:
                return True

        wait_until(raises_cursor_not_found, 'close cursor')

    def create_mock_monitor(self, responses, uri, expected_results):
        with client_knobs(heartbeat_frequency=0.1, events_queue_frequency=0.1):
            class MockMonitor(Monitor):
                def _check_with_socket(self, sock_info):
                    if isinstance(responses[1], Exception):
                        raise responses[1]
                    return IsMaster(responses[1]), 99

            m = single_client(h=uri,
                              event_listeners=(self.all_listener,),
                              _monitor_class=MockMonitor,
                              _pool_class=MockPool
                              )

            expected_len = len(expected_results)
            wait_until(lambda: len(self.all_listener.results) == expected_len,
                       "publish all events", timeout=15)

        try:
            for i in range(len(expected_results)):
                result = self.all_listener.results[i] if len(
                    self.all_listener.results) > i else None
                self.assertEqual(expected_results[i],
                                 result.__class__.__name__)
                self.assertEqual(result.connection_id,
                                 responses[0])
                if expected_results[i] != 'ServerHeartbeatStartedEvent':
                    if isinstance(result.reply, IsMaster):
                        self.assertEqual(result.duration, 99)
                        self.assertEqual(result.reply._doc, responses[1])
                    else:
                        self.assertEqual(result.reply, responses[1])

        finally:
            m.close()

    def test_failover(self):
        nthreads = 10
        client = connected(self.mock_client(localThresholdMS=0.001))
        wait_until(lambda: len(client.nodes) == 3, 'connect to all mongoses')

        # Our chosen mongos goes down.
        client.kill_host('a:1')

        # Trigger failover to higher-latency nodes. AutoReconnect should be
        # raised at most once in each thread.
        passed = []

        def f():
            try:
                client.db.command('ismaster')
            except AutoReconnect:
                # Second attempt succeeds.
                client.db.command('ismaster')

            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(nthreads, len(passed))

        # Down host removed from list.
        self.assertEqual(2, len(client.nodes))

    def test_local_threshold(self):
        client = connected(self.mock_client(localThresholdMS=30))
        self.assertEqual(30, client.local_threshold_ms)
        wait_until(lambda: len(client.nodes) == 3, 'connect to all mongoses')
        topology = client._topology

        # All are within a 30-ms latency window, see self.mock_client().
        self.assertEqual(set([('a', 1), ('b', 2), ('c', 3)]),
                         writable_addresses(topology))

        # No error
        client.admin.command('ismaster')

        client = connected(self.mock_client(localThresholdMS=0))
        self.assertEqual(0, client.local_threshold_ms)
        # No error
        client.db.command('ismaster')
        # Our chosen mongos goes down.
        client.kill_host('%s:%s' % next(iter(client.nodes)))
        try:
            client.db.command('ismaster')
        except:
            pass

        # We eventually connect to a new mongos.
        def connect_to_new_mongos():
            try:
                return client.db.command('ismaster')
            except AutoReconnect:
                pass
        wait_until(connect_to_new_mongos, 'connect to a new mongos')

    def test_local_threshold(self):
        client = connected(self.mock_client(localThresholdMS=30))
        self.assertEqual(30, client.local_threshold_ms)
        wait_until(lambda: len(client.nodes) == 3, 'connect to all mongoses')
        topology = client._topology

        # All are within a 30-ms latency window, see self.mock_client().
        self.assertEqual(set([('a', 1), ('b', 2), ('c', 3)]),
                         writable_addresses(topology))

        # No error
        client.db.collection.find_one()

        client = connected(self.mock_client(localThresholdMS=0))
        self.assertEqual(0, client.local_threshold_ms)
        # No error
        client.db.collection.find_one()
        # Our chosen mongos goes down.
        client.kill_host('%s:%s' % next(iter(client.nodes)))
        try:
            client.db.collection.find_one()
        except:
            pass
        # No error
        client.db.collection.find_one()

    def test_ipv6(self):
        c = MongoClient("mongodb://[::1]:%d" % (port,), replicaSet=self.name)

        # Client switches to IPv4 once it has first ismaster response.
        msg = 'discovered primary with IPv4 address "%r"' % (self.primary,)
        wait_until(lambda: c.primary == self.primary, msg)

        # Same outcome with both IPv4 and IPv6 seeds.
        c = MongoClient("[::1]:%d,localhost:%d" % (port, port),
                        replicaSet=self.name)

        wait_until(lambda: c.primary == self.primary, msg)

        if client_context.auth_enabled:
            auth_str = "%s:%[email protected]" % (db_user, db_pwd)
        else:
            auth_str = ""

        uri = "mongodb://%slocalhost:%d,[::1]:%d" % (auth_str, port, port)
        client = MongoClient(uri, replicaSet=self.name)
        client.pymongo_test.test.insert_one({"dummy": u("object")})
        client.pymongo_test_bernie.test.insert_one({"dummy": u("object")})

        dbs = client.database_names()
        self.assertTrue("pymongo_test" in dbs)
        self.assertTrue("pymongo_test_bernie" in dbs)
        client.close()

    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_unpin_for_non_transaction_operation(self):
        # Increase localThresholdMS and wait until both nodes are discovered
        # to avoid false positives.
        client = rs_client(client_context.mongos_seeds(),
                           localThresholdMS=1000)
        wait_until(lambda: len(client.nodes) > 1, "discover both mongoses")
        coll = client.test.test
        # Create the collection.
        coll.insert_one({})
        self.addCleanup(client.close)
        with client.start_session() as s:
            # Session is pinned to Mongos.
            with s.start_transaction():
                coll.insert_one({}, session=s)

            addresses = set()
            for _ in range(UNPIN_TEST_MAX_ATTEMPTS):
                cursor = coll.find({}, session=s)
                self.assertTrue(next(cursor))
                addresses.add(cursor.address)
                # Break early if we can.
                if len(addresses) > 1:
                    break

            self.assertGreater(len(addresses), 1)

    def test_read_with_failover(self):
        c = MongoClient(
            self.seed,
            replicaSet=self.name,
            serverSelectionTimeoutMS=self.server_selection_timeout)
        wait_until(lambda: c.primary, "discover primary")
        wait_until(lambda: len(c.secondaries) == 2, "discover secondaries")

        def iter_cursor(cursor):
            for _ in cursor:
                pass
            return True

        w = len(c.secondaries) + 1
        db = c.get_database("pymongo_test",
                            write_concern=WriteConcern(w=w))
        db.test.delete_many({})
        # Force replication
        db.test.insert_many([{'foo': i} for i in xrange(10)])
        self.assertEqual(10, db.test.count())

        db.read_preference = SECONDARY_PREFERRED
        cursor = db.test.find().batch_size(5)
        next(cursor)
        self.assertEqual(5, cursor._Cursor__retrieved)
        self.assertTrue(cursor.address in c.secondaries)
        ha_tools.kill_primary()
        # Primary failure shouldn't interrupt the cursor
        self.assertTrue(iter_cursor(cursor))
        self.assertEqual(10, cursor._Cursor__retrieved)

    def test_load_balancing(self):
        # Although the server selection JSON tests already prove that
        # select_servers works for sharded topologies, here we do an end-to-end
        # test of discovering servers' round trip times and configuring
        # localThresholdMS.
        client = connected(self.mock_client())
        wait_until(lambda: len(client.nodes) == 3, 'connect to all mongoses')

        # Prohibited for topology type Sharded.
        with self.assertRaises(InvalidOperation):
            client.address

        topology = client._topology
        self.assertEqual(TOPOLOGY_TYPE.Sharded,
                         topology.description.topology_type)

        # a and b are within the 15-ms latency window, see self.mock_client().
        self.assertEqual(set([('a', 1), ('b', 2)]),
                         writable_addresses(topology))

        client.mock_rtts['a:1'] = 0.040

        # Discover only b is within latency window.
        wait_until(lambda: set([('b', 2)]) == writable_addresses(topology),
                   'discover server "a" is too far')

    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 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_local_threshold(self):
        client = connected(self.mock_client(localThresholdMS=30))
        wait_until(lambda: len(client.nodes) == 3, 'connect to all mongoses')
        topology = client._topology

        # All are within a 30-ms latency window, see self.mock_client().
        self.assertEqual(set([('a', 1), ('b', 2), ('c', 3)]),
                         writable_addresses(topology))

    def test_bool(self):
        client = single_client()

        wait_until(lambda: client.address, "discover primary")
        selection = Selection.from_topology_description(
            client._topology.description)

        self.assertTrue(selection)
        self.assertFalse(selection.with_server_descriptions([]))

    def assertReadsFrom(self, expected, **kwargs):
        c = rs_client(**kwargs)
        wait_until(
            lambda: len(c.nodes - c.arbiters) == self.w,
            "discovered all nodes")

        used = self.read_from_which_kind(c)
        self.assertEqual(expected, used, 'Cursor used %s, expected %s' % (
            used, expected))

    def test_no_results_unordered_success(self):

        batch = self.coll.initialize_unordered_bulk_op()
        batch.insert({'_id': 1})
        batch.find({'_id': 3}).upsert().update_one({'$set': {'b': 1}})
        batch.insert({'_id': 2})
        batch.find({'_id': 1}).remove_one()
        self.assertTrue(batch.execute({'w': 0}) is None)
        wait_until(lambda: 2 == self.coll.count(),
                   'insert 2 documents')

    def test_insert_large_batch(self):
        # Tests legacy insert.
        db = self.client.test_insert_large_batch
        self.addCleanup(self.client.drop_database, 'test_insert_large_batch')
        max_bson_size = self.client.max_bson_size
        if client_context.version.at_least(2, 5, 4, -1):
            # Write commands are limited to 16MB + 16k per batch
            big_string = 'x' * int(max_bson_size / 2)
        else:
            big_string = 'x' * (max_bson_size - 100)

        # Batch insert that requires 2 batches.
        successful_insert = [{'x': big_string}, {'x': big_string},
                             {'x': big_string}, {'x': big_string}]
        db.collection_0.insert(successful_insert, w=1)
        self.assertEqual(4, db.collection_0.count())

        # Test that inserts fail after first error.
        insert_second_fails = [{'_id': 'id0', 'x': big_string},
                               {'_id': 'id0', 'x': big_string},
                               {'_id': 'id1', 'x': big_string},
                               {'_id': 'id2', 'x': big_string}]

        with self.assertRaises(DuplicateKeyError):
            db.collection_1.insert(insert_second_fails)

        self.assertEqual(1, db.collection_1.count())

        # 2 batches, 2nd insert fails, don't continue on error.
        self.assertTrue(db.collection_2.insert(insert_second_fails, w=0))
        wait_until(lambda: 1 == db.collection_2.count(),
                   'insert 1 document', timeout=60)

        # 2 batches, ids of docs 0 and 1 are dupes, ids of docs 2 and 3 are
        # dupes. Acknowledged, continue on error.
        insert_two_failures = [{'_id': 'id0', 'x': big_string},
                               {'_id': 'id0', 'x': big_string},
                               {'_id': 'id1', 'x': big_string},
                               {'_id': 'id1', 'x': big_string}]

        with self.assertRaises(OperationFailure) as context:
            db.collection_3.insert(insert_two_failures,
                                   continue_on_error=True, w=1)

        self.assertIn('id1', str(context.exception))

        # Only the first and third documents should be inserted.
        self.assertEqual(2, db.collection_3.count())

        # 2 batches, 2 errors, unacknowledged, continue on error.
        db.collection_4.insert(insert_two_failures, continue_on_error=True, w=0)

        # Only the first and third documents are inserted.
        wait_until(lambda: 2 == db.collection_4.count(),
                   'insert 2 documents', timeout=60)