PBM-1649 Tests for isRequiredForOplogSlicing

Author: jcechace

pbm/backup/delete_test.go

@@ -131,15 +131,15 @@ func (tenv *TestEnvironment) Cleanup(ctx context.Context) error {
 	return errors.Join(mongoErr, stgErr)
 }
 
-func (tenv *TestEnvironment) Reset(t *testing.T) {
+func (tenv *TestEnvironment) ResetWithConfig(t *testing.T, baseCfg *config.Config) {
 	if err := tenv.resetStorage(); err != nil {
 		assert.FailNow(t, "failed to reset storage: %v", err)
 	}
 	if err := tenv.resetMongo(t.Context()); err != nil {
 		assert.FailNow(t, "failed to reset mongo: %v", err)
 	}
 
-	cfg, err := tenv.SetConfigWithStorage(nil)
+	cfg, err := tenv.SetConfigWithStorage(baseCfg)
 	if err != nil {
 		assert.FailNow(t, "failed to reset config: %v", err)
 	}
@@ -150,6 +150,10 @@ func (tenv *TestEnvironment) Reset(t *testing.T) {
 	}
 }
 
+func (tenv *TestEnvironment) Reset(t *testing.T) {
+	tenv.ResetWithConfig(t, nil)
+}
+
 func (tenv *TestEnvironment) SetConfigWithStorage(cfg *config.Config) (*config.Config, error) {
 	if cfg == nil {
 		cfg = &config.Config{}
@@ -335,6 +339,87 @@ func NewTestStorage(stgType storage.Type) *TestStorage {
 	return &TestStorage{stgType: stgType}
 }
 
+func Test_isRequiredForOplogSlicing(t *testing.T) {
+	tm := NewRelativeTime(time.Now(), time.Minute)
+
+	tests := []struct {
+		name        string
+		lwt         time.Time
+		baseLWT     time.Time
+		backups     []bcp
+		chunks      chunks
+		pitrEnabled [2]bool
+		expected    bool
+	}{
+		{
+			name:        "more recent snapshot exists",
+			pitrEnabled: [2]bool{true, false},
+			lwt:         tm(20),
+			backups:     []bcp{{LwT: tm(25)}, {LwT: tm(20)}, {LwT: tm(15)}},
+			chunks: chunks{
+				{From: tm(14), To: tm(5)},
+			},
+			expected: false,
+		},
+		{
+			name:        "gap in pitr timeline",
+			pitrEnabled: [2]bool{true, false},
+			lwt:         tm(20),
+			backups:     []bcp{{LwT: tm(30)}, {LwT: tm(20)}},
+			chunks: chunks{
+				{From: tm(29), To: tm(25)},
+				{From: tm(22), To: tm(20)},
+			},
+			expected: false,
+		},
+		{
+			name:        "continuous pitr timeline (single chunk)",
+			pitrEnabled: [2]bool{true, false},
+			lwt:         tm(20),
+			backups:     []bcp{{LwT: tm(30)}, {LwT: tm(20)}},
+			chunks: chunks{
+				{From: tm(29), To: tm(20)},
+			},
+			expected: true,
+		},
+		{
+			name:        "continuous pitr timeline (multiple chunks)",
+			pitrEnabled: [2]bool{true, false},
+			lwt:         tm(20),
+			backups:     []bcp{{LwT: tm(30)}, {LwT: tm(20)}},
+			chunks: chunks{
+				{From: tm(29), To: tm(25)},
+				{From: tm(25), To: tm(20)},
+			},
+			expected: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg := &config.Config{
+				PITR: &config.PITRConf{
+					Enabled:   tt.pitrEnabled[0],
+					OplogOnly: tt.pitrEnabled[1],
+				},
+			}
+			TestEnv.ResetWithConfig(t, cfg)
+			insertTestBackupsStorage(t, TestEnv, TestEnv.StorageEnv, tt.backups)
+			insertTestChunks(t, TestEnv, tt.chunks)
+
+			lwtBson := primitive.Timestamp{T: uint32(tt.lwt.Unix())}
+			baseLWTBson := primitive.Timestamp{T: uint32(tt.baseLWT.Unix())}
+			if baseLWTBson.IsZero() {
+				baseLWTBson = primitive.Timestamp{}
+			}
+
+			actual, err := isRequiredForOplogSlicing(t.Context(), TestEnv.Mongo.Client, lwtBson, baseLWTBson)
+			assert.NoError(t, err)
+			assert.Equal(t, tt.expected, actual)
+		})
+	}
+}
+
 func Test_DeleteBackupUnsafe(t *testing.T) {
 	tests := []struct {
 		name   string
@@ -790,6 +875,17 @@ func stgsFromTestBackups(t *testing.T, env *TestEnvironment, backups bcps) map[s
 	return storages
 }
 
+func insertTestBackupsStorage(t *testing.T, env *TestEnvironment, stg Storage, bcps []bcp) []BackupMeta {
+	inserted := make([]BackupMeta, len(bcps))
+	for i, b := range bcps {
+		if b.Name == "" {
+			b.Name = fmt.Sprintf("backup-%s-%d", stg.Name, i)
+		}
+		inserted[i] = insertTestBcpMeta(t, env, stg, b)
+	}
+	return inserted
+}
+
 func insertTestBackups(
 	t *testing.T,
 	env *TestEnvironment,
@@ -800,12 +896,10 @@ func insertTestBackups(
 
 	tbd := make(map[string][]BackupMeta)
 	for profile, bcps := range profileBcps {
+		inserted := insertTestBackupsStorage(t, env, storages[profile], bcps)
 		for i, bcp := range bcps {
-			bcp.Name = fmt.Sprintf("backup-%s-%d", profile, i)
-			inserted := insertTestBcpMeta(t, env, storages[profile], bcp)
-
 			if bcp.Delete {
-				tbd[profile] = append(tbd[profile], inserted)
+				tbd[profile] = append(tbd[profile], inserted[i])
 			}
 		}
 	}