GODRIVER-1858 Use mock deployment for slow BulkWrite test (#584)

Author: Divjot Arora

mongo/integration/change_stream_test.go

@@ -236,6 +236,7 @@ func TestChangeStream_ReplicaSet(t *testing.T) {
 			Code:    errorHostUnreachable,
 			Name:    "foo",
 			Message: "bar",
+			Labels:  []string{"ResumableChangeStreamError"},
 		})
 		killCursorsRes := mtest.CreateCommandErrorResponse(mtest.CommandError{
 			Code:    errorInterrupted,

mongo/integration/collection_test.go

@@ -1661,9 +1661,10 @@ func TestCollection(t *testing.T) {
 				mt.Fatalf("expected BulkWrite error %v, got %v", mongo.ErrUnacknowledgedWrite, err)
 			}
 		})
-		mt.Run("insert and delete with batches", func(mt *mtest.T) {
+		mt.RunOpts("insert and delete with batches", mtest.NewOptions().ClientType(mtest.Mock), func(mt *mtest.T) {
 			// grouped together because delete requires the documents to be inserted
-			numDocs := 100050
+			maxBatchCount := int(mtest.MockDescription.MaxBatchCount)
+			numDocs := maxBatchCount + 50
 			var insertModels []mongo.WriteModel
 			var deleteModels []mongo.WriteModel
 			for i := 0; i < numDocs; i++ {
@@ -1675,6 +1676,21 @@ func TestCollection(t *testing.T) {
 				insertModels = append(insertModels, mongo.NewInsertOneModel().SetDocument(d))
 				deleteModels = append(deleteModels, mongo.NewDeleteOneModel().SetFilter(bson.D{}))
 			}
+
+			// Seed mock responses. Both insert and delete respones look like {ok: 1, n: <inserted/deleted count>}.
+			// This loop only creates one set of responses, but the sets for insert and delete should be equivalent,
+			// so we can duplicate the generated set before calling mt.AddMockResponses().
+			var responses []bson.D
+			for i := numDocs; i > 0; i -= maxBatchCount {
+				count := maxBatchCount
+				if i < maxBatchCount {
+					count = i
+				}
+				res := mtest.CreateSuccessResponse(bson.E{"n", count})
+				responses = append(responses, res)
+			}
+			mt.AddMockResponses(append(responses, responses...)...)
+
 			mt.ClearEvents()
 			res, err := mt.Coll.BulkWrite(mtest.Background, insertModels)
 			assert.Nil(mt, err, "BulkWrite error: %v", err)

mongo/integration/mtest/opmsg_deployment.go

@@ -15,6 +15,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/description"
 	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/x/mongo/driver"
+	"go.mongodb.org/mongo-driver/x/mongo/driver/topology"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/wiremessage"
 )
 
@@ -24,7 +25,22 @@ const (
 	maxMessageSize        uint32 = 48000000
 	maxBatchCount         uint32 = 100000
 	sessionTimeoutMinutes uint32 = 30
-	maxWireVersion        int32  = 8
+)
+
+var (
+	// MockDescription is the server description used for the mock deployment. Each mocked connection returns this
+	// value from its Description method.
+	MockDescription = description.Server{
+		CanonicalAddr:         serverAddress,
+		MaxDocumentSize:       maxDocumentSize,
+		MaxMessageSize:        maxMessageSize,
+		MaxBatchCount:         maxBatchCount,
+		SessionTimeoutMinutes: sessionTimeoutMinutes,
+		Kind:                  description.RSPrimary,
+		WireVersion: &description.VersionRange{
+			Max: topology.SupportedWireVersions.Max,
+		},
+	}
 )
 
 // connection implements the driver.Connection interface and responds to wire messages with pre-configured responses.
@@ -59,17 +75,7 @@ func (c *connection) ReadWireMessage(_ context.Context, dst []byte) ([]byte, err
 
 // Description returns a fixed server description for the connection.
 func (c *connection) Description() description.Server {
-	return description.Server{
-		CanonicalAddr:         serverAddress,
-		MaxDocumentSize:       maxDocumentSize,
-		MaxMessageSize:        maxMessageSize,
-		MaxBatchCount:         maxBatchCount,
-		SessionTimeoutMinutes: sessionTimeoutMinutes,
-		Kind:                  description.RSPrimary,
-		WireVersion: &description.VersionRange{
-			Max: maxWireVersion,
-		},
-	}
+	return MockDescription
 }
 
 // Close is a no-op operation.

x/mongo/driver/topology/fsm.go

@@ -16,8 +16,15 @@ import (
 	"go.mongodb.org/mongo-driver/mongo/description"
 )
 
-var supportedWireVersions = description.NewVersionRange(2, 9)
-var minSupportedMongoDBVersion = "2.6"
+var (
+	// SupportedWireVersions is the range of wire versions supported by the driver.
+	SupportedWireVersions = description.NewVersionRange(2, 9)
+)
+
+const (
+	// MinSupportedMongoDBVersion is the version string for the lowest MongoDB version supported by the driver.
+	MinSupportedMongoDBVersion = "2.6"
+)
 
 type fsm struct {
 	description.Topology
@@ -89,27 +96,27 @@ func (f *fsm) apply(s description.Server) (description.Topology, description.Ser
 
 	for _, server := range f.Servers {
 		if server.WireVersion != nil {
-			if server.WireVersion.Max < supportedWireVersions.Min {
+			if server.WireVersion.Max < SupportedWireVersions.Min {
 				f.compatible.Store(false)
 				f.compatibilityErr = fmt.Errorf(
 					"server at %s reports wire version %d, but this version of the Go driver requires "+
 						"at least %d (MongoDB %s)",
 					server.Addr.String(),
 					server.WireVersion.Max,
-					supportedWireVersions.Min,
-					minSupportedMongoDBVersion,
+					SupportedWireVersions.Min,
+					MinSupportedMongoDBVersion,
 				)
 				f.Topology.CompatibilityErr = f.compatibilityErr
 				return f.Topology, s, nil
 			}
 
-			if server.WireVersion.Min > supportedWireVersions.Max {
+			if server.WireVersion.Min > SupportedWireVersions.Max {
 				f.compatible.Store(false)
 				f.compatibilityErr = fmt.Errorf(
 					"server at %s requires wire version %d, but this version of the Go driver only supports up to %d",
 					server.Addr.String(),
 					server.WireVersion.Min,
-					supportedWireVersions.Max,
+					SupportedWireVersions.Max,
 				)
 				f.Topology.CompatibilityErr = f.compatibilityErr
 				return f.Topology, s, nil

x/mongo/driver/topology/server_test.go

@@ -576,7 +576,7 @@ func (p *processErrorTestConn) Stale() bool {
 func (p *processErrorTestConn) Description() description.Server {
 	return description.Server{
 		WireVersion: &description.VersionRange{
-			Max: supportedWireVersions.Max,
+			Max: SupportedWireVersions.Max,
 		},
 		TopologyVersion: p.tv,
 	}

x/mongo/driver/topology/topology_test.go

@@ -100,7 +100,7 @@ func TestServerSelection(t *testing.T) {
 			"server at %s requires wire version %d, but this version of the Go driver only supports up to %d",
 			desc.Servers[0].Addr.String(),
 			desc.Servers[0].WireVersion.Min,
-			supportedWireVersions.Max,
+			SupportedWireVersions.Max,
 		)
 		desc.CompatibilityErr = want
 		atomic.StoreInt32(&topo.connectionstate, connected)
@@ -124,8 +124,8 @@ func TestServerSelection(t *testing.T) {
 				"at least %d (MongoDB %s)",
 			desc.Servers[0].Addr.String(),
 			desc.Servers[0].WireVersion.Max,
-			supportedWireVersions.Min,
-			minSupportedMongoDBVersion,
+			SupportedWireVersions.Min,
+			MinSupportedMongoDBVersion,
 		)
 		desc.CompatibilityErr = want
 		atomic.StoreInt32(&topo.connectionstate, connected)