Merge pull request #5124 from danfengliu/add-migration-e2e-test

Add migration E2E test
2022-08-10 13:38:32 +08:00 · 2022-08-10 13:38:32 +08:00 · a71237cc64
parent 6951875053 fb445b3c0d
commit a71237cc64
28 changed files with 509 additions and 187 deletions
--- a/.gitignore
+++ b/.gitignore
@ -48,3 +48,5 @@ tilt-resources/velero_v1_backupstoragelocation.yaml
 tilt-resources/deployment.yaml
 tilt-resources/restic.yaml
 tilt-resources/cloud
+
+test/e2e/report.xml
--- a/pkg/client/client.go
+++ b/pkg/client/client.go
@ -27,15 +27,20 @@ import (
 	"github.com/vmware-tanzu/velero/pkg/buildinfo"
 )

+func buildConfigFromFlags(context, kubeconfigPath string, precedence []string) (*rest.Config, error) {
+	return clientcmd.NewNonInteractiveDeferredLoadingClientConfig(
+		&clientcmd.ClientConfigLoadingRules{ExplicitPath: kubeconfigPath, Precedence: precedence},
+		&clientcmd.ConfigOverrides{
+			CurrentContext: context,
+		}).ClientConfig()
+}
+
 // Config returns a *rest.Config, using either the kubeconfig (if specified) or an in-cluster
 // configuration.
 func Config(kubeconfig, kubecontext, baseName string, qps float32, burst int) (*rest.Config, error) {
 	loadingRules := clientcmd.NewDefaultClientConfigLoadingRules()
 	loadingRules.ExplicitPath = kubeconfig
-	configOverrides := &clientcmd.ConfigOverrides{CurrentContext: kubecontext}
-	kubeConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig(loadingRules, configOverrides)
-
-	clientConfig, err := kubeConfig.ClientConfig()
+	clientConfig, err := buildConfigFromFlags(kubecontext, kubeconfig, loadingRules.Precedence)
 	if err != nil {
 		return nil, errors.Wrap(err, "error finding Kubernetes API server config in --kubeconfig, $KUBECONFIG, or in-cluster configuration")
 	}
--- a/pkg/client/factory.go
+++ b/pkg/client/factory.go
@ -77,10 +77,11 @@ type factory struct {
 }

 // NewFactory returns a Factory.
-func NewFactory(baseName string, config VeleroConfig) Factory {
+func NewFactory(baseName, kubecontext string, config VeleroConfig) Factory {
 	f := &factory{
-		flags:    pflag.NewFlagSet("", pflag.ContinueOnError),
-		baseName: baseName,
+		flags:       pflag.NewFlagSet("", pflag.ContinueOnError),
+		baseName:    baseName,
+		kubecontext: kubecontext,
 	}

 	f.namespace = os.Getenv("VELERO_NAMESPACE")
@ -96,8 +97,7 @@ func NewFactory(baseName string, config VeleroConfig) Factory {

 	f.flags.StringVar(&f.kubeconfig, "kubeconfig", "", "Path to the kubeconfig file to use to talk to the Kubernetes apiserver. If unset, try the environment variable KUBECONFIG, as well as in-cluster configuration")
 	f.flags.StringVarP(&f.namespace, "namespace", "n", f.namespace, "The namespace in which Velero should operate")
-	f.flags.StringVar(&f.kubecontext, "kubecontext", "", "The context to use to talk to the Kubernetes apiserver. If unset defaults to whatever your current-context is (kubectl config current-context)")
-
+	//f.flags.StringVar(&f.kubecontext, "kubecontext", "", "The context to use to talk to the Kubernetes apiserver. If unset defaults to whatever your current-context is (kubectl config current-context)")
 	return f
 }

@ -127,7 +127,6 @@ func (f *factory) KubeClient() (kubernetes.Interface, error) {
 	if err != nil {
 		return nil, err
 	}
-
 	kubeClient, err := kubernetes.NewForConfig(clientConfig)
 	if err != nil {
 		return nil, errors.WithStack(err)
--- a/pkg/client/factory_test.go
+++ b/pkg/client/factory_test.go
@ -31,14 +31,14 @@ func TestFactory(t *testing.T) {

 	// Env variable should set the namespace if no config or argument are used
 	os.Setenv("VELERO_NAMESPACE", "env-velero")
-	f := NewFactory("velero", make(map[string]interface{}))
+	f := NewFactory("velero", "", make(map[string]interface{}))

 	assert.Equal(t, "env-velero", f.Namespace())

 	os.Unsetenv("VELERO_NAMESPACE")

 	// Argument should change the namespace
-	f = NewFactory("velero", make(map[string]interface{}))
+	f = NewFactory("velero", "", make(map[string]interface{}))
 	s := "flag-velero"
 	flags := new(pflag.FlagSet)

@ -50,7 +50,7 @@ func TestFactory(t *testing.T) {

 	// An argument overrides the env variable if both are set.
 	os.Setenv("VELERO_NAMESPACE", "env-velero")
-	f = NewFactory("velero", make(map[string]interface{}))
+	f = NewFactory("velero", "", make(map[string]interface{}))
 	flags = new(pflag.FlagSet)

 	f.BindFlags(flags)
--- a/pkg/cmd/velero/velero.go
+++ b/pkg/cmd/velero/velero.go
@ -89,7 +89,7 @@ operations can also be performed as 'velero backup get' and 'velero schedule cre
 		},
 	}

-	f := client.NewFactory(name, config)
+	f := client.NewFactory(name, "", config)
 	f.BindFlags(c.PersistentFlags())

 	// Bind features directly to the root command so it's available to all callers.
--- a/pkg/restore/prioritize_group_version.go
+++ b/pkg/restore/prioritize_group_version.go
@ -214,7 +214,7 @@ func userPriorityConfigMap() (*corev1.ConfigMap, error) {
 		return nil, errors.Wrap(err, "reading client config file")
 	}

-	fc := client.NewFactory("APIGroupVersionsRestore", cfg)
+	fc := client.NewFactory("APIGroupVersionsRestore", "", cfg)

 	kc, err := fc.KubeClient()
 	if err != nil {
--- a/test/e2e/Makefile
+++ b/test/e2e/Makefile
@ -62,6 +62,9 @@ UPGRADE_FROM_VELERO_VERSION ?= v1.7.1,v1.8.1
 # to the end, nil string will be set if UPGRADE_FROM_VELERO_CLI is shorter than UPGRADE_FROM_VELERO_VERSION
 UPGRADE_FROM_VELERO_CLI ?=

+MIGRATE_FROM_VELERO_VERSION ?= v1.8.1,self
+MIGRATE_FROM_VELERO_CLI ?=
+
 VELERO_NAMESPACE ?= velero
 CREDS_FILE ?=
 BSL_BUCKET ?=
@ -86,6 +89,10 @@ ADDITIONAL_BSL_CONFIG ?=
 FEATURES ?=
 DEBUG_E2E_TEST ?= false

+DEFAULT_CLUSTER ?=
+STANDBY_CLUSTER ?=
+
+
 .PHONY:ginkgo
 ginkgo: # Make sure ginkgo is in $GOPATH/bin
 	go get github.com/onsi/ginkgo/ginkgo
@ -105,6 +112,8 @@ run: ginkgo
 		-restic-helper-image=$(RESTIC_HELPER_IMAGE) \
 		-upgrade-from-velero-cli=$(UPGRADE_FROM_VELERO_CLI) \
 		-upgrade-from-velero-version=$(UPGRADE_FROM_VELERO_VERSION) \
+		-migrate-from-velero-cli=$(MIGRATE_FROM_VELERO_CLI) \
+		-migrate-from-velero-version=$(MIGRATE_FROM_VELERO_VERSION) \
 		-velero-namespace=$(VELERO_NAMESPACE) \
 		-credentials-file=$(CREDS_FILE) \
 		-bucket=$(BSL_BUCKET) \
@ -123,7 +132,9 @@ run: ginkgo
 		-install-velero=$(INSTALL_VELERO) \
 		-registry-credential-file=$(REGISTRY_CREDENTIAL_FILE) \
 		-kibishii-directory=$(KIBISHII_DIRECTORY) \
-		-debug-e2e-test=$(DEBUG_E2E_TEST)
+		-debug-e2e-test=$(DEBUG_E2E_TEST) \
+		-default-cluster=$(DEFAULT_CLUSTER) \
+		-standby-cluster=$(STANDBY_CLUSTER)

 build: ginkgo
 	mkdir -p $(OUTPUT_DIR)
--- a/test/e2e/backup/backup.go
+++ b/test/e2e/backup/backup.go
@ -42,14 +42,9 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 	kibishiiNamespace := "kibishii-workload"
 	var (
 		backupName, restoreName string
-		client                  TestClient
 		err                     error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
 	BeforeEach(func() {
 		if useVolumeSnapshots && VeleroCfg.CloudProvider == "kind" {
 			Skip("Volume snapshots not supported on kind")
@ -78,7 +73,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 			restoreName = "restore-" + UUIDgen.String()
 			// Even though we are using Velero's CloudProvider plugin for object storage, the kubernetes cluster is running on
 			// KinD. So use the kind installation for Kibishii.
-			Expect(RunKibishiiTests(client, VeleroCfg, backupName, restoreName, "", kibishiiNamespace, useVolumeSnapshots)).To(Succeed(),
+			Expect(RunKibishiiTests(*VeleroCfg.ClientToInstallVelero, VeleroCfg, backupName, restoreName, "", kibishiiNamespace, useVolumeSnapshots)).To(Succeed(),
 				"Failed to successfully backup and restore Kibishii namespace")
 		})

@ -106,7 +101,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 				secretKey: VeleroCfg.AdditionalBSLCredentials,
 			}

-			Expect(CreateSecretFromFiles(context.TODO(), client, VeleroCfg.VeleroNamespace, secretName, files)).To(Succeed())
+			Expect(CreateSecretFromFiles(context.TODO(), *VeleroCfg.ClientToInstallVelero, VeleroCfg.VeleroNamespace, secretName, files)).To(Succeed())

 			// Create additional BSL using credential
 			additionalBsl := fmt.Sprintf("bsl-%s", UUIDgen)
@ -133,7 +128,7 @@ func BackupRestoreTest(useVolumeSnapshots bool) {
 					backupName = fmt.Sprintf("%s-%s", backupName, UUIDgen)
 					restoreName = fmt.Sprintf("%s-%s", restoreName, UUIDgen)
 				}
-				Expect(RunKibishiiTests(client, VeleroCfg, backupName, restoreName, bsl, kibishiiNamespace, useVolumeSnapshots)).To(Succeed(),
+				Expect(RunKibishiiTests(*VeleroCfg.ClientToInstallVelero, VeleroCfg, backupName, restoreName, bsl, kibishiiNamespace, useVolumeSnapshots)).To(Succeed(),
 					"Failed to successfully backup and restore Kibishii namespace using BSL %s", bsl)
 			}
 		})
--- a/test/e2e/backups/deletion.go
+++ b/test/e2e/backups/deletion.go
@ -47,14 +47,9 @@ func BackupDeletionWithRestic() {
 func backup_deletion_test(useVolumeSnapshots bool) {
 	var (
 		backupName string
-		client     TestClient
 		err        error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
 	BeforeEach(func() {
 		if useVolumeSnapshots && VeleroCfg.CloudProvider == "kind" {
 			Skip("Volume snapshots not supported on kind")
@ -80,7 +75,7 @@ func backup_deletion_test(useVolumeSnapshots bool) {
 	When("kibishii is the sample workload", func() {
 		It("Deleted backups are deleted from object storage and backups deleted from object storage can be deleted locally", func() {
 			backupName = "backup-" + UUIDgen.String()
-			Expect(runBackupDeletionTests(client, VeleroCfg, backupName, "", useVolumeSnapshots, VeleroCfg.KibishiiDirectory)).To(Succeed(),
+			Expect(runBackupDeletionTests(*VeleroCfg.ClientToInstallVelero, VeleroCfg, backupName, "", useVolumeSnapshots, VeleroCfg.KibishiiDirectory)).To(Succeed(),
 				"Failed to run backup deletion test")
 		})
 	})
--- a/test/e2e/backups/sync_backups.go
+++ b/test/e2e/backups/sync_backups.go
@ -53,17 +53,9 @@ func (b *SyncBackups) Init() {
 func BackupsSyncTest() {
 	test := new(SyncBackups)
 	var (
-		client TestClient
-		err    error
+		err error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
-
-	Expect(err).To(Succeed(), "Failed to instantiate cluster client for backup tests")
-
 	BeforeEach(func() {
 		flag.Parse()
 		if VeleroCfg.InstallVelero {
@ -82,11 +74,11 @@ func BackupsSyncTest() {
 	It("Backups in object storage should be synced to a new Velero successfully", func() {
 		test.Init()
 		By(fmt.Sprintf("Prepare workload as target to backup by creating namespace %s namespace", test.testNS))
-		Expect(CreateNamespace(test.ctx, client, test.testNS)).To(Succeed(),
+		Expect(CreateNamespace(test.ctx, *VeleroCfg.ClientToInstallVelero, test.testNS)).To(Succeed(),
 			fmt.Sprintf("Failed to create %s namespace", test.testNS))

 		defer func() {
-			Expect(DeleteNamespace(test.ctx, client, test.testNS, false)).To(Succeed(), fmt.Sprintf("Failed to delete the namespace %s", test.testNS))
+			Expect(DeleteNamespace(test.ctx, *VeleroCfg.ClientToInstallVelero, test.testNS, false)).To(Succeed(), fmt.Sprintf("Failed to delete the namespace %s", test.testNS))
 		}()

 		var BackupCfg BackupConfig
@ -119,12 +111,12 @@ func BackupsSyncTest() {
 	It("Deleted backups in object storage are synced to be deleted in Velero", func() {
 		test.Init()
 		By(fmt.Sprintf("Prepare workload as target to backup by creating namespace in %s namespace", test.testNS), func() {
-			Expect(CreateNamespace(test.ctx, client, test.testNS)).To(Succeed(),
+			Expect(CreateNamespace(test.ctx, *VeleroCfg.ClientToInstallVelero, test.testNS)).To(Succeed(),
 				fmt.Sprintf("Failed to create %s namespace", test.testNS))
 		})
 		if !VeleroCfg.Debug {
 			defer func() {
-				Expect(DeleteNamespace(test.ctx, client, test.testNS, false)).To(Succeed(),
+				Expect(DeleteNamespace(test.ctx, *VeleroCfg.ClientToInstallVelero, test.testNS, false)).To(Succeed(),
 					fmt.Sprintf("Failed to delete the namespace %s", test.testNS))
 			}()
 		}
--- a/test/e2e/backups/ttl.go
+++ b/test/e2e/backups/ttl.go
@ -60,11 +60,11 @@ func (b *TTL) Init() {
 func TTLTest() {
 	useVolumeSnapshots := true
 	test := new(TTL)
-	client, err := NewTestClient()
+	client, err := NewTestClient(VeleroCfg.DefaultCluster)
 	if err != nil {
 		println(err.Error())
 	}
-	Expect(err).To(Succeed(), "Failed to instantiate cluster client for backup tests")
+	//Expect(err).To(Succeed(), "Failed to instantiate cluster client for backup tests")

 	BeforeEach(func() {
 		flag.Parse()
@ -142,7 +142,7 @@ func TTLTest() {

 		By(fmt.Sprintf("Restore %s", test.testNS), func() {
 			Expect(VeleroRestore(test.ctx, VeleroCfg.VeleroCLI,
-				VeleroCfg.VeleroNamespace, test.restoreName, test.backupName)).To(Succeed(), func() string {
+				VeleroCfg.VeleroNamespace, test.restoreName, test.backupName, "")).To(Succeed(), func() string {
 				RunDebug(test.ctx, VeleroCfg.VeleroCLI,
 					VeleroCfg.VeleroNamespace, "", test.restoreName)
 				return "Fail to restore workload"
--- a/test/e2e/basic/enable_api_group_versions.go
+++ b/test/e2e/basic/enable_api_group_versions.go
@ -45,14 +45,8 @@ func APIGropuVersionsTest() {
 		resource, group string
 		err             error
 		ctx             = context.Background()
-		client          TestClient
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
-
 	BeforeEach(func() {
 		resource = "rockbands"
 		group = "music.example.io"
@ -91,7 +85,7 @@ func APIGropuVersionsTest() {
 		It("Should back up API group version and restore by version priority", func() {
 			Expect(runEnableAPIGroupVersionsTests(
 				ctx,
-				client,
+				*VeleroCfg.ClientToInstallVelero,
 				resource,
 				group,
 			)).To(Succeed(), "Failed to successfully backup and restore multiple API Groups")
@ -288,7 +282,7 @@ func runEnableAPIGroupVersionsTests(ctx context.Context, client TestClient, reso
 		restore := "restore-rockbands-" + UUIDgen.String() + "-" + strconv.Itoa(i)

 		if tc.want != nil {
-			if err := VeleroRestore(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, restore, backup); err != nil {
+			if err := VeleroRestore(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, restore, backup, ""); err != nil {
 				RunDebug(context.Background(), VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, "", restore)
 				return errors.Wrapf(err, "restore %s namespaces on target cluster", namespacesStr)
 			}
@ -327,7 +321,7 @@ func runEnableAPIGroupVersionsTests(ctx context.Context, client TestClient, reso
 		} else {
 			// No custom resource should have been restored. Expect "no resource found"
 			// error during restore.
-			err := VeleroRestore(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, restore, backup)
+			err := VeleroRestore(ctx, VeleroCfg.VeleroCLI, VeleroCfg.VeleroNamespace, restore, backup, "")

 			if err.Error() != "Unexpected restore phase got PartiallyFailed, expecting Completed" {
 				return errors.New("expected error but not none")
--- a/test/e2e/bsl-mgmt/deletion.go
+++ b/test/e2e/bsl-mgmt/deletion.go
@ -53,15 +53,9 @@ func BslDeletionWithRestic() {
 }
 func BslDeletionTest(useVolumeSnapshots bool) {
 	var (
-		client TestClient
-		err    error
+		err error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
-
 	less := func(a, b string) bool { return a < b }
 	BeforeEach(func() {
 		if useVolumeSnapshots && VeleroCfg.CloudProvider == "kind" {
@ -79,7 +73,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 	AfterEach(func() {
 		if VeleroCfg.InstallVelero {
 			if !VeleroCfg.Debug {
-				Expect(DeleteNamespace(context.Background(), client, bslDeletionTestNs,
+				Expect(DeleteNamespace(context.Background(), *VeleroCfg.ClientToInstallVelero, bslDeletionTestNs,
 					true)).To(Succeed(), fmt.Sprintf("failed to delete the namespace %q",
 					bslDeletionTestNs))
 				Expect(VeleroUninstall(context.Background(), VeleroCfg.VeleroCLI,
@ -116,7 +110,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 			}

 			By(fmt.Sprintf("Create Secret for additional BSL %s", additionalBsl), func() {
-				Expect(CreateSecretFromFiles(context.TODO(), client, VeleroCfg.VeleroNamespace, secretName, files)).To(Succeed())
+				Expect(CreateSecretFromFiles(context.TODO(), *VeleroCfg.ClientToInstallVelero, VeleroCfg.VeleroNamespace, secretName, files)).To(Succeed())
 			})

 			By(fmt.Sprintf("Create additional BSL using credential %s", secretName), func() {
@ -147,11 +141,11 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 			//label_2 := "for!=1"
 			label_2 := "for=2"
 			By("Create namespace for sample workload", func() {
-				Expect(CreateNamespace(oneHourTimeout, client, bslDeletionTestNs)).To(Succeed())
+				Expect(CreateNamespace(oneHourTimeout, *VeleroCfg.ClientToInstallVelero, bslDeletionTestNs)).To(Succeed())
 			})

 			By("Deploy sample workload of Kibishii", func() {
-				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, client, VeleroCfg.CloudProvider,
+				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *VeleroCfg.ClientToInstallVelero, VeleroCfg.CloudProvider,
 					bslDeletionTestNs, VeleroCfg.RegistryCredentialFile, VeleroCfg.Features,
 					VeleroCfg.KibishiiDirectory, useVolumeSnapshots)).To(Succeed())
 			})
@ -218,14 +212,14 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 				var snapshotCheckPoint SnapshotCheckPoint
 				snapshotCheckPoint.NamespaceBackedUp = bslDeletionTestNs
 				By(fmt.Sprintf("Snapshot of bsl %s should be created in cloud object store", backupLocation_1), func() {
-					snapshotCheckPoint, err = GetSnapshotCheckPoint(client, VeleroCfg, 1, bslDeletionTestNs, backupName_1, []string{podName_1})
+					snapshotCheckPoint, err = GetSnapshotCheckPoint(*VeleroCfg.ClientToInstallVelero, VeleroCfg, 1, bslDeletionTestNs, backupName_1, []string{podName_1})
 					Expect(err).NotTo(HaveOccurred(), "Fail to get Azure CSI snapshot checkpoint")
 					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, err = GetSnapshotCheckPoint(client, VeleroCfg, 1, bslDeletionTestNs, backupName_2, []string{podName_2})
+					snapshotCheckPoint, err = GetSnapshotCheckPoint(*VeleroCfg.ClientToInstallVelero, VeleroCfg, 1, bslDeletionTestNs, backupName_2, []string{podName_2})
 					Expect(err).NotTo(HaveOccurred(), "Fail to get Azure CSI snapshot checkpoint")
 					var BSLCredentials, BSLConfig string
 					if VeleroCfg.CloudProvider == "vsphere" {
@ -242,11 +236,11 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 				})
 			} else { // For Restics
 				By(fmt.Sprintf("Resticrepositories for BSL %s should be created in Velero namespace", backupLocation_1), func() {
-					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+					Expect(BackupRepositoriesCountShouldBe(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(),
+					Expect(BackupRepositoriesCountShouldBe(context.Background(),
 						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_2, 1)).To(Succeed())
 				})
 			}
@ -317,7 +311,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 				snapshotCheckPoint.NamespaceBackedUp = bslDeletionTestNs
 				By(fmt.Sprintf("Snapshot should not be deleted in cloud object store after deleting bsl %s", backupLocation_1), func() {

-					snapshotCheckPoint, err = GetSnapshotCheckPoint(client, VeleroCfg, 1, bslDeletionTestNs, backupName_1, []string{podName_1})
+					snapshotCheckPoint, err = GetSnapshotCheckPoint(*VeleroCfg.ClientToInstallVelero, VeleroCfg, 1, bslDeletionTestNs, backupName_1, []string{podName_1})
 					Expect(err).NotTo(HaveOccurred(), "Fail to get Azure CSI snapshot checkpoint")
 					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
 						VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket,
@ -332,7 +326,7 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 						BSLCredentials = VeleroCfg.CloudCredentialsFile
 						BSLConfig = VeleroCfg.BSLConfig
 					}
-					snapshotCheckPoint, err = GetSnapshotCheckPoint(client, VeleroCfg, 1, bslDeletionTestNs, backupName_2, []string{podName_2})
+					snapshotCheckPoint, err = GetSnapshotCheckPoint(*VeleroCfg.ClientToInstallVelero, VeleroCfg, 1, bslDeletionTestNs, backupName_2, []string{podName_2})
 					Expect(err).NotTo(HaveOccurred(), "Fail to get Azure CSI snapshot checkpoint")
 					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
 						BSLCredentials, VeleroCfg.AdditionalBSLBucket,
@ -340,11 +334,11 @@ func BslDeletionTest(useVolumeSnapshots bool) {
 				})
 			} else { // For Restic
 				By(fmt.Sprintf("Resticrepositories for BSL %s should be deleted in Velero namespace", backupLocation_1), func() {
-					Expect(ResticRepositoriesCountShouldBe(context.Background(),
+					Expect(BackupRepositoriesCountShouldBe(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(),
+					Expect(BackupRepositoriesCountShouldBe(context.Background(),
 						VeleroCfg.VeleroNamespace, bslDeletionTestNs+"-"+backupLocation_2, 1)).To(Succeed())
 				})
 			}
--- a/test/e2e/e2e_suite_test.go
+++ b/test/e2e/e2e_suite_test.go
@ -17,7 +17,10 @@ limitations under the License.
 package e2e_test

 import (
+	"context"
+	"errors"
 	"flag"
+	"fmt"
 	"testing"

 	. "github.com/onsi/ginkgo"
@ -35,6 +38,9 @@ import (
 	. "github.com/vmware-tanzu/velero/test/e2e/resource-filtering"
 	. "github.com/vmware-tanzu/velero/test/e2e/scale"
 	. "github.com/vmware-tanzu/velero/test/e2e/upgrade"
+
+	. "github.com/vmware-tanzu/velero/test/e2e/migration"
+	. "github.com/vmware-tanzu/velero/test/e2e/util/k8s"
 )

 func init() {
@ -50,6 +56,8 @@ func init() {
 	flag.StringVar(&VeleroCfg.ResticHelperImage, "restic-helper-image", "", "image for the velero restic restore helper to be tested.")
 	flag.StringVar(&VeleroCfg.UpgradeFromVeleroCLI, "upgrade-from-velero-cli", "", "path to the pre-upgrade velero application to use.")
 	flag.StringVar(&VeleroCfg.UpgradeFromVeleroVersion, "upgrade-from-velero-version", "v1.7.1", "image for the pre-upgrade velero server to be tested.")
+	flag.StringVar(&VeleroCfg.MigrateFromVeleroCLI, "migrate-from-velero-cli", "", "path to the origin velero application to use.")
+	flag.StringVar(&VeleroCfg.MigrateFromVeleroVersion, "migrate-from-velero-version", "self", "image for the origin velero server to be tested.")
 	flag.StringVar(&VeleroCfg.BSLConfig, "bsl-config", "", "configuration to use for the backup storage location. Format is key1=value1,key2=value2")
 	flag.StringVar(&VeleroCfg.BSLPrefix, "prefix", "", "prefix under which all Velero data should be stored within the bucket. Optional.")
 	flag.StringVar(&VeleroCfg.VSLConfig, "vsl-config", "", "configuration to use for the volume snapshot location. Format is key1=value1,key2=value2")
@ -67,6 +75,8 @@ func init() {
 	flag.StringVar(&VeleroCfg.Features, "features", "", "Comma-separated list of features to enable for this Velero process.")
 	flag.BoolVar(&VeleroCfg.Debug, "debug-e2e-test", false, "Switch to control namespace cleaning.")
 	flag.StringVar(&VeleroCfg.GCFrequency, "garbage-collection-frequency", "", "Frequency of garbage collection.")
+	flag.StringVar(&VeleroCfg.DefaultCluster, "default-cluster", "", "Default cluster context for migration test.")
+	flag.StringVar(&VeleroCfg.StandbyCluster, "standby-cluster", "", "Standby cluster context for migration test.")
 }

 var _ = Describe("[APIGroup] Velero tests with various CRD API group versions", APIGropuVersionsTest)
@ -106,8 +116,41 @@ var _ = Describe("[PrivilegesMgmt][SSR] Velero test on ssr object when controlle
 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)

+var _ = Describe("[Migration][Restic]", MigrationWithRestic)
+
+var _ = Describe("[Migration][Snapshot]", MigrationWithSnapshots)
+
 var _ = Describe("[Schedule][OrederedResources] Backup resources should follow the specific order in schedule", ScheduleOrderedResources)

+func GetKubeconfigContext() error {
+	var err error
+	var tcDefault, tcStandby TestClient
+	tcDefault, err = NewTestClient(VeleroCfg.DefaultCluster)
+	VeleroCfg.DefaultClient = &tcDefault
+	VeleroCfg.ClientToInstallVelero = VeleroCfg.DefaultClient
+	if err != nil {
+		return err
+	}
+
+	if VeleroCfg.DefaultCluster != "" {
+		err = KubectlConfigUseContext(context.Background(), VeleroCfg.DefaultCluster)
+		if err != nil {
+			return err
+		}
+		if VeleroCfg.StandbyCluster != "" {
+			tcStandby, err = NewTestClient(VeleroCfg.StandbyCluster)
+			VeleroCfg.StandbyClient = &tcStandby
+			if err != nil {
+				return err
+			}
+		} else {
+			return errors.New("migration test needs 2 clusters to run")
+		}
+	}
+
+	return nil
+}
+
 func TestE2e(t *testing.T) {
 	// Skip running E2E tests when running only "short" tests because:
 	// 1. E2E tests are long running tests involving installation of Velero and performing backup and restore operations.
@ -116,6 +159,12 @@ func TestE2e(t *testing.T) {
 		t.Skip("Skipping E2E tests")
 	}

+	var err error
+	if err = GetKubeconfigContext(); err != nil {
+		fmt.Println(err)
+		t.FailNow()
+	}
+
 	RegisterFailHandler(Fail)
 	junitReporter := reporters.NewJUnitReporter("report.xml")
 	RunSpecsWithDefaultAndCustomReporters(t, "E2e Suite", []Reporter{junitReporter})
--- a/test/e2e/migration/migration.go
+++ b/test/e2e/migration/migration.go
@ -0,0 +1,256 @@
+/*
+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 migration
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"strings"
+	"time"
+
+	"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"
+)
+
+var migrationNamespace string
+
+func MigrationWithSnapshots() {
+	for _, veleroCLI2Version := range GetVersionList(VeleroCfg.MigrateFromVeleroCLI, VeleroCfg.MigrateFromVeleroVersion) {
+		MigrationTest(true, veleroCLI2Version)
+	}
+}
+
+func MigrationWithRestic() {
+	for _, veleroCLI2Version := range GetVersionList(VeleroCfg.MigrateFromVeleroCLI, VeleroCfg.MigrateFromVeleroVersion) {
+		MigrationTest(false, veleroCLI2Version)
+	}
+}
+
+func MigrationTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version) {
+	var (
+		backupName, restoreName     string
+		backupScName, restoreScName string
+		err                         error
+	)
+
+	BeforeEach(func() {
+		UUIDgen, err = uuid.NewRandom()
+		migrationNamespace = "migration-workload-" + UUIDgen.String()
+		if useVolumeSnapshots && VeleroCfg.CloudProvider == "kind" {
+			Skip("Volume snapshots not supported on kind")
+		}
+		if useVolumeSnapshots && VeleroCfg.CloudProvider == "aws" {
+			Skip("Volume snapshots migration not supported on AWS provisioned by Sheperd public pool")
+		}
+		if VeleroCfg.DefaultCluster == "" && VeleroCfg.StandbyCluster == "" {
+			Skip("Migration test needs 2 clusters")
+		}
+	})
+	AfterEach(func() {
+		if VeleroCfg.InstallVelero {
+			if !VeleroCfg.Debug {
+				By(fmt.Sprintf("Uninstall Velero and delete sample workload namespace %s", migrationNamespace), func() {
+					Expect(KubectlConfigUseContext(context.Background(), VeleroCfg.DefaultCluster)).To(Succeed())
+					Expect(VeleroUninstall(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace)).To(Succeed())
+					DeleteNamespace(context.Background(), *VeleroCfg.DefaultClient, migrationNamespace, true)
+
+					Expect(KubectlConfigUseContext(context.Background(), VeleroCfg.StandbyCluster)).To(Succeed())
+					Expect(VeleroUninstall(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace)).To(Succeed())
+					DeleteNamespace(context.Background(), *VeleroCfg.StandbyClient, migrationNamespace, true)
+				})
+				By(fmt.Sprintf("Switch to default kubeconfig context %s", VeleroCfg.DefaultClient), func() {
+					Expect(KubectlConfigUseContext(context.Background(), VeleroCfg.DefaultCluster)).To(Succeed())
+					VeleroCfg.ClientToInstallVelero = VeleroCfg.DefaultClient
+				})
+			}
+		}
+	})
+	When("kibishii is the sample workload", func() {
+		It("should be successfully backed up and restored to the default BackupStorageLocation", func() {
+			flag.Parse()
+			UUIDgen, err = uuid.NewRandom()
+			Expect(err).To(Succeed())
+
+			oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*10)
+
+			if veleroCLI2Version.VeleroCLI == "" {
+				//Assume tag of velero server image is identical to velero CLI version
+				//Download velero CLI if it's empty according to velero CLI version
+				By(fmt.Sprintf("Install the expected version Velero CLI (%s) for installing Velero",
+					veleroCLI2Version.VeleroVersion), func() {
+					if veleroCLI2Version.VeleroVersion == "self" {
+						veleroCLI2Version.VeleroCLI = VeleroCfg.VeleroCLI
+					} else {
+						veleroCLI2Version.VeleroCLI, err = InstallVeleroCLI(veleroCLI2Version.VeleroVersion)
+						Expect(err).To(Succeed())
+					}
+				})
+			}
+
+			By(fmt.Sprintf("Install Velero in cluster-A (%s) to backup workload", VeleroCfg.DefaultCluster), func() {
+				Expect(KubectlConfigUseContext(context.Background(), VeleroCfg.DefaultCluster)).To(Succeed())
+
+				OriginVeleroCfg := VeleroCfg
+				OriginVeleroCfg.MigrateFromVeleroVersion = veleroCLI2Version.VeleroVersion
+				OriginVeleroCfg.VeleroCLI = veleroCLI2Version.VeleroCLI
+				OriginVeleroCfg.ClientToInstallVelero = OriginVeleroCfg.DefaultClient
+				if veleroCLI2Version.VeleroVersion != "self" {
+					fmt.Printf("Using default images address of Velero CLI %s\n", veleroCLI2Version.VeleroVersion)
+					OriginVeleroCfg.VeleroImage = ""
+					OriginVeleroCfg.ResticHelperImage = ""
+					OriginVeleroCfg.Plugins = ""
+				}
+				fmt.Println(OriginVeleroCfg)
+				Expect(VeleroInstall(context.Background(), &OriginVeleroCfg, useVolumeSnapshots)).To(Succeed())
+				if veleroCLI2Version.VeleroVersion != "self" {
+					Expect(CheckVeleroVersion(context.Background(), OriginVeleroCfg.VeleroCLI,
+						OriginVeleroCfg.MigrateFromVeleroVersion)).To(Succeed())
+				}
+			})
+
+			backupName = "backup-" + UUIDgen.String()
+			backupScName = backupName + "-sc"
+			restoreName = "restore-" + UUIDgen.String()
+			restoreScName = restoreName + "-sc"
+
+			By("Create namespace for sample workload", func() {
+				Expect(CreateNamespace(oneHourTimeout, *VeleroCfg.DefaultClient, migrationNamespace)).To(Succeed(),
+					fmt.Sprintf("Failed to create namespace %s to install Kibishii workload", migrationNamespace))
+			})
+
+			By("Deploy sample workload of Kibishii", func() {
+				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *VeleroCfg.DefaultClient, VeleroCfg.CloudProvider,
+					migrationNamespace, VeleroCfg.RegistryCredentialFile, VeleroCfg.Features,
+					VeleroCfg.KibishiiDirectory, useVolumeSnapshots)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Backup namespace %s", migrationNamespace), func() {
+				var BackupStorageClassCfg BackupConfig
+				BackupStorageClassCfg.BackupName = backupScName
+				BackupStorageClassCfg.IncludeResources = "StorageClass"
+				BackupStorageClassCfg.IncludeClusterResources = true
+				Expect(VeleroBackupNamespace(context.Background(), VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, BackupStorageClassCfg)).ShouldNot(HaveOccurred(), func() string {
+					err = VeleroBackupLogs(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace, backupName)
+					return "Get backup logs"
+				})
+
+				var BackupCfg BackupConfig
+				BackupCfg.BackupName = backupName
+				BackupCfg.Namespace = migrationNamespace
+				BackupCfg.UseVolumeSnapshots = useVolumeSnapshots
+				BackupCfg.BackupLocation = ""
+				BackupCfg.Selector = ""
+				//BackupCfg.ExcludeResources = "tierentitlementbindings,tierentitlements,tiers,capabilities,customresourcedefinitions"
+				Expect(VeleroBackupNamespace(context.Background(), VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, BackupCfg)).ShouldNot(HaveOccurred(), func() string {
+					err = VeleroBackupLogs(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace, backupName)
+					return "Get backup logs"
+				})
+
+			})
+
+			if useVolumeSnapshots {
+				if VeleroCfg.CloudProvider == "vsphere" {
+					// TODO - remove after upload progress monitoring is implemented
+					By("Waiting for vSphere uploads to complete", func() {
+						Expect(WaitForVSphereUploadCompletion(context.Background(), time.Hour,
+							migrationNamespace)).To(Succeed())
+					})
+				}
+				var snapshotCheckPoint SnapshotCheckPoint
+				snapshotCheckPoint.NamespaceBackedUp = migrationNamespace
+				By("Snapshot should be created in cloud object store", func() {
+					snapshotCheckPoint, err := GetSnapshotCheckPoint(*VeleroCfg.DefaultClient, VeleroCfg, 2,
+						migrationNamespace, backupName, KibishiiPodNameList)
+					Expect(err).NotTo(HaveOccurred(), "Fail to get snapshot checkpoint")
+					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
+						VeleroCfg.CloudCredentialsFile, VeleroCfg.BSLBucket,
+						VeleroCfg.BSLConfig, backupName, snapshotCheckPoint)).To(Succeed())
+				})
+			}
+
+			if useVolumeSnapshots && VeleroCfg.CloudProvider == "azure" && strings.EqualFold(VeleroCfg.Features, "EnableCSI") {
+				// Upgrade test is not running daily since no CSI plugin v1.0 released, because builds before
+				//   v1.0 have issues to fail upgrade case.
+				By("Sleep 5 minutes to avoid snapshot recreated by unknown reason ", func() {
+					time.Sleep(5 * time.Minute)
+				})
+			}
+			// the snapshots of AWS may be still in pending status when do the restore, wait for a while
+			// to avoid this https://github.com/vmware-tanzu/velero/issues/1799
+			// TODO remove this after https://github.com/vmware-tanzu/velero/issues/3533 is fixed
+			if VeleroCfg.CloudProvider == "aws" && useVolumeSnapshots {
+				fmt.Println("Waiting 5 minutes to make sure the snapshots are ready...")
+				time.Sleep(5 * time.Minute)
+			}
+
+			By(fmt.Sprintf("Install Velero in cluster-B (%s) to restore workload", VeleroCfg.StandbyCluster), func() {
+				ns, err := GetNamespace(context.Background(), *VeleroCfg.DefaultClient, migrationNamespace)
+				Expect(ns.Name).To(Equal(migrationNamespace))
+				Expect(err).NotTo(HaveOccurred())
+
+				Expect(KubectlConfigUseContext(context.Background(), VeleroCfg.StandbyCluster)).To(Succeed())
+				_, err = GetNamespace(context.Background(), *VeleroCfg.StandbyClient, migrationNamespace)
+				Expect(err).To(HaveOccurred())
+				strings.Contains(fmt.Sprint(err), "namespaces \""+migrationNamespace+"\" not found")
+
+				fmt.Println(err)
+
+				VeleroCfg.ObjectStoreProvider = ""
+				VeleroCfg.ClientToInstallVelero = VeleroCfg.StandbyClient
+				Expect(VeleroInstall(context.Background(), &VeleroCfg, useVolumeSnapshots)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Waiting for backups sync to Velero in cluster-B (%s)", VeleroCfg.StandbyCluster), func() {
+				Expect(WaitForBackupToBeCreated(context.Background(), VeleroCfg.VeleroCLI, backupName, 5*time.Minute)).To(Succeed())
+				Expect(WaitForBackupToBeCreated(context.Background(), VeleroCfg.VeleroCLI, backupScName, 5*time.Minute)).To(Succeed())
+			})
+
+			By(fmt.Sprintf("Restore %s", migrationNamespace), func() {
+				Expect(VeleroRestore(context.Background(), VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, restoreScName, backupScName, "StorageClass")).To(Succeed(), func() string {
+					RunDebug(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace, "", restoreName)
+					return "Fail to restore workload"
+				})
+				Expect(VeleroRestore(context.Background(), VeleroCfg.VeleroCLI,
+					VeleroCfg.VeleroNamespace, restoreName, backupName, "")).To(Succeed(), func() string {
+					RunDebug(context.Background(), VeleroCfg.VeleroCLI,
+						VeleroCfg.VeleroNamespace, "", restoreName)
+					return "Fail to restore workload"
+				})
+			})
+
+			By(fmt.Sprintf("Verify workload %s after restore ", migrationNamespace), func() {
+				Expect(KibishiiVerifyAfterRestore(*VeleroCfg.StandbyClient, migrationNamespace,
+					oneHourTimeout)).To(Succeed(), "Fail to verify workload after restore")
+			})
+		})
+	})
+}
--- a/test/e2e/orderedresources/ordered_resources.go
+++ b/test/e2e/orderedresources/ordered_resources.go
@ -112,7 +112,7 @@ func ScheduleOrderedResources() {
 func (o *OrderedResources) Init() error {
 	rand.Seed(time.Now().UnixNano())
 	UUIDgen, _ = uuid.NewRandom()
-	client, err := NewTestClient()
+	client, err := NewTestClient(VeleroCfg.DefaultCluster)
 	if err != nil {
 		return fmt.Errorf("failed to init ordered resources test with err %v", err)
 	}
--- a/test/e2e/privilegesmgmt/ssr.go
+++ b/test/e2e/privilegesmgmt/ssr.go
@ -37,15 +37,9 @@ import (
 func SSRTest() {
 	testNS := "ssr-test"
 	var (
-		client TestClient
-		err    error
+		err error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
-
 	BeforeEach(func() {
 		flag.Parse()
 		if VeleroCfg.InstallVelero {
@ -62,10 +56,10 @@ func SSRTest() {
 	})

 	It(fmt.Sprintf("Should create an ssr object in the %s namespace and later removed by controller", VeleroCfg.VeleroNamespace), func() {
-		defer DeleteNamespace(context.TODO(), client, testNS, false)
+		defer DeleteNamespace(context.TODO(), *VeleroCfg.ClientToInstallVelero, testNS, false)
 		ctx, _ := context.WithTimeout(context.Background(), time.Duration(time.Minute*10))
 		By(fmt.Sprintf("Create %s namespace", testNS))
-		Expect(CreateNamespace(ctx, client, testNS)).To(Succeed(),
+		Expect(CreateNamespace(ctx, *VeleroCfg.ClientToInstallVelero, testNS)).To(Succeed(),
 			fmt.Sprintf("Failed to create %s namespace", testNS))

 		By(fmt.Sprintf("Get version in %s namespace", testNS))
@ -82,7 +76,7 @@ func SSRTest() {
 		By(fmt.Sprintf("Check ssr object in %s namespace", VeleroCfg.VeleroNamespace))
 		err = waitutil.PollImmediate(5*time.Second, time.Minute,
 			func() (bool, error) {
-				if err = client.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
+				if err = VeleroCfg.ClientToInstallVelero.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
 					return false, fmt.Errorf("failed to list ssr object in %s namespace with err %v", VeleroCfg.VeleroNamespace, err)
 				}
 				if len(ssrListResp.Items) != 1 {
@ -105,7 +99,7 @@ func SSRTest() {
 		Expect(err).To(Succeed(), fmt.Sprintf("Failed to check ssr object in %s namespace", VeleroCfg.VeleroNamespace))

 		By(fmt.Sprintf("Check ssr object in %s namespace", testNS))
-		Expect(client.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: testNS})).To(Succeed(),
+		Expect(VeleroCfg.ClientToInstallVelero.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: testNS})).To(Succeed(),
 			fmt.Sprintf("Failed to list ssr object in %s namespace", testNS))
 		Expect(len(ssrListResp.Items)).To(BeNumerically("==", 1),
 			fmt.Sprintf("Count of ssr object in %s namespace is not 1", testNS))
@ -117,7 +111,7 @@ func SSRTest() {
 		By(fmt.Sprintf("Waiting ssr object in %s namespace deleted", VeleroCfg.VeleroNamespace))
 		err = waitutil.PollImmediateInfinite(5*time.Second,
 			func() (bool, error) {
-				if err = client.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
+				if err = VeleroCfg.ClientToInstallVelero.Kubebuilder.List(ctx, ssrListResp, &kbclient.ListOptions{Namespace: VeleroCfg.VeleroNamespace}); err != nil {
 					if apierrors.IsNotFound(err) {
 						return true, nil
 					}
--- a/test/e2e/test/test.go
+++ b/test/e2e/test/test.go
@ -73,9 +73,7 @@ var TestClientInstance TestClient
 func TestFunc(test VeleroBackupRestoreTest) func() {
 	return func() {
 		By("Create test client instance", func() {
-			var err error
-			TestClientInstance, err = NewTestClient()
-			Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
+			TestClientInstance = *VeleroCfg.ClientToInstallVelero
 		})
 		Expect(test.Init()).To(Succeed(), "Failed to instantiate test cases")
 		BeforeEach(func() {
@ -99,13 +97,11 @@ func TestFunc(test VeleroBackupRestoreTest) func() {

 func TestFuncWithMultiIt(tests []VeleroBackupRestoreTest) func() {
 	return func() {
-		var err error
 		var countIt int
 		By("Create test client instance", func() {
-			TestClientInstance, err = NewTestClient()
-			Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
+			TestClientInstance = *VeleroCfg.ClientToInstallVelero
 		})
-		//Expect(err).To(Succeed(), "Failed to instantiate cluster client for backup tests")
+
 		for k := range tests {
 			Expect(tests[k].Init()).To(Succeed(), fmt.Sprintf("Failed to instantiate test %s case", tests[k].GetTestMsg().Desc))
 		}
--- a/test/e2e/types.go
+++ b/test/e2e/types.go
@ -20,6 +20,8 @@ import (
 	"time"

 	"github.com/google/uuid"
+
+	. "github.com/vmware-tanzu/velero/test/e2e/util/k8s"
 )

 var UUIDgen uuid.UUID
@ -47,6 +49,8 @@ type VerleroConfig struct {
 	ResticHelperImage        string
 	UpgradeFromVeleroVersion string
 	UpgradeFromVeleroCLI     string
+	MigrateFromVeleroVersion string
+	MigrateFromVeleroCLI     string
 	Plugins                  string
 	AddBSLPlugins            string
 	InstallVelero            bool
@ -54,6 +58,11 @@ type VerleroConfig struct {
 	Features                 string
 	Debug                    bool
 	GCFrequency              string
+	DefaultCluster           string
+	StandbyCluster           string
+	ClientToInstallVelero    *TestClient
+	DefaultClient            *TestClient
+	StandbyClient            *TestClient
 }

 type SnapshotCheckPoint struct {
@ -67,10 +76,19 @@ type SnapshotCheckPoint struct {
 }

 type BackupConfig struct {
-	BackupName         string
-	Namespace          string
-	BackupLocation     string
-	UseVolumeSnapshots bool
-	Selector           string
-	TTL                time.Duration
+	BackupName              string
+	Namespace               string
+	BackupLocation          string
+	UseVolumeSnapshots      bool
+	Selector                string
+	TTL                     time.Duration
+	IncludeResources        string
+	ExcludeResources        string
+	IncludeClusterResources bool
+	OrderedResources        string
+}
+
+type VeleroCLI2Version struct {
+	VeleroVersion string
+	VeleroCLI     string
 }
--- a/test/e2e/upgrade/upgrade.go
+++ b/test/e2e/upgrade/upgrade.go
@ -33,56 +33,28 @@ import (
 	. "github.com/vmware-tanzu/velero/test/e2e/util/velero"
 )

-type UpgradeFromVelero struct {
-	UpgradeFromVeleroVersion string
-	UpgradeFromVeleroCLI     string
-}
-
 const (
 	upgradeNamespace = "upgrade-workload"
 )

-func GetUpgradePathList() []UpgradeFromVelero {
-	var upgradeFromVeleroList []UpgradeFromVelero
-	UpgradeFromVeleroVersionList := strings.Split(VeleroCfg.UpgradeFromVeleroVersion, ",")
-	UpgradeFromVeleroCliList := strings.Split(VeleroCfg.UpgradeFromVeleroCLI, ",")
-
-	for _, upgradeFromVeleroVersion := range UpgradeFromVeleroVersionList {
-		upgradeFromVeleroList = append(upgradeFromVeleroList,
-			UpgradeFromVelero{upgradeFromVeleroVersion, ""})
-	}
-	for i, upgradeFromVeleroCli := range UpgradeFromVeleroCliList {
-		if i == len(UpgradeFromVeleroVersionList)-1 {
-			break
-		}
-		upgradeFromVeleroList[i].UpgradeFromVeleroCLI = upgradeFromVeleroCli
-	}
-	return upgradeFromVeleroList
-}
-
 func BackupUpgradeRestoreWithSnapshots() {
-	for _, upgradeFromVelero := range GetUpgradePathList() {
+	for _, upgradeFromVelero := range GetVersionList(VeleroCfg.UpgradeFromVeleroCLI, VeleroCfg.UpgradeFromVeleroVersion) {
 		BackupUpgradeRestoreTest(true, upgradeFromVelero)
 	}
 }

 func BackupUpgradeRestoreWithRestic() {
-	for _, upgradeFromVelero := range GetUpgradePathList() {
+	for _, upgradeFromVelero := range GetVersionList(VeleroCfg.UpgradeFromVeleroCLI, VeleroCfg.UpgradeFromVeleroVersion) {
 		BackupUpgradeRestoreTest(false, upgradeFromVelero)
 	}
 }

-func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero UpgradeFromVelero) {
+func BackupUpgradeRestoreTest(useVolumeSnapshots bool, veleroCLI2Version VeleroCLI2Version) {
 	var (
 		backupName, restoreName string
-		client                  TestClient
 		err                     error
 	)

-	By("Create test client instance", func() {
-		client, err = NewTestClient()
-		Expect(err).NotTo(HaveOccurred(), "Failed to instantiate cluster client for backup tests")
-	})
 	BeforeEach(func() {
 		if !VeleroCfg.InstallVelero {
 			Skip("Upgrade test should not be triggered if VeleroCfg.InstallVelero is set to false")
@ -101,7 +73,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 		if VeleroCfg.InstallVelero {
 			if !VeleroCfg.Debug {
 				By(fmt.Sprintf("Delete sample workload namespace %s", upgradeNamespace), func() {
-					DeleteNamespace(context.Background(), client, upgradeNamespace, true)
+					DeleteNamespace(context.Background(), *VeleroCfg.ClientToInstallVelero, upgradeNamespace, true)
 				})
 				By("Uninstall Velero", func() {
 					Expect(VeleroUninstall(context.Background(), VeleroCfg.VeleroCLI,
@ -115,24 +87,25 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 			flag.Parse()
 			UUIDgen, err = uuid.NewRandom()
 			Expect(err).To(Succeed())
+			oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*60)

-			if upgradeFromVelero.UpgradeFromVeleroCLI == "" {
+			if veleroCLI2Version.VeleroCLI == "" {
 				//Assume tag of velero server image is identical to velero CLI version
 				//Download velero CLI if it's empty according to velero CLI version
 				By(fmt.Sprintf("Install the expected old version Velero CLI (%s) for installing Velero",
-					upgradeFromVelero.UpgradeFromVeleroVersion), func() {
-					upgradeFromVelero.UpgradeFromVeleroCLI, err = InstallVeleroCLI(upgradeFromVelero.UpgradeFromVeleroVersion)
+					veleroCLI2Version.VeleroVersion), func() {
+					veleroCLI2Version.VeleroCLI, err = InstallVeleroCLI(veleroCLI2Version.VeleroVersion)
 					Expect(err).To(Succeed())
 				})
 			}
 			VeleroCfg.GCFrequency = ""
 			By(fmt.Sprintf("Install the expected old version Velero (%s) for upgrade",
-				upgradeFromVelero.UpgradeFromVeleroVersion), func() {
+				veleroCLI2Version.VeleroVersion), func() {
 				//Set VeleroImage and ResticHelperImage to blank
 				//VeleroImage and ResticHelperImage should be the default value in originalCli
 				tmpCfgForOldVeleroInstall := VeleroCfg
-				tmpCfgForOldVeleroInstall.UpgradeFromVeleroVersion = upgradeFromVelero.UpgradeFromVeleroVersion
-				tmpCfgForOldVeleroInstall.VeleroCLI = upgradeFromVelero.UpgradeFromVeleroCLI
+				tmpCfgForOldVeleroInstall.UpgradeFromVeleroVersion = veleroCLI2Version.VeleroVersion
+				tmpCfgForOldVeleroInstall.VeleroCLI = veleroCLI2Version.VeleroCLI
 				tmpCfgForOldVeleroInstall.VeleroImage = ""
 				tmpCfgForOldVeleroInstall.ResticHelperImage = ""
 				tmpCfgForOldVeleroInstall.Plugins = ""
@ -146,17 +119,16 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 			backupName = "backup-" + UUIDgen.String()
 			restoreName = "restore-" + UUIDgen.String()
 			tmpCfg := VeleroCfg
-			tmpCfg.UpgradeFromVeleroCLI = upgradeFromVelero.UpgradeFromVeleroCLI
-			tmpCfg.UpgradeFromVeleroVersion = upgradeFromVelero.UpgradeFromVeleroVersion
-			oneHourTimeout, _ := context.WithTimeout(context.Background(), time.Minute*60)
+			tmpCfg.UpgradeFromVeleroCLI = veleroCLI2Version.VeleroCLI
+			tmpCfg.UpgradeFromVeleroVersion = veleroCLI2Version.VeleroVersion

 			By("Create namespace for sample workload", func() {
-				Expect(CreateNamespace(oneHourTimeout, client, upgradeNamespace)).To(Succeed(),
+				Expect(CreateNamespace(oneHourTimeout, *VeleroCfg.ClientToInstallVelero, upgradeNamespace)).To(Succeed(),
 					fmt.Sprintf("Failed to create namespace %s to install Kibishii workload", upgradeNamespace))
 			})

 			By("Deploy sample workload of Kibishii", func() {
-				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, client, tmpCfg.CloudProvider,
+				Expect(KibishiiPrepareBeforeBackup(oneHourTimeout, *VeleroCfg.ClientToInstallVelero, tmpCfg.CloudProvider,
 					upgradeNamespace, tmpCfg.RegistryCredentialFile, tmpCfg.Features,
 					tmpCfg.KibishiiDirectory, useVolumeSnapshots)).To(Succeed())
 			})
@ -187,7 +159,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 				var snapshotCheckPoint SnapshotCheckPoint
 				snapshotCheckPoint.NamespaceBackedUp = upgradeNamespace
 				By("Snapshot should be created in cloud object store", func() {
-					snapshotCheckPoint, err := GetSnapshotCheckPoint(client, VeleroCfg, 2,
+					snapshotCheckPoint, err := GetSnapshotCheckPoint(*VeleroCfg.ClientToInstallVelero, VeleroCfg, 2,
 						upgradeNamespace, backupName, KibishiiPodNameList)
 					Expect(err).NotTo(HaveOccurred(), "Fail to get snapshot checkpoint")
 					Expect(SnapshotsShouldBeCreatedInCloud(VeleroCfg.CloudProvider,
@ -197,7 +169,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 			}

 			By(fmt.Sprintf("Simulating a disaster by removing namespace %s\n", upgradeNamespace), func() {
-				Expect(DeleteNamespace(oneHourTimeout, client, upgradeNamespace, true)).To(Succeed(),
+				Expect(DeleteNamespace(oneHourTimeout, *VeleroCfg.ClientToInstallVelero, upgradeNamespace, true)).To(Succeed(),
 					fmt.Sprintf("failed to delete namespace %s", upgradeNamespace))
 			})

@ -225,7 +197,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade

 			By(fmt.Sprintf("Restore %s", upgradeNamespace), func() {
 				Expect(VeleroRestore(oneHourTimeout, tmpCfg.VeleroCLI,
-					tmpCfg.VeleroNamespace, restoreName, backupName)).To(Succeed(), func() string {
+					tmpCfg.VeleroNamespace, restoreName, backupName, "")).To(Succeed(), func() string {
 					RunDebug(context.Background(), tmpCfg.VeleroCLI,
 						tmpCfg.VeleroNamespace, "", restoreName)
 					return "Fail to restore workload"
@ -233,7 +205,7 @@ func BackupUpgradeRestoreTest(useVolumeSnapshots bool, upgradeFromVelero Upgrade
 			})

 			By(fmt.Sprintf("Verify workload %s after restore ", upgradeNamespace), func() {
-				Expect(KibishiiVerifyAfterRestore(client, upgradeNamespace,
+				Expect(KibishiiVerifyAfterRestore(*VeleroCfg.ClientToInstallVelero, upgradeNamespace,
 					oneHourTimeout)).To(Succeed(), "Fail to verify workload after restore")
 			})
 		})
--- a/test/e2e/util/k8s/client.go
+++ b/test/e2e/util/k8s/client.go
@ -17,8 +17,6 @@ limitations under the License.
 package k8s

 import (
-	"sync"
-
 	"k8s.io/client-go/kubernetes"
 	kbclient "sigs.k8s.io/controller-runtime/pkg/client"

@ -48,28 +46,24 @@ type TestClient struct {
 }

 var (
-	once       sync.Once
-	testClient TestClient
-	err        error
+	err error
 )

-func NewTestClient() (TestClient, error) {
-	once.Do(func() { // <-- atomic, does not allow repeating
-		testClient, err = InitTestClient() // <-- thread safe
-	})
-	return testClient, err
+// NewTestClient returns a set of ready-to-use API clients.
+func NewTestClient(kubecontext string) (TestClient, error) {
+	return InitTestClient(kubecontext)
 }

-// NewTestClient returns a set of ready-to-use API clients.
-func InitTestClient() (TestClient, error) {
+func InitTestClient(kubecontext string) (TestClient, error) {
 	config, err := client.LoadConfig()
 	if err != nil {
 		return TestClient{}, err
 	}

-	f := client.NewFactory("e2e", config)
+	f := client.NewFactory("e2e", kubecontext, config)

 	clientGo, err := f.KubeClient()
+
 	if err != nil {
 		return TestClient{}, err
 	}
--- 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"
+	veleroexec "github.com/vmware-tanzu/velero/pkg/util/exec"
 	common "github.com/vmware-tanzu/velero/test/e2e/util/common"
 )

@ -56,11 +57,11 @@ func CreateSecretFromFiles(ctx context.Context, client TestClient, namespace str

 // WaitForPods waits until all of the pods have gone to PodRunning state
 func WaitForPods(ctx context.Context, client TestClient, namespace string, pods []string) error {
-	timeout := 10 * time.Minute
+	timeout := 5 * time.Minute
 	interval := 5 * time.Second
 	err := wait.PollImmediate(interval, timeout, func() (bool, error) {
 		for _, podName := range pods {
-			checkPod, err := client.ClientGo.CoreV1().Pods(namespace).Get(ctx, podName, metav1.GetOptions{})
+			checkPod, err := client.ClientGo.CoreV1().Pods(namespace).Get(context.TODO(), podName, metav1.GetOptions{})
 			if err != nil {
 				//Should ignore "etcdserver: request timed out" kind of errors, try to get pod status again before timeout.
 				fmt.Println(errors.Wrap(err, fmt.Sprintf("Failed to verify pod %s/%s is %s, try again...", namespace, podName, corev1api.PodRunning)))
@ -143,3 +144,13 @@ func KubectlApplyByFile(ctx context.Context, file string) error {
 	args := []string{"apply", "-f", file, "--force=true"}
 	return exec.CommandContext(ctx, "kubectl", args...).Run()
 }
+
+func KubectlConfigUseContext(ctx context.Context, kubectlContext string) error {
+	cmd := exec.CommandContext(ctx, "kubectl",
+		"config", "use-context", kubectlContext)
+	fmt.Printf("Kubectl config use-context cmd =%v\n", cmd)
+	stdout, stderr, err := veleroexec.RunCommand(cmd)
+	fmt.Print(stdout)
+	fmt.Print(stderr)
+	return err
+}
--- a/test/e2e/util/k8s/namespace.go
+++ b/test/e2e/util/k8s/namespace.go
@ -67,21 +67,22 @@ func GetNamespace(ctx context.Context, client TestClient, namespace string) (*co
 }

 func DeleteNamespace(ctx context.Context, client TestClient, namespace string, wait bool) error {
-	if err := client.ClientGo.CoreV1().Namespaces().Delete(ctx, namespace, metav1.DeleteOptions{}); err != nil {
+	oneMinuteTimeout, _ := context.WithTimeout(context.Background(), time.Minute*1)
+	if err := client.ClientGo.CoreV1().Namespaces().Delete(context.TODO(), namespace, metav1.DeleteOptions{}); err != nil {
 		return errors.Wrap(err, fmt.Sprintf("failed to delete the namespace %q", namespace))
 	}
 	if !wait {
 		return nil
 	}
-
 	return waitutil.PollImmediateInfinite(5*time.Second,
 		func() (bool, error) {
-			if _, err := client.ClientGo.CoreV1().Namespaces().Get(context.TODO(), namespace, metav1.GetOptions{}); err != nil {
+			if _, err := client.ClientGo.CoreV1().Namespaces().Get(oneMinuteTimeout, namespace, metav1.GetOptions{}); err != nil {
 				if apierrors.IsNotFound(err) {
 					return true, nil
 				}
 				return false, err
 			}
+			fmt.Printf("namespace %q is still being deleted...\n", namespace)
 			logrus.Debugf("namespace %q is still being deleted...", namespace)
 			return false, nil
 		})
--- a/test/e2e/util/kibishii/kibishii_utils.go
+++ b/test/e2e/util/kibishii/kibishii_utils.go
@ -118,7 +118,7 @@ func RunKibishiiTests(client TestClient, veleroCfg VerleroConfig, backupName, re
 		time.Sleep(5 * time.Minute)
 	}

-	if err := VeleroRestore(oneHourTimeout, veleroCLI, veleroNamespace, restoreName, backupName); err != nil {
+	if err := VeleroRestore(oneHourTimeout, veleroCLI, veleroNamespace, restoreName, backupName, ""); err != nil {
 		RunDebug(context.Background(), veleroCLI, veleroNamespace, "", restoreName)
 		return errors.Wrapf(err, "Restore %s failed from backup %s", restoreName, backupName)
 	}
@ -134,8 +134,7 @@ func RunKibishiiTests(client TestClient, veleroCfg VerleroConfig, backupName, re
 func installKibishii(ctx context.Context, namespace string, cloudPlatform, veleroFeatures,
 	kibishiiDirectory string, useVolumeSnapshots bool) error {
 	if strings.EqualFold(cloudPlatform, "azure") &&
-		strings.EqualFold(veleroFeatures, "EnableCSI") &&
-		useVolumeSnapshots {
+		strings.EqualFold(veleroFeatures, "EnableCSI") {
 		cloudPlatform = "azure-csi"
 	}
 	// We use kustomize to generate YAML for Kibishii from the checked-in yaml directories
@ -182,7 +181,8 @@ func generateData(ctx context.Context, namespace string, levels int, filesPerLev

 func verifyData(ctx context.Context, namespace string, levels int, filesPerLevel int, dirsPerLevel int, fileSize int,
 	blockSize int, passNum int, expectedNodes int) error {
-	kibishiiVerifyCmd := exec.CommandContext(ctx, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
+	timeout, _ := context.WithTimeout(context.Background(), time.Minute*5)
+	kibishiiVerifyCmd := exec.CommandContext(timeout, "kubectl", "exec", "-n", namespace, "jump-pad", "--",
 		"/usr/local/bin/verify.sh", strconv.Itoa(levels), strconv.Itoa(filesPerLevel), strconv.Itoa(dirsPerLevel), strconv.Itoa(fileSize),
 		strconv.Itoa(blockSize), strconv.Itoa(passNum), strconv.Itoa(expectedNodes))
 	fmt.Printf("kibishiiVerifyCmd cmd =%v\n", kibishiiVerifyCmd)
@ -238,7 +238,7 @@ func KibishiiVerifyAfterRestore(client TestClient, kibishiiNamespace string, one
 	if err := waitForKibishiiPods(oneHourTimeout, client, kibishiiNamespace); err != nil {
 		return errors.Wrapf(err, "Failed to wait for ready status of kibishii pods in %s", kibishiiNamespace)
 	}
-
+	time.Sleep(60 * time.Second)
 	// TODO - check that namespace exists
 	fmt.Printf("running kibishii verify\n")
 	if err := verifyData(oneHourTimeout, kibishiiNamespace, 2, 10, 10, 1024, 1024, 0, 2); err != nil {
--- a/test/e2e/util/providers/aws_utils.go
+++ b/test/e2e/util/providers/aws_utils.go
@ -181,7 +181,14 @@ func (s AWSStorage) IsSnapshotExisted(cloudCredentialsFile, bslConfig, backupObj

 	for _, n := range result.Snapshots {
 		fmt.Println(n.SnapshotId)
+		if n.SnapshotId != nil {
+			fmt.Println(*n.SnapshotId)
+		}
 		fmt.Println(n.Tags)
+		fmt.Println(n.VolumeId)
+		if n.VolumeId != nil {
+			fmt.Println(*n.VolumeId)
+		}
 	}
 	if len(result.Snapshots) != snapshotCheck.ExpectCount {
 		return errors.New(fmt.Sprintf("Snapshot count is not as expected %d", snapshotCheck.ExpectCount))
--- a/test/e2e/util/providers/common.go
+++ b/test/e2e/util/providers/common.go
@ -35,7 +35,7 @@ type ObjectsInStorage interface {
 }

 func ObjectsShouldBeInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix string) error {
-	fmt.Printf("|| VERIFICATION || - %s %s should exist in storage [%s]\n", subPrefix, backupName, bslPrefix)
+	fmt.Printf("|| VERIFICATION || - %s should exist in storage [%s %s]\n", backupName, bslPrefix, subPrefix)
 	exist, err := IsObjectsInBucket(cloudProvider, cloudCredentialsFile, bslBucket, bslPrefix, bslConfig, backupName, subPrefix)
 	if !exist {
 		return errors.Wrap(err, fmt.Sprintf("|| UNEXPECTED ||Backup object %s is not exist in object store after backup as expected\n", backupName))
--- a/test/e2e/util/velero/install.go
+++ b/test/e2e/util/velero/install.go
@ -76,7 +76,7 @@ func VeleroInstall(ctx context.Context, veleroCfg *VerleroConfig, useVolumeSnaps
 		// backup, but needed to pick up the provider plugins earlier.  vSphere plugin no longer needs a Volume
 		// Snapshot location specified
 		veleroCfg.ObjectStoreProvider = "aws"
-		if err := configvSpherePlugin(); err != nil {
+		if err := configvSpherePlugin(*veleroCfg.ClientToInstallVelero); err != nil {
 			return errors.WithMessagef(err, "Failed to config vsphere plugin")
 		}
 	}
@ -106,11 +106,8 @@ func VeleroInstall(ctx context.Context, veleroCfg *VerleroConfig, useVolumeSnaps
 }

 //configvSpherePlugin refers to https://github.com/vmware-tanzu/velero-plugin-for-vsphere/blob/v1.3.0/docs/vanilla.md
-func configvSpherePlugin() error {
-	cli, err := NewTestClient()
-	if err != nil {
-		return errors.WithMessagef(err, "Failed to instantiate cluster client to config vsphere plugin")
-	}
+func configvSpherePlugin(cli TestClient) error {
+	var err error
 	vsphereSecret := "velero-vsphere-config-secret"
 	configmaptName := "velero-vsphere-plugin-config"
 	if err := clearupvSpherePluginConfig(cli.ClientGo, VeleroCfg.VeleroNamespace, vsphereSecret, configmaptName); err != nil {
--- a/test/e2e/util/velero/velero_utils.go
+++ b/test/e2e/util/velero/velero_utils.go
@ -309,9 +309,11 @@ func VeleroBackupNamespace(ctx context.Context, veleroCLI, veleroNamespace strin
 	args := []string{
 		"--namespace", veleroNamespace,
 		"create", "backup", backupCfg.BackupName,
-		"--include-namespaces", backupCfg.Namespace,
 		"--wait",
 	}
+	if backupCfg.Namespace != "" {
+		args = append(args, "--include-namespaces", backupCfg.Namespace)
+	}
 	if backupCfg.Selector != "" {
 		args = append(args, "--selector", backupCfg.Selector)
 	}
@ -332,6 +334,23 @@ func VeleroBackupNamespace(ctx context.Context, veleroCLI, veleroNamespace strin
 	if backupCfg.TTL != 0 {
 		args = append(args, "--ttl", backupCfg.TTL.String())
 	}
+
+	if backupCfg.IncludeResources != "" {
+		args = append(args, "--include-resources", backupCfg.IncludeResources)
+	}
+
+	if backupCfg.ExcludeResources != "" {
+		args = append(args, "--exclude-resources", backupCfg.ExcludeResources)
+	}
+
+	if backupCfg.IncludeClusterResources {
+		args = append(args, "--include-cluster-resources")
+	}
+
+	if backupCfg.OrderedResources != "" {
+		args = append(args, "--ordered-resources", backupCfg.OrderedResources)
+	}
+
 	return VeleroBackupExec(ctx, veleroCLI, veleroNamespace, backupCfg.BackupName, args)
 }

@ -358,11 +377,14 @@ func VeleroBackupIncludeNamespaces(ctx context.Context, veleroCLI string, velero
 }

 // VeleroRestore uses the VeleroCLI to restore from a Velero backup.
-func VeleroRestore(ctx context.Context, veleroCLI string, veleroNamespace string, restoreName string, backupName string) error {
+func VeleroRestore(ctx context.Context, veleroCLI, veleroNamespace, restoreName, backupName, includeResources string) error {
 	args := []string{
 		"--namespace", veleroNamespace, "create", "restore", restoreName,
 		"--from-backup", backupName, "--wait",
 	}
+	if includeResources != "" {
+		args = append(args, "--include-resources", includeResources)
+	}
 	return VeleroRestoreExec(ctx, veleroCLI, veleroNamespace, restoreName, args)
 }

@ -761,7 +783,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)
+	fmt.Printf("Backup <%s> exist locally according to output \n[%s]\n", backupName, out)
 	return true, nil
 }

@ -785,9 +807,9 @@ func WaitForExpectedStateOfBackup(ctx context.Context, veleroCLI string, backupN
 		if exist, err := IsBackupExist(ctx, veleroCLI, backupName); err != nil {
 			return false, err
 		} else {
-			msg := "does not exist"
-			if existing {
-				msg = "was found"
+			msg := "does not exist as expect"
+			if exist {
+				msg = "was found as expect"
 			}
 			if exist == existing {
 				fmt.Println("Backup <" + backupName + "> " + msg)
@ -892,22 +914,22 @@ func SnapshotCRsCountShouldBe(ctx context.Context, namespace, backupName string,
 	}
 }

-func ResticRepositoriesCountShouldBe(ctx context.Context, veleroNamespace, targetNamespace string, expectedCount int) error {
+func BackupRepositoriesCountShouldBe(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")
+		return errors.Wrapf(err, "Fail to get BackupRepositories")
 	}
 	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))
+		return errors.New(fmt.Sprintf("BackupRepositories 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"},
+		Args: []string{"get", "-n", veleroNamespace, "BackupRepositories"},
 	}

 	CmdLine2 := &common.OsCommandLine{
@ -968,3 +990,21 @@ func GetBackupTTL(ctx context.Context, veleroNamespace, backupName string) (stri
 	// return complete, nil
 	return stdout, err
 }
+
+func GetVersionList(veleroCli, veleroVersion string) []VeleroCLI2Version {
+	var veleroCLI2VersionList []VeleroCLI2Version
+	veleroVersionList := strings.Split(veleroVersion, ",")
+	veleroCliList := strings.Split(veleroCli, ",")
+
+	for _, veleroVersion := range veleroVersionList {
+		veleroCLI2VersionList = append(veleroCLI2VersionList,
+			VeleroCLI2Version{veleroVersion, ""})
+	}
+	for i, veleroCli := range veleroCliList {
+		if i == len(veleroVersionList)-1 {
+			break
+		}
+		veleroCLI2VersionList[i].VeleroCLI = veleroCli
+	}
+	return veleroCLI2VersionList
+}