// copyright (C) MongoDB, Inc. 2014-present.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may
// not use this file except in compliance with the License. You may obtain
// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0

package mongodump

import (
	"bytes"
	"compress/gzip"
	"context"
	"crypto/sha1"
	"encoding/base64"
	"fmt"
	"io"
	"math/rand"
	"net/url"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"strings"
	"testing"
	"time"

	"github.com/mongodb/mongo-tools/common"
	"github.com/mongodb/mongo-tools/common/archive"
	"github.com/mongodb/mongo-tools/common/bsonutil"
	"github.com/mongodb/mongo-tools/common/db"
	"github.com/mongodb/mongo-tools/common/failpoint"
	"github.com/mongodb/mongo-tools/common/json"
	"github.com/mongodb/mongo-tools/common/log"
	"github.com/mongodb/mongo-tools/common/options"
	"github.com/mongodb/mongo-tools/common/testtype"
	"github.com/mongodb/mongo-tools/common/testutil"
	"github.com/pkg/errors"
	. "github.com/smartystreets/goconvey/convey"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	mopt "go.mongodb.org/mongo-driver/v2/mongo/options"
	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/xoptions"
)

var (
	// database with test data.
	testDB = "mongodump_test_db"
	// temp database used for restoring a DB.
	testRestoreDB       = "temp_mongodump_restore_test_db"
	testCollectionNames = []string{"coll1", "coll2", "coll/three"}
)

const (
	longPrefix = "aVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVery" +
		"VeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVery" +
		"VeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVeryVery"
	longCollectionName = longPrefix + "/Long/Collection/Name/Consisting/Of/Many/Characters"
	longBsonName       = longPrefix + "%2FLong%2FCollection%2FName%24FUVlwTrb2eHN1RUE1swI1fFzWmA.bson"
	longMetadataName   = longPrefix + "%2FLong%2FCollection%2FName%24FUVlwTrb2eHN1RUE1swI1fFzWmA.metadata.json"
)

func simpleMongoDumpInstance() (*MongoDump, error) {
	toolOptions, err := testutil.GetToolOptions()
	if err != nil {
		return nil, fmt.Errorf("error getting tool options to create a mongodump instance: %w", err)
	}

	// Limit ToolOptions to test database
	toolOptions.Namespace = &options.Namespace{DB: testDB}

	outputOptions := &OutputOptions{
		NumParallelCollections: 1,
	}
	inputOptions := &InputOptions{}

	log.SetVerbosity(toolOptions.Verbosity)

	return &MongoDump{
		ToolOptions:   toolOptions,
		InputOptions:  inputOptions,
		OutputOptions: outputOptions,
	}, nil
}

// returns the number of .bson files in a directory
// excluding system.indexes.bson.
func countNonIndexBSONFiles(dir string) (int, error) {
	files, err := listNonIndexBSONFiles(dir)
	if err != nil {
		return 0, err
	}
	return len(files), nil
}

func assertBSONEqual(t *testing.T, expected, actual any) {
	expectedJSON, err := bson.MarshalExtJSONIndent(expected, false, false, "", "    ")
	require.NoError(t, err)

	actualJSON, err := bson.MarshalExtJSONIndent(actual, false, false, "", "    ")
	require.NoError(t, err)

	assert.Equal(t, string(expectedJSON), string(actualJSON))
}

func listNonIndexBSONFiles(dir string) ([]string, error) {
	var files []string
	matchingFiles, err := getMatchingFiles(dir, ".*\\.bson")
	if err != nil {
		return nil, err
	}
	for _, fileName := range matchingFiles {
		if fileName != "system.indexes.bson" {
			files = append(files, fileName)
		}
	}
	return files, nil
}

// returns count of metadata files.
func countMetaDataFiles(dir string) (int, error) {
	matchingFiles, err := getMatchingFiles(dir, ".*\\.metadata\\.json")
	if err != nil {
		return 0, err
	}
	return len(matchingFiles), nil
}

// returns count of oplog entries with 'ui' field.
func countOplogUI(iter *db.DecodedBSONSource) int {
	var count int
	var doc bson.M
	for iter.Next(&doc) {
		count += countOpsWithUI(doc)
	}
	return count
}

func countOpsWithUI(doc bson.M) int {
	var count int
	switch doc["op"] {
	case "i", "u", "d":
		if _, ok := doc["ui"]; ok {
			count++
		}
	case "c":
		if _, ok := doc["ui"]; ok {
			count++
		} else if v, ok := doc["o"]; ok {
			opts, _ := v.(bson.M)
			if applyOps, ok := opts["applyOps"]; ok {
				//nolint:errcheck
				list := applyOps.([]bson.M)
				for _, v := range list {
					count += countOpsWithUI(v)
				}
			}
		}
	}
	return count
}

// returns filenames that match the given pattern.
func getMatchingFiles(dir, pattern string) ([]string, error) {
	fileInfos, err := os.ReadDir(dir)
	if err != nil {
		return nil, err
	}

	matchingFiles := []string{}
	var matched bool
	for _, fileInfo := range fileInfos {
		fileName := fileInfo.Name()
		if matched, err = regexp.MatchString(pattern, fileName); matched {
			matchingFiles = append(matchingFiles, fileName)
		}
		if err != nil {
			return nil, err
		}
	}
	return matchingFiles, nil
}

// read all the database bson documents from dir and put it into another DB
// ignore the indexes for now.
func readBSONIntoDatabase(t *testing.T, dir, restoreDBName string) error {
	if ok := fileDirExists(dir); !ok {
		return fmt.Errorf("error finding '%v' on local FS", dir)
	}

	session, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	fileInfos, err := os.ReadDir(dir)
	if err != nil {
		return err
	}

	for _, fileInfo := range fileInfos {
		fileName := fileInfo.Name()
		if !strings.HasSuffix(fileName, ".bson") || fileName == "system.indexes.bson" {
			continue
		}

		collectionName, err := url.QueryUnescape(
			fileName[:strings.LastIndex(fileName, ".bson")],
		)
		if err != nil {
			return err
		}

		collection := session.Database(restoreDBName).Collection(collectionName)

		file, err := os.Open(fmt.Sprintf("%s/%s", dir, fileName))
		if err != nil {
			return err
		}
		defer file.Close()

		bsonSource := db.NewDecodedBSONSource(db.NewBSONSource(file))
		defer bsonSource.Close()

		var result bson.D
		for bsonSource.Next(&result) {
			_, err = collection.InsertOne(t.Context(), result)
			if err != nil {
				return err
			}
		}
		if err = bsonSource.Err(); err != nil {
			return err
		}
	}

	return nil
}

func setUpMongoDumpTestData(t *testing.T) error {
	session, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	for i, collectionName := range testCollectionNames {
		coll := session.Database(testDB).Collection(collectionName)

		for j := 0; j < 10*(i+1); j++ {
			_, err = coll.InsertOne(
				t.Context(),
				bson.M{
					"collectionName": collectionName,
					"age":            j,
					"":               "foo",
					"coords":         bson.D{{"x", i}, {"y", j}},
				},
			)
			if err != nil {
				return err
			}

			idx := mongo.IndexModel{
				Keys: bson.M{`"`: 1},
			}
			_, err = coll.Indexes().CreateOne(t.Context(), idx)
			if err != nil {
				return err
			}
		}
	}

	return nil
}

func setupTimeseriesWithMixedSchema(t *testing.T, dbName string, collName string) {
	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err, "get session provider")

	serverVersion, err := sessionProvider.ServerVersionArray()
	require.NoError(t, err, "get server version")

	client, err := sessionProvider.GetSession()
	require.NoError(t, err, "get session")

	err = client.Database(dbName).Collection(collName).Drop(t.Context())
	require.NoError(t, err, "drop existing coll")

	createCmd := bson.D{
		{"create", collName},
		{"timeseries", bson.D{
			{"timeField", "t"},
			{"metaField", "m"},
		}},
	}

	createRes := sessionProvider.DB(dbName).RunCommand(t.Context(), createCmd)
	require.NoError(t, createRes.Err(), "create timeseries coll")

	// SERVER-84531 was only backported to 7.3.
	// TODO: Run collMod command on 6.0 and 7.0 (TOOLS-3597).
	clientFCV := testutil.GetFCV(client)

	shouldAccommodateMixedSchema := clientFCV == "6.0" || clientFCV == "7.0"
	if !shouldAccommodateMixedSchema {
		cmp, err := testutil.CompareFCV(clientFCV, "7.3")
		require.NoError(t, err, "compare client fcv (%s)", clientFCV)

		shouldAccommodateMixedSchema = cmp >= 0
	}

	if shouldAccommodateMixedSchema {
		res := sessionProvider.DB(dbName).RunCommand(t.Context(), bson.D{
			{"collMod", collName},
			{"timeseriesBucketsMayHaveMixedSchemaData", true},
		})
		require.NoError(t, res.Err(), "set mixed schema data")
	}

	bucketName := timeseriesCollName(serverVersion, collName)
	bucketColl := sessionProvider.DB(dbName).Collection(bucketName)
	bucketJSON := `{"_id":{"$oid":"65a6eb806ffc9fa4280ecac4"},"control":{"version":1,"min":{"_id":{"$oid":"65a6eba7e6d2e848e08c3750"},"t":{"$date":"2024-01-16T20:48:00Z"},"a":1},"max":{"_id":{"$oid":"65a6eba7e6d2e848e08c3751"},"t":{"$date":"2024-01-16T20:48:39.448Z"},"a":"a"}},"meta":0,"data":{"_id":{"0":{"$oid":"65a6eba7e6d2e848e08c3750"},"1":{"$oid":"65a6eba7e6d2e848e08c3751"}},"t":{"0":{"$date":"2024-01-16T20:48:39.448Z"},"1":{"$date":"2024-01-16T20:48:39.448Z"}},"a":{"0":"a","1":1}}}`

	var bucketDoc bson.D
	err = bson.UnmarshalExtJSON([]byte(bucketJSON), false, &bucketDoc)
	require.NoError(t, err, "unmarshal ext json")

	opts := mopt.InsertOne()
	if serverVersion.SupportsRawData() {
		err := xoptions.SetInternalInsertOneOptions(opts, "rawData", true)
		require.NoError(t, err)
	}

	_, err = bucketColl.InsertOne(t.Context(), bucketDoc, opts)
	require.NoError(t, err, "insert bucket doc")
}

func timeseriesCollName(version db.Version, base string) string {
	if version.SupportsRawData() {
		// viewless timeseries
		return base
	}

	return common.TimeseriesBucketPrefix + base
}

func setUpDBView(dbName string, colName string) error {
	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		return err
	}

	pipeline := []bson.M{{"$project": bson.M{"b": "$a"}}}
	createCmd := bson.D{
		{"create", "test view"},
		{"viewOn", colName},
		{"pipeline", pipeline},
	}
	var r2 bson.D
	err = sessionProvider.Run(createCmd, &r2, dbName)
	if err != nil {
		return err
	}
	return nil
}

func turnOnProfiling(dbName string) error {
	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		return err
	}

	profileCmd := bson.D{
		{"profile", 2},
	}

	var res bson.M
	return sessionProvider.Run(profileCmd, &res, dbName)
}

func countSnapshotCmds(
	t *testing.T,
	profileCollection *mongo.Collection,
	ns string,
) (int64, error) {
	return profileCollection.CountDocuments(t.Context(),
		bson.D{
			{"ns", ns},
			{"op", "query"},
			{"$or", []any{
				// 4.0+
				bson.D{{"command.hint._id", 1}},
				// 3.6
				bson.D{{"command.$snapshot", true}},
				bson.D{{"command.snapshot", true}},
				// 3.4 and previous
				bson.D{{"query.$snapshot", true}},
				bson.D{{"query.snapshot", true}},
				bson.D{{"query.hint._id", 1}},
			}},
		},
	)
}

// backgroundInsert inserts into random collections until provided done
// channel is closed.  The function closes the ready channel to signal that
// background insertion has started.  When the done channel is closed, the
// function returns.  Any errors are passed back on the errs channel.
func backgroundInsert(t *testing.T, ready, done chan struct{}, errs chan error) {
	defer close(errs)
	session, err := testutil.GetBareSession()
	if err != nil {
		errs <- err
		close(ready)
		return
	}

	colls := make([]*mongo.Collection, len(testCollectionNames))
	for i, v := range testCollectionNames {
		colls[i] = session.Database(testDB).Collection(v)
	}

	var n int

	// Insert a doc to ensure the DB is actually ready for inserts
	// and not pausing while a dropDatabase is processing.
	_, err = colls[0].InsertOne(t.Context(), bson.M{"n": n})
	if err != nil {
		errs <- err
		close(ready)
		return
	}
	close(ready)
	n++

	for {
		select {
		case <-done:
			return
		default:
			coll := colls[rand.Intn(len(colls))]
			_, err := coll.InsertOne(t.Context(), bson.M{"n": n})
			if err != nil {
				errs <- err
				return
			}
			n++
		}
	}
}

func tearDownMongoDumpTestData(t *testing.T) error {
	session, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	err = session.Database(testDB).Drop(t.Context())
	if err != nil {
		return err
	}
	return nil
}

func dropDB(t *testing.T, dbName string) error {
	session, err := testutil.GetBareSession()
	if err != nil {
		return err
	}

	err = session.Database(dbName).Drop(t.Context())
	if err != nil {
		return err
	}
	return nil
}

func fileDirExists(name string) bool {
	if _, err := os.Stat(name); err != nil {
		if os.IsNotExist(err) {
			return false
		}
	}
	return true
}

func testQuery(t *testing.T, md *MongoDump, session *mongo.Client) string {
	origDB := session.Database(testDB)
	restoredDB := session.Database(testRestoreDB)

	// query to test --query* flags
	bsonQuery := bson.M{"age": bson.M{"$lt": 10}}

	// we can only dump using query per collection
	for _, testCollName := range testCollectionNames {
		md.ToolOptions.Collection = testCollName

		err := md.Init()
		So(err, ShouldBeNil)

		err = md.Dump()
		So(err, ShouldBeNil)
	}

	path, err := os.Getwd()
	So(err, ShouldBeNil)

	dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	So(fileDirExists(dumpDir), ShouldBeTrue)
	So(fileDirExists(dumpDBDir), ShouldBeTrue)

	So(restoredDB.Drop(t.Context()), ShouldBeNil)
	err = readBSONIntoDatabase(t, dumpDBDir, testRestoreDB)
	So(err, ShouldBeNil)

	for _, testCollName := range testCollectionNames {
		// count filtered docs
		origDocCount, err := origDB.Collection(testCollName).
			CountDocuments(t.Context(), bsonQuery)
		So(err, ShouldBeNil)

		// count number of all restored documents
		restDocCount, err := restoredDB.Collection(testCollName).
			CountDocuments(t.Context(), bson.D{})
		So(err, ShouldBeNil)

		So(restDocCount, ShouldEqual, origDocCount)
	}
	return dumpDir
}

func testDumpOneCollection(t *testing.T, md *MongoDump, dumpDir string) {
	path, err := os.Getwd()
	So(err, ShouldBeNil)

	absDumpDir := filepath.FromSlash(filepath.Join(path, dumpDir))
	So(os.RemoveAll(absDumpDir), ShouldBeNil)
	So(fileDirExists(absDumpDir), ShouldBeFalse)

	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	So(fileDirExists(dumpDBDir), ShouldBeFalse)

	md.OutputOptions.Out = dumpDir
	err = md.Dump()
	So(err, ShouldBeNil)
	So(fileDirExists(dumpDBDir), ShouldBeTrue)

	session, err := testutil.GetBareSession()
	So(err, ShouldBeNil)

	countColls, err := countNonIndexBSONFiles(dumpDBDir)
	So(err, ShouldBeNil)
	So(countColls, ShouldEqual, 1)

	collOriginal := session.Database(testDB).Collection(md.ToolOptions.Collection)

	So(session.Database(testRestoreDB).Drop(t.Context()), ShouldBeNil)
	collRestore := session.Database(testRestoreDB).Collection(md.ToolOptions.Collection)

	err = readBSONIntoDatabase(t, dumpDBDir, testRestoreDB)
	So(err, ShouldBeNil)

	Convey("with the correct number of documents", func() {
		numDocsOrig, err := collOriginal.CountDocuments(t.Context(), bson.D{})
		So(err, ShouldBeNil)

		numDocsRestore, err := collRestore.CountDocuments(t.Context(), bson.D{})
		So(err, ShouldBeNil)

		So(numDocsRestore, ShouldEqual, numDocsOrig)
	})

	Convey("that are the same as the documents in the test database", func() {
		iter, err := collOriginal.Find(t.Context(), bson.D{})
		So(err, ShouldBeNil)

		var result bson.D
		for iter.Next(t.Context()) {
			So(iter.Decode(&result), ShouldBeNil)
			restoredCount, err := collRestore.CountDocuments(t.Context(), result)
			So(err, ShouldBeNil)
			So(restoredCount, ShouldNotEqual, 0)
		}
		So(iter.Err(), ShouldBeNil)
		So(iter.Close(t.Context()), ShouldBeNil)
	})
}

func TestMongoDumpValidateOptions(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.UnitTestType)

	t.Run("no db", func(t *testing.T) {
		md, err := simpleMongoDumpInstance()
		require.NoError(t, err)
		md.ToolOptions.Collection = "some_collection"
		md.ToolOptions.DB = ""

		err = md.ValidateOptions()
		require.Error(t, err)
		assert.ErrorContains(
			t,
			err,
			"cannot dump a collection without a specified database",
		)
	})

	t.Run("no collection name with query", func(t *testing.T) {
		md, err := simpleMongoDumpInstance()
		require.NoError(t, err)

		md.ToolOptions.Collection = ""
		md.InputOptions.Query = "{_id:\"\"}"

		err = md.ValidateOptions()
		require.Error(t, err)
		assert.ErrorContains(
			t,
			err,
			"cannot dump using a query without a specified collection",
		)
	})
}

func TestMongoDumpConnectedToAtlasProxy(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err)
	defer sessionProvider.Close()

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.isAtlasProxy = true
	md.OutputOptions.DumpDBUsersAndRoles = false
	md.SessionProvider = sessionProvider

	session, err := sessionProvider.GetSession()
	require.NoError(t, err)
	// This case shouldn't error and should instead not return that it will try to restore users and roles.
	_, err = session.Database("admin").
		Collection("testcol").
		InsertOne(t.Context(), bson.M{})
	require.NoError(t, err)
	dbNames, err := md.GetValidDbs()
	require.NoError(t, err)
	require.NotContains(t, dbNames, "admin")

	// This case should error because it has explicitly been set to dump users and roles for a DB, but thats
	// not possible with an atlas proxy.
	md.OutputOptions.DumpDBUsersAndRoles = true
	err = md.ValidateOptions()
	require.Error(
		t,
		err,
		"can't dump from admin database when connecting to a MongoDB Atlas free or shared cluster",
	)

	require.NoError(t, session.Database("admin").Collection("testcol").Drop(t.Context()))
}

func TestMongoDumpBSON(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		Convey(
			"testing that using MongoDump WITHOUT giving a query dumps everything in the database and/or collection",
			func() {
				md, err := simpleMongoDumpInstance()
				So(err, ShouldBeNil)

				md.InputOptions.Query = ""

				Convey("and that for a particular collection", func() {
					md.ToolOptions.Collection = testCollectionNames[0]
					err = md.Init()
					So(err, ShouldBeNil)

					Convey("it dumps to the default output directory", func() {
						testDumpOneCollection(t, md, "dump")
					})

					Convey("it dumps to a user-specified output directory", func() {
						testDumpOneCollection(t, md, "dump_user")
					})

					Convey("it dumps to standard output", func() {
						md.OutputOptions.Out = "-"
						stdoutBuf := &bytes.Buffer{}
						md.OutputWriter = stdoutBuf
						err = md.Dump()
						So(err, ShouldBeNil)
						var count int
						bsonSource := db.NewDecodedBSONSource(
							db.NewBSONSource(io.NopCloser(stdoutBuf)),
						)
						defer bsonSource.Close()

						var result bson.Raw
						for bsonSource.Next(&result) {
							count++
						}
						So(bsonSource.Err(), ShouldBeNil)
						So(count, ShouldEqual, 10) //The 0th collection has 10 documents

						Reset(func() {
						})

					})

				})

				Convey("and that it dumps a collection with a slash in its name", func() {
					md.ToolOptions.Collection = testCollectionNames[2]

					Convey("to the filesystem", func() {
						err = md.Init()
						So(err, ShouldBeNil)
						testDumpOneCollection(t, md, "dump_slash")
					})

					Convey("to an archive", func() {
						md.OutputOptions.Archive = "dump_slash.archive"
						err = md.Init()
						So(err, ShouldBeNil)
					})
				})

				Convey("for an entire database", func() {
					md.ToolOptions.Collection = ""
					err = md.Init()
					So(err, ShouldBeNil)

					Convey(
						"that exists. The dumped directory should contain the necessary bson files",
						func() {
							md.OutputOptions.Out = "dump"
							err = md.Dump()
							So(err, ShouldBeNil)
							path, err := os.Getwd()
							So(err, ShouldBeNil)

							dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
							dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
							So(fileDirExists(dumpDir), ShouldBeTrue)
							So(fileDirExists(dumpDBDir), ShouldBeTrue)

							countColls, err := countNonIndexBSONFiles(dumpDBDir)
							So(err, ShouldBeNil)
							So(countColls, ShouldEqual, len(testCollectionNames))

							Reset(func() {
								So(os.RemoveAll(dumpDir), ShouldBeNil)
							})

						},
					)

					Convey(
						"that does not exist. The dumped directory shouldn't be created",
						func() {
							md.OutputOptions.Out = "dump"
							md.ToolOptions.DB = "nottestdb"
							err = md.Dump()
							So(err, ShouldBeNil)

							path, err := os.Getwd()
							So(err, ShouldBeNil)

							dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
							dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, "nottestdb"))

							So(fileDirExists(dumpDir), ShouldBeFalse)
							So(fileDirExists(dumpDBDir), ShouldBeFalse)
						},
					)

				})
			},
		)

		Convey(
			"testing that using MongoDump WITH a query dumps a subset of documents in a database and/or collection",
			func() {
				session, err := testutil.GetBareSession()
				So(err, ShouldBeNil)
				md, err := simpleMongoDumpInstance()
				So(err, ShouldBeNil)

				// expect 10 documents per collection
				bsonQuery := bson.M{"age": bson.M{"$lt": 10}}
				jsonQuery, err := bsonutil.ConvertBSONValueToLegacyExtJSON(bsonQuery)
				So(err, ShouldBeNil)
				jsonQueryBytes, err := json.Marshal(jsonQuery)
				So(err, ShouldBeNil)

				Convey("using --query for all the collections in the database", func() {
					md.InputOptions.Query = string(jsonQueryBytes)
					md.ToolOptions.DB = testDB
					md.OutputOptions.Out = "dump"
					dumpDir := testQuery(t, md, session)

					Reset(func() {
						So(session.Database(testRestoreDB).Drop(t.Context()), ShouldBeNil)
						So(os.RemoveAll(dumpDir), ShouldBeNil)
					})

				})

				Convey("using --queryFile for all the collections in the database", func() {
					err = os.WriteFile("example.json", jsonQueryBytes, 0777)
					So(err, ShouldBeNil)
					md.InputOptions.QueryFile = "example.json"
					md.ToolOptions.DB = testDB
					md.OutputOptions.Out = "dump"
					dumpDir := testQuery(t, md, session)

					Reset(func() {
						So(session.Database(testRestoreDB).Drop(t.Context()), ShouldBeNil)
						So(os.RemoveAll(dumpDir), ShouldBeNil)
						So(os.Remove("example.json"), ShouldBeNil)
					})

				})
			},
		)

		Convey("using MongoDump against a collection that doesn't exist succeeds", func() {
			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.ToolOptions.DB = "nonExistentDB"
			md.ToolOptions.Collection = "nonExistentColl"

			err = md.Init()
			So(err, ShouldBeNil)
			err = md.Dump()
			So(err, ShouldBeNil)
		})

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})
	})
}

func TestMongoDumpBSONLongCollectionName(t *testing.T) {
	// Disabled: see TOOLS-2658
	t.Skip()

	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	session, err := testutil.GetBareSession()
	if err != nil {
		t.Fatalf("No server available")
	}
	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "4.4"); err != nil || cmp < 0 {
		t.Skipf("Requires server with FCV 4.4 or later; found %v", fcv)
	}

	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		err = setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		md, err := simpleMongoDumpInstance()
		So(err, ShouldBeNil)

		Convey(
			"testing that it dumps a collection with a name >238 bytes in the right format",
			func() {
				coll := session.Database(testDB).Collection(longCollectionName)
				_, err = coll.InsertOne(t.Context(), bson.M{"a": 1})
				So(err, ShouldBeNil)
				//nolint:errcheck
				defer coll.Drop(t.Context())

				md.ToolOptions.Collection = longCollectionName
				err = md.Init()
				So(err, ShouldBeNil)

				path, err := os.Getwd()
				So(err, ShouldBeNil)

				absDumpDir := filepath.FromSlash(filepath.Join(path, "dump_slash"))
				So(os.RemoveAll(absDumpDir), ShouldBeNil)
				So(fileDirExists(absDumpDir), ShouldBeFalse)

				dumpDBDir := filepath.FromSlash(filepath.Join("dump_slash", testDB))
				So(fileDirExists(dumpDBDir), ShouldBeFalse)

				md.OutputOptions.Out = "dump_slash"
				err = md.Dump()
				So(err, ShouldBeNil)
				So(fileDirExists(dumpDBDir), ShouldBeTrue)

				Convey("to a bson file", func() {
					oneBsonFile, err := os.Open(
						filepath.FromSlash(filepath.Join(dumpDBDir, longBsonName)),
					)
					So(err, ShouldBeNil)
					oneBsonFile.Close()
				})

				Convey("to a metadata file", func() {
					oneMetaFile, err := os.Open(
						filepath.FromSlash(filepath.Join(dumpDBDir, longMetadataName)),
					)
					So(err, ShouldBeNil)
					oneMetaFile.Close()
				})
			},
		)

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})
	})
}

func testPreludeMetadata(md *MongoDump, dir string, serverVersion string) {
	So(fileDirExists(dir), ShouldBeFalse)
	err := md.Init()
	So(err, ShouldBeNil)

	err = md.Dump()
	So(err, ShouldBeNil)

	preludeFilepath := filepath.Join(dir, "prelude.json")
	if md.OutputOptions.Gzip {
		preludeFilepath += ".gz"
	}
	So(fileDirExists(preludeFilepath), ShouldBeTrue)
	var reader io.Reader
	preludeFile, err := os.Open(filepath.FromSlash(preludeFilepath))
	So(err, ShouldBeNil)
	reader = preludeFile
	defer preludeFile.Close()
	if md.OutputOptions.Gzip {
		zipfile, err := gzip.NewReader(preludeFile)
		So(err, ShouldBeNil)
		defer zipfile.Close()
		reader = zipfile
	}
	contents, err := io.ReadAll(reader)
	So(err, ShouldBeNil)
	var jsonResult map[string]any
	err = json.Unmarshal(contents, &jsonResult)
	So(err, ShouldBeNil)
	So(jsonResult["ServerVersion"], ShouldEqual, serverVersion)
}

func TestDumpPreludeMetadataJson(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		path, err := os.Getwd()
		So(err, ShouldBeNil)

		err = setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		sessionProvider, _, _ := testutil.GetBareSessionProvider()
		So(sessionProvider, ShouldNotBeNil)
		serverVersion, err := sessionProvider.ServerVersion()
		So(err, ShouldBeNil)

		Convey("when dumping all databases", func() {
			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.ToolOptions.DB = ""
			md.ToolOptions.Collection = ""

			Convey("when dumping to the default directory", func() {
				dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
				So(os.RemoveAll(dumpDir), ShouldBeNil)

				Convey("writes prelude.json to dump directory", func() {
					testPreludeMetadata(md, dumpDir, serverVersion)
				})

				Convey("writes prelude.json.gz to dump directory when --gzip is used", func() {
					md.OutputOptions.Gzip = true
					testPreludeMetadata(md, dumpDir, serverVersion)
				})

				Reset(func() {
					So(os.RemoveAll(dumpDir), ShouldBeNil)
				})
			})

			Convey("when output directory is specified", func() {
				dumpDir := filepath.FromSlash(filepath.Join(path, "dump_output"))
				So(os.RemoveAll(dumpDir), ShouldBeNil)

				Convey("writes prelude.json to output directory", func() {
					md.OutputOptions.Out = "dump_output"
					testPreludeMetadata(md, dumpDir, serverVersion)
				})

				Reset(func() {
					So(os.RemoveAll(dumpDir), ShouldBeNil)
				})
			})
		})

		Convey("when dumping one db", func() {
			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
			So(os.RemoveAll(dumpDir), ShouldBeNil)

			Convey("writes prelude.json to dump directory", func() {
				testPreludeMetadata(md, dumpDBDir, serverVersion)
			})

			Reset(func() {
				So(os.RemoveAll(dumpDir), ShouldBeNil)
			})
		})

		Convey("when the dump directory is not created", func() {

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, "nottestdb"))

			Convey("the dump does not fail and prelude.json should not be created", func() {
				md, err := simpleMongoDumpInstance()
				So(err, ShouldBeNil)

				md.ToolOptions.DB = "nonExistentDB"

				err = md.Init()
				So(err, ShouldBeNil)
				err = md.Dump()
				So(err, ShouldBeNil)

				So(fileDirExists(dumpDir), ShouldBeFalse)
				So(fileDirExists(dumpDBDir), ShouldBeFalse)
				So(fileDirExists(filepath.Join(dumpDir, "prelude.json")), ShouldBeFalse)
				So(fileDirExists(filepath.Join(dumpDBDir, "prelude.json")), ShouldBeFalse)
			})

			Reset(func() {
				So(os.RemoveAll(dumpDir), ShouldBeNil)
			})
		})

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})

	})
}

func TestMongoDumpMetaData(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		session, err := testutil.GetBareSession()
		So(session, ShouldNotBeNil)
		So(err, ShouldBeNil)

		err = setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		Convey("testing that the dumped directory contains information about indexes", func() {

			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.OutputOptions.Out = "dump"
			err = md.Init()
			So(err, ShouldBeNil)

			err = md.Dump()
			So(err, ShouldBeNil)

			path, err := os.Getwd()
			So(err, ShouldBeNil)
			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
			So(fileDirExists(dumpDir), ShouldBeTrue)
			So(fileDirExists(dumpDBDir), ShouldBeTrue)

			Convey("having one metadata file per collection", func() {
				c1, err := countNonIndexBSONFiles(dumpDBDir)
				So(err, ShouldBeNil)

				c2, err := countMetaDataFiles(dumpDBDir)
				So(err, ShouldBeNil)

				So(c1, ShouldEqual, c2)

				Convey("and that the JSON in a metadata file is valid", func() {
					metaFiles, err := getMatchingFiles(dumpDBDir, ".*\\.metadata\\.json")
					So(err, ShouldBeNil)
					So(len(metaFiles), ShouldBeGreaterThan, 0)

					oneMetaFile, err := os.Open(
						filepath.FromSlash(filepath.Join(dumpDBDir, metaFiles[0])),
					)
					defer oneMetaFile.Close()
					So(err, ShouldBeNil)
					contents, err := io.ReadAll(oneMetaFile)
					So(err, ShouldBeNil)
					var jsonResult map[string]any
					err = json.Unmarshal(contents, &jsonResult)
					So(err, ShouldBeNil)

					Convey("and contains an 'indexes' key", func() {
						_, ok := jsonResult["indexes"]
						So(ok, ShouldBeTrue)
					})

					Convey("and contains a 'collectionName' key", func() {
						_, ok := jsonResult["collectionName"]
						So(ok, ShouldBeTrue)
					})

					fcv := testutil.GetFCV(session)
					cmp, err := testutil.CompareFCV(fcv, "3.6")
					So(err, ShouldBeNil)
					if cmp >= 0 {
						Convey("and on FCV 3.6+, contains a 'uuid' key", func() {
							uuid, ok := jsonResult["uuid"]
							So(ok, ShouldBeTrue)
							checkUUID := regexp.MustCompile(`(?i)^[a-z0-9]{32}$`)

							uuidStr, ok := uuid.(string)
							So(ok, ShouldBeTrue)

							So(checkUUID.MatchString(uuidStr), ShouldBeTrue)
							// XXX useless -- xdg, 2018-09-21
							So(err, ShouldBeNil)
						})
					}

				})

			})

			Reset(func() {
				So(os.RemoveAll(dumpDir), ShouldBeNil)
			})
		})

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})

	})
}

func TestMongoDumpOplog(t *testing.T) {
	// Disabled: see TOOLS-2657
	t.Skip()

	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		t.Fatalf("No cluster available: %v", err)
	}
	session, err := sessionProvider.GetSession()
	if err != nil {
		t.Fatalf("No client available: %v", err)
	}
	if ok, _ := sessionProvider.IsReplicaSet(); !ok {
		t.SkipNow()
	}
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {

		Convey("testing that the dumped directory contains an oplog", func() {

			// Start with clean filesystem
			path, err := os.Getwd()
			So(err, ShouldBeNil)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpOplogFile := filepath.FromSlash(filepath.Join(dumpDir, "oplog.bson"))

			err = os.RemoveAll(dumpDir)
			So(err, ShouldBeNil)
			So(fileDirExists(dumpDir), ShouldBeFalse)

			// Start with clean database
			So(tearDownMongoDumpTestData(t), ShouldBeNil)

			// Prepare mongodump with options
			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.OutputOptions.Oplog = true
			md.ToolOptions.Namespace = &options.Namespace{}
			err = md.Init()
			So(err, ShouldBeNil)

			// Start inserting docs in the background so the oplog has data
			ready := make(chan struct{})
			done := make(chan struct{})
			errs := make(chan error, 1)
			go backgroundInsert(t, ready, done, errs)
			<-ready

			// Run mongodump
			err = md.Dump()
			So(err, ShouldBeNil)

			// Stop background insertion
			close(done)
			err = <-errs
			So(err, ShouldBeNil)

			// Check for and read the oplog file
			So(fileDirExists(dumpDir), ShouldBeTrue)
			So(fileDirExists(dumpOplogFile), ShouldBeTrue)

			oplogFile, err := os.Open(dumpOplogFile)
			defer oplogFile.Close()
			So(err, ShouldBeNil)

			rdr := db.NewBSONSource(oplogFile)
			iter := db.NewDecodedBSONSource(rdr)

			fcv := testutil.GetFCV(session)
			cmp, err := testutil.CompareFCV(fcv, "3.6")
			So(err, ShouldBeNil)

			withUI := countOplogUI(iter)
			So(iter.Err(), ShouldBeNil)

			if cmp >= 0 {
				// for FCV 3.6+, should have 'ui' field in oplog entries
				So(withUI, ShouldBeGreaterThan, 0)
			} else {
				// for FCV <3.6, should no have 'ui' field in oplog entries
				So(withUI, ShouldEqual, 0)
			}

			// Cleanup
			So(os.RemoveAll(dumpDir), ShouldBeNil)
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})

	})
}

// Test dumping a collection with autoIndexId:false.  As of MongoDB 4.0,
// this is only allowed on the 'local' database.
func TestMongoDumpTOOLS2174(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		t.Fatalf("No cluster available: %v", err)
	}

	serverVersion, err := sessionProvider.ServerVersionArray()
	if err != nil {
		t.Fatalf("Could not get Server version: %v", err)
	}
	if serverVersion.GTE(db.Version{8, 2, 0}) {
		t.Skipf(
			"createCollection no longer accepts autoIndexID as of Server version 8.2.0; testing with %s",
			serverVersion.String(),
		)
	}

	collName := "tools-2174"
	dbName := "local"

	var r1 bson.M
	err = sessionProvider.Run(bson.D{{"drop", collName}}, &r1, dbName)
	if err != nil {
		var commandErr mongo.CommandError
		if !errors.As(err, &commandErr) || commandErr.Code != 26 {
			t.Fatalf("Failed to run drop: %v", err)
		}
	}

	createCmd := bson.D{
		{"create", collName},
		{"autoIndexId", false},
	}
	var r2 bson.M
	err = sessionProvider.Run(createCmd, &r2, dbName)
	if err != nil {
		t.Fatalf("Error creating capped, no-autoIndexId collection: %v", err)
	}

	Convey("testing dumping a capped, autoIndexId:false collection", t, func() {
		md, err := simpleMongoDumpInstance()
		So(err, ShouldBeNil)

		md.ToolOptions.Collection = collName
		md.ToolOptions.DB = dbName
		md.OutputOptions.Out = "dump"
		err = md.Init()
		So(err, ShouldBeNil)
		err = md.Dump()
		So(err, ShouldBeNil)
	})
}

// Test dumping a collection while respecting no index scan for wired tiger.
func TestMongoDumpTOOLS1952(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		t.Fatalf("No cluster available: %v", err)
	}

	session, err := sessionProvider.GetSession()
	if err != nil {
		t.Fatalf("Failed to get session: %v", err)
	}

	collName := "tools-1952-dump"
	dbName := "test"
	ns := dbName + "." + collName

	var r1 bson.M

	dbStruct := session.Database(dbName)

	err = sessionProvider.Run(bson.D{{"drop", collName}}, &r1, dbName)
	if err != nil {
		var commandErr mongo.CommandError
		if !errors.As(err, &commandErr) || commandErr.Code != 26 {
			t.Fatalf("Failed to run drop: %v", err)
		}
	}

	createCmd := bson.D{
		{"create", collName},
	}
	var r2 bson.M
	err = sessionProvider.Run(createCmd, &r2, dbName)
	if err != nil {
		t.Fatalf("Error creating collection: %v", err)
	}

	// Turn on profiling.
	if err = turnOnProfiling(dbName); err != nil {
		t.Fatalf("Failed to turn on profiling: %v", err)
	}

	profileCollection := dbStruct.Collection("system.profile")

	Convey("testing dumping a collection query hints", t, func() {
		md, err := simpleMongoDumpInstance()
		So(err, ShouldBeNil)

		md.ToolOptions.Collection = collName
		md.ToolOptions.DB = dbName
		md.OutputOptions.Out = "dump"
		err = md.Init()
		So(err, ShouldBeNil)
		err = md.Dump()
		So(err, ShouldBeNil)

		count, err := countSnapshotCmds(t, profileCollection, ns)
		So(err, ShouldBeNil)

		// On modern storage engines, there should be no query that matches.
		So(count, ShouldEqual, 0)
	})
}

// Test the fix for nil pointer bug when getCollectionInfo failed.
func TestMongoDumpTOOLS2498(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		t.Fatalf("No cluster available: %v", err)
	}

	collName := "tools-2498-dump"
	dbName := "test"

	var r1 bson.M
	err = sessionProvider.Run(bson.D{{"drop", collName}}, &r1, dbName)
	if err != nil {
		var commandErr mongo.CommandError
		if !errors.As(err, &commandErr) || commandErr.Code != 26 {
			t.Fatalf("Failed to run drop: %v", err)
		}
	}

	createCmd := bson.D{
		{"create", collName},
	}
	var r2 bson.M
	err = sessionProvider.Run(createCmd, &r2, dbName)
	if err != nil {
		t.Fatalf("Error creating collection: %v", err)
	}

	Convey("failing to get collection info should error, but not panic", t, func() {
		md, err := simpleMongoDumpInstance()
		So(err, ShouldBeNil)

		md.ToolOptions.Collection = collName
		md.ToolOptions.DB = dbName
		md.OutputOptions.Out = "dump"
		err = md.Init()
		So(err, ShouldBeNil)

		require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
		defer failpoint.DefaultManager.Reset()

		dumpErrCh := make(chan error, 1)
		go func() { dumpErrCh <- md.Dump() }()

		fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
		So(ok, ShouldBeTrue)
		require.NoError(t, fp.Reached(context.TODO()))
		session, _ := md.SessionProvider.GetSession()
		disconnectErr := session.Disconnect(t.Context())
		So(disconnectErr, ShouldBeNil)
		fp.Signal()

		err = <-dumpErrCh
		// Mongodump should not panic, but return correct the error if getCollectionInfo failed.
		So(err, ShouldNotBeNil)
		So(err.Error(), ShouldContainSubstring, "client is disconnected")
	})
}

func TestMongoDumpOrderedQuery(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)
		path, err := os.Getwd()
		So(err, ShouldBeNil)
		dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))

		Convey("testing that --query is order-preserving", func() {
			// If order is not preserved, probabilistically, some of these
			// loops will fail.
			for i := 0; i < 100; i++ {
				So(os.RemoveAll(dumpDir), ShouldBeNil)

				md, err := simpleMongoDumpInstance()
				So(err, ShouldBeNil)

				md.InputOptions.Query = `{"coords":{"x":0,"y":1}}`
				md.ToolOptions.Collection = testCollectionNames[0]
				md.ToolOptions.DB = testDB
				md.OutputOptions.Out = "dump"
				err = md.Init()
				So(err, ShouldBeNil)
				err = md.Dump()
				So(err, ShouldBeNil)

				dumpBSON := filepath.FromSlash(
					filepath.Join(dumpDir, testDB, testCollectionNames[0]+".bson"),
				)

				file, err := os.Open(dumpBSON)
				So(err, ShouldBeNil)

				bsonSource := db.NewDecodedBSONSource(db.NewBSONSource(file))

				var count int
				var result bson.M
				for bsonSource.Next(&result) {
					count++
				}
				So(bsonSource.Err(), ShouldBeNil)

				So(count, ShouldEqual, 1)

				bsonSource.Close()
				file.Close()
			}
		})

		Reset(func() {
			So(os.RemoveAll(dumpDir), ShouldBeNil)
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})
	})
}

func TestMongoDumpViewsAsCollections(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		colName := "dump_view_as_collection"
		dbName := testDB
		err = setUpDBView(dbName, colName)
		So(err, ShouldBeNil)

		err = turnOnProfiling(testDB)
		So(err, ShouldBeNil)

		Convey("testing that the dumped directory contains information about metadata", func() {
			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.ToolOptions.DB = testDB
			md.OutputOptions.Out = "dump"
			md.OutputOptions.ViewsAsCollections = true

			err = md.Init()
			So(err, ShouldBeNil)

			err = md.Dump()
			So(err, ShouldBeNil)

			path, err := os.Getwd()
			So(err, ShouldBeNil)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
			So(fileDirExists(dumpDir), ShouldBeTrue)
			So(fileDirExists(dumpDBDir), ShouldBeTrue)

			Convey("having one metadata file per read-only view", func() {
				c1, err := countNonIndexBSONFiles(dumpDBDir)
				So(err, ShouldBeNil)

				c2, err := countMetaDataFiles(dumpDBDir)
				So(err, ShouldBeNil)

				So(c1, ShouldEqual, c2)

			})

			Convey("testing dumping a view, we should not hint index", func() {
				session, err := testutil.GetBareSession()
				So(err, ShouldBeNil)

				dbStruct := session.Database(dbName)
				profileCollection := dbStruct.Collection("system.profile")
				ns := dbName + "." + colName
				count, err := countSnapshotCmds(t, profileCollection, ns)
				So(err, ShouldBeNil)

				// view dump should not do collection scan
				So(count, ShouldEqual, 0)
			})

			Reset(func() {
				So(os.RemoveAll(dumpDir), ShouldBeNil)
			})
		})

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})

	})
}

func TestMongoDumpViews(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	Convey("With a MongoDump instance", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		colName := "dump_views"
		dbName := testDB
		err = setUpDBView(dbName, colName)
		So(err, ShouldBeNil)

		Convey("testing that the dumped directory contains information about metadata", func() {

			md, err := simpleMongoDumpInstance()
			So(err, ShouldBeNil)

			md.ToolOptions.DB = testDB
			md.OutputOptions.Out = "dump"

			err = md.Init()
			So(err, ShouldBeNil)

			err = md.Dump()
			So(err, ShouldBeNil)

			path, err := os.Getwd()
			So(err, ShouldBeNil)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
			So(fileDirExists(dumpDir), ShouldBeTrue)
			So(fileDirExists(dumpDBDir), ShouldBeTrue)

			Convey("having one metadata file per view", func() {

				c1, err := countMetaDataFiles(dumpDBDir)
				So(err, ShouldBeNil)

				So(c1, ShouldBeGreaterThan, 0)

			})

			Convey("testing dumping a view, we should not hint index", func() {
				session, err := testutil.GetBareSession()
				So(err, ShouldBeNil)

				dbStruct := session.Database(dbName)
				profileCollection := dbStruct.Collection("system.profile")
				ns := dbName + "." + colName
				count, err := countSnapshotCmds(t, profileCollection, ns)
				So(err, ShouldBeNil)

				// view dump should not do collection scan
				So(count, ShouldEqual, 0)
			})

			Reset(func() {
				So(os.RemoveAll(dumpDir), ShouldBeNil)
			})
		})

		Reset(func() {
			So(tearDownMongoDumpTestData(t), ShouldBeNil)
		})

	})
}

func TestMongoDumpCollectionOutputPath(t *testing.T) {
	t.Skip("disabled: see TOOLS-2658")

	testtype.SkipUnlessTestType(t, testtype.UnitTestType)
	log.SetWriter(io.Discard)

	newDump := func() *MongoDump {
		md, err := simpleMongoDumpInstance()
		require.NoError(t, err)

		md.OutputOptions.Out = "dump"
		return md
	}

	t.Run("paths less than 255 bytes", func(t *testing.T) {
		md := newDump()

		// 26 bytes < 255 bytes
		// (output path will be under 255 bytes, regardless of file extension)
		colName := "abcdefghijklmnopqrstuvwxyz"

		fileComponents := strings.Split(md.outputPath(testDB, colName), "/")
		assert.Len(t, fileComponents, 3)

		filePath := fileComponents[len(fileComponents)-1]
		assert.Equal(t, colName, filePath)
		assert.NotContains(t, filePath, "%24")
	})

	t.Run("paths equal 255 bytes", func(t *testing.T) {
		md := newDump()

		// 17 bytes * 14 = 238 bytes
		// (output would be exactly 255 bytes with longest possible file extension of .metadata.json.gz)
		colName := strings.Repeat("abcdefghijklmnopq", 14)

		fileComponents := strings.Split(md.outputPath(testDB, colName), "/")
		assert.Len(t, fileComponents, 3)

		filePath := fileComponents[len(fileComponents)-1]
		assert.Equal(t, colName, filePath)
		assert.NotContains(t, filePath, "%24")
	})

	t.Run("path longer than 255 bytes", func(t *testing.T) {
		t.Run("without special characters", func(t *testing.T) {
			md := newDump()

			// 26 bytes * 10 = 260 bytes > 238 bytes
			// (output path is already over the file name limit of 255, regardless of file extension)
			colName := strings.Repeat("abcdefghijklmnopqrstuvwxyz", 10)

			fileComponents := strings.Split(md.outputPath(testDB, colName), "/")
			assert.Len(t, fileComponents, 3)

			filePath := fileComponents[len(fileComponents)-1]
			assert.Equal(t, colName[:208]+"%24", filePath[:211])

			hashDecoded, _ := base64.RawURLEncoding.DecodeString(filePath[211:])
			hash := sha1.Sum([]byte(colName))
			assert.Equal(t, hash[:], hashDecoded)
		})

		t.Run("with special characters", func(t *testing.T) {
			md := newDump()

			// (26 bytes + 3 special bytes) * 8 = 232 bytes < 238 bytes
			// (output path is under the limit, but will go over when we escape the special symbols)
			colName := strings.Repeat("abcdefghijklmnopqrstuvwxyz+/@", 8)

			fileComponents := strings.Split(md.outputPath(testDB, colName), "/")
			assert.Len(t, fileComponents, 3)

			filePath := fileComponents[len(fileComponents)-1]
			assert.Equal(t, url.QueryEscape(colName)[:208]+"%24", filePath[:211])

			hashDecoded, _ := base64.RawURLEncoding.DecodeString(filePath[211:])
			hash := sha1.Sum([]byte(colName))
			assert.Equal(t, hash[:], hashDecoded)
		})
	})
}

func TestCount(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	Convey("test count collection", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		session, err := testutil.GetBareSession()
		So(err, ShouldBeNil)

		collection := session.Database(testDB).Collection(testCollectionNames[0])
		restoredDB := session.Database(testDB)
		//nolint:errcheck
		defer restoredDB.Drop(t.Context())

		Convey("count collection without filter", func() {
			findQuery := &db.DeferredQuery{Coll: collection}
			cnt, err := findQuery.Count(false)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 10)

			findQuery = &db.DeferredQuery{Coll: collection, Filter: bson.M{}}
			cnt, err = findQuery.Count(false)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 10)

			findQuery = &db.DeferredQuery{Coll: collection, Filter: bson.D{}}
			cnt, err = findQuery.Count(false)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 10)
		})

		Convey("count collection with filter in BSON.M", func() {
			findQuery := &db.DeferredQuery{Coll: collection, Filter: bson.M{"age": 1}}
			cnt, err := findQuery.Count(false)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 1)
		})

		Convey("count collection with filter in BSON.D", func() {
			findQuery := &db.DeferredQuery{Coll: collection, Filter: bson.D{{"age", 1}}}
			cnt, err := findQuery.Count(false)
			So(err, ShouldBeNil)
			So(cnt, ShouldEqual, 1)
		})
	})
}

func TestTimeseriesCollections(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	session, err := testutil.GetBareSession()
	require.NoError(t, err, "get session")

	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "5.0"); err != nil || cmp < 0 {
		t.Skipf("Requires server with FCV 5.0 or later; found %v", fcv)
	}

	sp, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err, "get session provider")

	serverVersion, err := sp.ServerVersionArray()
	require.NoError(t, err, "get server version")

	colName := "timeseriesColl"
	dbName := "timeseries_test_DB"
	testutil.SetUpTimeseries(t, dbName, colName)

	defer func() {
		err := dropDB(t, dbName)
		require.NoError(t, err, "drop timeseries coll")
	}()

	setup := func(kind string) *MongoDump {
		md, err := simpleMongoDumpInstance()
		require.NoError(t, err)

		md.ToolOptions.DB = dbName

		switch kind {
		case "archive":
			md.OutputOptions.Out = ""
			md.OutputOptions.Archive = "dump.archive"
		case "dir":
			md.OutputOptions.Out = "dump"
		default:
			panic("unknown setup kind " + kind)
		}

		return md
	}

	t.Run("dump to archive", func(t *testing.T) {
		runArchiveTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.NoError(t, err)

			path, err := os.Getwd()
			require.NoError(t, err)

			archiveFilePath := filepath.FromSlash(filepath.Join(path, "dump.archive"))

			archiveFile, err := os.Open(archiveFilePath)
			require.NoError(t, err)

			archiveReader := &archive.Reader{
				In:      archiveFile,
				Prelude: &archive.Prelude{},
			}

			err = archiveReader.Prelude.Read(archiveReader.In)
			require.NoError(t, err)

			collectionMetadatas, ok := archiveReader.Prelude.NamespaceMetadatasByDB[dbName]
			assert.True(t, ok)

			require.Len(t, collectionMetadatas, 1)
			assert.Equal(t, colName, collectionMetadatas[0].Collection)

			pe, err := archiveReader.Prelude.NewPreludeExplorer()
			require.NoError(t, err)

			archiveContents, err := pe.ReadDir()
			require.NoError(t, err)

			expectedCollFile := timeseriesCollName(serverVersion, colName) + ".bson"

			for _, dirlike := range archiveContents {
				if !dirlike.IsDir() || dirlike.Name() != dbName {
					continue
				}

				dbContents, err := dirlike.ReadDir()
				require.NoError(t, err)

				assert.Len(t, dbContents, 2)

				for _, file := range dbContents {
					assert.Contains(
						t,
						[]string{
							colName + ".metadata.json",
							expectedCollFile,
						},
						file.Name(),
					)
				}
			}

			require.NoError(t, archiveFile.Close())
			require.NoError(t, os.RemoveAll(archiveFilePath))
		}

		t.Run("whole db", func(t *testing.T) {
			md := setup("archive")
			runArchiveTest(t, md)
		})

		t.Run("specified collection", func(t *testing.T) {
			md := setup("archive")
			md.ToolOptions.DB = dbName
			md.ToolOptions.Collection = colName
			runArchiveTest(t, md)
		})

		t.Run("exclude system.buckets.coll", func(t *testing.T) {
			md := setup("archive")
			md.OutputOptions.ExcludedCollections = []string{common.TimeseriesBucketPrefix + colName}
			runArchiveTest(t, md)
		})

		t.Run("exclude system.buckets prefix", func(t *testing.T) {
			md := setup("archive")
			md.OutputOptions.ExcludedCollectionPrefixes = []string{common.TimeseriesBucketPrefix}
			runArchiveTest(t, md)
		})
	})

	t.Run("dump to directory", func(t *testing.T) {
		runDirTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.NoError(t, err)

			path, err := os.Getwd()
			require.NoError(t, err)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, dbName))
			metadataFile := filepath.FromSlash(
				filepath.Join(dumpDBDir, colName+".metadata.json"),
			)

			expectedCollFile := timeseriesCollName(serverVersion, colName)
			bsonFile := filepath.FromSlash(
				filepath.Join(dumpDBDir, expectedCollFile+".bson"),
			)
			assert.True(t, fileDirExists(dumpDir))
			assert.True(t, fileDirExists(dumpDBDir))
			assert.True(t, fileDirExists(metadataFile))
			assert.True(t, fileDirExists(bsonFile), bsonFile)

			allFiles, err := getMatchingFiles(dumpDBDir, ".*"+colName+".*")
			require.NoError(t, err)
			assert.Len(t, allFiles, 2)

			info, err := os.Stat(bsonFile)
			require.NoError(t, err)
			assert.NotZero(t, info.Size())

			require.NoError(t, os.RemoveAll(dumpDir))
		}

		t.Run("whole db", func(t *testing.T) {
			md := setup("dir")
			runDirTest(t, md)
		})

		t.Run("specified collection", func(t *testing.T) {
			md := setup("dir")
			md.ToolOptions.DB = dbName
			md.ToolOptions.Collection = colName
			runDirTest(t, md)
		})

		t.Run("exclude system.buckets.coll", func(t *testing.T) {
			md := setup("dir")
			md.OutputOptions.ExcludedCollections = []string{common.TimeseriesBucketPrefix + colName}
			runDirTest(t, md)
		})

		t.Run("exclude system.buckets prefix", func(t *testing.T) {
			md := setup("dir")
			md.OutputOptions.ExcludedCollectionPrefixes = []string{common.TimeseriesBucketPrefix}
			runDirTest(t, md)
		})
	})

	t.Run("exclude to archive", func(t *testing.T) {
		runArchiveTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.NoError(t, err)

			path, err := os.Getwd()
			require.NoError(t, err)

			archiveFilePath := filepath.FromSlash(filepath.Join(path, "dump.archive"))

			archiveFile, err := os.Open(archiveFilePath)
			require.NoError(t, err)
			archiveReader := &archive.Reader{
				In:      archiveFile,
				Prelude: &archive.Prelude{},
			}

			err = archiveReader.Prelude.Read(archiveReader.In)
			require.NoError(t, err)

			_, ok := archiveReader.Prelude.NamespaceMetadatasByDB[dbName]
			assert.False(t, ok)

			require.NoError(t, archiveFile.Close())
			require.NoError(t, os.RemoveAll(archiveFilePath))
		}

		t.Run("explicit exclude", func(t *testing.T) {
			md := setup("archive")
			md.OutputOptions.ExcludedCollections = []string{colName}
			runArchiveTest(t, md)
		})

		t.Run("exclude by prefix", func(t *testing.T) {
			md := setup("archive")
			md.OutputOptions.ExcludedCollectionPrefixes = []string{colName[:5]}
			runArchiveTest(t, md)
		})
	})

	t.Run("exclude to directory", func(t *testing.T) {
		runDirTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.NoError(t, err)

			path, err := os.Getwd()
			require.NoError(t, err)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			assert.False(t, fileDirExists(dumpDir))

			require.NoError(t, os.RemoveAll(dumpDir))
		}

		t.Run("explicit exclude", func(t *testing.T) {
			md := setup("dir")
			md.OutputOptions.ExcludedCollections = []string{colName}
			runDirTest(t, md)
		})

		t.Run("exclude by prefix", func(t *testing.T) {
			md := setup("dir")
			md.OutputOptions.ExcludedCollectionPrefixes = []string{colName[:5]}
			runDirTest(t, md)
		})
	})

	t.Run("specify buckets collection explicitly", func(t *testing.T) {
		md := setup("dir")
		md.ToolOptions.DB = dbName
		md.ToolOptions.Collection = common.TimeseriesBucketPrefix + colName

		err = md.Init()
		require.Error(t, err)
		assert.ErrorContains(
			t,
			err,
			"cannot specify a system.buckets collection in --collection. "+
				"Specifying the timeseries collection will dump the system.buckets collection",
		)
	})

	t.Run("query by metadata field", func(t *testing.T) {
		runQueryTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.NoError(t, err)

			path, err := os.Getwd()
			require.NoError(t, err)

			dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
			dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, dbName))
			metadataFile := filepath.FromSlash(
				filepath.Join(dumpDBDir, colName+".metadata.json"),
			)
			bsonFile := filepath.FromSlash(
				filepath.Join(dumpDBDir, timeseriesCollName(serverVersion, colName)+".bson"),
			)
			assert.True(t, fileDirExists(dumpDir))
			assert.True(t, fileDirExists(dumpDBDir))
			assert.True(t, fileDirExists(metadataFile))
			assert.True(t, fileDirExists(bsonFile))

			allFiles, err := getMatchingFiles(dumpDBDir, ".*"+colName+".*")
			require.NoError(t, err)
			assert.Len(t, allFiles, 2)

			info, err := os.Stat(bsonFile)
			require.NoError(t, err)
			assert.NotZero(t, info.Size())

			fd, err := os.Open(bsonFile)
			require.NoError(t, err)

			bsonSource := db.NewBSONSource(fd)

			matchedDoc := bson.Raw(bsonSource.LoadNext())
			rawVal, err := matchedDoc.LookupErr("meta", "device")
			require.NoError(t, err)

			val, ok := rawVal.Int32OK()
			require.True(t, ok)
			assert.EqualValues(t, 1, val)

			// only one bucket document should be matched
			require.Nil(t, bsonSource.LoadNext())

			require.NoError(t, fd.Close())
			require.NoError(t, os.RemoveAll(dumpDir))
			os.Remove("ts_query.json")
		}

		t.Run("query", func(t *testing.T) {
			md := setup("dir")
			md.ToolOptions.Collection = colName
			md.InputOptions.Query = "{\"my_meta.device\": 1}"
			runQueryTest(t, md)
		})

		t.Run("queryFile", func(t *testing.T) {
			md := setup("dir")
			md.ToolOptions.Collection = colName

			err = os.WriteFile("ts_query.json", []byte("{\"my_meta.device\": 1}"), 0777)
			require.NoError(t, err)

			md.InputOptions.QueryFile = "ts_query.json"
			runQueryTest(t, md)
		})
	})

	t.Run("query by non-metadata field", func(t *testing.T) {
		runQueryTest := func(t *testing.T, md *MongoDump) {
			err = md.Init()
			require.NoError(t, err)

			err = md.Dump()
			require.Error(t, err)
			assert.ErrorContains(
				t,
				err,
				"mongodump only processes queries on metadata fields for timeseries collections",
			)

			os.Remove("ts_query.json")
		}

		t.Run("query", func(t *testing.T) {
			md := setup("dir")
			md.ToolOptions.Collection = colName
			md.InputOptions.Query = "{\"wrong.device\": 1}"
			runQueryTest(t, md)
		})

		t.Run("queryFile", func(t *testing.T) {
			md := setup("dir")
			md.ToolOptions.Collection = colName

			err = os.WriteFile("ts_query.json", []byte("{\"wrong.device\": 1}"), 0777)
			require.NoError(t, err)

			md.InputOptions.QueryFile = "ts_query.json"
			runQueryTest(t, md)
		})
	})

}

func TestDumpTimeseriesCollectionsWithMixedSchema(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	session, err := testutil.GetBareSession()
	require.NoError(t, err, "get session")

	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "5.0"); err != nil || cmp < 0 {
		t.Skipf("Requires server with FCV 5.0 or later; found %v", fcv)
	}

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err, "get session provider")

	serverVersion, err := sessionProvider.ServerVersionArray()
	require.NoError(t, err, "get server version from session provider")

	colName := "timeseries_mixed_schema"
	dbName := "timeseries_test_DB"

	setupTimeseriesWithMixedSchema(t, dbName, colName)

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = dbName
	md.OutputOptions.Out = "dump"
	md.OutputOptions.Out = ""
	md.OutputOptions.Archive = "dump.archive"

	require.NoError(t, md.Init())

	require.NoError(t, md.Dump())

	path, err := os.Getwd()
	require.NoError(t, err)

	archiveFilePath := filepath.FromSlash(filepath.Join(path, "dump.archive"))

	archiveFile, err := os.Open(archiveFilePath)
	require.NoError(t, err)
	archiveReader := &archive.Reader{
		In:      archiveFile,
		Prelude: &archive.Prelude{},
	}

	require.NoError(t, archiveReader.Prelude.Read(archiveReader.In))

	collectionMetadatas, ok := archiveReader.Prelude.NamespaceMetadatasByDB[dbName]
	require.True(t, ok)

	require.Len(t, collectionMetadatas, 1)
	require.Equal(t, colName, collectionMetadatas[0].Collection)

	pe, err := archiveReader.Prelude.NewPreludeExplorer()
	require.NoError(t, err)

	archiveContents, err := pe.ReadDir()
	require.NoError(t, err)

	expectedCollFile := timeseriesCollName(serverVersion, colName) + ".bson"

	for _, dirlike := range archiveContents {
		if dirlike.IsDir() && dirlike.Name() == dbName {
			dbContents, err := dirlike.ReadDir()
			require.NoError(t, err)

			require.Len(t, dbContents, 2)

			for _, file := range dbContents {
				require.Contains(
					t,
					[]string{colName + ".metadata.json", expectedCollFile},
					file.Name(),
				)
			}
		}
	}

	require.NoError(t, archiveFile.Close())
	require.NoError(t, os.Remove(archiveFilePath))
	require.NoError(t, session.Database(dbName).Collection(colName).Drop(t.Context()))
}

func TestFailDuringResharding(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	testutil.SkipForAtlasCluster(
		t,
		"this test requires permissions that may not be available for an Atlas cluster",
	)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	if err != nil {
		t.Fatalf("Failed to get session provider: %v", err)
	}

	session, err := sessionProvider.GetSession()
	if err != nil {
		t.Fatalf("Failed to get session: %v", err)
	}

	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "4.9"); err != nil || cmp < 0 {
		t.Skipf("Requires server with FCV 4.9 or later; found %v", fcv)
	}

	ctx := t.Context()

	if ok, _ := sessionProvider.IsReplicaSet(); !ok {
		t.Skipf("Not for replica sets")
	}

	Convey("With a MongoDump instance", t, func() {
		err := setUpMongoDumpTestData(t)
		So(err, ShouldBeNil)

		md, err := simpleMongoDumpInstance()
		So(err, ShouldBeNil)

		md.OutputOptions.Oplog = true
		md.ToolOptions.Namespace = &options.Namespace{}
		err = md.Init()
		So(err, ShouldBeNil)

		DefaultErrorMsg := "detected resharding in progress. Cannot dump with --oplog while resharding"
		OplogErrorMsg := "cannot dump with oplog while resharding"

		Convey("dump should fail if config.reshardingOperations exists on source", func() {
			err = session.Database("config").CreateCollection(ctx, "reshardingOperations")
			So(err, ShouldBeNil)
			//nolint:errcheck
			defer session.Database("config").Collection("reshardingOperations").Drop(ctx)

			err = md.Dump()
			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, DefaultErrorMsg)
		})

		Convey(
			"dump should fail if config.localReshardingOperations.donor exists on source",
			func() {
				err = session.Database("config").
					CreateCollection(ctx, "localReshardingOperations.donor")
				So(err, ShouldBeNil)
				//nolint:errcheck
				defer session.Database("config").
					Collection("localReshardingOperations.donor").
					Drop(ctx)

				err = md.Dump()
				So(err, ShouldNotBeNil)
				So(err.Error(), ShouldContainSubstring, DefaultErrorMsg)
			},
		)

		Convey(
			"dump should fail if config.localReshardingOperations.recipient exists on source",
			func() {
				err = session.Database("config").
					CreateCollection(ctx, "localReshardingOperations.recipient")
				So(err, ShouldBeNil)
				//nolint:errcheck
				defer session.Database("config").
					Collection("localReshardingOperations.recipient").
					Drop(ctx)

				err = md.Dump()
				So(err, ShouldNotBeNil)
				So(err.Error(), ShouldContainSubstring, DefaultErrorMsg)
			},
		)

		Convey("dump should fail if config.reshardingOperations created in oplog", func() {
			require.NoError(t, failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()))
			defer failpoint.DefaultManager.Reset()

			dumpErrCh := make(chan error, 1)
			go func() { dumpErrCh <- md.Dump() }()

			fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
			So(ok, ShouldBeTrue)
			require.NoError(t, fp.Reached(context.TODO()))
			sessErr1 := session.Database("config").CreateCollection(ctx, "reshardingOperations")
			sessErr2 := session.Database("config").Collection("reshardingOperations").Drop(ctx)
			fp.Signal()

			err = <-dumpErrCh

			So(err, ShouldNotBeNil)
			So(err.Error(), ShouldContainSubstring, OplogErrorMsg)
			So(sessErr1, ShouldBeNil)
			So(sessErr2, ShouldBeNil)
		})

		Convey(
			"dump should fail if config.localReshardingOperations.donor created in oplog",
			func() {
				require.NoError(
					t,
					failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()),
				)
				defer failpoint.DefaultManager.Reset()

				dumpErrCh := make(chan error, 1)
				go func() { dumpErrCh <- md.Dump() }()

				fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
				So(ok, ShouldBeTrue)
				require.NoError(t, fp.Reached(context.TODO()))
				sessErr1 := session.Database("config").
					CreateCollection(ctx, "localReshardingOperations.donor")
				sessErr2 := session.Database("config").
					Collection("localReshardingOperations.donor").
					Drop(ctx)
				fp.Signal()

				err = <-dumpErrCh

				So(err, ShouldNotBeNil)
				So(err.Error(), ShouldContainSubstring, OplogErrorMsg)
				So(sessErr1, ShouldBeNil)
				So(sessErr2, ShouldBeNil)
			},
		)

		Convey(
			"dump should fail if config.localReshardingOperations.recipient created in oplog",
			func() {
				require.NoError(
					t,
					failpoint.DefaultManager.Parse(failpoint.PauseUntilResumed.String()),
				)
				defer failpoint.DefaultManager.Reset()

				dumpErrCh := make(chan error, 1)
				go func() { dumpErrCh <- md.Dump() }()

				fp, ok := failpoint.DefaultManager.Get(failpoint.PauseUntilResumed)
				So(ok, ShouldBeTrue)
				require.NoError(t, fp.Reached(context.TODO()))
				sessErr1 := session.Database("config").
					CreateCollection(ctx, "localReshardingOperations.recipient")
				sessErr2 := session.Database("config").
					Collection("localReshardingOperations.recipient").
					Drop(ctx)
				fp.Signal()

				err = <-dumpErrCh

				So(err, ShouldNotBeNil)
				So(err.Error(), ShouldContainSubstring, OplogErrorMsg)
				So(sessErr1, ShouldBeNil)
				So(sessErr2, ShouldBeNil)
			},
		)

	})
}

func TestOptionsOrderIsPreserved(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err)

	collName := "students"
	viewName := "studentsView"

	pipeline := bson.A{
		bson.D{{Key: "$group", Value: bson.D{
			{Key: "_id", Value: bson.D{
				{Key: "year", Value: "$year"},
				{Key: "name", Value: "$name"},
			}},
			{Key: "highest", Value: bson.D{
				{Key: "$max", Value: "$score"},
			}},
		}}},
		bson.D{{Key: "$project", Value: bson.D{
			{Key: "_id", Value: 0},
		}}},
		bson.D{{Key: "$sort", Value: bson.D{
			{Key: "year", Value: 1},
			{Key: "sID", Value: -1},
			{Key: "name", Value: 1},
			{Key: "score", Value: 1},
		}}},
	}

	createViewCmd := bson.D{
		{"create", viewName},
		{"viewOn", collName},
		{"pipeline", pipeline},
	}

	var result bson.D
	err = sessionProvider.Run(createViewCmd, &result, testDB)
	require.NoError(t, err)

	// The check should be run a few times due to the probabilistic nature
	// of TOOLS-3411
	for i := 0; i < 10; i++ {
		dumpAndCheckPipelineOrder(t, collName, pipeline)
	}

	err = tearDownMongoDumpTestData(t)
	require.NoError(t, err)
}

func dumpAndCheckPipelineOrder(t *testing.T, collName string, pipeline bson.A) {
	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)

	md.ToolOptions.DB = testDB
	md.OutputOptions.Out = "dump"

	require.NoError(t, md.Init())
	require.NoError(t, md.Dump())

	path, err := os.Getwd()
	require.NoError(t, err)

	dumpDir := filepath.FromSlash(filepath.Join(path, "dump"))
	dumpDBDir := filepath.FromSlash(filepath.Join(dumpDir, testDB))
	require.True(t, fileDirExists(dumpDir))
	require.True(t, fileDirExists(dumpDBDir))

	metaFiles, err := getMatchingFiles(dumpDBDir, "studentsView\\.metadata\\.json")
	require.NoError(t, err)
	require.Equal(t, len(metaFiles), 1)

	metaFile, err := os.Open(filepath.FromSlash(filepath.Join(dumpDBDir, metaFiles[0])))

	require.NoError(t, err)
	contents, err := io.ReadAll(metaFile)
	require.NoError(t, err)

	var bsonResult bson.D
	err = bson.UnmarshalExtJSON(contents, true, &bsonResult)
	require.NoError(t, err)
	options, err := bsonutil.FindSubdocumentByKey("options", &bsonResult)
	require.NoError(t, err)

	assertBSONEqual(t, options, bson.D{
		{Key: "viewOn", Value: collName},
		{Key: "pipeline", Value: pipeline},
	})

	os.RemoveAll(dumpDir)
	metaFile.Close()
}

// TestBrokenPipe verifies that mongodump handles a broken pipe gracefully
// (exits with a write error rather than being killed by SIGPIPE).
func TestBrokenPipe(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	const (
		dbName   = "mongodump_broken_pipe_test"
		collName = "docs"
	)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err)
	client, err := sessionProvider.GetSession()
	require.NoError(t, err)
	t.Cleanup(func() {
		_ = client.Database(dbName).Drop(context.Background())
	})

	// Insert 2000 docs so the archive output exceeds the pipe buffer.
	docs := make([]any, 2000)
	for i := range 2000 {
		docs[i] = bson.D{{"_id", int32(i)}, {"data", strings.Repeat("x", 500)}}
	}
	_, err = client.Database(dbName).Collection(collName).InsertMany(context.Background(), docs)
	require.NoError(t, err)

	args := append(
		[]string{"run", filepath.Join("..", "mongodump", "main")},
		testutil.GetBareArgs()...,
	)
	args = append(args, "--db", dbName, "--archive=-")
	testutil.AssertBrokenPipeHandled(t, exec.Command("go", args...))
}

// TestMongoDumpMaxEstimatedScanBytes verifies that mongodump can dump a collection larger than the
// server's maxEstimatedScanBytes threshold (SERVER-127688, added in 9.0). An unfiltered dump plans
// an unbounded COLLSCAN, which the server rejects with NoQueryExecutionPlans unless the query
// carries a $natural hint. The view case additionally covers the count that mongodump issues
// before dumping, which goes through CountDocuments rather than the find.
func TestMongoDumpMaxEstimatedScanBytes(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)
	log.SetWriter(io.Discard)

	sessionProvider, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err, "get session provider")
	defer sessionProvider.Close()

	serverVersion, err := sessionProvider.ServerVersionArray()
	require.NoError(t, err, "get server version")
	if serverVersion.LT(db.Version{9, 0, 0}) {
		t.Skipf("maxEstimatedScanBytes was added in 9.0; testing with %s", serverVersion.String())
	}

	// setParameter on a mongos does not reach the shards, so the knob would never be applied to
	// the nodes that actually plan the query.
	isMongos, err := sessionProvider.IsMongos()
	require.NoError(t, err, "check for mongos")
	if isMongos {
		t.Skip("setParameter does not propagate from mongos to shards")
	}

	session, err := sessionProvider.GetSession()
	require.NoError(t, err, "get session")

	const collName = "max_estimated_scan_bytes"
	const viewName = "max_estimated_scan_bytes_view"
	const numDocs = 100

	coll := session.Database(testDB).Collection(collName)
	require.NoError(t, coll.Drop(t.Context()), "drop existing collection")
	_ = session.Database(testDB).Collection(viewName).Drop(t.Context())

	docs := make([]any, numDocs)
	for i := range docs {
		docs[i] = bson.D{{"_id", i}, {"pad", strings.Repeat("x", 1024)}}
	}
	_, err = coll.InsertMany(t.Context(), docs)
	require.NoError(t, err, "insert test documents")

	err = session.Database(testDB).
		CreateView(t.Context(), viewName, collName, mongo.Pipeline{})
	require.NoError(t, err, "create view")

	// The collection is ~100KB, so any threshold well below that triggers the rejection.
	setKnob := func(value any) {
		var res bson.M
		err := sessionProvider.Run(
			bson.D{{"setParameter", 1}, {"maxEstimatedScanBytes", value}},
			&res,
			"admin",
		)
		require.NoError(t, err, "set maxEstimatedScanBytes to %v", value)
	}
	setKnob(int64(1024))
	defer setKnob(int64(-1))

	countDumpedDocs := func(t *testing.T, dumpDir, name string) int {
		file, err := os.Open(filepath.Join(dumpDir, testDB, name+".bson"))
		require.NoError(t, err, "open dumped %s", name)
		defer file.Close()

		bsonSource := db.NewDecodedBSONSource(db.NewBSONSource(file))
		defer bsonSource.Close()

		count := 0
		var result bson.D
		for bsonSource.Next(&result) {
			count++
		}
		require.NoError(t, bsonSource.Err(), "read dumped %s", name)
		return count
	}

	// --viewsAsCollections omits standard collections, so the find path and the count path need
	// separate dumps.
	t.Run("collection", func(t *testing.T) {
		dumpDir, cleanupDir := testutil.MakeTempDir(t)
		defer cleanupDir()

		md, err := simpleMongoDumpInstance()
		require.NoError(t, err, "create mongodump instance")
		md.OutputOptions.Out = dumpDir
		require.NoError(t, md.Init(), "mongodump init")
		require.NoError(t, md.Dump(), "dump a collection larger than maxEstimatedScanBytes")

		assert.Equal(
			t,
			numDocs,
			countDumpedDocs(t, dumpDir, collName),
			"all documents were dumped from the collection",
		)
	})

	t.Run("view", func(t *testing.T) {
		dumpDir, cleanupDir := testutil.MakeTempDir(t)
		defer cleanupDir()

		md, err := simpleMongoDumpInstance()
		require.NoError(t, err, "create mongodump instance")
		md.OutputOptions.Out = dumpDir
		md.OutputOptions.ViewsAsCollections = true
		require.NoError(t, md.Init(), "mongodump init")
		require.NoError(
			t,
			md.Dump(),
			"dump a view over a collection larger than maxEstimatedScanBytes",
		)

		assert.Equal(
			t,
			numDocs,
			countDumpedDocs(t, dumpDir, viewName),
			"all documents were dumped from the view",
		)
	})
}

// TestTimeseriesDumpConcurrentWithSetFCV exercises the open item from the TOOLS-4182
// investigation: a long-running mongodump that overlaps a setFCV, which on 9.0+ converts
// timeseries collections between their viewless and viewful forms (SERVER-114505). The
// server does not currently expect the collection format to change mid-operation, so the
// point of this test is to pin down that the dump either completes with every measurement
// or fails outright — it must never silently produce a truncated dump.
func TestTimeseriesDumpConcurrentWithSetFCV(t *testing.T) {
	testtype.SkipUnlessTestType(t, testtype.IntegrationTestType)

	ctx := t.Context()

	session, err := testutil.GetBareSession()
	require.NoError(t, err, "get session")

	sp, _, err := testutil.GetBareSessionProvider()
	require.NoError(t, err, "get session provider")

	serverVersion, err := sp.ServerVersionArray()
	require.NoError(t, err, "get server version")
	if serverVersion.LT(db.Version{9, 0, 0}) {
		t.Skipf(
			"the viewless timeseries conversion requires a 9.0 server; found %v",
			serverVersion,
		)
	}

	fcv := testutil.GetFCV(session)
	if cmp, err := testutil.CompareFCV(fcv, "9.0"); err != nil || cmp < 0 {
		t.Skipf("Requires FCV 9.0 so that setFCV downgrades to a viewful format; found %v", fcv)
	}

	const dbName = "ts_setfcv_test_DB"
	const collName = "timeseriesColl"
	// Enough distinct meta values to produce more buckets than fit in one cursor batch, so
	// the dump must issue a getMore that we can block on.
	const numBuckets = 500

	require.NoError(t, session.Database(dbName).Drop(ctx))
	t.Cleanup(func() { _ = session.Database(dbName).Drop(context.Background()) })

	setUpBucketedTimeseries(t, session, dbName, collName, numBuckets)

	// The failpoint targets this app name only, so the test's own commands — including the
	// setFCV below — are not blocked.
	const dumpAppName = "mongodump_setfcv_test"

	md, err := simpleMongoDumpInstance()
	require.NoError(t, err)
	md.ToolOptions.DB = dbName
	md.ToolOptions.AppName = dumpAppName
	md.OutputOptions.Out = t.TempDir()

	require.NoError(t, md.Init())

	blockGetMoreForAppName(t, session, dumpAppName)

	dumpErr := make(chan error, 1)
	go func() { dumpErr <- md.Dump() }()

	waitForBlockedGetMore(t, session, dumpAppName)

	// This is the conversion under test: downgrading from 9.0 turns the viewless timeseries
	// collection back into a view plus a system.buckets collection, underneath the dump.
	var fcvResult bson.D
	setFCVErr := session.Database("admin").RunCommand(ctx, bson.D{
		{"setFeatureCompatibilityVersion", "8.0"},
		{"confirm", true},
	}).Decode(&fcvResult)
	t.Cleanup(func() {
		_ = session.Database("admin").RunCommand(context.Background(), bson.D{
			{"setFeatureCompatibilityVersion", "9.0"},
			{"confirm", true},
		}).Err()
	})

	err = <-dumpErr

	// A failed setFCV means the conversion never happened, so there is nothing to assert.
	require.NoError(t, setFCVErr, "setFCV should succeed while a dump is running")

	if err != nil {
		// The server interrupts reads whose collection is converted underneath them, and
		// mongodump surfaces that as a dump failure. Losing the race is the expected outcome;
		// what matters is that it is reported rather than silently truncating the dump.
		require.ErrorContains(
			t,
			err,
			"InterruptedDueToTimeseriesUpgradeDowngrade",
			"a dump interrupted by the conversion should fail with the server's specific error",
		)
		return
	}

	dumped := countDumpedMeasurements(t, filepath.Join(md.OutputOptions.Out, dbName))
	assert.EqualValues(
		t,
		numBuckets,
		dumped,
		"a dump that reports success must contain every measurement, even though the "+
			"collection format changed underneath it",
	)
}

// setUpBucketedTimeseries creates a timeseries collection holding one measurement per distinct
// meta value, so that the collection contains numBuckets buckets.
func setUpBucketedTimeseries(
	t *testing.T,
	session *mongo.Client,
	dbName, collName string,
	numBuckets int,
) {
	t.Helper()

	ctx := t.Context()

	require.NoError(t, session.Database(dbName).RunCommand(ctx, bson.D{
		{"create", collName},
		{"timeseries", bson.D{{"timeField", "ts"}, {"metaField", "my_meta"}}},
	}).Err(), "create timeseries collection")

	docs := make([]any, 0, numBuckets)
	for i := range numBuckets {
		docs = append(docs, bson.D{
			{"ts", bson.NewDateTimeFromTime(time.Now())},
			{"my_meta", bson.D{{"device", i}}},
			{"measurement", i},
		})
	}

	_, err := session.Database(dbName).Collection(collName).InsertMany(ctx, docs)
	require.NoError(t, err, "insert measurements")
}

// blockGetMoreForAppName makes every getMore from the given app name hang, so that a dump from
// that app name stays open long enough for a concurrent setFCV to run.
func blockGetMoreForAppName(t *testing.T, session *mongo.Client, appName string) {
	t.Helper()

	err := session.Database("admin").RunCommand(t.Context(), bson.D{
		{"configureFailPoint", "failCommand"},
		{"mode", "alwaysOn"},
		{"data", bson.D{
			{"failCommands", bson.A{"getMore"}},
			{"appName", appName},
			{"blockConnection", true},
			{"blockTimeMS", 10000},
		}},
	}).Err()

	var cmdErr mongo.CommandError
	if errors.As(err, &cmdErr) && cmdErr.Name == "CommandNotFound" {
		t.Skip("Requires a server started with enableTestCommands=1")
	}
	require.NoError(t, err, "enable the getMore failpoint")

	t.Cleanup(func() {
		_ = session.Database("admin").RunCommand(context.Background(), bson.D{
			{"configureFailPoint", "failCommand"},
			{"mode", "off"},
		}).Err()
	})
}

// waitForBlockedGetMore blocks until the dump's getMore is visible in currentOp, which means
// the dump is mid-read and it is safe to change the collection format underneath it.
func waitForBlockedGetMore(t *testing.T, session *mongo.Client, appName string) {
	t.Helper()

	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		var result struct {
			Inprog []bson.D `bson:"inprog"`
		}
		err := session.Database("admin").RunCommand(t.Context(), bson.D{
			{"currentOp", 1},
			{"appName", appName},
			{"op", "getmore"},
		}).Decode(&result)
		require.NoError(t, err, "run currentOp")

		if len(result.Inprog) > 0 {
			return
		}

		time.Sleep(100 * time.Millisecond)
	}

	t.Fatalf("timed out waiting for a blocked getMore from app name %#q", appName)
}

// countDumpedMeasurements sums the measurement counts of the bucket documents in the timeseries
// BSON file in the given dump directory.
func countDumpedMeasurements(t *testing.T, dumpDir string) int {
	t.Helper()

	entries, err := os.ReadDir(dumpDir)
	require.NoError(t, err, "read dump directory")

	total := 0
	for _, entry := range entries {
		if !strings.HasSuffix(entry.Name(), ".bson") {
			continue
		}

		file, err := os.Open(filepath.Join(dumpDir, entry.Name()))
		require.NoError(t, err)
		defer file.Close()

		bsonSource := db.NewDecodedBSONSource(db.NewBSONSource(file))
		defer bsonSource.Close()

		var bucket struct {
			Control struct {
				Count int `bson:"count"`
			} `bson:"control"`
		}
		for bsonSource.Next(&bucket) {
			total += bucket.Control.Count
		}
		require.NoError(t, bsonSource.Err(), "read dumped buckets")
	}

	return total
}
