Merge pull request #4736 from danfengliu/add-e2e-test-of-bsl-deletion

Add E2E test of bsl deletion
2025-12-23 06:15:21 +00:00 · 2022-03-23 14:34:00 +08:00
parent 33026a0fbe 09cdf41d97
commit 38c83d608a
16 changed files with 620 additions and 65 deletions
--- a/go.mod
+++ b/go.mod
@@ -17,6 +17,7 @@ require (
 	github.com/gobwas/glob v0.2.3
 	github.com/gofrs/uuid v3.2.0+incompatible
 	github.com/golang/protobuf v1.5.2
+	github.com/google/go-cmp v0.5.6
 	github.com/google/uuid v1.2.0
 	github.com/hashicorp/go-hclog v0.12.0
 	github.com/hashicorp/go-plugin v0.0.0-20190610192547-a1bc61569a26
@@ -71,7 +72,6 @@ require (
 	github.com/gobuffalo/flect v0.2.3 // indirect
 	github.com/gogo/protobuf v1.3.2 // indirect
 	github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
-	github.com/google/go-cmp v0.5.6 // indirect
 	github.com/google/gofuzz v1.2.0 // indirect
 	github.com/googleapis/gax-go/v2 v2.1.0 // indirect
 	github.com/googleapis/gnostic v0.5.5 // indirect
@@ -108,6 +108,7 @@ require (
 	golang.org/x/term v0.0.0-20210220032956-6a3ed077a48d // indirect
 	golang.org/x/text v0.3.7 // indirect
 	golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac // indirect
+	golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
 	gomodules.xyz/jsonpatch/v2 v2.2.0 // indirect
 	google.golang.org/appengine v1.6.7 // indirect
 	google.golang.org/genproto v0.0.0-20210828152312-66f60bf46e71 // indirect
--- a/test/e2e/backups/deletion.go
+++ b/test/e2e/backups/deletion.go
@@ -102,7 +102,7 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 	if err != nil {
 		return err
 	}
-	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, deletionTest, backupLocation, useVolumeSnapshots); err != nil {
+	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, deletionTest, backupLocation, useVolumeSnapshots, ""); err != nil {
 		// TODO currently, the upgrade case covers the upgrade path from 1.6 to main and the velero v1.6 doesn't support "debug" command
 		// TODO move to "runDebug" after we bump up to 1.7 in the upgrade case
 		VeleroBackupLogs(context.Background(), VeleroCfg.UpgradeFromVeleroCLI, veleroNamespace, backupName)
@@ -125,7 +125,7 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 		var snapshotCheckPoint SnapshotCheckPoint
 		snapshotCheckPoint.ExpectCount = 2
 		snapshotCheckPoint.NamespaceBackedUp = deletionTest
-		err = SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix, snapshotCheckPoint)
+		err = SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslConfig, backupName, snapshotCheckPoint)
 		if err != nil {
 			return err
 		}
@@ -140,14 +140,14 @@ func runBackupDeletionTests(client TestClient, veleroCLI, providerName, veleroNa
 		return err
 	}
 	if useVolumeSnapshots {
-		err = SnapshotsShouldNotExistInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslPrefix, bslConfig, backupName, BackupObjectsPrefix)
+		err = SnapshotsShouldNotExistInCloud(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket, bslConfig, backupName)
 		if err != nil {
 			return err
 		}
 	}

 	backupName = "backup-1-" + UUIDgen.String()
-	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, deletionTest, backupLocation, useVolumeSnapshots); err != nil {
+	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, deletionTest, backupLocation, useVolumeSnapshots, ""); err != nil {
 		// TODO currently, the upgrade case covers the upgrade path from 1.6 to main and the velero v1.6 doesn't support "debug" command
 		// TODO move to "runDebug" after we bump up to 1.7 in the upgrade case
 		VeleroBackupLogs(context.Background(), VeleroCfg.UpgradeFromVeleroCLI, veleroNamespace, backupName)
--- a/test/e2e/backups/sync_backups.go
+++ b/test/e2e/backups/sync_backups.go
@@ -29,7 +29,6 @@ import (
 	"github.com/google/uuid"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
-	"k8s.io/apimachinery/pkg/util/wait"

 	. "github.com/vmware-tanzu/velero/test/e2e"
 	. "github.com/vmware-tanzu/velero/test/e2e/util/k8s"
@@ -83,7 +82,7 @@ func BackupsSyncTest() {
 		}()

 		By(fmt.Sprintf("Backup the workload in %s namespace", test.testNS), func() {
-			if err = VeleroBackupNamespace(test.ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, test.testNS, "", false); err != nil {
+			if err = VeleroBackupNamespace(test.ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, test.testNS, "", false, ""); err != nil {
 				RunDebug(context.Background(), VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, "")
 			}
 			Expect(err).To(Succeed(), fmt.Sprintf("Failed to backup %s namespace", test.testNS))
@@ -116,7 +115,7 @@ func BackupsSyncTest() {
 		}()

 		By(fmt.Sprintf("Backup the workload in %s namespace", test.testNS), func() {
-			if err = VeleroBackupNamespace(test.ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, test.testNS, "", false); err != nil {
+			if err = VeleroBackupNamespace(test.ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, test.testNS, "", false, ""); err != nil {
 				RunDebug(context.Background(), VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, test.backupName, "")
 			}
 			Expect(err).To(Succeed(), fmt.Sprintf("Failed to backup %s namespace", test.testNS))
@@ -141,15 +140,5 @@ func BackupsSyncTest() {
 }

 func (b *SyncBackups) IsBackupsSynced() error {
-	return wait.PollImmediate(10*time.Second, 10*time.Minute, func() (bool, error) {
-		if exist, err := IsBackupExist(b.ctx, VeleroCfg.VeleroCLI, b.backupName); err != nil {
-			return false, err
-		} else {
-			if exist {
-				return true, nil
-			} else {
-				return false, nil
-			}
-		}
-	})
+	return WaitForBackupCreated(b.ctx, VeleroCfg.VeleroCLI, b.backupName, 10*time.Minute)
 }
--- a/test/e2e/basic/enable_api_group_versions.go
+++ b/test/e2e/basic/enable_api_group_versions.go
@@ -236,7 +236,7 @@ func runEnableAPIGroupVersionsTests(ctx context.Context, client TestClient, reso
 		backup := "backup-rockbands-" + UUIDgen.String() + "-" + strconv.Itoa(i)
 		namespacesStr := strings.Join(tc.namespaces, ",")

-		err = VeleroBackupNamespace(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, backup, namespacesStr, "", false)
+		err = VeleroBackupNamespace(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, backup, namespacesStr, "", false, "")
 		if err != nil {
 			RunDebug(context.Background(), VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, backup, "")
 			return errors.Wrapf(err, "back up %s namespaces on source cluster", namespacesStr)
--- a/test/e2e/bsl-mgmt/deletion.go
+++ b/test/e2e/bsl-mgmt/deletion.go
@@ -0,0 +1,330 @@
+/*
+Copyright the Velero contributors.
+
+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
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+package bslmgmt
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"time"
+
+	"github.com/google/go-cmp/cmp"
+	"github.com/google/go-cmp/cmp/cmpopts"
+	"github.com/google/uuid"
+	. "github.com/onsi/ginkgo"
+	. "github.com/onsi/gomega"
+
+	. "github.com/vmware-tanzu/velero/test/e2e"
+	. "github.com/vmware-tanzu/velero/test/e2e/util/k8s"
+	. "github.com/vmware-tanzu/velero/test/e2e/util/kibishii"
+
+	. "github.com/vmware-tanzu/velero/test/e2e/util/providers"
+	. "github.com/vmware-tanzu/velero/test/e2e/util/velero"
+)
+
+const (
+	// Please make sure length of this namespace should be shorter,
+	// otherwise ResticRepositories name verification will be wrong
+	// when making combination of ResticRepositories name(max length is 63)
+	bslDeletionTestNs = "bsl-deletion"
+)
+
+// Test backup and restore of Kibishi using restic
+
+func BslDeletionWithSnapshots() {
+	BslDeletionTest(true)
+}
+
+func BslDeletionWithRestic() {
+	BslDeletionTest(false)
+}
+func BslDeletionTest(useVolumeSnapshots bool) {
+	client, err := NewTestClient()
+	Expect(err).To(Succeed(), "Failed to instantiate cluster client for backup deletion tests")
+	less := func(a, b string) bool { return a < b }
+	BeforeEach(func() {
+		if useVolumeSnapshots && VeleroCfg.CloudProvider == "kind" {
+			Skip("Volume snapshots not supported on kind")
+		}
+		var err error
+		flag.Parse()
+		UUIDgen, err = uuid.NewRandom()
+		Expect(err).To(Succeed())
+		if VeleroCfg.InstallVelero {
+			Expect(VeleroInstall(context.Background(), &VeleroCfg, "", useVolumeSnapshots)).To(Succeed())
+		}
+	})
+
+	AfterEach(func() {
+		if VeleroCfg.InstallVelero {
+			Expect(DeleteNamespace(context.Background(), client, bslDeletionTestNs,
+				true)).To(Succeed(), fmt.Sprintf("failed to delete the namespace %q",
+				bslDeletionTestNs))
+			Expect(VeleroUninstall(context.Background(), VeleroCfg.VeleroCLI,
+				VeleroCfg.VeleroNamespace)).To(Succeed())
+		}
+	})
+
+	When("kibishii is the sample workload", func() {
+		It("Local backups and restic repos (if Velero was installed with Restic) will be deleted once the corresponding backup storage location is deleted", func() {
+			if VeleroCfg.AdditionalBSLProvider == "" {
+				Skip("no additional BSL provider given, not running multiple BackupStorageLocation with unique credentials tests")
+			}
+
+			if VeleroCfg.AdditionalBSLBucket == "" {
+				Skip("no additional BSL bucket given, not running multiple BackupStorageLocation with unique credentials tests")
+			}
+
+			if VeleroCfg.AdditionalBSLCredentials == "" {
+				Skip("no additional BSL credentials given, not running multiple BackupStorageLocation with unique credentials tests")
+			}
+
+			By(fmt.Sprintf("Add an additional plugin for provider %s", VeleroCfg.AdditionalBSLProvider), func() {
+				Expect(VeleroAddPluginsForProvider(context.TODO(), VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, VeleroCfg.AdditionalBSLProvider, VeleroCfg.AddBSLPlugins)).To(Succeed())
+			})
+
+			additionalBsl := fmt.Sprintf("bsl-%s", UUIDgen)
+			secretName := fmt.Sprintf("bsl-credentials-%s", UUIDgen)
+			secretKey := fmt.Sprintf("creds-%s", VeleroCfg.AdditionalBSLProvider)
+			files := map[string]string{
+				secretKey: VeleroCfg.AdditionalBSLCredentials,
+			}
+
+			By(fmt.Sprintf("Create Secret for additional BSL %s", additionalBsl), func() {
+				Expect(CreateSecretFromFiles(context.TODO(), client, VeleroCfg.VeleroNamespace, secretName, files)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Create additional BSL using credential %s", secretName), func() {
+				Expect(VeleroCreateBackupLocation(context.TODO(),
+					VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace,
+					additionalBsl,
+					VeleroCfg.AdditionalBSLProvider,
+					VeleroCfg.AdditionalBSLBucket,
+					VeleroCfg.AdditionalBSLPrefix,
+					VeleroCfg.AdditionalBSLConfig,
+					secretName,
+					secretKey,
+				)).To(Succeed())
+			})
+
+			backupName_1 := "backup1-" + UUIDgen.String()
+			backupName_2 := "backup2-" + UUIDgen.String()
+			oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*60)
+
+			backupLocation_1 := "default"
+			backupLocation_2 := additionalBsl
+			podName_1 := "kibishii-deployment-0"
+			podName_2 := "kibishii-deployment-1"
+
+			label_1 := "for=1"
+			// TODO remove when issue https://github.com/vmware-tanzu/velero/issues/4724 is fixed
+			//label_2 := "for!=1"
+			label_2 := "for=2"
+			By("Create namespace for sample workload", func() {
+				Expect(CreateNamespace(oneHourTimeout, client, bslDeletionTestNs)).To(Succeed())
+			})
+
+			By("Deploy sample workload of Kibishii", func() {
+				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, client, VeleroCfg.CloudProvider,
+					bslDeletionTestNs, VeleroCfg.RegistryCredentialFile)).To(Succeed())
+			})
+
+			// Restic can not backup PV only, so pod need to be labeled also
+			By("Label all 2 worker-pods of Kibishii", func() {
+				Expect(AddLabelToPod(context.Background(), podName_1, bslDeletionTestNs, label_1)).To(Succeed())
+				Expect(AddLabelToPod(context.Background(), "kibishii-deployment-1", bslDeletionTestNs, label_2)).To(Succeed())
+			})
+
+			By("Get all 2 PVCs of Kibishii and label them seprately ", func() {
+				pvc, err := GetPvcByPodName(context.Background(), bslDeletionTestNs, podName_1)
+				Expect(err).To(Succeed())
+				fmt.Println(pvc)
+				Expect(len(pvc)).To(Equal(1))
+				pvc1 := pvc[0]
+				pvc, err = GetPvcByPodName(context.Background(), bslDeletionTestNs, podName_2)
+				Expect(err).To(Succeed())
+				fmt.Println(pvc)
+				Expect(len(pvc)).To(Equal(1))
+				pvc2 := pvc[0]
+				Expect(AddLabelToPvc(context.Background(), pvc1, bslDeletionTestNs, label_1)).To(Succeed())
+				Expect(AddLabelToPvc(context.Background(), pvc2, bslDeletionTestNs, label_2)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Backup one of PV of sample workload by label-1 - Kibishii by the first BSL %s", backupLocation_1), func() {
+				// TODO currently, the upgrade case covers the upgrade path from 1.6 to main and the velero v1.6 doesn't support "debug" command
+				// TODO move to "runDebug" after we bump up to 1.7 in the upgrade case
+				Expect(VeleroBackupNamespace(oneHourTimeout, VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, backupName_1, bslDeletionTestNs,
+					backupLocation_1, useVolumeSnapshots, label_1)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Back up the other one PV of sample workload with label-2 into the additional BSL %s", backupLocation_2), func() {
+				Expect(VeleroBackupNamespace(oneHourTimeout, VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, backupName_2, bslDeletionTestNs,
+					backupLocation_2, useVolumeSnapshots, label_2)).To(Succeed())
+			})
+
+			if useVolumeSnapshots {
+				if VeleroCfg.CloudProvider == "vsphere" {
+					// TODO - remove after upload progress monitoring is implemented
+					By("Waiting for vSphere uploads to complete", func() {
+						Expect(WaitForVSphereUploadCompletion(oneHourTimeout, time.Hour,
+							bslDeletionTestNs)).To(Succeed())
+					})
+					By(fmt.Sprintf("Snapshot CR in backup %s should be created", backupName_1), func() {
+						Expect(SnapshotCRsCountShouldBe(context.Background(), bslDeletionTestNs,
+							backupName_1, 1)).To(Succeed())
+					})
+					By(fmt.Sprintf("Snapshot CR in backup %s should be created", backupName_2), func() {
+						Expect(SnapshotCRsCountShouldBe(context.Background(), bslDeletionTestNs,
+							backupName_2, 1)).To(Succeed())
+					})
+				}
+				var snapshotCheckPoint SnapshotCheckPoint
+				snapshotCheckPoint.NamespaceBackedUp = bslDeletionTestNs
+				By(fmt.Sprintf("Snapshot of bsl %s should be created in cloud object store", backupLocation_1), func() {
+					snapshotCheckPoint.ExpectCount = 1
+					snapshotCheckPoint.PodName = podName_1
+					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
+						VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket,
+						VeleroCfg.BSLConfig, backupName_1, snapshotCheckPoint)).To(Succeed())
+				})
+				By(fmt.Sprintf("Snapshot of bsl %s should be created in cloud object store", backupLocation_2), func() {
+					snapshotCheckPoint.ExpectCount = 1
+					snapshotCheckPoint.PodName = podName_2
+					var BSLCredentials, BSLConfig string
+					if VeleroCfg.CloudProvider == "vsphere" {
+						BSLCredentials = VeleroCfg.AdditionalBSLCredentials
+						BSLConfig = VeleroCfg.AdditionalBSLConfig
+					} else {
+						BSLCredentials = VeleroCfg.CloudCredentialsFile
+						BSLConfig = VeleroCfg.BSLConfig
+					}
+					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
+						BSLCredentials, VeleroCfg.AdditionalBSLBucket,
+						BSLConfig, backupName_2, snapshotCheckPoint)).To(Succeed())
+				})
+			} else { // For Restics
+				By(fmt.Sprintf("Resticrepositories for BSL %s should be created in Velero namespace", backupLocation_1), func() {
+					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_1, 1)).To(Succeed())
+				})
+				By(fmt.Sprintf("Resticrepositories for BSL %s should be created in Velero namespace", backupLocation_2), func() {
+					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_2, 1)).To(Succeed())
+				})
+			}
+
+			By(fmt.Sprintf("Verify if backup %s is created or not", backupName_1), func() {
+				Expect(WaitForBackupCreated(context.Background(), VeleroCfg.VeleroCLI,
+					backupName_1, 10*time.Minute)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Verify if backup %s is created or not", backupName_2), func() {
+				Expect(WaitForBackupCreated(context.Background(), VeleroCfg.VeleroCLI,
+					backupName_2, 10*time.Minute)).To(Succeed())
+			})
+
+			backupsInBSL1, err := GetBackupsFromBsl(context.Background(), VeleroCfg.VeleroCLI, backupLocation_1)
+			Expect(err).To(Succeed())
+			backupsInBSL2, err := GetBackupsFromBsl(context.Background(), VeleroCfg.VeleroCLI, backupLocation_2)
+			Expect(err).To(Succeed())
+			backupsInBsl1AndBsl2 := append(backupsInBSL1, backupsInBSL2...)
+
+			By(fmt.Sprintf("Get all backups from 2 BSLs %s before deleting one of them", backupLocation_1), func() {
+				backupsBeforeDel, err := GetAllBackups(context.Background(), VeleroCfg.VeleroCLI)
+				Expect(err).To(Succeed())
+				Expect(cmp.Diff(backupsInBsl1AndBsl2, backupsBeforeDel, cmpopts.SortSlices(less))).Should(BeEmpty())
+
+				By(fmt.Sprintf("Backup1 %s should exist in cloud object store before bsl deletion", backupName_1), func() {
+					Expect(ObjectsShouldBeInBucket(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile,
+						VeleroCfg.BSLBucket, VeleroCfg.BSLPrefix, VeleroCfg.BSLConfig,
+						backupName_1, BackupObjectsPrefix)).To(Succeed())
+				})
+
+				By(fmt.Sprintf("Delete one of backup locations - %s", backupLocation_1), func() {
+					Expect(DeleteBslResource(context.Background(), VeleroCfg.VeleroCLI, backupLocation_1)).To(Succeed())
+				})
+
+				By("Get all backups from 2 BSLs after deleting one of them", func() {
+					backupsAfterDel, err := GetAllBackups(context.Background(), VeleroCfg.VeleroCLI)
+					Expect(err).To(Succeed())
+					// Default BSL is deleted, so backups in additional BSL should be left only
+					Expect(cmp.Diff(backupsInBSL2, backupsAfterDel, cmpopts.SortSlices(less))).Should(BeEmpty())
+				})
+			})
+
+			By(fmt.Sprintf("Backup1 %s should still exist in cloud object store after bsl deletion", backupName_1), func() {
+				Expect(ObjectsShouldBeInBucket(VeleroCfg.CloudProvider, VeleroCfg.CloudCredentialsFile,
+					VeleroCfg.BSLBucket, VeleroCfg.BSLPrefix, VeleroCfg.BSLConfig,
+					backupName_1, BackupObjectsPrefix)).To(Succeed())
+			})
+
+			// TODO: Choose additional BSL to be deleted as an new test case
+			// By(fmt.Sprintf("Backup %s should still exist in cloud object store", backupName_2), func() {
+			// 	Expect(ObjectsShouldBeInBucket(VeleroCfg.CloudProvider, VeleroCfg.AdditionalBSLCredentials,
+			// 		VeleroCfg.AdditionalBSLBucket, VeleroCfg.AdditionalBSLPrefix, VeleroCfg.AdditionalBSLConfig,
+			// 		backupName_2, BackupObjectsPrefix)).To(Succeed())
+			// })
+
+			if useVolumeSnapshots {
+				if VeleroCfg.CloudProvider == "vsphere" {
+					By(fmt.Sprintf("Snapshot in backup %s should still exist, because snapshot CR will be deleted 24 hours later if the status is a success", backupName_2), func() {
+						Expect(SnapshotCRsCountShouldBe(context.Background(), bslDeletionTestNs,
+							backupName_1, 1)).To(Succeed())
+						Expect(SnapshotCRsCountShouldBe(context.Background(), bslDeletionTestNs,
+							backupName_2, 1)).To(Succeed())
+					})
+				}
+				var snapshotCheckPoint SnapshotCheckPoint
+				snapshotCheckPoint.NamespaceBackedUp = bslDeletionTestNs
+				By(fmt.Sprintf("Snapshot should not be deleted in cloud object store after deleting bsl %s", backupLocation_1), func() {
+					snapshotCheckPoint.ExpectCount = 1
+					snapshotCheckPoint.PodName = podName_1
+					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
+						VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket,
+						VeleroCfg.BSLConfig, backupName_1, snapshotCheckPoint)).To(Succeed())
+				})
+				By(fmt.Sprintf("Snapshot should not be deleted in cloud object store after deleting bsl %s", backupLocation_2), func() {
+					snapshotCheckPoint.ExpectCount = 1
+					snapshotCheckPoint.PodName = podName_2
+					var BSLCredentials, BSLConfig string
+					if VeleroCfg.CloudProvider == "vsphere" {
+						BSLCredentials = VeleroCfg.AdditionalBSLCredentials
+						BSLConfig = VeleroCfg.AdditionalBSLConfig
+					} else {
+						BSLCredentials = VeleroCfg.CloudCredentialsFile
+						BSLConfig = VeleroCfg.BSLConfig
+					}
+					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
+						BSLCredentials, VeleroCfg.AdditionalBSLBucket,
+						BSLConfig, backupName_2, snapshotCheckPoint)).To(Succeed())
+				})
+			} else { // For Restic
+				By(fmt.Sprintf("Resticrepositories for BSL %s should be deleted in Velero namespace", backupLocation_1), func() {
+					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_1, 0)).To(Succeed())
+				})
+				By(fmt.Sprintf("Resticrepositories for BSL %s should still exist in Velero namespace", backupLocation_2), func() {
+					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_2, 1)).To(Succeed())
+				})
+			}
+			fmt.Printf("|| EXPECTED || - Backup deletion test completed successfully\n")
+		})
+	})
+}
--- a/test/e2e/e2e_suite_test.go
+++ b/test/e2e/e2e_suite_test.go
@@ -29,6 +29,7 @@ import (
 	. "github.com/vmware-tanzu/velero/test/e2e/backups"
 	. "github.com/vmware-tanzu/velero/test/e2e/basic"
 	. "github.com/vmware-tanzu/velero/test/e2e/basic/resources-check"
+	. "github.com/vmware-tanzu/velero/test/e2e/bsl-mgmt"
 	. "github.com/vmware-tanzu/velero/test/e2e/privilegesmgmt"
 	. "github.com/vmware-tanzu/velero/test/e2e/resource-filtering"
 	. "github.com/vmware-tanzu/velero/test/e2e/scale"
@@ -93,6 +94,8 @@ var _ = Describe("[Backups][Deletion][Restic] Velero tests of Restic backup dele
 var _ = Describe("[Backups][Deletion][Snapshot] Velero tests of snapshot backup deletion", BackupDeletionWithSnapshots)
 var _ = Describe("[PrivilegesMgmt][SSR] Velero test on ssr object when controller namespace mix-ups", SSRTest)
 var _ = Describe("[Backups][BackupsSync] Backups in object storage are synced to a new Velero and deleted backups in object storage are synced to be deleted in Velero", BackupsSyncTest)
+var _ = Describe("[BSL][Deletion][Snapshot] Local backups will be deleted once the corresponding backup storage location is deleted", BslDeletionWithSnapshots)
+var _ = Describe("[BSL][Deletion][Restic] Local backups and restic repos will be deleted once the corresponding backup storage location is deleted", BslDeletionWithRestic)

 func TestE2e(t *testing.T) {
 	// Skip running E2E tests when running only "short" tests because:
--- a/test/e2e/types.go
+++ b/test/e2e/types.go
@@ -54,4 +54,5 @@ type SnapshotCheckPoint struct {
 	NamespaceBackedUp string
 	SnapshotIDList    []string
 	ExpectCount       int
+	PodName           string
 }
--- a/test/e2e/upgrade/upgrade.go
+++ b/test/e2e/upgrade/upgrade.go
@@ -126,7 +126,7 @@ func runUpgradeTests(client TestClient, veleroCfg *VerleroConfig, backupName, re
 		return errors.Wrapf(err, "Failed to install and prepare data for kibishii %s", upgradeNamespace)
 	}

-	if err := VeleroBackupNamespace(oneHourTimeout, veleroCfg.UpgradeFromVeleroCLI, veleroCfg.VeleroNamespace, backupName, upgradeNamespace, backupLocation, useVolumeSnapshots); err != nil {
+	if err := VeleroBackupNamespace(oneHourTimeout, veleroCfg.UpgradeFromVeleroCLI, veleroCfg.VeleroNamespace, backupName, upgradeNamespace, backupLocation, useVolumeSnapshots, ""); err != nil {
 		// TODO currently, the upgrade case covers the upgrade path from 1.6 to main and the velero v1.6 doesn't support "debug" command
 		// TODO move to "RunDebug" after we bump up to 1.7 in the upgrade case
 		VeleroBackupLogs(context.Background(), veleroCfg.UpgradeFromVeleroCLI, veleroCfg.VeleroNamespace, backupName)
--- a/test/e2e/util/common/common.go
+++ b/test/e2e/util/common/common.go
@@ -0,0 +1,54 @@
+package common
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"os/exec"
+)
+
+type OsCommandLine struct {
+	Cmd  string
+	Args []string
+}
+
+func GetListBy2Pipes(ctx context.Context, cmdline1, cmdline2, cmdline3 OsCommandLine) ([]string, error) {
+	var b2 bytes.Buffer
+	var errVelero, errAwk error
+
+	c1 := exec.CommandContext(ctx, cmdline1.Cmd, cmdline1.Args...)
+	c2 := exec.Command(cmdline2.Cmd, cmdline2.Args...)
+	c3 := exec.Command(cmdline3.Cmd, cmdline3.Args...)
+	fmt.Println(c1)
+	fmt.Println(c2)
+	fmt.Println(c3)
+	c2.Stdin, errVelero = c1.StdoutPipe()
+	if errVelero != nil {
+		return nil, errVelero
+	}
+	c3.Stdin, errAwk = c2.StdoutPipe()
+	if errAwk != nil {
+		return nil, errAwk
+	}
+	c3.Stdout = &b2
+	_ = c3.Start()
+	_ = c2.Start()
+	_ = c1.Run()
+	_ = c2.Wait()
+	_ = c3.Wait()
+
+	fmt.Println(&b2)
+	scanner := bufio.NewScanner(&b2)
+	var ret []string
+	for scanner.Scan() {
+		fmt.Printf("line: %s\n", scanner.Text())
+		ret = append(ret, scanner.Text())
+	}
+
+	if err := scanner.Err(); err != nil {
+		return nil, err
+	}
+
+	return ret, nil
+}
--- a/test/e2e/util/k8s/common.go
+++ b/test/e2e/util/k8s/common.go
@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/util/wait"

 	"github.com/vmware-tanzu/velero/pkg/builder"
+	common "github.com/vmware-tanzu/velero/test/e2e/util/common"
 )

 // ensureClusterExists returns whether or not a kubernetes cluster exists for tests to be run on.
@@ -77,3 +78,61 @@ func WaitForPods(ctx context.Context, client TestClient, namespace string, pods
 	}
 	return nil
 }
+
+func GetPvcByPodName(ctx context.Context, namespace, podName string) ([]string, error) {
+	// Example:
+	//    NAME                                  STATUS   VOLUME                                     CAPACITY   ACCESS MODES   STORAGECLASS             AGE
+	//    kibishii-data-kibishii-deployment-0   Bound    pvc-94b9fdf2-c30f-4a7b-87bf-06eadca0d5b6   1Gi        RWO            kibishii-storage-class   115s
+	CmdLine1 := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", "pvc", "-n", namespace},
+	}
+	CmdLine2 := &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{podName},
+	}
+	CmdLine3 := &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+
+	return common.GetListBy2Pipes(ctx, *CmdLine1, *CmdLine2, *CmdLine3)
+}
+
+func GetPvByPvc(ctx context.Context, pvc string) ([]string, error) {
+	// Example:
+	// 	  NAME                                       CAPACITY   ACCESS MODES   RECLAIM POLICY   STATUS   CLAIM                                              STORAGECLASS             REASON   AGE
+	//    pvc-3f784366-58db-40b2-8fec-77307807e74b   1Gi        RWO            Delete           Bound    bsl-deletion/kibishii-data-kibishii-deployment-0   kibishii-storage-class            6h41m
+	CmdLine1 := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", "pv"},
+	}
+
+	CmdLine2 := &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{pvc},
+	}
+
+	CmdLine3 := &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+
+	return common.GetListBy2Pipes(ctx, *CmdLine1, *CmdLine2, *CmdLine3)
+}
+
+func AddLabelToPv(ctx context.Context, pv, label string) error {
+	return exec.CommandContext(ctx, "kubectl", "label", "pv", pv, label).Run()
+}
+
+func AddLabelToPvc(ctx context.Context, pvc, namespace, label string) error {
+	args := []string{"label", "pvc", pvc, "-n", namespace, label}
+	fmt.Println(args)
+	return exec.CommandContext(ctx, "kubectl", args...).Run()
+}
+
+func AddLabelToPod(ctx context.Context, podName, namespace, label string) error {
+	args := []string{"label", "pod", podName, "-n", namespace, label}
+	fmt.Println(args)
+	return exec.CommandContext(ctx, "kubectl", args...).Run()
+}
--- a/test/e2e/util/kibishii/kibishii_utils.go
+++ b/test/e2e/util/kibishii/kibishii_utils.go
@@ -51,7 +51,7 @@ func RunKibishiiTests(client TestClient, providerName, veleroCLI, veleroNamespac
 		return errors.Wrapf(err, "Failed to install and prepare data for kibishii %s", kibishiiNamespace)
 	}

-	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, kibishiiNamespace, backupLocation, useVolumeSnapshots); err != nil {
+	if err := VeleroBackupNamespace(oneHourTimeout, veleroCLI, veleroNamespace, backupName, kibishiiNamespace, backupLocation, useVolumeSnapshots, ""); err != nil {
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, backupName, "")
 		return errors.Wrapf(err, "Failed to backup kibishii namespace %s", kibishiiNamespace)
 	}
--- a/test/e2e/util/providers/aws_utils.go
+++ b/test/e2e/util/providers/aws_utils.go
@@ -29,7 +29,7 @@ import (
 	"github.com/pkg/errors"

 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
-	. "github.com/vmware-tanzu/velero/test/e2e"
+	e2e "github.com/vmware-tanzu/velero/test/e2e"
 )

 type AWSStorage string
@@ -62,10 +62,7 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 	if bslPrefix != "" {
 		objectsInput.Prefix = aws.String(bslPrefix)
 	}
-	s3Config := &aws.Config{
-		Region:      aws.String(region),
-		Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
-	}
+	var s3Config *aws.Config
 	if region == "minio" {
 		s3url = config.Data()["s3Url"]
 		s3Config = &aws.Config{
@@ -75,8 +72,12 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 			DisableSSL:       aws.Bool(true),
 			S3ForcePathStyle: aws.Bool(true),
 		}
+	} else {
+		s3Config = &aws.Config{
+			Region:      aws.String(region),
+			Credentials: credentials.NewSharedCredentials(cloudCredentialsFile, ""),
+		}
 	}
-
 	sess, err := session.NewSession(s3Config)

 	if err != nil {
@@ -97,7 +98,7 @@ func (s AWSStorage) IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix
 		fmt.Println("item:")
 		fmt.Println(item)
 		backupNameInStorage = strings.TrimPrefix(*item.Prefix, strings.Trim(bslPrefix, "/")+"/")
-		fmt.Println("backupNameInStorage:" + backupNameInStorage)
+		fmt.Println("backupNameInStorage:" + backupNameInStorage + " backupObject:" + backupObject)
 		if strings.Contains(backupNameInStorage, backupObject) {
 			fmt.Printf("Backup %s was found under prefix %s \n", backupObject, bslPrefix)
 			return true, nil
@@ -144,7 +145,7 @@ func (s AWSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	return nil
 }

-func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObject string, snapshotCheck e2e.SnapshotCheckPoint) error {

 	config := flag.NewMap()
 	config.Set(bslConfig)
@@ -172,10 +173,16 @@ func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix
 			},
 		},
 	}
+
 	result, err := svc.DescribeSnapshots(params)
 	if err != nil {
 		fmt.Println(err)
 	}
+
+	for _, n := range result.Snapshots {
+		fmt.Println(n.SnapshotId)
+		fmt.Println(n.Tags)
+	}
 	if len(result.Snapshots) != snapshotCheck.ExpectCount {
 		return errors.New(fmt.Sprintf("Snapshot count is not as expected %d", snapshotCheck.ExpectCount))
 	} else {
--- a/test/e2e/util/providers/azure_utils.go
+++ b/test/e2e/util/providers/azure_utils.go
@@ -317,18 +317,9 @@ func mapLookup(data map[string]string) func(string) string {
 		return data[key]
 	}
 }
-func (s AzureStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+func (s AzureStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {

 	ctx := context.Background()
-	config := flag.NewMap()
-	config.Set(bslConfig)
-	if err := validateConfigKeys(config.Data(),
-		resourceGroupConfigKey,
-		subscriptionIDConfigKey,
-		storageAccount,
-	); err != nil {
-		return err
-	}

 	if err := loadCredentialsIntoEnv(cloudCredentialsFile); err != nil {
 		return err
@@ -348,13 +339,7 @@ func (s AzureStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPref

 	// set a different subscriptionId for snapshots if specified
 	snapshotsSubscriptionID := envVars[subscriptionIDEnvVar]
-	if val := config.Data()[subscriptionIDConfigKey]; val != "" {
-		// if subscription was set in config, it is required to also set the resource group
-		if _, err := getRequiredValues(mapLookup(config.Data()), resourceGroupConfigKey); err != nil {
-			return errors.Wrap(err, "resourceGroup not specified, but is a requirement when backing up to a different subscription")
-		}
-		snapshotsSubscriptionID = val
-	}
+
 	// set up clients
 	snapsClient := disk.NewSnapshotsClientWithBaseURI(env.ResourceManagerEndpoint, snapshotsSubscriptionID)
 	snapsClient.PollingDelay = 5 * time.Second
--- a/test/e2e/util/providers/common.go
+++ b/test/e2e/util/providers/common.go
@@ -31,7 +31,7 @@ import (
 type ObjectsInStorage interface {
 	IsObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) (bool, error)
 	DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string) error
-	IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error
+	IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error
 }

 func ObjectsShouldBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
@@ -109,11 +109,11 @@ func DeleteObjectsInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPr
 	return nil
 }

-func SnapshotsShouldNotExistInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
+func SnapshotsShouldNotExistInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslConfig, backupName string) error {
 	fmt.Printf("|| VERIFICATION || - Snapshots should not exist in cloud, backup %s\n", backupName)
 	var snapshotCheckPoint SnapshotCheckPoint
 	snapshotCheckPoint.ExpectCount = 0
-	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix, snapshotCheckPoint)
+	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslConfig, backupName, snapshotCheckPoint)
 	if err != nil {
 		return errors.Wrapf(err, fmt.Sprintf("|| UNEXPECTED ||Snapshots %s is existed in cloud after backup as expected", backupName))
 	}
@@ -121,9 +121,9 @@ func SnapshotsShouldNotExistInCloud(cloudProvider, cloudCredentialsFile, bslBuck
 	return nil
 }

-func SnapshotsShouldBeCreatedInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string, snapshotCheckPoint SnapshotCheckPoint) error {
+func SnapshotsShouldBeCreatedInCloud(cloudProvider, cloudCredentialsFile, bslBucket, bslConfig, backupName string, snapshotCheckPoint SnapshotCheckPoint) error {
 	fmt.Printf("|| VERIFICATION || - Snapshots should exist in cloud, backup %s\n", backupName)
-	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix, snapshotCheckPoint)
+	err := IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslConfig, backupName, snapshotCheckPoint)
 	if err != nil {
 		return errors.Wrapf(err, fmt.Sprintf("|| UNEXPECTED ||Snapshots %s are not existed in cloud after backup as expected", backupName))
 	}
@@ -131,8 +131,8 @@ func SnapshotsShouldBeCreatedInCloud(cloudProvider, cloudCredentialsFile, bslBuc
 	return nil
 }

-func IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string, snapshotCheck SnapshotCheckPoint) error {
-	bslPrefix = getFullPrefix(bslPrefix, subPrefix)
+func IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslConfig, backupName string, snapshotCheck SnapshotCheckPoint) error {
+
 	s, err := getProvider(cloudProvider)
 	if err != nil {
 		return errors.Wrapf(err, fmt.Sprintf("Cloud provider %s is not valid", cloudProvider))
@@ -140,12 +140,13 @@ func IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix
 	if cloudProvider == "vsphere" {
 		var retSnapshotIDs []string
 		ctx, _ := context.WithTimeout(context.Background(), time.Minute*2)
-		retSnapshotIDs, err = velero.GetVsphereSnapshotIDs(ctx, time.Hour, snapshotCheck.NamespaceBackedUp)
+		retSnapshotIDs, err = velero.GetVsphereSnapshotIDs(ctx, time.Hour, snapshotCheck.NamespaceBackedUp, snapshotCheck.PodName)
 		if err != nil {
 			return errors.Wrapf(err, fmt.Sprintf("Fail to get snapshot CRs of backup%s", backupName))
 		}
-		bslPrefix = "plugins"
-		subPrefix = "vsphere-astrolabe-repo/ivd/data"
+
+		bslPrefix := "plugins"
+		subPrefix := "vsphere-astrolabe-repo/ivd/data"
 		if snapshotCheck.ExpectCount == 0 {
 			for _, snapshotID := range retSnapshotIDs {
 				err := ObjectsShouldNotBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, snapshotID, subPrefix, 5)
@@ -165,7 +166,7 @@ func IsSnapshotExisted(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix
 			}
 		}
 	} else {
-		err = s.IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, snapshotCheck)
+		err = s.IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupName, snapshotCheck)
 		if err != nil {
 			return errors.Wrapf(err, fmt.Sprintf("Fail to get snapshot of backup%s", backupName))
 		}
--- a/test/e2e/util/providers/gcloud_utils.go
+++ b/test/e2e/util/providers/gcloud_utils.go
@@ -105,7 +105,7 @@ func (s GCSStorage) DeleteObjectsInBucket(cloudCredentialsFile, bslBucket, bslPr
 	}
 }

-func (s GCSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
+func (s GCSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObject string, snapshotCheck SnapshotCheckPoint) error {
 	ctx := context.Background()
 	data, err := ioutil.ReadFile(cloudCredentialsFile)
 	if err != nil {
@@ -137,7 +137,8 @@ func (s GCSStorage) IsSnapshotExisted(cloudCredentialsFile, bslBucket, bslPrefix
 	}); err != nil {
 		return errors.Wrapf(err, "Failed listing snapshot pages")
 	}
-	if snapshotCountFound != len(snapshotCheck.SnapshotIDList) {
+
+	if snapshotCountFound != snapshotCheck.ExpectCount {
 		return errors.New(fmt.Sprintf("Snapshot count %d is not as expected %d\n", snapshotCountFound, len(snapshotCheck.SnapshotIDList)))
 	} else {
 		fmt.Printf("Snapshot count %d is as expected %d\n", snapshotCountFound, len(snapshotCheck.SnapshotIDList))
--- a/test/e2e/util/velero/velero_utils.go
+++ b/test/e2e/util/velero/velero_utils.go
@@ -40,6 +40,7 @@ import (
 	cliinstall "github.com/vmware-tanzu/velero/pkg/cmd/cli/install"
 	"github.com/vmware-tanzu/velero/pkg/cmd/util/flag"
 	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
+	common "github.com/vmware-tanzu/velero/test/e2e/util/common"
 )

 const BackupObjectsPrefix = "backups"
@@ -233,14 +234,17 @@ func checkRestorePhase(ctx context.Context, veleroCLI string, veleroNamespace st
 }

 // VeleroBackupNamespace uses the veleroCLI to backup a namespace.
-func VeleroBackupNamespace(ctx context.Context, veleroCLI string, veleroNamespace string, backupName string, namespace string, backupLocation string,
-	useVolumeSnapshots bool) error {
+func VeleroBackupNamespace(ctx context.Context, veleroCLI, veleroNamespace, backupName, namespace, backupLocation string,
+	useVolumeSnapshots bool, selector string) error {
 	args := []string{
 		"--namespace", veleroNamespace,
 		"create", "backup", backupName,
 		"--include-namespaces", namespace,
 		"--wait",
 	}
+	if selector != "" {
+		args = append(args, "--selector", selector)
+	}

 	if useVolumeSnapshots {
 		args = append(args, "--snapshot-volumes")
@@ -478,7 +482,7 @@ func WaitForVSphereUploadCompletion(ctx context.Context, timeout time.Duration,
 	return err
 }

-func GetVsphereSnapshotIDs(ctx context.Context, timeout time.Duration, namespace string) ([]string, error) {
+func GetVsphereSnapshotIDs(ctx context.Context, timeout time.Duration, namespace, podName string) ([]string, error) {
 	checkSnapshotCmd := exec.CommandContext(ctx, "kubectl",
 		"get", "-n", namespace, "snapshots.backupdriver.cnsdp.vmware.com", "-o=jsonpath='{range .items[*]}{.spec.resourceHandle.name}{\"=\"}{.status.snapshotID}{\"\\n\"}{end}'")
 	fmt.Printf("checkSnapshotCmd cmd =%v\n", checkSnapshotCmd)
@@ -498,6 +502,9 @@ func GetVsphereSnapshotIDs(ctx context.Context, timeout time.Duration, namespace
 		if len(curLine) == 0 {
 			continue
 		}
+		if podName != "" && !strings.Contains(curLine, podName) {
+			continue
+		}
 		snapshotID := curLine[strings.LastIndex(curLine, ":")+1:]
 		fmt.Println("snapshotID:" + snapshotID)
 		snapshotIDDec, _ := b64.StdEncoding.DecodeString(snapshotID)
@@ -626,7 +633,7 @@ func DeleteBackupResource(ctx context.Context, veleroCLI string, backupName stri
 				fmt.Printf("|| EXPECTED || - Backup %s was deleted successfully according to message %s\n", backupName, stderr)
 				return nil
 			}
-			return errors.Wrapf(err, "Fail to get delete backup, stdout=%s, stderr=%s", stdout, stderr)
+			return errors.Wrapf(err, "Fail to perform get backup, stdout=%s, stderr=%s", stdout, stderr)
 		}
 		time.Sleep(1 * time.Minute)
 	}
@@ -640,7 +647,8 @@ func GetBackup(ctx context.Context, veleroCLI string, backupName string) (string
 }

 func IsBackupExist(ctx context.Context, veleroCLI string, backupName string) (bool, error) {
-	if _, outerr, err := GetBackup(ctx, veleroCLI, backupName); err != nil {
+	out, outerr, err := GetBackup(ctx, veleroCLI, backupName)
+	if err != nil {
 		if err != nil {
 			if strings.Contains(outerr, "not found") {
 				return false, nil
@@ -648,6 +656,7 @@ func IsBackupExist(ctx context.Context, veleroCLI string, backupName string) (bo
 			return false, err
 		}
 	}
+	fmt.Printf("Backup %s exist locally according to output %s", backupName, out)
 	return true, nil
 }

@@ -664,3 +673,118 @@ func WaitBackupDeleted(ctx context.Context, veleroCLI string, backupName string,
 		}
 	})
 }
+
+func WaitForBackupCreated(ctx context.Context, veleroCLI string, backupName string, timeout time.Duration) error {
+	return wait.PollImmediate(10*time.Second, timeout, func() (bool, error) {
+		if exist, err := IsBackupExist(ctx, veleroCLI, backupName); err != nil {
+			return false, err
+		} else {
+			if exist {
+				return true, nil
+			} else {
+				return false, nil
+			}
+		}
+	})
+}
+func GetBackupsFromBsl(ctx context.Context, veleroCLI, bslName string) ([]string, error) {
+	args1 := []string{"get", "backups"}
+	if strings.TrimSpace(bslName) != "" {
+		args1 = append(args1, "-l", "velero.io/storage-location="+bslName)
+	}
+	CmdLine1 := &common.OsCommandLine{
+		Cmd:  veleroCLI,
+		Args: args1,
+	}
+
+	CmdLine2 := &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+
+	CmdLine3 := &common.OsCommandLine{
+		Cmd:  "tail",
+		Args: []string{"-n", "+2"},
+	}
+
+	return common.GetListBy2Pipes(ctx, *CmdLine1, *CmdLine2, *CmdLine3)
+}
+
+func GetAllBackups(ctx context.Context, veleroCLI string) ([]string, error) {
+	return GetBackupsFromBsl(ctx, veleroCLI, "")
+}
+func DeleteBslResource(ctx context.Context, veleroCLI string, bslName string) error {
+	args := []string{"backup-location", "delete", bslName, "--confirm"}
+
+	cmd := exec.CommandContext(ctx, veleroCLI, args...)
+	fmt.Println("Delete backup location Command:" + cmd.String())
+	stdout, stderr, err := veleroexec.RunCommand(cmd)
+	if err != nil {
+		return errors.Wrapf(err, "Fail to get delete location, stdout=%s, stderr=%s", stdout, stderr)
+	}
+
+	output := strings.Replace(stdout, "\n", " ", -1)
+	fmt.Println("Backup location delete command output:" + output)
+
+	fmt.Println(stdout)
+	fmt.Println(stderr)
+	return nil
+}
+
+func SnapshotCRsCountShouldBe(ctx context.Context, namespace, backupName string, expectedCount int) error {
+
+	checkSnapshotCmd := exec.CommandContext(ctx, "kubectl",
+		"get", "-n", namespace, "snapshots.backupdriver.cnsdp.vmware.com", "-o=jsonpath='{range .items[*]}{.metadata.labels.velero\\.io\\/backup-name}{\"\\n\"}{end}'")
+	fmt.Printf("checkSnapshotCmd cmd =%v\n", checkSnapshotCmd)
+	stdout, stderr, err := veleroexec.RunCommand(checkSnapshotCmd)
+	if err != nil {
+		fmt.Print(stdout)
+		fmt.Print(stderr)
+		return errors.Wrap(err, fmt.Sprintf("Failed getting snapshot CR of backup %s in namespace %d", backupName, expectedCount))
+	}
+	count := 0
+	stdout = strings.Replace(stdout, "'", "", -1)
+	arr := strings.Split(stdout, "\n")
+	for _, bn := range arr {
+		fmt.Println("Snapshot CR:" + bn)
+		if strings.Contains(bn, backupName) {
+			count++
+		}
+	}
+	if count == expectedCount {
+		return nil
+	} else {
+		return errors.New(fmt.Sprintf("SnapshotCR count %d of backup %s in namespace %s is not as expected %d", count, backupName, namespace, expectedCount))
+	}
+}
+
+func ResticRepositoriesCountShouldBe(ctx context.Context, veleroNamespace, targetNamespace string, expectedCount int) error {
+	resticArr, err := GetResticRepositories(ctx, veleroNamespace, targetNamespace)
+	if err != nil {
+		return errors.Wrapf(err, "Fail to get GetResticRepositories")
+	}
+	if len(resticArr) == expectedCount {
+		return nil
+	} else {
+		return errors.New(fmt.Sprintf("Resticrepositories count %d in namespace %s is not as expected %d", len(resticArr), targetNamespace, expectedCount))
+	}
+}
+
+func GetResticRepositories(ctx context.Context, veleroNamespace, targetNamespace string) ([]string, error) {
+	CmdLine1 := &common.OsCommandLine{
+		Cmd:  "kubectl",
+		Args: []string{"get", "-n", veleroNamespace, "resticrepositories"},
+	}
+
+	CmdLine2 := &common.OsCommandLine{
+		Cmd:  "grep",
+		Args: []string{targetNamespace},
+	}
+
+	CmdLine3 := &common.OsCommandLine{
+		Cmd:  "awk",
+		Args: []string{"{print $1}"},
+	}
+
+	return common.GetListBy2Pipes(ctx, *CmdLine1, *CmdLine2, *CmdLine3)
+}