def test_tag_sets_validation(self):
        # Can't use tags with PRIMARY
        self.assertRaises(ConfigurationError, _ServerMode,
                          0, tag_sets=[{'k': 'v'}])

        # ... but empty tag sets are ok with PRIMARY
        self.assertRaises(ConfigurationError, _ServerMode,
                          0, tag_sets=[{}])

        S = Secondary(tag_sets=[{}])
        self.assertEqual(
            [{}],
            rs_client(read_preference=S).read_preference.tag_sets)

        S = Secondary(tag_sets=[{'k': 'v'}])
        self.assertEqual(
            [{'k': 'v'}],
            rs_client(read_preference=S).read_preference.tag_sets)

        S = Secondary(tag_sets=[{'k': 'v'}, {}])
        self.assertEqual(
            [{'k': 'v'}, {}],
            rs_client(read_preference=S).read_preference.tag_sets)

        self.assertRaises(ValueError, Secondary, tag_sets=[])

        # One dict not ok, must be a list of dicts
        self.assertRaises(TypeError, Secondary, tag_sets={'k': 'v'})

        self.assertRaises(TypeError, Secondary, tag_sets='foo')

        self.assertRaises(TypeError, Secondary, tag_sets=['foo'])

    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_ipv6(self):
        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("[::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_threshold_validation(self):
        self.assertEqual(17, rs_client(
            localThresholdMS=17
        ).local_threshold_ms)

        self.assertEqual(42, rs_client(
            localThresholdMS=42
        ).local_threshold_ms)

        self.assertEqual(666, rs_client(
            localthresholdms=666
        ).local_threshold_ms)

    def test_commit_not_retried_after_timeout(self):
        listener = OvertCommandListener()
        client = rs_client(event_listeners=[listener])
        coll = client[self.db.name].test

        def callback(session):
            coll.insert_one({}, session=session)

        # Create the collection.
        coll.insert_one({})
        self.set_fail_point({
            'configureFailPoint': 'failCommand', 'mode': {'times': 2},
            'data': {
                'failCommands': ['commitTransaction'],
                'closeConnection': True}})
        self.addCleanup(self.set_fail_point, {
            'configureFailPoint': 'failCommand', 'mode': 'off'})
        listener.results.clear()

        with client.start_session() as s:
            with PatchSessionTimeout(0):
                with self.assertRaises(ConnectionFailure):
                    s.with_transaction(callback)

        # One insert for the callback and two commits (includes the automatic
        # retry).
        self.assertEqual(listener.started_command_names(),
                         ['insert', 'commitTransaction', 'commitTransaction'])

    def test_callback_not_retried_after_commit_timeout(self):
        listener = OvertCommandListener()
        client = rs_client(event_listeners=[listener])
        coll = client[self.db.name].test

        def callback(session):
            coll.insert_one({}, session=session)

        # Create the collection.
        coll.insert_one({})
        self.set_fail_point({
            'configureFailPoint': 'failCommand', 'mode': {'times': 1},
            'data': {
                'failCommands': ['commitTransaction'],
                'errorCode': 251,  # NoSuchTransaction
            }})
        self.addCleanup(self.set_fail_point, {
            'configureFailPoint': 'failCommand', 'mode': 'off'})
        listener.results.clear()

        with client.start_session() as s:
            with PatchSessionTimeout(0):
                with self.assertRaises(OperationFailure):
                    s.with_transaction(callback)

        self.assertEqual(listener.started_command_names(),
                         ['insert', 'commitTransaction'])

    def test_callback_not_retried_after_timeout(self):
        listener = OvertCommandListener()
        client = rs_client(event_listeners=[listener])
        coll = client[self.db.name].test

        def callback(session):
            coll.insert_one({}, session=session)
            err = {
                'ok': 0,
                'errmsg': 'Transaction 7819 has been aborted.',
                'code': 251,
                'codeName': 'NoSuchTransaction',
                'errorLabels': ['TransientTransactionError'],
            }
            raise OperationFailure(err['errmsg'], err['code'], err)

        # Create the collection.
        coll.insert_one({})
        listener.results.clear()
        with client.start_session() as s:
            with PatchSessionTimeout(0):
                with self.assertRaises(OperationFailure):
                    s.with_transaction(callback)

        self.assertEqual(listener.started_command_names(),
                         ['insert', 'abortTransaction'])

    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_timeout_does_not_mark_member_down(self):
        # If a query times out, the client shouldn't mark the member "down".

        # Disable background refresh.
        with client_knobs(heartbeat_frequency=999999):
            c = rs_client(socketTimeoutMS=3000, w=self.w)
            collection = c.pymongo_test.test
            collection.insert_one({})

            # Query the primary.
            self.assertRaises(
                NetworkTimeout,
                collection.find_one,
                {'$where': delay(5)})

            self.assertTrue(c.primary)
            collection.find_one()  # No error.

            coll = collection.with_options(
                read_preference=ReadPreference.SECONDARY)

            # Query the secondary.
            self.assertRaises(
                NetworkTimeout,
                coll.find_one,
                {'$where': delay(5)})

            self.assertTrue(c.secondaries)

            # No error.
            coll.find_one()

    def test_gridfs_replica_set(self):
        rsc = rs_client(w=self.w, wtimeout=5000, read_preference=ReadPreference.SECONDARY)

        fs = gridfs.GridFS(rsc.pymongo_test)
        oid = fs.put(b"foo")
        content = fs.get(oid).read()
        self.assertEqual(b"foo", content)

    def test_nearest(self):
        # With high localThresholdMS, expect to read from any
        # member
        c = rs_client(
            read_preference=ReadPreference.NEAREST,
            localThresholdMS=10000)  # 10 seconds

        data_members = set(self.hosts).difference(set(self.arbiters))

        # This is a probabilistic test; track which members we've read from so
        # far, and keep reading until we've used all the members or give up.
        # Chance of using only 2 of 3 members 10k times if there's no bug =
        # 3 * (2/3)**10000, very low.
        used = set()
        i = 0
        while data_members.difference(used) and i < 10000:
            address = self.read_from_which_host(c)
            used.add(address)
            i += 1

        not_used = data_members.difference(used)
        latencies = ', '.join(
            '%s: %dms' % (server.description.address,
                          server.description.round_trip_time)
            for server in c._get_topology().select_servers(
                readable_server_selector))

        self.assertFalse(
            not_used,
            "Expected to use primary and all secondaries for mode NEAREST,"
            " but didn't use %s\nlatencies: %s" % (not_used, latencies))

    def test_gridfs_replica_set(self):
        rsc = rs_client(
            w=self.w,
            read_preference=ReadPreference.SECONDARY)

        gfs = gridfs.GridFSBucket(rsc.pymongo_test)
        oid = gfs.upload_from_stream("test_filename", b'foo')
        content = gfs.open_download_stream(oid).read()
        self.assertEqual(b'foo', content)

 def create_targets(self, *args, **kwargs):
     codec_options = kwargs.pop('codec_options', None)
     if codec_options:
         kwargs['type_registry'] = codec_options.type_registry
         kwargs['document_class'] = codec_options.document_class
     self.watched_target = rs_client(*args, **kwargs)
     self.input_target = self.watched_target[self.db.name].test
     # Insert a record to ensure db, coll are created.
     self.input_target.insert_one({'data': 'dummy'})

    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_threshold_validation(self):
        self.assertEqual(17, rs_client(
            localThresholdMS=17
        ).local_threshold_ms)

        self.assertEqual(42, rs_client(
            localThresholdMS=42
        ).local_threshold_ms)

        self.assertEqual(666, rs_client(
            localthresholdms=666
        ).local_threshold_ms)

        self.assertEqual(0, rs_client(
            localthresholdms=0
        ).local_threshold_ms)

        self.assertRaises(ValueError,
                          rs_client,
                          localthresholdms=-1)

    def test_properties(self):
        c = client_context.client
        c.admin.command('ping')

        wait_until(lambda: c.primary == self.primary, "discover primary")
        wait_until(lambda: c.arbiters == self.arbiters, "discover arbiters")
        wait_until(lambda: c.secondaries == self.secondaries,
                   "discover secondaries")

        self.assertEqual(c.primary, self.primary)
        self.assertEqual(c.secondaries, self.secondaries)
        self.assertEqual(c.arbiters, self.arbiters)
        self.assertEqual(c.max_pool_size, 100)

        # Make sure MongoClient's properties are copied to Database and
        # Collection.
        for obj in c, c.pymongo_test, c.pymongo_test.test:
            self.assertEqual(obj.codec_options, CodecOptions())
            self.assertEqual(obj.read_preference, ReadPreference.PRIMARY)
            self.assertEqual(obj.write_concern, WriteConcern())

        cursor = c.pymongo_test.test.find()
        self.assertEqual(
            ReadPreference.PRIMARY, cursor._Cursor__read_preference)

        tag_sets = [{'dc': 'la', 'rack': '2'}, {'foo': 'bar'}]
        secondary = Secondary(tag_sets=tag_sets)
        c = rs_client(
            maxPoolSize=25,
            document_class=SON,
            tz_aware=True,
            read_preference=secondary,
            localThresholdMS=77,
            j=True)

        self.assertEqual(c.max_pool_size, 25)

        for obj in c, c.pymongo_test, c.pymongo_test.test:
            self.assertEqual(obj.codec_options, CodecOptions(SON, True))
            self.assertEqual(obj.read_preference, secondary)
            self.assertEqual(obj.write_concern, WriteConcern(j=True))

        cursor = c.pymongo_test.test.find()
        self.assertEqual(
            secondary, cursor._Cursor__read_preference)

        nearest = Nearest(tag_sets=[{'dc': 'ny'}, {}])
        cursor = c.pymongo_test.get_collection(
            "test", read_preference=nearest).find()

        self.assertEqual(nearest, cursor._Cursor__read_preference)
        self.assertEqual(c.max_bson_size, 16777216)
        c.close()

    def test_mode_validation(self):
        for mode in (ReadPreference.PRIMARY,
                     ReadPreference.PRIMARY_PREFERRED,
                     ReadPreference.SECONDARY,
                     ReadPreference.SECONDARY_PREFERRED,
                     ReadPreference.NEAREST):
            self.assertEqual(
                mode,
                rs_client(read_preference=mode).read_preference)

        self.assertRaises(
            TypeError,
            rs_client, read_preference='foo')

    def test_gridfs_replica_set(self):
        rsc = rs_client(
            w=self.w,
            read_preference=ReadPreference.SECONDARY)

        fs = gridfs.GridFS(rsc.pymongo_test)

        gin = fs.new_file()
        self.assertEqual(gin._coll.read_preference, ReadPreference.PRIMARY)

        oid = fs.put(b'foo')
        content = fs.get(oid).read()
        self.assertEqual(b'foo', content)

    def test_transaction_write_concern_override(self):
        """Test txn overrides Client/Database/Collection write_concern."""
        client = rs_client(w=0)
        self.addCleanup(client.close)
        db = client.test
        coll = db.test
        coll.insert_one({})
        with client.start_session() as s:
            with s.start_transaction(write_concern=WriteConcern(w=1)):
                self.assertTrue(coll.insert_one({}, session=s).acknowledged)
                self.assertTrue(coll.insert_many(
                    [{}, {}], session=s).acknowledged)
                self.assertTrue(coll.bulk_write(
                    [InsertOne({})], session=s).acknowledged)
                self.assertTrue(coll.replace_one(
                    {}, {}, session=s).acknowledged)
                self.assertTrue(coll.update_one(
                    {}, {"$set": {"a": 1}}, session=s).acknowledged)
                self.assertTrue(coll.update_many(
                    {}, {"$set": {"a": 1}}, session=s).acknowledged)
                self.assertTrue(coll.delete_one({}, session=s).acknowledged)
                self.assertTrue(coll.delete_many({}, session=s).acknowledged)
                coll.find_one_and_delete({}, session=s)
                coll.find_one_and_replace({}, {}, session=s)
                coll.find_one_and_update({}, {"$set": {"a": 1}}, session=s)

        unsupported_txn_writes = [
            (client.drop_database, [db.name], {}),
            (db.create_collection, ['collection'], {}),
            (db.drop_collection, ['collection'], {}),
            (coll.drop, [], {}),
            (coll.map_reduce,
             ['function() {}', 'function() {}', 'output'], {}),
            (coll.rename, ['collection2'], {}),
            # Drop collection2 between tests of "rename", above.
            (coll.database.drop_collection, ['collection2'], {}),
            (coll.create_indexes, [[IndexModel('a')]], {}),
            (coll.create_index, ['a'], {}),
            (coll.drop_index, ['a_1'], {}),
            (coll.drop_indexes, [], {}),
            (coll.aggregate, [[{"$out": "aggout"}]], {}),
        ]
        for op in unsupported_txn_writes:
            op, args, kwargs = op
            with client.start_session() as s:
                kwargs['session'] = s
                s.start_transaction(write_concern=WriteConcern(w=1))
                with self.assertRaises(OperationFailure):
                    op(*args, **kwargs)
                s.abort_transaction()

    def test_tag_sets_validation(self):
        S = Secondary(tag_sets=[{}])
        self.assertEqual(
            [{}],
            rs_client(read_preference=S).read_preference.tag_sets)

        S = Secondary(tag_sets=[{'k': 'v'}])
        self.assertEqual(
            [{'k': 'v'}],
            rs_client(read_preference=S).read_preference.tag_sets)

        S = Secondary(tag_sets=[{'k': 'v'}, {}])
        self.assertEqual(
            [{'k': 'v'}, {}],
            rs_client(read_preference=S).read_preference.tag_sets)

        self.assertRaises(ValueError, Secondary, tag_sets=[])

        # One dict not ok, must be a list of dicts
        self.assertRaises(TypeError, Secondary, tag_sets={'k': 'v'})

        self.assertRaises(TypeError, Secondary, tag_sets='foo')

        self.assertRaises(TypeError, Secondary, tag_sets=['foo'])