def test_backport_latency_validation(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertEqual(17, self._get_client(
                localThresholdMS=17
            ).secondary_acceptable_latency_ms)

            self.assertEqual(42, self._get_client(
                secondaryAcceptableLatencyMS=42
            ).local_threshold_ms)

            self.assertEqual(666, self._get_client(
                localThresholdMS=666
            ).local_threshold_ms)

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

            self.assertRaises(ConfigurationError,
                              self._get_client,
                              localthresholdms=-1)
        finally:
            ctx.exit()

    def test_client_read_preference(self):
        preferences = (
            Primary(), Secondary([{'dc': 'ny'}]), Nearest([{'dc': 'ny'}, {}]))
        tags = [{'foo': 'bar'}]
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            for pref in preferences:
                client = MongoClient(read_preference=pref, connect=False)
                self.assertEqual(client.read_preference, pref.mode)
                self.assertEqual(client.tag_sets, pref.tag_sets)

                client = MongoClient(
                    read_preference=pref, tag_sets=tags, connect=False)
                self.assertEqual(client.read_preference, pref.mode)
                self.assertEqual(client.tag_sets, pref.tag_sets)

                client = MongoReplicaSetClient(
                    read_preference=pref, replicaSet='foo', connect=False)
                self.assertEqual(client.read_preference, pref.mode)
                self.assertEqual(client.tag_sets, pref.tag_sets)

                client = MongoReplicaSetClient(
                    read_preference=pref, tag_sets=tags,
                    replicaSet='foo', connect=False)
                self.assertEqual(client.read_preference, pref.mode)
                self.assertEqual(client.tag_sets, pref.tag_sets)
        finally:
            ctx.exit()

    def test_backport_localthresholdms_uri(self):
        uri = "mongodb://%s:%s" % (host, port)
        lt_uri = "mongodb://%s:%d/?localThresholdMS=10" % (host, port)
        sl_uri = ("mongodb://%s:%d/?secondaryAcceptableLatencyMS=10" %
                  (host, port))
        lt_sl_uri = ("mongodb://%s:%d/?localThresholdMS=10;"
                     "secondaryAcceptableLatencyMS=8" % (host, port))

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            # Just localThresholdMS
            client = MongoClient(uri)
            self.assertEqual(client.secondary_acceptable_latency_ms, 15)
            self.assertEqual(client.local_threshold_ms, 15)
            client = MongoClient(uri, localThresholdMS=10)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(uri, localThresholdMS=10,
                                 secondaryAcceptableLatencyMS=8)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)

            # URI options take precedence over kwargs but localThresholdMS
            # takes precedence over secondaryAcceptableLatencyMS always. Test
            # to make sure the precedence is correct between URI vs. kwargs.
            client = MongoClient(lt_uri)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(lt_uri, localThresholdMS=8)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(lt_uri, secondaryAcceptableLatencyMS=8)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(lt_uri, localThresholdMS=8,
                                 secondaryAcceptableLatencyMS=6)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)

            client = MongoClient(sl_uri, secondaryAcceptableLatencyMS=8)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(sl_uri, localThresholdMS=10)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(sl_uri, localThresholdMS=10,
                                 secondaryAcceptableLatencyMS=6)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)

            client = MongoClient(lt_sl_uri)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
            client = MongoClient(lt_sl_uri, localThresholdMS=8,
                                 secondaryAcceptableLatencyMS=4)
            self.assertEqual(client.secondary_acceptable_latency_ms, 10)
            self.assertEqual(client.local_threshold_ms, 10)
        finally:
            ctx.exit()

    def test_base_object(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            c = self.client
            self.assertFalse(c.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(c.safe)
            self.assertEqual({}, c.get_lasterror_options())
            db = c.pymongo_test
            self.assertFalse(db.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(db.safe)
            self.assertEqual({}, db.get_lasterror_options())
            coll = db.test
            coll.drop()
            self.assertFalse(coll.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(coll.safe)
            self.assertEqual({}, coll.get_lasterror_options())
            cursor = coll.find()
            self.assertFalse(cursor._Cursor__slave_okay)
            self.assertTrue(bool(cursor._Cursor__read_preference))

            w = 1 + len(self.slaves)
            wtimeout=10000 # Wait 10 seconds for replication to complete
            c.set_lasterror_options(w=w, wtimeout=wtimeout)
            self.assertFalse(c.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(c.safe)
            self.assertEqual({'w': w, 'wtimeout': wtimeout}, c.get_lasterror_options())
            db = c.pymongo_test
            self.assertFalse(db.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(db.safe)
            self.assertEqual({'w': w, 'wtimeout': wtimeout}, db.get_lasterror_options())
            coll = db.test
            self.assertFalse(coll.slave_okay)
            self.assertTrue(bool(c.read_preference))
            self.assertTrue(coll.safe)
            self.assertEqual({'w': w, 'wtimeout': wtimeout},
                             coll.get_lasterror_options())
            cursor = coll.find()
            self.assertFalse(cursor._Cursor__slave_okay)
            self.assertTrue(bool(cursor._Cursor__read_preference))

            coll.insert({'foo': 'bar'})
            self.assertEqual(1, coll.find({'foo': 'bar'}).count())
            self.assertTrue(coll.find({'foo': 'bar'}))
            coll.remove({'foo': 'bar'})
            self.assertEqual(0, coll.find({'foo': 'bar'}).count())

            c.safe = False
            c.unset_lasterror_options()
            self.assertFalse(self.client.slave_okay)
            self.assertTrue(bool(self.client.read_preference))
            self.assertFalse(self.client.safe)
            self.assertEqual({}, self.client.get_lasterror_options())
        finally:
            ctx.exit()

    def test_init_disconnected(self):
        c = MongoClient(host, port, _connect=False)

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertIsInstance(c.is_primary, bool)
            self.assertIsInstance(c.is_mongos, bool)
            self.assertIsInstance(c.max_pool_size, int)
            self.assertIsInstance(c.use_greenlets, bool)
            self.assertIsInstance(c.nodes, frozenset)
            self.assertIsInstance(c.auto_start_request, bool)
            self.assertEqual(dict, c.get_document_class())
            self.assertIsInstance(c.tz_aware, bool)
            self.assertIsInstance(c.max_bson_size, int)
            self.assertIsInstance(c.min_wire_version, int)
            self.assertIsInstance(c.max_wire_version, int)
            self.assertIsInstance(c.max_write_batch_size, int)
            self.assertEqual(None, c.host)
            self.assertEqual(None, c.port)
        finally:
            ctx.exit()

        c.pymongo_test.test.find_one()  # Auto-connect.
        self.assertEqual((host, port), c.address)

        if version.at_least(c, (2, 5, 4, -1)):
            self.assertTrue(c.max_wire_version > 0)
        else:
            self.assertEqual(c.max_wire_version, 0)
        self.assertTrue(c.min_wire_version >= 0)

        bad_host = "somedomainthatdoesntexist.org"
        c = MongoClient(bad_host, port, connectTimeoutMS=1, _connect=False)
        self.assertRaises(ConnectionFailure, c.pymongo_test.test.find_one)

    def test_aggregate_command_with_out(self):
        if not version.at_least(self.c, (2, 5, 2)):
            raise SkipTest("Aggregation with $out requires MongoDB >= 2.5.2")

        # Tests aggregate command when pipeline contains $out.
        self.c.pymongo_test.test.insert({"x": 1, "y": 1}, w=self.w)
        self.c.pymongo_test.test.insert({"x": 1, "y": 2}, w=self.w)
        self.c.pymongo_test.test.insert({"x": 2, "y": 1}, w=self.w)
        self.c.pymongo_test.test.insert({"x": 2, "y": 2}, w=self.w)

        # Aggregate with $out always goes to primary, doesn't obey read prefs.
        # Test aggregate command sent directly to db.command.
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", UserWarning)
            self._test_fn(False, lambda: self.c.pymongo_test.command(
                "aggregate", "test",
                pipeline=[{"$match": {"x": 1}}, {"$out": "agg_out"}]
            ))

            # Test aggregate when sent through the collection aggregate function.
            self._test_fn(False, lambda: self.c.pymongo_test.test.aggregate(
                [{"$match": {"x": 2}}, {"$out": "agg_out"}]
            ))
        finally:
            ctx.exit()

        self.c.pymongo_test.drop_collection("test")
        self.c.pymongo_test.drop_collection("agg_out")

    def test_command_read_pref_warning(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("error", UserWarning)
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertRaises(
                UserWarning, self.c.pymongo_test.command, "ping", read_preference=ReadPreference.SECONDARY
            )
            try:
                self.c.pymongo_test.command("dbStats", read_preference=ReadPreference.SECONDARY_PREFERRED)
            except UserWarning:
                self.fail("Shouldn't have raised UserWarning.")

            primary = MongoClient(host, port)
            try:
                primary.pymongo_test.command("ping", read_preference=ReadPreference.SECONDARY_PREFERRED)
            except UserWarning:
                self.fail("Shouldn't have raised UserWarning.")

            secondary_addr = iter(self.c.secondaries).next()
            secondary = MongoClient(*secondary_addr)
            msclient = MasterSlaveConnection(primary, [secondary])
            self.assertRaises(
                UserWarning, msclient.pymongo_test.command, "ping", read_preference=ReadPreference.SECONDARY
            )
            try:
                msclient.pymongo_test.command("dbStats", read_preference=ReadPreference.SECONDARY_PREFERRED)
            except UserWarning:
                self.fail("Shouldn't have raised UserWarning.")
        finally:
            ctx.exit()

    def test_cursor_transfer(self):

        # This is just a test, don't try this at home...
        self.db.test.remove({})
        self.db.test.insert({'_id': i} for i in xrange(200))

        class CManager(CursorManager):
            def __init__(self, connection):
                super(CManager, self).__init__(connection)

            def close(self, dummy):
                # Do absolutely nothing...
                pass

        client = self.db.connection
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            client.set_cursor_manager(CManager)

            docs = []
            cursor = self.db.test.find().batch_size(10)
            docs.append(cursor.next())
            cursor.close()
            docs.extend(cursor)
            self.assertEqual(len(docs), 10)
            cmd_cursor = {'id': cursor.cursor_id, 'firstBatch': []}
            ccursor = CommandCursor(cursor.collection, cmd_cursor,
                                    cursor.conn_id, retrieved=cursor.retrieved)
            docs.extend(ccursor)
            self.assertEqual(len(docs), 200)
        finally:
            client.set_cursor_manager(CursorManager)
            ctx.exit()

    def test_document_class(self):
        c = MongoClient(host, port)
        db = c.pymongo_test
        db.test.insert({"x": 1})

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertEqual(dict, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), dict))
            self.assertFalse(isinstance(db.test.find_one(), SON))

            c.document_class = SON
            db = c.pymongo_test

            self.assertEqual(SON, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), SON))
            self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

            c = MongoClient(host, port, document_class=SON)
            db = c.pymongo_test

            self.assertEqual(SON, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), SON))
            self.assertFalse(isinstance(db.test.find_one(as_class=dict), SON))

            c.document_class = dict
            db = c.pymongo_test

            self.assertEqual(dict, c.document_class)
            self.assertTrue(isinstance(db.test.find_one(), dict))
            self.assertFalse(isinstance(db.test.find_one(), SON))
        finally:
            ctx.exit()

    def test_auto_start_request(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            for bad_horrible_value in (None, 5, 'hi!'):
                self.assertRaises(
                    (TypeError, ConfigurationError),
                    lambda: get_client(auto_start_request=bad_horrible_value)
                )

            # auto_start_request should default to False
            client = get_client()
            self.assertFalse(client.auto_start_request)

            client = get_client(auto_start_request=True)
            self.assertTrue(client.auto_start_request)

            # Assure we acquire a request socket.
            client.pymongo_test.test.find_one()
            self.assertTrue(client.in_request())
            pool = get_pool(client)
            self.assertRequestSocket(pool)
            self.assertSameSock(pool)

            client.end_request()
            self.assertNoRequest(pool)
            self.assertDifferentSock(pool)

            # Trigger auto_start_request
            client.pymongo_test.test.find_one()
            self.assertRequestSocket(pool)
            self.assertSameSock(pool)
        finally:
            ctx.exit()

    def test_drop_collection(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", UserWarning)
            self._test_fn(False, lambda: self.c.pymongo_test.drop_collection("some_collection"))

            self._test_fn(False, lambda: self.c.pymongo_test.some_collection.drop())
        finally:
            ctx.exit()

    def test_alive(self):
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertTrue(get_client().alive())

            client = MongoClient('doesnt exist', _connect=False)
            self.assertFalse(client.alive())
        finally:
            ctx.exit()

 def test_use_greenlets(self):
     ctx = catch_warnings()
     try:
         warnings.simplefilter("ignore", DeprecationWarning)
         self.assertFalse(MongoClient(host, port).use_greenlets)
         if thread_util.have_gevent:
             self.assertTrue(
                 MongoClient(
                     host, port, use_greenlets=True).use_greenlets)
     finally:
         ctx.exit()

 def test_primary_with_tags(self):
     # Tags not allowed with PRIMARY
     ctx = catch_warnings()
     try:
         warnings.simplefilter("ignore", DeprecationWarning)
         self.assertRaises(
             ConfigurationError,
             self._get_client,
             tag_sets=[{'dc': 'ny'}])
     finally:
         ctx.exit()

    def test_create_collection(self):
        # Collections should be created on primary, obviously
        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", UserWarning)
            self._test_fn(False, lambda: self.c.pymongo_test.command(
                'create', 'some_collection%s' % random.randint(0, sys.maxint)))

            self._test_fn(False, lambda: self.c.pymongo_test.create_collection(
                'some_collection%s' % random.randint(0, sys.maxint)))
        finally:
            ctx.exit()

 def test_command_read_pref_warning(self):
     ctx = catch_warnings()
     try:
         warnings.simplefilter("error", UserWarning)
         self.assertRaises(UserWarning, self.client.pymongo_test.command,
                           'ping', read_preference=ReadPreference.SECONDARY)
         try:
             self.client.pymongo_test.command('dbStats',
                 read_preference=ReadPreference.SECONDARY_PREFERRED)
         except UserWarning:
             self.fail("Shouldn't have raised UserWarning.")
     finally:
         ctx.exit()

    def test_from_uri(self):
        c = MongoClient(host, port)

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            self.assertEqual(c, MongoClient("mongodb://%s:%d" % (host, port)))
            self.assertTrue(MongoClient(
                "mongodb://%s:%d" % (host, port), slave_okay=True).slave_okay)
            self.assertTrue(MongoClient(
                "mongodb://%s:%d/?slaveok=true;w=2" % (host, port)).slave_okay)
        finally:
            ctx.exit()

    def test_map_reduce(self):
        # mapreduce fails if no collection
        self.c.pymongo_test.test.insert({}, w=self.w)

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", UserWarning)
            self._test_fn(False, lambda: self.c.pymongo_test.test.map_reduce(
                'function() { }', 'function() { }', 'mr_out'))
        finally:
            ctx.exit()

        self._test_fn(True, lambda: self.c.pymongo_test.test.map_reduce(
            'function() { }', 'function() { }', {'inline': 1}))

    def test_last_status(self):
        db = self.client.pymongo_test

        db.test.remove({})
        db.test.save({"i": 1})

        ctx = catch_warnings()
        try:
            warnings.simplefilter("ignore", DeprecationWarning)
            db.test.update({"i": 1}, {"$set": {"i": 2}}, w=0)
            self.assertTrue(db.last_status()["updatedExisting"])

            db.test.update({"i": 1}, {"$set": {"i": 500}}, w=0)
            self.assertFalse(db.last_status()["updatedExisting"])
        finally:
            ctx.exit()

    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()