catch expanded events

Author: FGasper

internal/verifier/change_stream_test.go

@@ -5,6 +5,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/10gen/migration-verifier/mslices"
 	"github.com/pkg/errors"
 	"github.com/samber/lo"
 	"github.com/stretchr/testify/require"
@@ -35,6 +36,12 @@ func TestChangeStreamFilter(t *testing.T) {
 // terminates that verifier, updates the source cluster, starts a new
 // verifier with change stream, and confirms that things look as they should.
 func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
+	suite.Require().NoError(
+		suite.srcMongoClient.
+			Database(suite.DBNameForTest()).
+			CreateCollection(suite.Context(), "testColl"),
+	)
+
 	func() {
 		verifier1 := suite.BuildVerifier()
 		ctx, cancel := context.WithCancel(context.Background())
@@ -43,7 +50,7 @@ func (suite *IntegrationTestSuite) TestChangeStreamResumability() {
 		suite.Require().NoError(err)
 	}()
 
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(suite.Context())
 	defer cancel()
 
 	_, err := suite.srcMongoClient.
@@ -219,14 +226,21 @@ func (suite *IntegrationTestSuite) TestNoStartAtTime() {
 }
 
 func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
-	verifier := suite.BuildVerifier()
+	ctx := suite.Context()
 
-	ctx, cancel := context.WithCancel(context.Background())
-	defer cancel()
+	db := suite.srcMongoClient.Database(suite.DBNameForTest())
+	coll1 := db.Collection("testColl1")
+	coll2 := db.Collection("testColl2")
+
+	for _, coll := range mslices.Of(coll1, coll2) {
+		suite.Require().NoError(db.CreateCollection(ctx, coll.Name()))
+	}
+
+	verifier := suite.BuildVerifier()
 
 	suite.Require().NoError(verifier.StartChangeStream(ctx))
 
-	_, err := suite.srcMongoClient.Database("testDb").Collection("testColl1").InsertOne(ctx, bson.D{{"_id", 1}})
+	_, err := coll1.InsertOne(ctx, bson.D{{"_id", 1}})
 	suite.Require().NoError(err)
 	_, err = suite.srcMongoClient.Database("testDb").Collection("testColl1").InsertOne(ctx, bson.D{{"_id", 2}})
 	suite.Require().NoError(err)
@@ -245,6 +259,7 @@ func (suite *IntegrationTestSuite) TestWithChangeEventsBatching() {
 		500*time.Millisecond,
 		"the verifier should flush a recheck doc after a batch",
 	)
+
 }
 
 func (suite *IntegrationTestSuite) TestManyInsertsBeforeWritesOff() {