diff --git a/Makefile b/Makefile
index 48136d5abdd..abcde958c76 100644
--- a/Makefile
+++ b/Makefile
@@ -31,6 +31,7 @@ ARTIFACT_DIR ?= /tmp
 OC_CLI = $(shell which oc)
 TEST_VIRT ?= false
 TEST_UPGRADE ?= false
+TEST_CLI ?= false
 
 ifdef CLI_DIR
 	OC_CLI = ${CLI_DIR}/oc
@@ -494,6 +495,11 @@ ifeq ($(TEST_UPGRADE),true)
 else
 	TEST_FILTER += && (! upgrade)
 endif
+ifeq ($(TEST_CLI),true)
+	TEST_FILTER += && (cli)
+else
+	TEST_FILTER += && (! cli)
+endif
 SETTINGS_TMP=/tmp/test-settings
 
 .PHONY: test-e2e-setup
diff --git a/build/ci-Dockerfile b/build/ci-Dockerfile
index 8208510d941..b35d6b7e8e3 100644
--- a/build/ci-Dockerfile
+++ b/build/ci-Dockerfile
@@ -6,6 +6,21 @@ WORKDIR /go/src/github.com/openshift/oadp-operator
 COPY ./ .
 
 RUN go install -mod=mod github.com/onsi/ginkgo/v2/ginkgo
+
+# Clone and install oadp-cli pinned to oadp-1.3
+RUN git clone --depth=1 --branch oadp-1.3 https://github.com/migtools/oadp-cli.git /tmp/oadp-cli && \
+    cd /tmp/oadp-cli && \
+    make build && \
+    cp kubectl-oadp /usr/local/bin/ && \
+    chmod +x /usr/local/bin/kubectl-oadp && \
+    rm -rf /tmp/oadp-cli
+
+# Install kubectl (multi-arch)
+RUN ARCH=$(uname -m | sed 's/x86_64/amd64/;s/aarch64/arm64/') && \
+    curl -LO "https://dl.k8s.io/release/$(curl -L -s https://dl.k8s.io/release/stable.txt)/bin/linux/${ARCH}/kubectl" && \
+    chmod +x kubectl && \
+    mv kubectl /usr/local/bin/
+
 RUN go mod download
 RUN chmod -R 777 ./
 RUN chmod -R 777 $(go env GOPATH)
diff --git a/docs/developer/testing/TESTING.md b/docs/developer/testing/TESTING.md
index 316ceee3931..3cbeafd55e3 100644
--- a/docs/developer/testing/TESTING.md
+++ b/docs/developer/testing/TESTING.md
@@ -24,6 +24,7 @@ To get started, you need to provide the following **required** environment varia
 | `MUST_GATHER_IMAGE` | Container image to use for must-gather collection via `oc adm must-gather` | `quay.io/konveyor/oadp-must-gather:oadp-1.3` | false |
 | `MUST_GATHER_REPO` | GitHub repo (e.g., `openshift/oadp-must-gather`) to build must-gather from source. Sets `MUST_GATHER_IMAGE` to a ttl.sh image automatically | - | false |
 | `MUST_GATHER_BRANCH` | Branch to use when building from `MUST_GATHER_REPO` | `oadp-1.3` | false |
+| `TEST_CLI` | Exclusively run CLI-based backup/restore testing | `false` | false |
 
 The expected format for `OADP_CRED_FILE` and `CI_CRED_FILE` files is:
 ```
diff --git a/tests/e2e/backup_restore_cli_suite_test.go b/tests/e2e/backup_restore_cli_suite_test.go
new file mode 100644
index 00000000000..9b638b41857
--- /dev/null
+++ b/tests/e2e/backup_restore_cli_suite_test.go
@@ -0,0 +1,366 @@
+package e2e_test
+
+import (
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	"github.com/onsi/ginkgo/v2"
+	"github.com/onsi/gomega"
+
+	"github.com/openshift/oadp-operator/tests/e2e/lib"
+)
+
+// CLI-specific backup execution
+func runBackupViaCLI(brCase BackupRestoreCase, backupName string) bool {
+	nsRequiresResticDCWorkaround, err := lib.NamespaceRequiresResticDCWorkaround(dpaCR.Client, brCase.Namespace)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	if strings.Contains(brCase.Name, "twovol") {
+		volumeSyncDelay := 30 * time.Second
+		log.Printf("Sleeping for %v to allow volume to be in sync with /tmp/log/ for case %s", volumeSyncDelay, brCase.Name)
+		time.Sleep(volumeSyncDelay)
+	}
+
+	// Create backup via CLI
+	log.Printf("Creating backup %s for case %s via CLI", backupName, brCase.Name)
+	err = lib.CreateBackupForNamespacesViaCLI(backupName, []string{brCase.Namespace}, brCase.BackupRestoreType == lib.RESTIC || brCase.BackupRestoreType == lib.KOPIA, brCase.BackupRestoreType == lib.CSIDataMover)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	// Wait for backup via CLI
+	gomega.Eventually(lib.IsBackupDoneViaCLI(backupName), brCase.BackupTimeout, time.Second*10).Should(gomega.BeTrue())
+
+	// Get backup details via CLI
+	describeBackup := lib.DescribeBackupViaCLI(backupName)
+	ginkgo.GinkgoWriter.Println(describeBackup)
+
+	backupLogs, err := lib.BackupLogsViaCLI(backupName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	backupErrorLogs := lib.BackupErrorLogsViaCLI(backupName)
+	accumulatedTestLogs = append(accumulatedTestLogs, describeBackup, backupLogs)
+
+	if !brCase.SkipVerifyLogs {
+		gomega.Expect(backupErrorLogs).Should(gomega.Equal([]string{}))
+	}
+
+	// Check if backup succeeded
+	succeeded, err := lib.IsBackupCompletedSuccessfullyViaCLI(backupName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	gomega.Expect(succeeded).To(gomega.Equal(true))
+
+	log.Printf("Backup for case %s succeeded via CLI", brCase.Name)
+
+	return nsRequiresResticDCWorkaround
+}
+
+// CLI-specific restore execution
+func runRestoreViaCLI(brCase BackupRestoreCase, backupName, restoreName string, nsRequiresResticDCWorkaround bool) {
+	log.Printf("Creating restore %s for case %s via CLI", restoreName, brCase.Name)
+	err := lib.CreateRestoreFromBackupViaCLI(backupName, restoreName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	gomega.Eventually(lib.IsRestoreDoneViaCLI(restoreName), time.Minute*60, time.Second*10).Should(gomega.BeTrue())
+
+	// Get restore details via CLI
+	describeRestore := lib.DescribeRestoreViaCLI(restoreName)
+	ginkgo.GinkgoWriter.Println(describeRestore)
+
+	restoreLogs, err := lib.RestoreLogsViaCLI(restoreName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	restoreErrorLogs := lib.RestoreErrorLogsViaCLI(restoreName)
+	accumulatedTestLogs = append(accumulatedTestLogs, describeRestore, restoreLogs)
+
+	if !brCase.SkipVerifyLogs {
+		gomega.Expect(restoreErrorLogs).Should(gomega.Equal([]string{}))
+	}
+
+	// Check if restore succeeded
+	succeeded, err := lib.IsRestoreCompletedSuccessfullyViaCLI(restoreName)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	gomega.Expect(succeeded).To(gomega.Equal(true))
+
+	if nsRequiresResticDCWorkaround {
+		log.Printf("Running dc-post-restore.sh script.")
+		err = lib.RunDcPostRestoreScript(restoreName)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	}
+}
+
+// CLI-specific application backup and restore
+func runApplicationBackupAndRestoreViaCLI(brCase ApplicationBackupRestoreCase, updateLastBRcase func(brCase ApplicationBackupRestoreCase), updateLastInstallTime func()) {
+	updateLastBRcase(brCase)
+
+	// create DPA (still using K8s client for setup)
+	backupName, restoreName := prepareBackupAndRestore(brCase.BackupRestoreCase, updateLastInstallTime)
+
+	// Ensure that an existing backup repository is deleted (still using K8s client)
+	brerr := lib.DeleteBackupRepositories(runTimeClientForSuiteRun, namespace)
+	gomega.Expect(brerr).ToNot(gomega.HaveOccurred())
+
+	// install app (still using K8s client for setup)
+	updateLastInstallTime()
+	log.Printf("Installing application for case %s", brCase.Name)
+	err := lib.InstallApplication(dpaCR.Client, brCase.ApplicationTemplate)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	if brCase.BackupRestoreType == lib.CSI || brCase.BackupRestoreType == lib.CSIDataMover {
+		log.Printf("Creating pvc for case %s", brCase.Name)
+		var pvcName string
+		var pvcPath string
+
+		pvcName = provider
+		if brCase.PvcSuffixName != "" {
+			pvcName += brCase.PvcSuffixName
+		}
+
+		pvcPathFormat := "./sample-applications/%s/pvc/%s.yaml"
+		if strings.Contains(brCase.Name, "twovol") {
+			pvcPathFormat = "./sample-applications/%s/pvc-twoVol/%s.yaml"
+		}
+
+		pvcPath = fmt.Sprintf(pvcPathFormat, brCase.Namespace, pvcName)
+
+		err = lib.InstallApplication(dpaCR.Client, pvcPath)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	}
+
+	// Run optional custom verification
+	if brCase.PreBackupVerify != nil {
+		log.Printf("Running pre-backup custom function for case %s", brCase.Name)
+		err := brCase.PreBackupVerify(dpaCR.Client, brCase.Namespace)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	}
+
+	// do the backup via CLI
+	nsRequiredResticDCWorkaround := runBackupViaCLI(brCase.BackupRestoreCase, backupName)
+
+	// uninstall app (still using K8s client)
+	log.Printf("Uninstalling app for case %s", brCase.Name)
+	err = lib.UninstallApplication(dpaCR.Client, brCase.ApplicationTemplate)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	// Wait for namespace to be deleted
+	gomega.Eventually(lib.IsNamespaceDeleted(kubernetesClientForSuiteRun, brCase.Namespace), time.Minute*4, time.Second*5).Should(gomega.BeTrue())
+
+	updateLastInstallTime()
+
+	// run restore via CLI
+	runRestoreViaCLI(brCase.BackupRestoreCase, backupName, restoreName, nsRequiredResticDCWorkaround)
+
+	// Run optional custom verification
+	if brCase.PostRestoreVerify != nil {
+		log.Printf("Running post-restore custom function for case %s", brCase.Name)
+		err = brCase.PostRestoreVerify(dpaCR.Client, brCase.Namespace)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	}
+}
+
+// CLI teardown function
+func tearDownBackupAndRestoreViaCLI(brCase BackupRestoreCase, installTime time.Time, report ginkgo.SpecReport) {
+	log.Println("Post backup and restore state (CLI): ", report.State.String())
+
+	if report.Failed() {
+		knownFlake = lib.CheckIfFlakeOccurred(accumulatedTestLogs)
+		accumulatedTestLogs = nil
+		getFailedTestLogs(namespace, brCase.Namespace, installTime, report)
+	}
+
+	if brCase.BackupRestoreType == lib.CSI || brCase.BackupRestoreType == lib.CSIDataMover {
+		log.Printf("Deleting VolumeSnapshot for CSI backuprestore of %s", brCase.Name)
+		snapshotClassPath := fmt.Sprintf("./sample-applications/snapclass-csi/%s.yaml", provider)
+		err := lib.UninstallApplication(dpaCR.Client, snapshotClassPath)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	}
+
+	err := dpaCR.Delete(runTimeClientForSuiteRun)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+	err = lib.DeleteNamespace(kubernetesClientForSuiteRun, brCase.Namespace)
+	gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	gomega.Eventually(lib.IsNamespaceDeleted(kubernetesClientForSuiteRun, brCase.Namespace), time.Minute*5, time.Second*5).Should(gomega.BeTrue())
+}
+
+// CLI Test Suite
+var _ = ginkgo.Describe("Backup and restore tests via OADP CLI", ginkgo.Label("cli"), ginkgo.Ordered, func() {
+	var lastBRCase ApplicationBackupRestoreCase
+	var lastInstallTime time.Time
+	updateLastBRcase := func(brCase ApplicationBackupRestoreCase) {
+		lastBRCase = brCase
+	}
+	updateLastInstallTime := func() {
+		lastInstallTime = time.Now()
+	}
+
+	ginkgo.BeforeAll(func() {
+
+		cliSetup := lib.NewOADPCLISetup()
+		if err := cliSetup.Install(); err != nil {
+			ginkgo.Fail(fmt.Sprintf("OADP CLI setup failed: %v", err))
+		}
+	})
+
+	var _ = ginkgo.AfterEach(func(ctx ginkgo.SpecContext) {
+		tearDownBackupAndRestoreViaCLI(lastBRCase.BackupRestoreCase, lastInstallTime, ctx.SpecReport())
+	})
+
+	var _ = ginkgo.AfterAll(func() {
+		// Same cleanup as original
+		waitOADPReadiness(lib.KOPIA)
+
+		var err error
+		log.Printf("skipMustGather: %v", skipMustGather)
+		if !skipMustGather {
+			log.Printf("Running OADP must-gather")
+			err = lib.RunMustGather(artifact_dir, dpaCR.Client)
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+		}
+
+		err = dpaCR.Delete(runTimeClientForSuiteRun)
+		gomega.Expect(err).ToNot(gomega.HaveOccurred())
+	})
+
+	ginkgo.DescribeTable("Backup and restore applications via OADP CLI",
+		func(brCase ApplicationBackupRestoreCase, expectedErr error) {
+			if ginkgo.CurrentSpecReport().NumAttempts > 1 && !knownFlake {
+				ginkgo.Fail("No known FLAKE found in a previous run, marking test as failed.")
+			}
+			runApplicationBackupAndRestoreViaCLI(brCase, updateLastBRcase, updateLastInstallTime)
+		},
+		ginkgo.Entry("MySQL application CSI via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent-csi.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-csi-cli-e2e",
+				BackupRestoreType: lib.CSI,
+				PreBackupVerify:   todoListReady(true, false, "mysql"),
+				PostRestoreVerify: todoListReady(false, false, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application CSI via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent-csi.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-csi-cli-e2e",
+				BackupRestoreType: lib.CSI,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("MySQL application two Vol CSI via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent-twovol-csi.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-twovol-csi-cli-e2e",
+				BackupRestoreType: lib.CSI,
+				PreBackupVerify:   todoListReady(true, true, "mysql"),
+				PostRestoreVerify: todoListReady(false, true, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application RESTIC via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-restic-cli-e2e",
+				BackupRestoreType: lib.RESTIC,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("MySQL application RESTIC via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-restic-cli-e2e",
+				BackupRestoreType: lib.RESTIC,
+				PreBackupVerify:   todoListReady(true, false, "mysql"),
+				PostRestoreVerify: todoListReady(false, false, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application KOPIA via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-kopia-cli-e2e",
+				BackupRestoreType: lib.KOPIA,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("MySQL application KOPIA via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-kopia-cli-e2e",
+				BackupRestoreType: lib.KOPIA,
+				PreBackupVerify:   todoListReady(true, false, "mysql"),
+				PostRestoreVerify: todoListReady(false, false, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application DATAMOVER via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent-csi.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-datamover-cli-e2e",
+				BackupRestoreType: lib.CSIDataMover,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("MySQL application DATAMOVER via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent-csi.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-datamover-cli-e2e",
+				BackupRestoreType: lib.CSIDataMover,
+				PreBackupVerify:   todoListReady(true, false, "mysql"),
+				PostRestoreVerify: todoListReady(false, false, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application BlockDevice DATAMOVER via CLI", ginkgo.FlakeAttempts(flakeAttempts), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent-block.yaml",
+			PvcSuffixName:       "-block-mode",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-blockdevice-cli-e2e",
+				BackupRestoreType: lib.CSIDataMover,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("MySQL application Native-Snapshots via CLI", ginkgo.FlakeAttempts(flakeAttempts), ginkgo.Label("aws", "azure", "gcp"), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mysql-persistent/mysql-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mysql-persistent",
+				Name:              "mysql-native-snapshots-cli-e2e",
+				BackupRestoreType: lib.NativeSnapshots,
+				PreBackupVerify:   todoListReady(true, false, "mysql"),
+				PostRestoreVerify: todoListReady(false, false, "mysql"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+		ginkgo.Entry("Mongo application Native-Snapshots via CLI", ginkgo.FlakeAttempts(flakeAttempts), ginkgo.Label("aws", "azure", "gcp"), ApplicationBackupRestoreCase{
+			ApplicationTemplate: "./sample-applications/mongo-persistent/mongo-persistent.yaml",
+			BackupRestoreCase: BackupRestoreCase{
+				Namespace:         "mongo-persistent",
+				Name:              "mongo-native-snapshots-cli-e2e",
+				BackupRestoreType: lib.NativeSnapshots,
+				PreBackupVerify:   todoListReady(true, false, "mongo"),
+				PostRestoreVerify: todoListReady(false, false, "mongo"),
+				BackupTimeout:     20 * time.Minute,
+			},
+		}, nil),
+	)
+})
diff --git a/tests/e2e/backup_restore_suite_test.go b/tests/e2e/backup_restore_suite_test.go
index 1dff67c0acc..e10f3a6f421 100755
--- a/tests/e2e/backup_restore_suite_test.go
+++ b/tests/e2e/backup_restore_suite_test.go
@@ -24,6 +24,23 @@ type VerificationFunction func(client.Client, string) error
 type appVerificationFunction func(bool, bool, BackupRestoreType) VerificationFunction
 
 // TODO duplications with mongoready
+func todoListReady(preBackupState bool, twoVol bool, database string) VerificationFunction {
+	return VerificationFunction(func(ocClient client.Client, namespace string) error {
+		log.Printf("checking for the NAMESPACE: %s", namespace)
+		Eventually(IsDeploymentReady(ocClient, namespace, database), timeoutMultiplier*time.Minute*10, time.Second*10).Should(BeTrue())
+		Eventually(IsDCReady(ocClient, namespace, "todolist"), timeoutMultiplier*time.Minute*10, time.Second*10).Should(BeTrue())
+		Eventually(AreApplicationPodsRunning(kubernetesClientForSuiteRun, namespace), timeoutMultiplier*time.Minute*9, time.Second*5).Should(BeTrue())
+		exists, err := DoesSCCExist(ocClient, database+"-persistent-scc")
+		if err != nil {
+			return err
+		}
+		if !exists {
+			return errors.New("did not find " + database + " scc")
+		}
+		return nil
+	})
+}
+
 func mongoready(preBackupState bool, twoVol bool, backupRestoreType BackupRestoreType) VerificationFunction {
 	return VerificationFunction(func(ocClient client.Client, namespace string) error {
 		Eventually(IsDCReady(ocClient, namespace, "todolist"), timeoutMultiplier*time.Minute*10, time.Second*10).Should(BeTrue())
@@ -60,11 +77,20 @@ type BackupRestoreCase struct {
 	ApplicationTemplate  string
 	PvcSuffixName        string
 	ApplicationNamespace string
+	Namespace            string
 	Name                 string
 	BackupRestoreType    BackupRestoreType
 	PreBackupVerify      VerificationFunction
 	PostRestoreVerify    VerificationFunction
 	AppReadyDelay        time.Duration
+	BackupTimeout        time.Duration
+	SkipVerifyLogs       bool
+}
+
+type ApplicationBackupRestoreCase struct {
+	BackupRestoreCase
+	ApplicationTemplate string
+	PvcSuffixName       string
 }
 
 func waitOADPReadiness(backupRestoreType BackupRestoreType) {
@@ -233,6 +259,45 @@ func runBackupAndRestore(brCase BackupRestoreCase, expectedErr error, updateLast
 	Expect(err).ToNot(HaveOccurred())
 }
 
+func prepareBackupAndRestore(brCase BackupRestoreCase, updateLastInstallTime func()) (string, string) {
+	updateLastInstallTime()
+	waitOADPReadiness(brCase.BackupRestoreType)
+
+	if brCase.BackupRestoreType == CSI || brCase.BackupRestoreType == CSIDataMover {
+		if provider == "aws" || provider == "ibmcloud" || provider == "gcp" || provider == "azure" {
+			log.Printf("Creating VolumeSnapshotClass for CSI backuprestore of %s", brCase.Name)
+			snapshotClassPath := fmt.Sprintf("./sample-applications/snapclass-csi/%s.yaml", provider)
+			err := InstallApplication(dpaCR.Client, snapshotClassPath)
+			Expect(err).ToNot(HaveOccurred())
+		}
+	}
+
+	backupUid, _ := uuid.NewUUID()
+	restoreUid, _ := uuid.NewUUID()
+	backupName := fmt.Sprintf("%s-%s", brCase.Name, backupUid.String())
+	restoreName := fmt.Sprintf("%s-%s", brCase.Name, restoreUid.String())
+	return backupName, restoreName
+}
+
+func getFailedTestLogs(oadpNamespace string, appNamespace string, installTime time.Time, report SpecReport) {
+	baseReportDir := artifact_dir + "/" + report.LeafNodeText
+	log.Println("Storing failed test logs in: ", baseReportDir)
+	err := os.MkdirAll(baseReportDir, 0755)
+	Expect(err).NotTo(HaveOccurred())
+
+	log.Println("Printing OADP namespace events")
+	PrintNamespaceEventsAfterTime(kubernetesClientForSuiteRun, oadpNamespace, installTime)
+	err = SavePodLogs(kubernetesClientForSuiteRun, oadpNamespace, baseReportDir)
+	Expect(err).NotTo(HaveOccurred())
+
+	if appNamespace != "" {
+		log.Println("Printing app namespace events")
+		PrintNamespaceEventsAfterTime(kubernetesClientForSuiteRun, appNamespace, installTime)
+		err = SavePodLogs(kubernetesClientForSuiteRun, appNamespace, baseReportDir)
+		Expect(err).NotTo(HaveOccurred())
+	}
+}
+
 func tearDownBackupAndRestore(brCase BackupRestoreCase, installTime time.Time, report SpecReport) {
 	log.Println("Post backup and restore state: ", report.State.String())
 	knownFlake = false
diff --git a/tests/e2e/e2e_suite_test.go b/tests/e2e/e2e_suite_test.go
index f5f87bcb689..902a5525ba4 100755
--- a/tests/e2e/e2e_suite_test.go
+++ b/tests/e2e/e2e_suite_test.go
@@ -24,6 +24,7 @@ import (
 var bslCredFile, namespace, credSecretRef, instanceName, provider, vslCredFile, settings, artifact_dir, stream string
 var timeoutMultiplierInput, flakeAttempts int64
 var timeoutMultiplier time.Duration
+var skipMustGather bool
 
 func init() {
 	// TODO better descriptions to flags
@@ -40,6 +41,7 @@ func init() {
 	flag.Int64Var(&timeoutMultiplierInput, "timeout_multiplier", 1, "Customize timeout multiplier from default (1)")
 	timeoutMultiplier = time.Duration(timeoutMultiplierInput)
 	flag.Int64Var(&flakeAttempts, "flakeAttempts", 3, "Customize the number of flake retries (3)")
+	flag.BoolVar(&skipMustGather, "skipMustGather", false, "avoid errors with local execution and cluster architecture")
 
 	// helps with launching debug sessions from IDE
 	if os.Getenv("E2E_USE_ENV_FLAGS") == "true" {
diff --git a/tests/e2e/lib/backup.go b/tests/e2e/lib/backup.go
index 2ddf236547d..3467b940330 100755
--- a/tests/e2e/lib/backup.go
+++ b/tests/e2e/lib/backup.go
@@ -46,8 +46,8 @@ func IsBackupDone(ocClient client.Client, veleroNamespace, name string) wait.Con
 			velero.BackupPhaseInProgress,
 			velero.BackupPhaseWaitingForPluginOperations,
 			velero.BackupPhaseWaitingForPluginOperationsPartiallyFailed,
-			velero.BackupPhaseFinalizing,
-			velero.BackupPhaseFinalizingPartiallyFailed,
+			"Finalizing",
+			"FinalizingPartiallyFailed",
 			"",
 		}
 		for _, notDonePhase := range phasesNotDone {
@@ -77,3 +77,78 @@ func IsBackupCompletedSuccessfully(c *kubernetes.Clientset, ocClient client.Clie
 		GetVeleroContainerFailureLogs(c, backup.Namespace),
 	)
 }
+
+// backupPhasesNotDone is the shared list of backup phases that indicate a backup is still in progress.
+var backupPhasesNotDone = []velero.BackupPhase{
+	velero.BackupPhaseNew,
+	"Queued",
+	"ReadyToStart",
+	velero.BackupPhaseInProgress,
+	velero.BackupPhaseWaitingForPluginOperations,
+	velero.BackupPhaseWaitingForPluginOperationsPartiallyFailed,
+	"Finalizing",
+	"FinalizingPartiallyFailed",
+	"",
+}
+
+// IsBackupPhaseNotDone returns true if the given phase string represents a backup still in progress.
+func IsBackupPhaseNotDone(phase string) bool {
+	for _, notDonePhase := range backupPhasesNotDone {
+		if phase == string(notDonePhase) {
+			return true
+		}
+	}
+	return false
+}
+
+func GetBackupRepositoryList(c client.Client, namespace string) (*velero.BackupRepositoryList, error) {
+	backupRepositoryList := &velero.BackupRepositoryList{
+		Items: []velero.BackupRepository{},
+	}
+	err := c.List(context.Background(), backupRepositoryList, client.InNamespace(namespace))
+	if err != nil {
+		log.Printf("error getting BackupRepository list: %v", err)
+		return nil, err
+	}
+	return backupRepositoryList, nil
+}
+
+func DeleteBackupRepository(c client.Client, namespace string, name string) error {
+	backupRepository := &velero.BackupRepository{
+		ObjectMeta: metav1.ObjectMeta{
+			Namespace: namespace,
+			Name:      name,
+		},
+	}
+	err := c.Delete(context.Background(), backupRepository)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+// DeleteBackupRepositories deletes all BackupRepositories in the given namespace.
+func DeleteBackupRepositories(c client.Client, namespace string) error {
+	log.Printf("Checking if backuprepository's exist in %s", namespace)
+
+	backupRepos, err := GetBackupRepositoryList(c, namespace)
+	if err != nil {
+		return fmt.Errorf("failed to get BackupRepository list: %v", err)
+	}
+	if len(backupRepos.Items) == 0 {
+		log.Printf("No BackupRepositories found in namespace %s", namespace)
+		return nil
+	}
+
+	for _, repo := range backupRepos.Items {
+		log.Printf("backuprepository name is %s", repo.Name)
+		err := DeleteBackupRepository(c, namespace, repo.Name)
+		if err != nil {
+			log.Printf("failed to delete BackupRepository %s: ", repo.Name)
+			return err
+		}
+		log.Printf("Successfully deleted BackupRepository: %s", repo.Name)
+	}
+
+	return nil
+}
diff --git a/tests/e2e/lib/backup_cli.go b/tests/e2e/lib/backup_cli.go
new file mode 100644
index 00000000000..50da9f6d1cd
--- /dev/null
+++ b/tests/e2e/lib/backup_cli.go
@@ -0,0 +1,281 @@
+package lib
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	velero "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+// CreateBackupForNamespacesViaCLI creates a backup using the OADP CLI
+func CreateBackupForNamespacesViaCLI(backupName string, namespaces []string, defaultVolumesToFsBackup bool, snapshotMoveData bool) error {
+	var options []string
+
+	// Add included namespaces (comma-separated)
+	if len(namespaces) > 0 {
+		options = append(options, "--include-namespaces", strings.Join(namespaces, ","))
+	}
+
+	// Add volume backup options
+	if defaultVolumesToFsBackup {
+		options = append(options, "--default-volumes-to-fs-backup")
+	}
+
+	// Add snapshot move data option
+	if snapshotMoveData {
+		options = append(options, "--snapshot-move-data")
+	}
+
+	// Execute CLI command
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "create",
+		Name:     backupName,
+		Options:  options,
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to create backup via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Backup created via CLI: %s", backupName)
+	return nil
+}
+
+// CreateCustomBackupForNamespacesViaCLI creates a custom backup using the OADP CLI
+func CreateCustomBackupForNamespacesViaCLI(backupName string, namespaces []string, includedResources, excludedResources []string, defaultVolumesToFsBackup bool, snapshotMoveData bool) error {
+	var options []string
+
+	// Add included namespaces (comma-separated)
+	if len(namespaces) > 0 {
+		options = append(options, "--include-namespaces", strings.Join(namespaces, ","))
+	}
+
+	// Add included resources
+	if len(includedResources) > 0 {
+		options = append(options, "--include-resources", strings.Join(includedResources, ","))
+	}
+
+	// Add excluded resources
+	if len(excludedResources) > 0 {
+		options = append(options, "--exclude-resources", strings.Join(excludedResources, ","))
+	}
+
+	// Add volume backup options
+	if defaultVolumesToFsBackup {
+		options = append(options, "--default-volumes-to-fs-backup")
+	}
+
+	// Add snapshot move data option
+	if snapshotMoveData {
+		options = append(options, "--snapshot-move-data")
+	}
+
+	// Execute CLI command
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "create",
+		Name:     backupName,
+		Options:  options,
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to create custom backup via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Custom backup created via CLI: %s", backupName)
+	return nil
+}
+
+// GetBackupViaCLI gets backup details using the OADP CLI
+func GetBackupViaCLI(name string) (*velero.Backup, error) {
+	if name == "" {
+		return nil, fmt.Errorf("backup name cannot be empty")
+	}
+
+	// Use CLI to get backup details in JSON format
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "get",
+		Name:     name,
+		Options:  []string{"-o", "json"},
+	}
+	output, err := cmd.ExecuteOutput()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get backup via CLI: %v", err)
+	}
+
+	// Parse the JSON output back to velero.Backup struct
+	var backup velero.Backup
+	if err := json.Unmarshal(output, &backup); err != nil {
+		return nil, fmt.Errorf("failed to parse backup JSON: %v", err)
+	}
+
+	return &backup, nil
+}
+
+// IsBackupDoneViaCLI checks if backup is done using the OADP CLI
+func IsBackupDoneViaCLI(name string) wait.ConditionFunc {
+	return func() (bool, error) {
+		cmd := &CLICommand{
+			Resource: "backup",
+			Action:   "get",
+			Name:     name,
+			Options:  []string{"-o", "yaml"},
+		}
+		output, err := cmd.ExecuteOutput()
+		if err != nil {
+			return false, fmt.Errorf("failed to get backup status via CLI: %v", err)
+		}
+
+		phase := ParsePhaseFromYAML(string(output))
+
+		if len(phase) > 0 {
+			log.Printf("backup phase: %s", phase)
+		}
+
+		return !IsBackupPhaseNotDone(phase), nil
+	}
+}
+
+// IsBackupCompletedSuccessfullyViaCLI checks if backup completed successfully using the OADP CLI
+func IsBackupCompletedSuccessfullyViaCLI(name string) (bool, error) {
+	// Use CLI to get backup status
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "get",
+		Name:     name,
+		Options:  []string{"-o", "yaml"},
+	}
+	output, err := cmd.ExecuteOutput()
+	if err != nil {
+		return false, fmt.Errorf("failed to get backup status via CLI: %v", err)
+	}
+
+	// Parse phase from YAML output
+	phase := ParsePhaseFromYAML(string(output))
+
+	if phase == string(velero.BackupPhaseCompleted) {
+		return true, nil
+	}
+
+	// Get additional failure information using CLI
+	backupLogs, logsErr := BackupLogsViaCLI(name)
+	if logsErr != nil {
+		backupLogs = fmt.Sprintf("Failed to get logs: %v", logsErr)
+	}
+
+	return false, fmt.Errorf(
+		"backup phase is: %s; expected: %s\nvelero failure logs: %v",
+		phase, string(velero.BackupPhaseCompleted), backupLogs,
+	)
+}
+
+// DescribeBackupViaCLI describes backup using the OADP CLI with default timeout and retry.
+// The timeout prevents the command from hanging when retrieving backup details from object storage.
+func DescribeBackupViaCLI(name string) (backupDescription string) {
+	return DescribeBackupViaCLIWithOptions(name, DefaultCLITimeout, DefaultCLIRetries, DefaultCLIRetryDelay)
+}
+
+// DescribeBackupViaCLIWithOptions describes backup using the OADP CLI with specified timeout and retry options.
+func DescribeBackupViaCLIWithOptions(name string, timeout time.Duration, maxRetries int, retryDelay time.Duration) (backupDescription string) {
+	// Use CLI to describe backup
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "describe",
+		Name:     name,
+		Options:  []string{"--details"},
+	}
+
+	var output []byte
+	var err error
+
+	if maxRetries > 1 {
+		output, err = cmd.ExecuteWithTimeoutAndRetry(timeout, maxRetries, retryDelay)
+	} else {
+		output, err = cmd.ExecuteWithTimeout(timeout)
+	}
+
+	if err != nil {
+		return fmt.Sprintf("could not describe backup via CLI: %v, output: %s", err, string(output))
+	}
+
+	return string(output)
+}
+
+// BackupLogsViaCLI gets backup logs using the OADP CLI with default timeout and retry.
+// The timeout prevents the command from hanging indefinitely when streaming logs from object storage.
+// Retry logic helps handle transient network issues.
+func BackupLogsViaCLI(name string) (backupLogs string, err error) {
+	return BackupLogsViaCLIWithOptions(name, DefaultCLITimeout, DefaultCLIRetries, DefaultCLIRetryDelay)
+}
+
+// BackupLogsViaCLIWithTimeout gets backup logs using the OADP CLI with a specified timeout (no retry).
+func BackupLogsViaCLIWithTimeout(name string, timeout time.Duration) (backupLogs string, err error) {
+	return BackupLogsViaCLIWithOptions(name, timeout, 1, 0)
+}
+
+// BackupLogsViaCLIWithOptions gets backup logs using the OADP CLI with specified timeout and retry options.
+func BackupLogsViaCLIWithOptions(name string, timeout time.Duration, maxRetries int, retryDelay time.Duration) (backupLogs string, err error) {
+	if name == "" {
+		return "", fmt.Errorf("backup name cannot be empty")
+	}
+
+	// Use CLI to get backup logs with timeout and retry to prevent hanging
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "logs",
+		Name:     name,
+		Options:  []string{},
+	}
+
+	var output []byte
+	var cmdErr error
+
+	if maxRetries > 1 {
+		output, cmdErr = cmd.ExecuteOutputWithTimeoutAndRetry(timeout, maxRetries, retryDelay)
+	} else {
+		output, cmdErr = cmd.ExecuteOutputWithTimeout(timeout)
+	}
+
+	if cmdErr != nil {
+		return "", fmt.Errorf("failed to get backup logs via CLI: %v", cmdErr)
+	}
+
+	return string(output), nil
+}
+
+// BackupErrorLogsViaCLI gets backup error logs using the OADP CLI
+func BackupErrorLogsViaCLI(name string) []string {
+	bl, err := BackupLogsViaCLI(name)
+	if err != nil {
+		return []string{err.Error()}
+	}
+	return errorLogsExcludingIgnored(bl)
+}
+
+// DeleteBackupViaCLI deletes a backup using the OADP CLI
+func DeleteBackupViaCLI(name string) error {
+	if name == "" {
+		return fmt.Errorf("backup name cannot be empty")
+	}
+
+	// Use CLI to delete backup
+	cmd := &CLICommand{
+		Resource: "backup",
+		Action:   "delete",
+		Name:     name,
+		Options:  []string{},
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to delete backup via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Backup deleted via CLI: %s", name)
+	return nil
+}
diff --git a/tests/e2e/lib/cli_common.go b/tests/e2e/lib/cli_common.go
new file mode 100644
index 00000000000..3011bc6aef2
--- /dev/null
+++ b/tests/e2e/lib/cli_common.go
@@ -0,0 +1,340 @@
+package lib
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+)
+
+// Default timeout for CLI commands that may hang (e.g., log streaming)
+const DefaultCLITimeout = 5 * time.Minute
+
+// Default retry settings for CLI commands
+const (
+	DefaultCLIRetries    = 3
+	DefaultCLIRetryDelay = 10 * time.Second
+)
+
+type CLICommand struct {
+	Resource string // "backup" or "restore"
+	Action   string // "create", "get", "delete", etc.
+	Name     string
+	Options  []string
+	Timeout  time.Duration // Optional timeout for commands that may hang
+}
+
+func (c *CLICommand) Execute() ([]byte, error) {
+	args := []string{"oadp", c.Resource, c.Action}
+	if c.Name != "" {
+		args = append(args, c.Name)
+	}
+	args = append(args, c.Options...)
+
+	c.LogCLICommand()
+	cmd := exec.Command("kubectl", args...)
+	return cmd.CombinedOutput()
+}
+
+func (c *CLICommand) ExecuteOutput() ([]byte, error) {
+	args := []string{"oadp", c.Resource, c.Action}
+	if c.Name != "" {
+		args = append(args, c.Name)
+	}
+	args = append(args, c.Options...)
+
+	c.LogCLICommand()
+	cmd := exec.Command("kubectl", args...)
+	return cmd.Output()
+}
+
+// ExecuteWithTimeout executes the CLI command with a timeout.
+// If the timeout is exceeded, the command is killed and an error is returned.
+func (c *CLICommand) ExecuteWithTimeout(timeout time.Duration) ([]byte, error) {
+	args := []string{"oadp", c.Resource, c.Action}
+	if c.Name != "" {
+		args = append(args, c.Name)
+	}
+	args = append(args, c.Options...)
+
+	c.LogCLICommand()
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "kubectl", args...)
+	output, err := cmd.CombinedOutput()
+
+	if ctx.Err() == context.DeadlineExceeded {
+		return output, fmt.Errorf("command timed out after %v: kubectl %s", timeout, strings.Join(args, " "))
+	}
+
+	return output, err
+}
+
+// ExecuteOutputWithTimeout executes the CLI command with a timeout and returns stdout only.
+// If the timeout is exceeded, the command is killed and an error is returned.
+func (c *CLICommand) ExecuteOutputWithTimeout(timeout time.Duration) ([]byte, error) {
+	args := []string{"oadp", c.Resource, c.Action}
+	if c.Name != "" {
+		args = append(args, c.Name)
+	}
+	args = append(args, c.Options...)
+
+	c.LogCLICommand()
+
+	ctx, cancel := context.WithTimeout(context.Background(), timeout)
+	defer cancel()
+
+	cmd := exec.CommandContext(ctx, "kubectl", args...)
+	output, err := cmd.Output()
+
+	if ctx.Err() == context.DeadlineExceeded {
+		return output, fmt.Errorf("command timed out after %v: kubectl %s", timeout, strings.Join(args, " "))
+	}
+
+	return output, err
+}
+
+// ExecuteOutputWithTimeoutAndRetry executes the CLI command with a timeout and retry logic.
+// It retries the command up to maxRetries times with a delay between attempts.
+// This is useful for commands that may fail due to transient issues (e.g., network problems).
+func (c *CLICommand) ExecuteOutputWithTimeoutAndRetry(timeout time.Duration, maxRetries int, retryDelay time.Duration) ([]byte, error) {
+	var lastErr error
+	var lastOutput []byte
+
+	for attempt := 1; attempt <= maxRetries; attempt++ {
+		output, err := c.ExecuteOutputWithTimeout(timeout)
+		if err == nil {
+			return output, nil
+		}
+
+		lastErr = err
+		lastOutput = output
+
+		if attempt < maxRetries {
+			log.Printf("CLI command failed (attempt %d/%d): %v. Retrying in %v...", attempt, maxRetries, err, retryDelay)
+			time.Sleep(retryDelay)
+		}
+	}
+
+	return lastOutput, fmt.Errorf("CLI command failed after %d attempts: %v", maxRetries, lastErr)
+}
+
+// ExecuteWithTimeoutAndRetry executes the CLI command with a timeout and retry logic.
+// It retries the command up to maxRetries times with a delay between attempts.
+func (c *CLICommand) ExecuteWithTimeoutAndRetry(timeout time.Duration, maxRetries int, retryDelay time.Duration) ([]byte, error) {
+	var lastErr error
+	var lastOutput []byte
+
+	for attempt := 1; attempt <= maxRetries; attempt++ {
+		output, err := c.ExecuteWithTimeout(timeout)
+		if err == nil {
+			return output, nil
+		}
+
+		lastErr = err
+		lastOutput = output
+
+		if attempt < maxRetries {
+			log.Printf("CLI command failed (attempt %d/%d): %v. Retrying in %v...", attempt, maxRetries, err, retryDelay)
+			time.Sleep(retryDelay)
+		}
+	}
+
+	return lastOutput, fmt.Errorf("CLI command failed after %d attempts: %v", maxRetries, lastErr)
+}
+
+func (c *CLICommand) LogCLICommand() {
+	args := []string{"kubectl", "oadp", c.Resource, c.Action}
+	if c.Name != "" {
+		args = append(args, c.Name)
+	}
+	args = append(args, c.Options...)
+	log.Printf("Executing CLI command: %s", strings.Join(args, " "))
+}
+
+func ParsePhaseFromYAML(yamlOutput string) string {
+	lines := strings.Split(yamlOutput, "\n")
+	for _, line := range lines {
+		line = strings.TrimSpace(line)
+		if strings.HasPrefix(line, "phase:") {
+			return strings.TrimSpace(strings.TrimPrefix(line, "phase:"))
+		}
+	}
+	return ""
+}
+
+type CLISetup struct {
+	repoURL     string
+	repoBranch  string
+	installArgs []string
+	namespace   string
+}
+
+func NewOADPCLISetup() *CLISetup {
+	return &CLISetup{
+		repoURL:     "https://github.com/migtools/oadp-cli.git",
+		repoBranch:  "oadp-1.3",
+		installArgs: []string{"build"},
+		namespace:   "openshift-adp",
+	}
+}
+
+func (c *CLISetup) Install() error {
+	tmpDir, err := c.createTempDir()
+	if err != nil {
+		return err
+	}
+	defer os.RemoveAll(tmpDir)
+
+	cloneDir := filepath.Join(tmpDir, "oadp-cli")
+
+	steps := []struct {
+		name string
+		fn   func() error
+	}{
+		{"Cloning repository", func() error { return c.cloneRepo(cloneDir) }},
+		{"Building and installing", func() error { return c.buildAndInstall(cloneDir) }},
+		{"Verifying installation", func() error { return c.verifyInstallation() }},
+		{"Configuring namespace", func() error { return c.configureNamespace() }},
+	}
+
+	for _, step := range steps {
+		log.Printf("OADP CLI Setup: %s...", step.name)
+		if err := step.fn(); err != nil {
+			return fmt.Errorf("%s failed: %w", step.name, err)
+		}
+	}
+
+	log.Print("OADP CLI setup completed successfully")
+	return nil
+}
+
+func (c *CLISetup) createTempDir() (string, error) {
+	tmpDir, err := os.MkdirTemp("", "oadp-cli-*")
+	if err != nil {
+		return "", fmt.Errorf("failed to create temp directory: %w", err)
+	}
+	return tmpDir, nil
+}
+
+func (c *CLISetup) cloneRepo(cloneDir string) error {
+	return runCommand("git", []string{"clone", "--branch", c.repoBranch, "--depth", "1", c.repoURL, cloneDir}, "")
+}
+
+func (c *CLISetup) buildAndInstall(cloneDir string) error {
+	// Build the binary
+	log.Print("Building OADP CLI...")
+	if err := runCommand("make", c.installArgs, cloneDir); err != nil {
+		return fmt.Errorf("build failed: %w", err)
+	}
+
+	// Verify the binary was created
+	binaryPath := filepath.Join(cloneDir, "kubectl-oadp")
+	if _, err := os.Stat(binaryPath); err != nil {
+		return fmt.Errorf("kubectl-oadp binary not found at %s: %w", binaryPath, err)
+	}
+
+	// Try multiple target locations, starting with user-writable paths
+	targetPaths := []string{
+		fmt.Sprintf("%s/bin/kubectl-oadp", os.Getenv("HOME")),
+		"/usr/local/bin/kubectl-oadp",
+	}
+
+	var targetPath string
+	var moveErr error
+
+	for _, tp := range targetPaths {
+		targetPath = tp
+		log.Printf("Attempting to move binary from %s to %s", binaryPath, targetPath)
+
+		// Create directory if it doesn't exist (for ~/bin)
+		if targetPath == fmt.Sprintf("%s/bin/kubectl-oadp", os.Getenv("HOME")) {
+			binDir := filepath.Dir(targetPath)
+			if err := os.MkdirAll(binDir, 0755); err != nil {
+				log.Printf("Failed to create directory %s: %v", binDir, err)
+				continue
+			}
+		}
+
+		if err := runCommand("mv", []string{binaryPath, targetPath}, ""); err != nil {
+			log.Printf("Failed to move to %s: %v", targetPath, err)
+			moveErr = err
+			continue
+		}
+
+		// Success!
+		moveErr = nil
+		break
+	}
+
+	if moveErr != nil {
+		return fmt.Errorf("failed to move binary to any location: %w", moveErr)
+	}
+
+	// Make it executable
+	if err := runCommand("chmod", []string{"+x", targetPath}, ""); err != nil {
+		return fmt.Errorf("failed to make binary executable: %w", err)
+	}
+
+	// If we installed to ~/bin, ensure it's in PATH
+	if targetPath == fmt.Sprintf("%s/bin/kubectl-oadp", os.Getenv("HOME")) {
+		homeBin := fmt.Sprintf("%s/bin", os.Getenv("HOME"))
+		currentPath := os.Getenv("PATH")
+		if !strings.Contains(currentPath, homeBin) {
+			newPath := fmt.Sprintf("%s:%s", homeBin, currentPath)
+			os.Setenv("PATH", newPath)
+			log.Printf("Added %s to PATH", homeBin)
+		}
+	}
+
+	log.Printf("Successfully installed kubectl-oadp to %s", targetPath)
+	return nil
+}
+
+func (c *CLISetup) verifyInstallation() error {
+	// Check current PATH
+	cmd := exec.Command("bash", "-c", "echo $PATH")
+	if output, err := cmd.CombinedOutput(); err == nil {
+		log.Printf("Current PATH: %s", string(output))
+	}
+
+	// Try to find kubectl-oadp binary
+	cmd = exec.Command("which", "kubectl-oadp")
+	if output, err := cmd.CombinedOutput(); err == nil {
+		log.Printf("kubectl-oadp found at: %s", string(output))
+	} else {
+		log.Printf("kubectl-oadp not found in PATH: %v", err)
+	}
+
+	// List kubectl plugins
+	cmd = exec.Command("kubectl", "plugin", "list")
+	if output, err := cmd.CombinedOutput(); err == nil {
+		log.Printf("Available kubectl plugins: %s", string(output))
+	}
+
+	return runCommand("kubectl", []string{"oadp", "version"}, "")
+}
+
+func (c *CLISetup) configureNamespace() error {
+	return runCommand("kubectl", []string{"oadp", "client", "config", "set", fmt.Sprintf("namespace=%s", c.namespace)}, "")
+}
+
+func runCommand(name string, args []string, dir string) error {
+	cmd := exec.Command(name, args...)
+	if dir != "" {
+		cmd.Dir = dir
+	}
+
+	output, err := cmd.CombinedOutput()
+	if err != nil {
+		return fmt.Errorf("command '%s %s' failed: %v\nOutput: %s",
+			name, strings.Join(args, " "), err, string(output))
+	}
+	return nil
+}
diff --git a/tests/e2e/lib/dpa_helpers.go b/tests/e2e/lib/dpa_helpers.go
index 04f83dff006..28651ae287f 100755
--- a/tests/e2e/lib/dpa_helpers.go
+++ b/tests/e2e/lib/dpa_helpers.go
@@ -38,10 +38,11 @@ import (
 type BackupRestoreType string
 
 const (
-	CSI          BackupRestoreType = "csi"
-	CSIDataMover BackupRestoreType = "csi-datamover"
-	RESTIC       BackupRestoreType = "restic"
-	KOPIA        BackupRestoreType = "kopia"
+	CSI             BackupRestoreType = "csi"
+	CSIDataMover    BackupRestoreType = "csi-datamover"
+	RESTIC          BackupRestoreType = "restic"
+	KOPIA           BackupRestoreType = "kopia"
+	NativeSnapshots BackupRestoreType = "native-snapshots"
 )
 
 type DpaCustomResource struct {
diff --git a/tests/e2e/lib/flakes.go b/tests/e2e/lib/flakes.go
index fe3e3a10442..b0d3836d7e5 100644
--- a/tests/e2e/lib/flakes.go
+++ b/tests/e2e/lib/flakes.go
@@ -3,6 +3,7 @@ package lib
 import (
 	"log"
 	"regexp"
+	"strings"
 )
 
 var errorIgnorePatterns = []string{
@@ -19,6 +20,7 @@ var errorIgnorePatterns = []string{
 	// Ignore managed fields errors per https://github.com/vmware-tanzu/velero/pull/6110 and avoid e2e failure.
 	// https://prow.ci.openshift.org/view/gs/origin-ci-test/pr-logs/pull/openshift_oadp-operator/1126/pull-ci-openshift-oadp-operator-master-4.10-operator-e2e-aws/1690109468546699264#1:build-log.txt%3A686
 	"level=error msg=\"error patch for managed fields ",
+	"VolumeSnapshot has a temporary error Failed to create snapshot: error updating status for volume snapshot content snapcontent-",
 }
 
 type FlakePattern struct {
@@ -27,35 +29,47 @@ type FlakePattern struct {
 	StringSearchPattern string
 }
 
-// CheckIfFlakeOccured checks for known flake patterns in the provided input string (typically log from the test ran).
-// It updates the value pointed to by knownFlake based on whether a known flake pattern is found.
-//
-// Parameters:
-//
-//	input (string):     The input string to be examined for known flake patterns.
-//	knownFlake (*bool): A pointer to a boolean variable that will be updated based on whether a known flake pattern is found in the input.
-func CheckIfFlakeOccured(input string, knownFlake *bool) {
-	flakePatterns := []FlakePattern{
-		{
-			Issue:               "https://github.com/kubernetes-csi/external-snapshotter/pull/876",
-			Description:         "Race condition in the VolumeSnapshotBeingCreated",
-			StringSearchPattern: "Failed to check and update snapshot content: failed to remove VolumeSnapshotBeingCreated annotation on the content snapcontent-",
-		},
-		{
-			Issue:               "https://github.com/vmware-tanzu/velero/issues/5856",
-			Description:         "Transient S3 bucket errors and limits",
-			StringSearchPattern: "Error copying image: writing blob: uploading layer chunked: received unexpected HTTP status: 500 Internal Server Error",
-		},
-	}
+var flakePatterns = []FlakePattern{
+	{
+		Issue:               "https://github.com/kubernetes-csi/external-snapshotter/pull/876",
+		Description:         "Race condition in the VolumeSnapshotBeingCreated",
+		StringSearchPattern: "Failed to check and update snapshot content: failed to remove VolumeSnapshotBeingCreated annotation on the content snapcontent-",
+	},
+	{
+		Issue:               "https://github.com/vmware-tanzu/velero/issues/5856",
+		Description:         "Transient S3 bucket errors and limits",
+		StringSearchPattern: "Error copying image: writing blob: uploading layer chunked: received unexpected HTTP status: 500 Internal Server Error",
+	},
+}
 
+func checkFlakePatterns(input string) bool {
 	for _, pattern := range flakePatterns {
 		re := regexp.MustCompile(pattern.StringSearchPattern)
 		if re.MatchString(input) {
 			log.Printf("FLAKE DETECTION: Match found for issue %s: %s\n", pattern.Issue, pattern.Description)
-			*knownFlake = true
-			return
+			return true
 		}
 	}
 	log.Println("FLAKE DETECTION: No known flakes found.")
-	*knownFlake = false
+	return false
+}
+
+// CheckIfFlakeOccurred checks for known flake patterns in the provided logs (typically logs from the test ran).
+//
+// Parameters:
+//
+//	logs ([]string): Logs to be examined for known flake patterns.
+func CheckIfFlakeOccurred(logs []string) bool {
+	return checkFlakePatterns(strings.Join(logs, "\n"))
+}
+
+// CheckIfFlakeOccured checks for known flake patterns in the provided input string (typically log from the test ran).
+// It updates the value pointed to by knownFlake based on whether a known flake pattern is found.
+//
+// Parameters:
+//
+//	input (string):     The input string to be examined for known flake patterns.
+//	knownFlake (*bool): A pointer to a boolean variable that will be updated based on whether a known flake pattern is found in the input.
+func CheckIfFlakeOccured(input string, knownFlake *bool) {
+	*knownFlake = checkFlakePatterns(input)
 }
diff --git a/tests/e2e/lib/kube_helpers.go b/tests/e2e/lib/kube_helpers.go
index 28ffa958634..c36a91ae060 100755
--- a/tests/e2e/lib/kube_helpers.go
+++ b/tests/e2e/lib/kube_helpers.go
@@ -134,3 +134,11 @@ func GetPodContainerLogs(clientset *kubernetes.Clientset, namespace, podname, co
 	}
 	return buf.String(), nil
 }
+
+func DeleteNamespace(clientset *kubernetes.Clientset, namespace string) error {
+	err := clientset.CoreV1().Namespaces().Delete(context.Background(), namespace, metav1.DeleteOptions{})
+	if apierrors.IsNotFound(err) {
+		return nil
+	}
+	return err
+}
diff --git a/tests/e2e/lib/restore.go b/tests/e2e/lib/restore.go
index 5000374d5bd..791fea585c3 100755
--- a/tests/e2e/lib/restore.go
+++ b/tests/e2e/lib/restore.go
@@ -74,3 +74,25 @@ func IsRestoreCompletedSuccessfully(c *kubernetes.Clientset, ocClient client.Cli
 		GetVeleroContainerFailureLogs(c, veleroNamespace),
 	)
 }
+
+// restorePhasesNotDone is the shared list of restore phases that indicate a restore is still in progress.
+var restorePhasesNotDone = []velero.RestorePhase{
+	velero.RestorePhaseNew,
+	"ReadyToStart",
+	velero.RestorePhaseInProgress,
+	velero.RestorePhaseWaitingForPluginOperations,
+	velero.RestorePhaseWaitingForPluginOperationsPartiallyFailed,
+	"Finalizing",
+	"FinalizingPartiallyFailed",
+	"",
+}
+
+// IsRestorePhaseNotDone returns true if the given phase string represents a restore still in progress.
+func IsRestorePhaseNotDone(phase string) bool {
+	for _, notDonePhase := range restorePhasesNotDone {
+		if phase == string(notDonePhase) {
+			return true
+		}
+	}
+	return false
+}
diff --git a/tests/e2e/lib/restore_cli.go b/tests/e2e/lib/restore_cli.go
new file mode 100644
index 00000000000..0439a3db3bd
--- /dev/null
+++ b/tests/e2e/lib/restore_cli.go
@@ -0,0 +1,283 @@
+package lib
+
+import (
+	"encoding/json"
+	"fmt"
+	"log"
+	"strings"
+	"time"
+
+	velero "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+// CreateRestoreFromBackupViaCLI creates a restore using the OADP CLI
+func CreateRestoreFromBackupViaCLI(backupName, restoreName string) error {
+	options := []string{"--from-backup", backupName}
+
+	// Execute CLI command
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "create",
+		Name:     restoreName,
+		Options:  options,
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to create restore via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Restore created via CLI: %s", restoreName)
+	return nil
+}
+
+// CreateRestoreFromBackupWithOptionsViaCLI creates a restore with options using the OADP CLI
+func CreateRestoreFromBackupWithOptionsViaCLI(backupName, restoreName string, restoreVolumes bool, includeNamespaces, excludeNamespaces []string) error {
+	options := []string{"--from-backup", backupName}
+
+	// Add restore volumes option
+	if restoreVolumes {
+		options = append(options, "--restore-volumes")
+	}
+
+	// Add included namespaces
+	if len(includeNamespaces) > 0 {
+		options = append(options, "--include-namespaces", strings.Join(includeNamespaces, ","))
+	}
+
+	// Add excluded namespaces
+	if len(excludeNamespaces) > 0 {
+		options = append(options, "--exclude-namespaces", strings.Join(excludeNamespaces, ","))
+	}
+
+	// Execute CLI command
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "create",
+		Name:     restoreName,
+		Options:  options,
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to create restore with options via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Restore with options created via CLI: %s", restoreName)
+	return nil
+}
+
+// GetRestoreViaCLI gets restore details using the OADP CLI
+func GetRestoreViaCLI(name string) (*velero.Restore, error) {
+	if name == "" {
+		return nil, fmt.Errorf("restore name cannot be empty")
+	}
+
+	// Use CLI to get restore details in JSON format
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "get",
+		Name:     name,
+		Options:  []string{"-o", "json"},
+	}
+	output, err := cmd.ExecuteOutput()
+	if err != nil {
+		return nil, fmt.Errorf("failed to get restore via CLI: %v", err)
+	}
+
+	// Parse the JSON output back to velero.Restore struct
+	var restore velero.Restore
+	if err := json.Unmarshal(output, &restore); err != nil {
+		return nil, fmt.Errorf("failed to parse restore JSON: %v", err)
+	}
+
+	return &restore, nil
+}
+
+// IsRestoreDoneViaCLI checks if restore is done using the OADP CLI
+func IsRestoreDoneViaCLI(name string) wait.ConditionFunc {
+	return func() (bool, error) {
+		cmd := &CLICommand{
+			Resource: "restore",
+			Action:   "get",
+			Name:     name,
+			Options:  []string{"-o", "yaml"},
+		}
+		output, err := cmd.ExecuteOutput()
+		if err != nil {
+			return false, fmt.Errorf("failed to get restore status via CLI: %v", err)
+		}
+
+		phase := ParsePhaseFromYAML(string(output))
+
+		if len(phase) > 0 {
+			log.Printf("restore phase: %s", phase)
+		}
+
+		return !IsRestorePhaseNotDone(phase), nil
+	}
+}
+
+// IsRestoreCompletedSuccessfullyViaCLI checks if restore completed successfully using the OADP CLI
+func IsRestoreCompletedSuccessfullyViaCLI(name string) (bool, error) {
+	// Use CLI to get restore status
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "get",
+		Name:     name,
+		Options:  []string{"-o", "yaml"},
+	}
+	output, err := cmd.ExecuteOutput()
+	if err != nil {
+		return false, fmt.Errorf("failed to get restore status via CLI: %v", err)
+	}
+
+	// Parse phase from YAML output
+	phase := ParsePhaseFromYAML(string(output))
+
+	if phase == string(velero.RestorePhaseCompleted) {
+		return true, nil
+	}
+
+	// Get additional failure information using CLI
+	restoreLogs, logsErr := RestoreLogsViaCLI(name)
+	if logsErr != nil {
+		restoreLogs = fmt.Sprintf("Failed to get logs: %v", logsErr)
+	}
+
+	return false, fmt.Errorf(
+		"restore phase is: %s; expected: %s\nvelero failure logs: %v",
+		phase, string(velero.RestorePhaseCompleted), restoreLogs,
+	)
+}
+
+// DescribeRestoreViaCLI describes restore using the OADP CLI with default timeout and retry.
+// The timeout prevents the command from hanging when retrieving restore details from object storage.
+func DescribeRestoreViaCLI(name string) string {
+	return DescribeRestoreViaCLIWithOptions(name, DefaultCLITimeout, DefaultCLIRetries, DefaultCLIRetryDelay)
+}
+
+// DescribeRestoreViaCLIWithOptions describes restore using the OADP CLI with specified timeout and retry options.
+func DescribeRestoreViaCLIWithOptions(name string, timeout time.Duration, maxRetries int, retryDelay time.Duration) string {
+	// Use CLI to describe restore
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "describe",
+		Name:     name,
+		Options:  []string{"--details"},
+	}
+
+	var output []byte
+	var err error
+
+	if maxRetries > 1 {
+		output, err = cmd.ExecuteWithTimeoutAndRetry(timeout, maxRetries, retryDelay)
+	} else {
+		output, err = cmd.ExecuteWithTimeout(timeout)
+	}
+
+	if err != nil {
+		return fmt.Sprintf("could not describe restore via CLI: %v, output: %s", err, string(output))
+	}
+
+	return string(output)
+}
+
+// RestoreLogsViaCLI gets restore logs using the OADP CLI with default timeout and retry.
+// The timeout prevents the command from hanging indefinitely when streaming logs from object storage.
+// Retry logic helps handle transient network issues.
+func RestoreLogsViaCLI(name string) (restoreLogs string, err error) {
+	return RestoreLogsViaCLIWithOptions(name, DefaultCLITimeout, DefaultCLIRetries, DefaultCLIRetryDelay)
+}
+
+// RestoreLogsViaCLIWithTimeout gets restore logs using the OADP CLI with a specified timeout (no retry).
+func RestoreLogsViaCLIWithTimeout(name string, timeout time.Duration) (restoreLogs string, err error) {
+	return RestoreLogsViaCLIWithOptions(name, timeout, 1, 0)
+}
+
+// RestoreLogsViaCLIWithOptions gets restore logs using the OADP CLI with specified timeout and retry options.
+func RestoreLogsViaCLIWithOptions(name string, timeout time.Duration, maxRetries int, retryDelay time.Duration) (restoreLogs string, err error) {
+	if name == "" {
+		return "", fmt.Errorf("restore name cannot be empty")
+	}
+
+	// Use CLI to get restore logs with timeout and retry to prevent hanging
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "logs",
+		Name:     name,
+		Options:  []string{},
+	}
+
+	var output []byte
+	var cmdErr error
+
+	if maxRetries > 1 {
+		output, cmdErr = cmd.ExecuteOutputWithTimeoutAndRetry(timeout, maxRetries, retryDelay)
+	} else {
+		output, cmdErr = cmd.ExecuteOutputWithTimeout(timeout)
+	}
+
+	if cmdErr != nil {
+		return "", fmt.Errorf("failed to get restore logs via CLI: %v", cmdErr)
+	}
+
+	return string(output), nil
+}
+
+// RestoreErrorLogsViaCLI gets restore error logs using the OADP CLI
+func RestoreErrorLogsViaCLI(name string) []string {
+	rl, err := RestoreLogsViaCLI(name)
+	if err != nil {
+		return []string{err.Error()}
+	}
+	return errorLogsExcludingIgnored(rl)
+}
+
+// DeleteRestoreViaCLI deletes a restore using the OADP CLI
+func DeleteRestoreViaCLI(name string) error {
+	if name == "" {
+		return fmt.Errorf("restore name cannot be empty")
+	}
+
+	// Use CLI to delete restore
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "delete",
+		Name:     name,
+		Options:  []string{},
+	}
+	output, err := cmd.Execute()
+	if err != nil {
+		return fmt.Errorf("failed to delete restore via CLI: %v, output: %s", err, string(output))
+	}
+
+	log.Printf("Restore deleted via CLI: %s", name)
+	return nil
+}
+
+// ListRestoresViaCLI lists all restores using the OADP CLI
+func ListRestoresViaCLI() ([]string, error) {
+	// Use CLI to list restores
+	cmd := &CLICommand{
+		Resource: "restore",
+		Action:   "list",
+		Name:     "", // list commands don't require a name
+		Options:  []string{"-o", "name"},
+	}
+	output, err := cmd.ExecuteOutput()
+	if err != nil {
+		return nil, fmt.Errorf("failed to list restores via CLI: %v", err)
+	}
+
+	lines := strings.Split(strings.TrimSpace(string(output)), "\n")
+	var restores []string
+	for _, line := range lines {
+		if line != "" {
+			// Remove "restore.velero.io/" prefix if present
+			restore := strings.TrimPrefix(line, "restore.velero.io/")
+			restores = append(restores, restore)
+		}
+	}
+
+	return restores, nil
+}