def test_auth_from_uri(self):
        self.client.admin.add_user("admin", "pass", roles=["root"])
        self.addCleanup(self.client.admin.remove_user, "admin")
        self.addCleanup(remove_all_users, self.client.pymongo_test)

        self.client.pymongo_test.add_user("user", "pass", roles=["userAdmin", "readWrite"])

        with self.assertRaises(OperationFailure):
            connected(rs_or_single_client("mongodb://a:[email protected]%s:%d" % (host, port)))

        # No error.
        connected(rs_or_single_client_noauth("mongodb://admin:[email protected]%s:%d" % (host, port)))

        # Wrong database.
        uri = "mongodb://admin:[email protected]%s:%d/pymongo_test" % (host, port)
        with self.assertRaises(OperationFailure):
            connected(rs_or_single_client(uri))

        # No error.
        connected(rs_or_single_client_noauth("mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port)))

        # Auth with lazy connection.
        rs_or_single_client(
            "mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port), connect=False
        ).pymongo_test.test.find_one()

        # Wrong password.
        bad_client = rs_or_single_client("mongodb://user:[email protected]%s:%d/pymongo_test" % (host, port), connect=False)

        self.assertRaises(OperationFailure, bad_client.pymongo_test.test.find_one)

    def test_mongo_client(self):
        pair = client_context.pair
        m = rs_or_single_client(w=0)
        coll = m.pymongo_test.write_concern_test
        coll.drop()
        doc = {"_id": ObjectId()}
        coll.insert_one(doc)
        self.assertTrue(coll.insert_one(doc))
        coll = coll.with_options(write_concern=WriteConcern(w=1))
        self.assertRaises(OperationFailure, coll.insert_one, doc)

        m = rs_or_single_client()
        coll = m.pymongo_test.write_concern_test
        new_coll = coll.with_options(write_concern=WriteConcern(w=0))
        self.assertTrue(new_coll.insert_one(doc))
        self.assertRaises(OperationFailure, coll.insert_one, doc)

        m = rs_or_single_client("mongodb://%s/" % (pair,),
                                replicaSet=client_context.replica_set_name)

        coll = m.pymongo_test.write_concern_test
        self.assertRaises(OperationFailure, coll.insert_one, doc)
        m = rs_or_single_client("mongodb://%s/?w=0" % (pair,),
                                replicaSet=client_context.replica_set_name)

        coll = m.pymongo_test.write_concern_test
        coll.insert_one(doc)

        # Equality tests
        direct = connected(single_client(w=0))
        direct2 = connected(single_client("mongodb://%s/?w=0" % (pair,),
                                          **self.credentials))
        self.assertEqual(direct, direct2)
        self.assertFalse(direct != direct2)

    def test_socket_timeout_ms_validation(self):
        c = rs_or_single_client(socketTimeoutMS=10 * 1000)
        self.assertEqual(10, get_pool(c).opts.socket_timeout)

        c = connected(rs_or_single_client(socketTimeoutMS=None))
        self.assertEqual(None, get_pool(c).opts.socket_timeout)

        self.assertRaises(ValueError, rs_or_single_client, socketTimeoutMS=0)

        self.assertRaises(ValueError, rs_or_single_client, socketTimeoutMS=-1)

        self.assertRaises(ValueError, rs_or_single_client, socketTimeoutMS=1e10)

        self.assertRaises(ValueError, rs_or_single_client, socketTimeoutMS="foo")

    def test_lazy_connect_w0(self):
        # Ensure that connect-on-demand works when the first operation is
        # an unacknowledged write. This exercises _writable_max_wire_version().

        # Use a separate collection to avoid races where we're still
        # completing an operation on a collection while the next test begins.
        client = rs_or_single_client(connect=False, w=0)
        client.test_lazy_connect_w0.test.insert_one({})

        client = rs_or_single_client(connect=False)
        client.test_lazy_connect_w0.test.update_one({}, {'$set': {'x': 1}})

        client = rs_or_single_client(connect=False)
        client.test_lazy_connect_w0.test.delete_one({})

    def test_max_staleness_float(self):
        with self.assertRaises(TypeError) as ctx:
            rs_or_single_client(maxStalenessSeconds=1.5,
                                readPreference="nearest")

        self.assertIn("must be an integer", str(ctx.exception))

        with warnings.catch_warnings(record=True) as ctx:
            warnings.simplefilter("always")
            client = MongoClient("mongodb://host/?maxStalenessSeconds=1.5"
                                 "&readPreference=nearest")

            # Option was ignored.
            self.assertEqual(-1, client.read_preference.max_staleness)
            self.assertIn("must be an integer", str(ctx[0]))

 def setUpClass(cls):
     cls.listener = EventListener()
     cls.saved_listeners = monitoring._LISTENERS
     monitoring._LISTENERS = monitoring._Listeners([], [], [], [])
     cls.client = rs_or_single_client(event_listeners=[cls.listener])
     cls.db = cls.client.pymongo_test
     cls.collation = Collation('en_US')

    def test_init_disconnected(self):
        c = rs_or_single_client(connect=False)

        self.assertIsInstance(c.is_primary, bool)
        self.assertIsInstance(c.is_mongos, bool)
        self.assertIsInstance(c.max_pool_size, int)
        self.assertIsInstance(c.nodes, frozenset)

        self.assertEqual(c.codec_options, CodecOptions())
        self.assertIsInstance(c.max_bson_size, int)
        self.assertIsInstance(c.max_write_batch_size, int)
        self.assertFalse(c.primary)
        self.assertFalse(c.secondaries)

        c.pymongo_test.command('ismaster')  # Auto-connect.

        if client_context.is_rs:
            # The primary's host and port are from the replica set config.
            self.assertIsNotNone(c.address)
        else:
            self.assertEqual(c.address, (host, port))

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

 def setUp(self):
     self.listener = SessionTestListener()
     self.session_checker_listener = SessionTestListener()
     self.client = rs_or_single_client(
         event_listeners=[self.listener, self.session_checker_listener])
     self.db = self.client.pymongo_test
     self.initial_lsids = set(s['id'] for s in session_ids(self.client))

    def test_implicit_session_logout(self):
        listener = SessionTestListener()

        # Changing auth doesn't invalidate the session. Start as root.
        client = rs_or_single_client(event_listeners=[listener])
        db = client.pymongo_test

        for name, f in [
            ('bulk_write', lambda: db.collection.bulk_write([InsertOne({})])),
            ('collection_names', db.collection_names),
            ('find_one', db.collection.find_one),
            ('aggregate', lambda: list(db.collection.aggregate([])))
        ]:
            def sub_test():
                listener.results.clear()
                f()
                for event in listener.results['started']:
                    self.assertIn(
                        'lsid', event.command,
                        "%s sent no lsid with %s" % (
                            name, event.command_name))

            # We switch auth without clearing the pool of session ids. The
            # server considers these to be new sessions since it's a new user.
            # The old sessions time out on the server after 30 minutes.
            client.admin.logout()
            db.authenticate('second-user', 'pass')
            sub_test()
            db.logout()
            client.admin.authenticate(db_user, db_pwd)
            sub_test()

    def test_close_kills_cursor_synchronously(self):
        # Kill any cursors possibly queued up by previous tests.
        gc.collect()
        self.client._process_periodic_tasks()

        listener = WhiteListEventListener("killCursors")
        results = listener.results
        client = rs_or_single_client(event_listeners=[listener])
        self.addCleanup(client.close)
        coll = client[self.db.name].test_close_kills_cursors

        # Add some test data.
        docs_inserted = 1000
        coll.insert_many([{"i": i} for i in range(docs_inserted)])

        results.clear()

        # Close the cursor while it's still open on the server.
        cursor = coll.find().batch_size(10)
        self.assertTrue(bool(next(cursor)))
        self.assertLess(cursor.retrieved, docs_inserted)
        cursor.close()

        # Test that the cursor was closed.
        self.assertEqual(1, len(results["started"]))
        self.assertEqual("killCursors", results["started"][0].command_name)
        self.assertEqual(1, len(results["succeeded"]))
        self.assertEqual("killCursors", results["succeeded"][0].command_name)

    def test_list_collection_names_filter(self):
        listener = OvertCommandListener()
        results = listener.results
        client = rs_or_single_client(event_listeners=[listener])
        db = client[self.db.name]
        db.capped.drop()
        db.create_collection("capped", capped=True, size=4096)
        db.capped.insert_one({})
        db.non_capped.insert_one({})
        self.addCleanup(client.drop_database, db.name)

        # Should not send nameOnly.
        for filter in ({'options.capped': True},
                       {'options.capped': True, 'name': 'capped'}):
            results.clear()
            names = db.list_collection_names(filter=filter)
            self.assertEqual(names, ["capped"])
            self.assertNotIn("nameOnly", results["started"][0].command)

        # Should send nameOnly (except on 2.6).
        for filter in (None, {}, {'name': {'$in': ['capped', 'non_capped']}}):
            results.clear()
            names = db.list_collection_names(filter=filter)
            self.assertIn("capped", names)
            self.assertIn("non_capped", names)
            command = results["started"][0].command
            if client_context.version >= (3, 0):
                self.assertIn("nameOnly", command)
                self.assertTrue(command["nameOnly"])
            else:
                self.assertNotIn("nameOnly", command)

    def test_errors(self):
        with ignore_deprecations():
            # We must call getlasterror, etc. on same socket as last operation.
            db = rs_or_single_client(maxPoolSize=1).pymongo_test
            db.reset_error_history()
            self.assertEqual(None, db.error())
            self.assertEqual(None, db.previous_error())

            db.command("forceerror", check=False)
            self.assertTrue(db.error())
            self.assertTrue(db.previous_error())

            db.command("forceerror", check=False)
            self.assertTrue(db.error())
            prev_error = db.previous_error()
            self.assertEqual(prev_error["nPrev"], 1)
            del prev_error["nPrev"]
            prev_error.pop("lastOp", None)
            error = db.error()
            error.pop("lastOp", None)
            # getLastError includes "connectionId" in recent
            # server versions, getPrevError does not.
            error.pop("connectionId", None)
            self.assertEqual(error, prev_error)

            db.test.find_one()
            self.assertEqual(None, db.error())
            self.assertTrue(db.previous_error())
            self.assertEqual(db.previous_error()["nPrev"], 2)

            db.reset_error_history()
            self.assertEqual(None, db.error())
            self.assertEqual(None, db.previous_error())

    def test_cursor_transfer(self):

        # This is just a test, don't try this at home...

        client = rs_or_single_client()
        db = client.pymongo_test

        db.test.delete_many({})
        db.test.insert_many([{'_id': i} for i in range(200)])

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

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

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

    def test_auth_network_error(self):
        # Make sure there's no semaphore leak if we get a network error
        # when authenticating a new socket with cached credentials.

        # Get a client with one socket so we detect if it's leaked.
        c = connected(rs_or_single_client(maxPoolSize=1,
                                          waitQueueTimeoutMS=1))

        # Simulate an authenticate() call on a different socket.
        credentials = auth._build_credentials_tuple(
            'DEFAULT', 'admin', db_user, db_pwd, {})

        c._cache_credentials('test', credentials, connect=False)

        # Cause a network error on the actual socket.
        pool = get_pool(c)
        socket_info = one(pool.sockets)
        socket_info.sock.close()

        # SocketInfo.check_auth logs in with the new credential, but gets a
        # socket.error. Should be reraised as AutoReconnect.
        self.assertRaises(AutoReconnect, c.test.collection.find_one)

        # No semaphore leak, the pool is allowed to make a new socket.
        c.test.collection.find_one()

    def test_retry_timeout_raises_original_error(self):
        """A ServerSelectionTimeoutError on the retry attempt raises the
        original error.
        """
        listener = OvertCommandListener()
        client = rs_or_single_client(
            retryWrites=True, event_listeners=[listener])
        self.addCleanup(client.close)
        topology = client._topology
        select_server = topology.select_server

        def mock_select_server(*args, **kwargs):
            server = select_server(*args, **kwargs)

            def raise_error(*args, **kwargs):
                raise ServerSelectionTimeoutError(
                    'No primary available for writes')
            # Raise ServerSelectionTimeout on the retry attempt.
            topology.select_server = raise_error
            return server

        for method, args, kwargs in retryable_single_statement_ops(
                client.db.retryable_write_test):
            msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
            listener.results.clear()
            topology.select_server = mock_select_server
            with self.assertRaises(ConnectionFailure, msg=msg):
                method(*args, **kwargs)
            self.assertEqual(len(listener.results['started']), 1, msg)

    def test_max_pool_size(self):
        max_pool_size = 4
        c = rs_or_single_client(maxPoolSize=max_pool_size)
        collection = c[DB].test

        # Need one document.
        collection.drop()
        collection.insert_one({})

        # nthreads had better be much larger than max_pool_size to ensure that
        # max_pool_size sockets are actually required at some point in this
        # test's execution.
        cx_pool = get_pool(c)
        nthreads = 10
        threads = []
        lock = threading.Lock()
        self.n_passed = 0

        def f():
            for _ in range(5):
                collection.find_one({'$where': delay(0.1)})
                assert len(cx_pool.sockets) <= max_pool_size

            with lock:
                self.n_passed += 1

        for i in range(nthreads):
            t = threading.Thread(target=f)
            threads.append(t)
            t.start()

        joinall(threads)
        self.assertEqual(nthreads, self.n_passed)
        self.assertTrue(len(cx_pool.sockets) > 1)
        self.assertEqual(max_pool_size, cx_pool._socket_semaphore.counter)

    def test_max_pool_size_none(self):
        c = rs_or_single_client(maxPoolSize=None)
        collection = c[DB].test

        # Need one document.
        collection.drop()
        collection.insert_one({})

        cx_pool = get_pool(c)
        nthreads = 10
        threads = []
        lock = threading.Lock()
        self.n_passed = 0

        def f():
            for _ in range(5):
                collection.find_one({'$where': delay(0.1)})

            with lock:
                self.n_passed += 1

        for i in range(nthreads):
            t = threading.Thread(target=f)
            threads.append(t)
            t.start()

        joinall(threads)
        self.assertEqual(nthreads, self.n_passed)
        self.assertTrue(len(cx_pool.sockets) > 1)

 def setUp(self):
     self.c = rs_or_single_client()
     db = self.c[DB]
     db.unique.drop()
     db.test.drop()
     db.unique.insert_one({"_id": "jesse"})
     db.test.insert_many([{} for _ in range(10)])

    def test_increment_transaction_id_without_sending_command(self):
        """Test that the txnNumber field is properly incremented, even when
        the first attempt fails before sending the command.
        """
        listener = OvertCommandListener()
        client = rs_or_single_client(
            retryWrites=True, event_listeners=[listener])
        topology = client._topology
        select_server = topology.select_server

        def raise_connection_err_select_server(*args, **kwargs):
            # Raise ConnectionFailure on the first attempt and perform
            # normal selection on the retry attempt.
            topology.select_server = select_server
            raise ConnectionFailure('Connection refused')

        for method, args, kwargs in _retryable_single_statement_ops(
                client.db.retryable_write_test):
            listener.results.clear()
            topology.select_server = raise_connection_err_select_server
            with client.start_session() as session:
                kwargs = copy.deepcopy(kwargs)
                kwargs['session'] = session
                msg = '%s(*%r, **%r)' % (method.__name__, args, kwargs)
                initial_txn_id = session._server_session.transaction_id

                # Each operation should fail on the first attempt and succeed
                # on the second.
                method(*args, **kwargs)
                self.assertEqual(len(listener.results['started']), 1, msg)
                retry_cmd = listener.results['started'][0].command
                sent_txn_id = retry_cmd['txnNumber']
                final_txn_id = session._server_session.transaction_id
                self.assertEqual(Int64(initial_txn_id + 1), sent_txn_id, msg)
                self.assertEqual(sent_txn_id, final_txn_id, msg)

    def test_survive_cursor_not_found(self):
        # By default the find command returns 101 documents in the first batch.
        # Use 102 batches to cause a single getMore.
        chunk_size = 1024
        data = b'd' * (102 * chunk_size)
        listener = EventListener()
        client = rs_or_single_client(event_listeners=[listener])
        db = client.pymongo_test
        with GridIn(db.fs, chunk_size=chunk_size) as infile:
            infile.write(data)

        with GridOut(db.fs, infile._id) as outfile:
            self.assertEqual(len(outfile.readchunk()), chunk_size)

            # Kill the cursor to simulate the cursor timing out on the server
            # when an application spends a long time between two calls to
            # readchunk().
            client._close_cursor_now(
                outfile._GridOut__chunk_iter._cursor.cursor_id,
                _CursorAddress(client.address, db.fs.chunks.full_name))

            # Read the rest of the file without error.
            self.assertEqual(len(outfile.read()), len(data) - chunk_size)

        # Paranoid, ensure that a getMore was actually sent.
        self.assertIn("getMore", listener.started_command_names())