/*
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 controller

import (
	"bytes"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"reflect"
	"time"

	snapshotv1api "github.com/kubernetes-csi/external-snapshotter/client/v8/apis/volumesnapshot/v1"

	"context"

	"github.com/sirupsen/logrus"
	"github.com/stretchr/testify/assert"

	"strings"
	"testing"

	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	corev1api "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"k8s.io/apimachinery/pkg/util/sets"
	ctrl "sigs.k8s.io/controller-runtime"
	"sigs.k8s.io/controller-runtime/pkg/client"

	"github.com/vmware-tanzu/velero/internal/volume"
	velerov1api "github.com/vmware-tanzu/velero/pkg/apis/velero/v1"
	pkgbackup "github.com/vmware-tanzu/velero/pkg/backup"
	"github.com/vmware-tanzu/velero/pkg/builder"
	"github.com/vmware-tanzu/velero/pkg/metrics"
	persistencemocks "github.com/vmware-tanzu/velero/pkg/persistence/mocks"
	"github.com/vmware-tanzu/velero/pkg/plugin/clientmgmt"
	pluginmocks "github.com/vmware-tanzu/velero/pkg/plugin/mocks"
	"github.com/vmware-tanzu/velero/pkg/plugin/velero"
	"github.com/vmware-tanzu/velero/pkg/plugin/velero/mocks"
	"github.com/vmware-tanzu/velero/pkg/repository"
	repomanager "github.com/vmware-tanzu/velero/pkg/repository/manager"
	repomocks "github.com/vmware-tanzu/velero/pkg/repository/mocks"
	repotypes "github.com/vmware-tanzu/velero/pkg/repository/types"
	velerotest "github.com/vmware-tanzu/velero/pkg/test"
)

type backupDeletionControllerTestData struct {
	fakeClient        client.Client
	volumeSnapshotter *velerotest.FakeVolumeSnapshotter
	backupStore       *persistencemocks.BackupStore
	controller        *backupDeletionReconciler
	req               ctrl.Request
}

func defaultTestDbr() *velerov1api.DeleteBackupRequest {
	req := pkgbackup.NewDeleteBackupRequest("foo", "uid")
	req.Namespace = velerov1api.DefaultNamespace
	req.Name = "foo-abcde"
	return req
}

func setupBackupDeletionControllerTest(t *testing.T, req *velerov1api.DeleteBackupRequest, objects ...runtime.Object) *backupDeletionControllerTestData {
	t.Helper()
	var (
		fakeClient        = velerotest.NewFakeControllerRuntimeClient(t, append(objects, req)...)
		volumeSnapshotter = &velerotest.FakeVolumeSnapshotter{SnapshotsTaken: sets.NewString()}
		pluginManager     = &pluginmocks.Manager{}
		backupStore       = &persistencemocks.BackupStore{}
	)

	data := &backupDeletionControllerTestData{
		fakeClient:        fakeClient,
		volumeSnapshotter: volumeSnapshotter,
		backupStore:       backupStore,
		controller: NewBackupDeletionReconciler(
			velerotest.NewLogger(),
			fakeClient,
			NewBackupTracker(),
			nil, // repository manager
			metrics.NewServerMetrics(),
			nil, // discovery helper
			func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
			NewFakeSingleObjectBackupStoreGetter(backupStore),
			velerotest.NewFakeCredentialsFileStore("", nil),
			nil,
		),
		req: ctrl.Request{NamespacedName: types.NamespacedName{Namespace: req.Namespace, Name: req.Name}},
	}

	pluginManager.On("CleanupClients").Return(nil)
	return data
}

func TestBackupDeletionControllerReconcile(t *testing.T) {
	t.Run("failed to get backup store", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()
		location := &velerov1api.BackupStorageLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      backup.Spec.StorageLocation,
			},
			Spec: velerov1api.BackupStorageLocationSpec{
				Provider: "objStoreProvider",
				StorageType: velerov1api.StorageType{
					ObjectStorage: &velerov1api.ObjectStorageLocation{
						Bucket: "bucket",
					},
				},
			},
			Status: velerov1api.BackupStorageLocationStatus{
				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
			},
		}
		dbr := defaultTestDbr()
		td := setupBackupDeletionControllerTest(t, dbr, location, backup)
		td.controller.backupStoreGetter = &fakeErrorBackupStoreGetter{}
		_, err := td.controller.Reconcile(ctx, td.req)
		require.NoError(t, err)
		res := &velerov1api.DeleteBackupRequest{}
		td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.True(t, strings.HasPrefix(res.Status.Errors[0], "error getting the backup store"))
	})

	t.Run("missing spec.backupName", func(t *testing.T) {
		dbr := defaultTestDbr()
		dbr.Spec.BackupName = ""
		td := setupBackupDeletionControllerTest(t, dbr)

		_, err := td.controller.Reconcile(ctx, td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "spec.backupName is required", res.Status.Errors[0])
	})

	t.Run("existing deletion requests for the backup are deleted", func(t *testing.T) {
		input := defaultTestDbr()
		td := setupBackupDeletionControllerTest(t, input)

		// add the backup to the tracker so the execution of reconcile doesn't progress
		// past checking for an in-progress backup. this makes validation easier.
		td.controller.backupTracker.Add(td.req.Namespace, input.Spec.BackupName)
		existing := &velerov1api.DeleteBackupRequest{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: td.req.Namespace,
				Name:      "bar",
				Labels: map[string]string{
					velerov1api.BackupNameLabel: input.Spec.BackupName,
				},
			},
			Spec: velerov1api.DeleteBackupRequestSpec{
				BackupName: input.Spec.BackupName,
			},
		}
		err := td.fakeClient.Create(t.Context(), existing)
		require.NoError(t, err)
		existing2 :=
			&velerov1api.DeleteBackupRequest{
				ObjectMeta: metav1.ObjectMeta{
					Namespace: td.req.Namespace,
					Name:      "bar-2",
					Labels: map[string]string{
						velerov1api.BackupNameLabel: "some-other-backup",
					},
				},
				Spec: velerov1api.DeleteBackupRequestSpec{
					BackupName: "some-other-backup",
				},
			}
		err = td.fakeClient.Create(t.Context(), existing2)
		require.NoError(t, err)
		_, err = td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)
		// verify "existing" is deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: existing.Namespace,
			Name:      existing.Name,
		}, &velerov1api.DeleteBackupRequest{})

		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		// verify "existing2" remains
		assert.NoError(t, td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: existing2.Namespace,
			Name:      existing2.Name,
		}, &velerov1api.DeleteBackupRequest{}))
	})
	t.Run("deleting an in progress backup isn't allowed", func(t *testing.T) {
		dbr := defaultTestDbr()
		td := setupBackupDeletionControllerTest(t, dbr)

		td.controller.backupTracker.Add(td.req.Namespace, dbr.Spec.BackupName)
		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "backup is still in progress", res.Status.Errors[0])
	})

	t.Run("unable to find backup", func(t *testing.T) {
		td := setupBackupDeletionControllerTest(t, defaultTestDbr())

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "backup not found", res.Status.Errors[0])
	})
	t.Run("unable to find backup storage location", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()

		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), backup)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "backup storage location default not found", res.Status.Errors[0])
	})

	t.Run("backup storage location is in read-only mode", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()
		location := builder.ForBackupStorageLocation("velero", "default").Phase(velerov1api.BackupStorageLocationPhaseAvailable).AccessMode(velerov1api.BackupStorageLocationAccessModeReadOnly).Result()

		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), location, backup)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "cannot delete backup because backup storage location default is currently in read-only mode", res.Status.Errors[0])
	})

	t.Run("backup storage location is in unavailable state", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").StorageLocation("default").Result()
		location := builder.ForBackupStorageLocation("velero", "default").Phase(velerov1api.BackupStorageLocationPhaseUnavailable).Result()

		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), location, backup)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "cannot delete backup because backup storage location default is currently in Unavailable state", res.Status.Errors[0])
	})

	t.Run("full delete, no errors", func(t *testing.T) {
		input := defaultTestDbr()

		// Clear out resource labels to make sure the controller adds them and does not
		// panic when encountering a nil Labels map
		// (https://github.com/vmware-tanzu/velero/issues/1546)
		input.Labels = nil

		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").Result()
		backup.UID = "uid"
		backup.Spec.StorageLocation = "primary"

		restore1 := builder.ForRestore(velerov1api.DefaultNamespace, "restore-1").Phase(velerov1api.RestorePhaseCompleted).Backup("foo").Result()
		restore2 := builder.ForRestore(velerov1api.DefaultNamespace, "restore-2").Phase(velerov1api.RestorePhaseCompleted).Backup("foo").Result()
		restore3 := builder.ForRestore(velerov1api.DefaultNamespace, "restore-3").Phase(velerov1api.RestorePhaseCompleted).Backup("some-other-backup").Result()

		location := &velerov1api.BackupStorageLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      backup.Spec.StorageLocation,
			},
			Spec: velerov1api.BackupStorageLocationSpec{
				Provider: "objStoreProvider",
				StorageType: velerov1api.StorageType{
					ObjectStorage: &velerov1api.ObjectStorageLocation{
						Bucket: "bucket",
					},
				},
			},
			Status: velerov1api.BackupStorageLocationStatus{
				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
			},
		}

		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      "vsl-1",
			},
			Spec: velerov1api.VolumeSnapshotLocationSpec{
				Provider: "provider-1",
			},
		}
		td := setupBackupDeletionControllerTest(t, input, backup, restore1, restore2, restore3, location, snapshotLocation)

		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")

		snapshots := []*volume.Snapshot{
			{
				Spec: volume.SnapshotSpec{
					Location: "vsl-1",
				},
				Status: volume.SnapshotStatus{
					ProviderSnapshotID: "snap-1",
				},
			},
		}

		pluginManager := &pluginmocks.Manager{}
		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
		pluginManager.On("GetDeleteItemActions").Return(nil, nil)
		pluginManager.On("CleanupClients")
		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }

		td.backupStore.On("GetBackupVolumeSnapshots", input.Spec.BackupName).Return(snapshots, nil)
		td.backupStore.On("GetBackupContents", input.Spec.BackupName).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
		td.backupStore.On("DeleteBackup", input.Spec.BackupName).Return(nil)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		// the dbr should be deleted
		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)
		if err == nil {
			t.Logf("status of the dbr: %s, errors in dbr: %v", res.Status.Phase, res.Status.Errors)
		}

		// backup CR, restore CR restore-1 and restore-2 should be deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      backup.Name,
		}, &velerov1api.Backup{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-1",
		}, &velerov1api.Restore{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-2",
		}, &velerov1api.Restore{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		// restore-3 should remain
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-3",
		}, &velerov1api.Restore{})
		require.NoError(t, err)

		td.backupStore.AssertCalled(t, "DeleteBackup", input.Spec.BackupName)

		// Make sure snapshot was deleted
		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
	})
	t.Run("full delete, no errors, with backup name greater than 63 chars", func(t *testing.T) {
		backup := defaultBackup().
			ObjectMeta(
				builder.WithName("the-really-long-backup-name-that-is-much-more-than-63-characters"),
			).
			Result()
		backup.UID = "uid"
		backup.Spec.StorageLocation = "primary"

		restore1 := builder.ForRestore(backup.Namespace, "restore-1").
			Phase(velerov1api.RestorePhaseCompleted).
			Backup(backup.Name).
			Result()
		restore2 := builder.ForRestore(backup.Namespace, "restore-2").
			Phase(velerov1api.RestorePhaseCompleted).
			Backup(backup.Name).
			Result()
		restore3 := builder.ForRestore(backup.Namespace, "restore-3").
			Phase(velerov1api.RestorePhaseCompleted).
			Backup("some-other-backup").
			Result()

		dbr := pkgbackup.NewDeleteBackupRequest(backup.Name, "uid")
		dbr.Namespace = velerov1api.DefaultNamespace
		dbr.Name = "foo-abcde"
		// Clear out resource labels to make sure the controller adds them
		dbr.Labels = nil

		location := &velerov1api.BackupStorageLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      backup.Spec.StorageLocation,
			},
			Spec: velerov1api.BackupStorageLocationSpec{
				Provider: "objStoreProvider",
				StorageType: velerov1api.StorageType{
					ObjectStorage: &velerov1api.ObjectStorageLocation{
						Bucket: "bucket",
					},
				},
			},
			Status: velerov1api.BackupStorageLocationStatus{
				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
			},
		}

		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      "vsl-1",
			},
			Spec: velerov1api.VolumeSnapshotLocationSpec{
				Provider: "provider-1",
			},
		}
		td := setupBackupDeletionControllerTest(t, dbr, backup, restore1, restore2, restore3, location, snapshotLocation)

		snapshots := []*volume.Snapshot{
			{
				Spec: volume.SnapshotSpec{
					Location: "vsl-1",
				},
				Status: volume.SnapshotStatus{
					ProviderSnapshotID: "snap-1",
				},
			},
		}

		pluginManager := &pluginmocks.Manager{}
		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
		pluginManager.On("GetDeleteItemActions").Return(nil, nil)
		pluginManager.On("CleanupClients")
		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }

		td.backupStore.On("GetBackupVolumeSnapshots", dbr.Spec.BackupName).Return(snapshots, nil)
		td.backupStore.On("GetBackupContents", dbr.Spec.BackupName).Return(io.NopCloser(bytes.NewReader([]byte("hello world"))), nil)
		td.backupStore.On("DeleteBackup", dbr.Spec.BackupName).Return(nil)
		td.backupStore.On("DeleteRestore", "restore-1").Return(nil)
		td.backupStore.On("DeleteRestore", "restore-2").Return(nil)

		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		// the dbr should be deleted
		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)
		if err == nil {
			t.Logf("status of the dbr: %s, errors in dbr: %v", res.Status.Phase, res.Status.Errors)
		}

		// backup CR, restore CR restore-1 and restore-2 should be deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      backup.Name,
		}, &velerov1api.Backup{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-1",
		}, &velerov1api.Restore{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-2",
		}, &velerov1api.Restore{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		// restore-3 should remain
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      "restore-3",
		}, &velerov1api.Restore{})
		require.NoError(t, err)

		// Make sure snapshot was deleted
		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
	})
	t.Run("backup is not downloaded when there are no DeleteItemAction plugins", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").Result()
		backup.UID = "uid"
		backup.Spec.StorageLocation = "primary"

		input := defaultTestDbr()
		// Clear out resource labels to make sure the controller adds them and does not
		// panic when encountering a nil Labels map
		// (https://github.com/vmware-tanzu/velero/issues/1546)
		input.Labels = nil

		location := &velerov1api.BackupStorageLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      backup.Spec.StorageLocation,
			},
			Spec: velerov1api.BackupStorageLocationSpec{
				Provider: "objStoreProvider",
				StorageType: velerov1api.StorageType{
					ObjectStorage: &velerov1api.ObjectStorageLocation{
						Bucket: "bucket",
					},
				},
			},
			Status: velerov1api.BackupStorageLocationStatus{
				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
			},
		}

		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      "vsl-1",
			},
			Spec: velerov1api.VolumeSnapshotLocationSpec{
				Provider: "provider-1",
			},
		}
		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), backup, location, snapshotLocation)
		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")

		snapshots := []*volume.Snapshot{
			{
				Spec: volume.SnapshotSpec{
					Location: "vsl-1",
				},
				Status: volume.SnapshotStatus{
					ProviderSnapshotID: "snap-1",
				},
			},
		}

		pluginManager := &pluginmocks.Manager{}
		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
		pluginManager.On("GetDeleteItemActions").Return([]velero.DeleteItemAction{}, nil)
		pluginManager.On("CleanupClients")
		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }

		td.backupStore.On("GetBackupVolumeSnapshots", input.Spec.BackupName).Return(snapshots, nil)
		td.backupStore.On("DeleteBackup", input.Spec.BackupName).Return(nil)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		td.backupStore.AssertNotCalled(t, "GetBackupContents", mock.Anything)
		td.backupStore.AssertCalled(t, "DeleteBackup", input.Spec.BackupName)

		// the dbr should be deleted
		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)
		if err == nil {
			t.Logf("status of the dbr: %s, errors in dbr: %v", res.Status.Phase, res.Status.Errors)
		}

		// backup CR should be deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      backup.Name,
		}, &velerov1api.Backup{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		// Make sure snapshot was deleted
		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
	})
	t.Run("backup is still deleted if downloading tarball fails for DeleteItemAction plugins", func(t *testing.T) {
		backup := builder.ForBackup(velerov1api.DefaultNamespace, "foo").Result()
		backup.UID = "uid"
		backup.Spec.StorageLocation = "primary"

		input := defaultTestDbr()
		// Clear out resource labels to make sure the controller adds them and does not
		// panic when encountering a nil Labels map
		// (https://github.com/vmware-tanzu/velero/issues/1546)
		input.Labels = nil

		location := &velerov1api.BackupStorageLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      backup.Spec.StorageLocation,
			},
			Spec: velerov1api.BackupStorageLocationSpec{
				Provider: "objStoreProvider",
				StorageType: velerov1api.StorageType{
					ObjectStorage: &velerov1api.ObjectStorageLocation{
						Bucket: "bucket",
					},
				},
			},
			Status: velerov1api.BackupStorageLocationStatus{
				Phase: velerov1api.BackupStorageLocationPhaseAvailable,
			},
		}

		snapshotLocation := &velerov1api.VolumeSnapshotLocation{
			ObjectMeta: metav1.ObjectMeta{
				Namespace: backup.Namespace,
				Name:      "vsl-1",
			},
			Spec: velerov1api.VolumeSnapshotLocationSpec{
				Provider: "provider-1",
			},
		}

		csiSnapshot := builder.ForVolumeSnapshot("user-ns", "vs-1").ObjectMeta(
			builder.WithLabelsMap(map[string]string{
				"velero.io/backup-name": "foo",
			})).SourcePVC("some-pvc").Result()

		td := setupBackupDeletionControllerTest(t, defaultTestDbr(), backup, location, snapshotLocation, csiSnapshot)
		td.volumeSnapshotter.SnapshotsTaken.Insert("snap-1")

		snapshots := []*volume.Snapshot{
			{
				Spec: volume.SnapshotSpec{
					Location: "vsl-1",
				},
				Status: volume.SnapshotStatus{
					ProviderSnapshotID: "snap-1",
				},
			},
		}

		pluginManager := &pluginmocks.Manager{}
		pluginManager.On("GetVolumeSnapshotter", "provider-1").Return(td.volumeSnapshotter, nil)
		pluginManager.On("GetDeleteItemActions").Return([]velero.DeleteItemAction{new(mocks.DeleteItemAction)}, nil)
		pluginManager.On("CleanupClients")
		td.controller.newPluginManager = func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager }

		td.backupStore.On("GetBackupVolumeSnapshots", input.Spec.BackupName).Return(snapshots, nil)
		td.backupStore.On("GetBackupContents", input.Spec.BackupName).Return(nil, fmt.Errorf("error downloading tarball"))
		td.backupStore.On("DeleteBackup", input.Spec.BackupName).Return(nil)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		td.backupStore.AssertCalled(t, "GetBackupContents", input.Spec.BackupName)
		td.backupStore.AssertCalled(t, "DeleteBackup", input.Spec.BackupName)

		// the dbr should be deleted
		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)
		if err == nil {
			t.Logf("status of the dbr: %s, errors in dbr: %v", res.Status.Phase, res.Status.Errors)
		}

		// backup CR should be deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: velerov1api.DefaultNamespace,
			Name:      backup.Name,
		}, &velerov1api.Backup{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)

		// leaked CSI snapshot should be deleted
		err = td.fakeClient.Get(t.Context(), types.NamespacedName{
			Namespace: "user-ns",
			Name:      "vs-1",
		}, &snapshotv1api.VolumeSnapshot{})
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error for the leaked CSI snapshot, but actual value of error: %v", err)

		// Make sure snapshot was deleted
		assert.Equal(t, 0, td.volumeSnapshotter.SnapshotsTaken.Len())
	})
	t.Run("Expired request will be deleted if the status is processed", func(t *testing.T) {
		expired := time.Date(2018, 4, 3, 12, 0, 0, 0, time.UTC)
		input := defaultTestDbr()
		input.CreationTimestamp = metav1.Time{
			Time: expired,
		}
		input.Status.Phase = velerov1api.DeleteBackupRequestPhaseProcessed
		td := setupBackupDeletionControllerTest(t, input)
		td.backupStore.On("DeleteBackup", mock.Anything).Return(nil)
		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		assert.True(t, apierrors.IsNotFound(err), "Expected not found error, but actual value of error: %v", err)
		td.backupStore.AssertNotCalled(t, "DeleteBackup", mock.Anything)
	})

	t.Run("Expired request will not be deleted if the status is not processed", func(t *testing.T) {
		expired := time.Date(2018, 4, 3, 12, 0, 0, 0, time.UTC)
		input := defaultTestDbr()
		input.CreationTimestamp = metav1.Time{
			Time: expired,
		}
		input.Status.Phase = velerov1api.DeleteBackupRequestPhaseNew
		td := setupBackupDeletionControllerTest(t, input)
		td.backupStore.On("DeleteBackup", mock.Anything).Return(nil)

		_, err := td.controller.Reconcile(t.Context(), td.req)
		require.NoError(t, err)

		res := &velerov1api.DeleteBackupRequest{}
		err = td.fakeClient.Get(ctx, td.req.NamespacedName, res)
		require.NoError(t, err)
		assert.Equal(t, "Processed", string(res.Status.Phase))
		assert.Len(t, res.Status.Errors, 1)
		assert.Equal(t, "backup not found", res.Status.Errors[0])
	})
}

func TestGetSnapshotsInBackup(t *testing.T) {
	tests := []struct {
		name                  string
		podVolumeBackups      []velerov1api.PodVolumeBackup
		expected              map[string][]repotypes.SnapshotIdentifier
		longBackupNameEnabled bool
	}{
		{
			name:             "no pod volume backups",
			podVolumeBackups: nil,
			expected:         map[string][]repotypes.SnapshotIdentifier{},
		},
		{
			name: "no pod volume backups with matching label",
			podVolumeBackups: []velerov1api.PodVolumeBackup{
				{
					ObjectMeta: metav1.ObjectMeta{Name: "foo", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-1"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "bar", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-2"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-2", Namespace: "ns-2"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-2"},
				},
			},
			expected: map[string][]repotypes.SnapshotIdentifier{},
		},
		{
			name: "some pod volume backups with matching label",
			podVolumeBackups: []velerov1api.PodVolumeBackup{
				{
					ObjectMeta: metav1.ObjectMeta{Name: "foo", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-1"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "bar", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-2"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-2", Namespace: "ns-2"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-2"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-3"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb-2", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-4"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "incomplete-or-failed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-2"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: ""},
				},
			},
			expected: map[string][]repotypes.SnapshotIdentifier{
				"ns-1": {
					{
						VolumeNamespace: "ns-1",
						SnapshotID:      "snap-3",
						RepositoryType:  "restic",
					},
					{
						VolumeNamespace: "ns-1",
						SnapshotID:      "snap-4",
						RepositoryType:  "restic",
					},
				},
			},
		},
		{
			name:                  "some pod volume backups with matching label and backup name greater than 63 chars",
			longBackupNameEnabled: true,
			podVolumeBackups: []velerov1api.PodVolumeBackup{
				{
					ObjectMeta: metav1.ObjectMeta{Name: "foo", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-1"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "bar", Labels: map[string]string{velerov1api.BackupNameLabel: "non-matching-backup-2"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-2", Namespace: "ns-2"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-2"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "the-really-long-backup-name-that-is-much-more-than-63-cha6ca4bc"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-3"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "completed-pvb-2", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-1"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: "snap-4"},
				},
				{
					ObjectMeta: metav1.ObjectMeta{Name: "incomplete-or-failed-pvb", Labels: map[string]string{velerov1api.BackupNameLabel: "backup-1"}},
					Spec: velerov1api.PodVolumeBackupSpec{
						Pod: corev1api.ObjectReference{Name: "pod-1", Namespace: "ns-2"},
					},
					Status: velerov1api.PodVolumeBackupStatus{SnapshotID: ""},
				},
			},
			expected: map[string][]repotypes.SnapshotIdentifier{
				"ns-1": {
					{
						VolumeNamespace: "ns-1",
						SnapshotID:      "snap-3",
						RepositoryType:  "restic",
					},
				},
			},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			var (
				clientBuilder = velerotest.NewFakeControllerRuntimeClientBuilder(t)
				veleroBackup  = &velerov1api.Backup{}
			)

			veleroBackup.Name = "backup-1"

			if test.longBackupNameEnabled {
				veleroBackup.Name = "the-really-long-backup-name-that-is-much-more-than-63-characters"
			}
			clientBuilder.WithLists(&velerov1api.PodVolumeBackupList{
				Items: test.podVolumeBackups,
			})

			res, err := getSnapshotsInBackup(t.Context(), veleroBackup, clientBuilder.Build())
			require.NoError(t, err)

			assert.True(t, reflect.DeepEqual(res, test.expected))
		})
	}
}

func batchDeleteSucceed(ctx context.Context, repoEnsurer *repository.Ensurer, repoMgr repomanager.Manager, directSnapshots map[string][]repotypes.SnapshotIdentifier, backup *velerov1api.Backup, logger logrus.FieldLogger) []error {
	return nil
}

func batchDeleteFail(ctx context.Context, repoEnsurer *repository.Ensurer, repoMgr repomanager.Manager, directSnapshots map[string][]repotypes.SnapshotIdentifier, backup *velerov1api.Backup, logger logrus.FieldLogger) []error {
	return []error{
		errors.New("fake-delete-1"),
		errors.New("fake-delete-2"),
	}
}

func generateSnapshotData(snapshot *repotypes.SnapshotIdentifier) (map[string]string, error) {
	if snapshot == nil {
		return nil, nil
	}

	b, err := json.Marshal(snapshot)
	if err != nil {
		return nil, err
	}

	data := make(map[string]string)
	if err := json.Unmarshal(b, &data); err != nil {
		return nil, err
	}

	return data, nil
}

func TestDeleteMovedSnapshots(t *testing.T) {
	tests := []struct {
		name               string
		repoMgr            repomanager.Manager
		batchDeleteSucceed bool
		backupName         string
		snapshots          []*repotypes.SnapshotIdentifier
		expected           []string
	}{
		{
			name: "repoMgr is nil",
		},
		{
			name:    "no cm",
			repoMgr: repomocks.NewManager(t),
		},
		{
			name:       "bad cm info",
			repoMgr:    repomocks.NewManager(t),
			backupName: "backup-01",
			snapshots:  []*repotypes.SnapshotIdentifier{nil},
			expected:   []string{"no snapshot info in config"},
		},
		{
			name:       "invalid snapshots",
			repoMgr:    repomocks.NewManager(t),
			backupName: "backup-01",
			snapshots: []*repotypes.SnapshotIdentifier{
				{
					RepositoryType:  "repo-1",
					VolumeNamespace: "ns-1",
				},
				{
					SnapshotID:      "snapshot-1",
					VolumeNamespace: "ns-1",
				},
				{
					SnapshotID:     "snapshot-1",
					RepositoryType: "repo-1",
				},
			},
			batchDeleteSucceed: true,
			expected: []string{
				"invalid snapshot, ID , namespace ns-1, repository repo-1",
				"invalid snapshot, ID snapshot-1, namespace ns-1, repository ",
				"invalid snapshot, ID snapshot-1, namespace , repository repo-1",
			},
		},
		{
			name:       "batch delete succeed",
			repoMgr:    repomocks.NewManager(t),
			backupName: "backup-01",
			snapshots: []*repotypes.SnapshotIdentifier{

				{
					SnapshotID:      "snapshot-1",
					RepositoryType:  "repo-1",
					VolumeNamespace: "ns-1",
				},
			},
			batchDeleteSucceed: true,
			expected:           []string{},
		},
		{
			name:       "batch delete fail",
			repoMgr:    repomocks.NewManager(t),
			backupName: "backup-01",
			snapshots: []*repotypes.SnapshotIdentifier{
				{
					RepositoryType:  "repo-1",
					VolumeNamespace: "ns-1",
				},
				{
					SnapshotID:      "snapshot-1",
					RepositoryType:  "repo-1",
					VolumeNamespace: "ns-1",
				},
			},
			expected: []string{"invalid snapshot, ID , namespace ns-1, repository repo-1", "fake-delete-1", "fake-delete-2"},
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			objs := []runtime.Object{}
			for i, snapshot := range test.snapshots {
				snapshotData, err := generateSnapshotData(snapshot)
				require.NoError(t, err)

				cm := corev1api.ConfigMap{
					TypeMeta: metav1.TypeMeta{
						APIVersion: corev1api.SchemeGroupVersion.String(),
						Kind:       "ConfigMap",
					},
					ObjectMeta: metav1.ObjectMeta{
						Namespace: "velero",
						Name:      fmt.Sprintf("du-info-%d", i),
						Labels: map[string]string{
							velerov1api.BackupNameLabel:             test.backupName,
							velerov1api.DataUploadSnapshotInfoLabel: "true",
						},
					},
					Data: snapshotData,
				}

				objs = append(objs, &cm)
			}

			veleroBackup := &velerov1api.Backup{}
			controller := NewBackupDeletionReconciler(
				velerotest.NewLogger(),
				velerotest.NewFakeControllerRuntimeClient(t, objs...),
				NewBackupTracker(),
				test.repoMgr,
				metrics.NewServerMetrics(),
				nil, // discovery helper
				func(logrus.FieldLogger) clientmgmt.Manager { return pluginManager },
				NewFakeSingleObjectBackupStoreGetter(backupStore),
				velerotest.NewFakeCredentialsFileStore("", nil),
				nil,
			)

			veleroBackup.Name = test.backupName

			if test.batchDeleteSucceed {
				batchDeleteSnapshotFunc = batchDeleteSucceed
			} else {
				batchDeleteSnapshotFunc = batchDeleteFail
			}

			errs := controller.deleteMovedSnapshots(t.Context(), veleroBackup)
			if test.expected == nil {
				assert.Nil(t, errs)
			} else {
				assert.Len(t, errs, len(test.expected))
				for i := range test.expected {
					assert.EqualError(t, errs[i], test.expected[i])
				}
			}
		})
	}
}