diff --git a/pkg/restore/restore_new_test.go b/pkg/restore/restore_new_test.go
index 544abad56..e82dd930c 100644
--- a/pkg/restore/restore_new_test.go
+++ b/pkg/restore/restore_new_test.go
@@ -27,6 +27,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pkg/errors"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -35,6 +36,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
 	"k8s.io/client-go/dynamic"
 	kubetesting "k8s.io/client-go/testing"
@@ -43,6 +45,7 @@ import (
 	"github.com/heptio/velero/pkg/backup"
 	"github.com/heptio/velero/pkg/client"
 	"github.com/heptio/velero/pkg/discovery"
+	"github.com/heptio/velero/pkg/kuberesource"
 	"github.com/heptio/velero/pkg/plugin/velero"
 	"github.com/heptio/velero/pkg/test"
 	"github.com/heptio/velero/pkg/util/encode"
@@ -917,43 +920,7 @@ func TestRestoreItems(t *testing.T) {
 			)
 
 			assertEmptyResults(t, warnings, errs)
-
-			for _, resource := range tc.want {
-				resourceClient := h.DynamicClient.Resource(resource.GVR())
-				for _, item := range resource.Items {
-					var client dynamic.ResourceInterface
-					if item.GetNamespace() != "" {
-						client = resourceClient.Namespace(item.GetNamespace())
-					} else {
-						client = resourceClient
-					}
-
-					res, err := client.Get(item.GetName(), metav1.GetOptions{})
-					if !assert.NoError(t, err) {
-						continue
-					}
-
-					itemJSON, err := json.Marshal(item)
-					if !assert.NoError(t, err) {
-						continue
-					}
-
-					t.Logf("%v", string(itemJSON))
-
-					u := make(map[string]interface{})
-					if !assert.NoError(t, json.Unmarshal(itemJSON, &u)) {
-						continue
-					}
-					want := &unstructured.Unstructured{Object: u}
-
-					// These fields get non-nil zero values in the unstructured objects if they're
-					// empty in the structured objects. Remove them to make comparison easier.
-					unstructured.RemoveNestedField(want.Object, "metadata", "creationTimestamp")
-					unstructured.RemoveNestedField(want.Object, "status")
-
-					assert.Equal(t, want, res)
-				}
-			}
+			assertRestoredItems(t, h, tc.want)
 		})
 	}
 }
@@ -1153,6 +1120,548 @@ func TestRestoreActionsRunForCorrectItems(t *testing.T) {
 	}
 }
 
+// pluggableAction is a restore item action that can be plugged with an Execute
+// function body at runtime.
+type pluggableAction struct {
+	selector    velero.ResourceSelector
+	executeFunc func(*velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error)
+}
+
+func (a *pluggableAction) Execute(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+	if a.executeFunc == nil {
+		return &velero.RestoreItemActionExecuteOutput{
+			UpdatedItem: input.Item,
+		}, nil
+	}
+
+	return a.executeFunc(input)
+}
+
+func (a *pluggableAction) AppliesTo() (velero.ResourceSelector, error) {
+	return a.selector, nil
+}
+
+// TestRestoreActionModifications runs restores with restore item actions that modify resources, and
+// verifies that that the modified item is correctly created in the API. Verification is done by looking
+// at the full object in the API.
+func TestRestoreActionModifications(t *testing.T) {
+	// modifyingActionGetter is a helper function that returns a *pluggableAction, whose Execute(...)
+	// method modifies the item being passed in by calling the 'modify' function on it.
+	modifyingActionGetter := func(modify func(*unstructured.Unstructured)) *pluggableAction {
+		return &pluggableAction{
+			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+				obj, ok := input.Item.(*unstructured.Unstructured)
+				if !ok {
+					return nil, errors.Errorf("unexpected type %T", input.Item)
+				}
+
+				res := obj.DeepCopy()
+				modify(res)
+
+				return &velero.RestoreItemActionExecuteOutput{
+					UpdatedItem: res,
+				}, nil
+			},
+		}
+	}
+
+	tests := []struct {
+		name         string
+		restore      *velerov1api.Restore
+		backup       *velerov1api.Backup
+		apiResources []*test.APIResource
+		tarball      io.Reader
+		actions      []velero.RestoreItemAction
+		want         []*test.APIResource
+	}{
+		{
+			name:         "action that adds a label to item gets restored",
+			restore:      defaultRestore().Restore(),
+			backup:       defaultBackup().Backup(),
+			tarball:      newTarWriter(t).addItems("pods", test.NewPod("ns-1", "pod-1")).done(),
+			apiResources: []*test.APIResource{test.Pods()},
+			actions: []velero.RestoreItemAction{
+				modifyingActionGetter(func(item *unstructured.Unstructured) {
+					item.SetLabels(map[string]string{"updated": "true"})
+				}),
+			},
+			want: []*test.APIResource{
+				test.Pods(
+					test.NewPod("ns-1", "pod-1", test.WithLabels("updated", "true"))),
+			},
+		},
+		{
+			name:         "action that removes a label to item gets restored",
+			restore:      defaultRestore().Restore(),
+			backup:       defaultBackup().Backup(),
+			tarball:      newTarWriter(t).addItems("pods", test.NewPod("ns-1", "pod-1", test.WithLabels("should-be-removed", "true"))).done(),
+			apiResources: []*test.APIResource{test.Pods()},
+			actions: []velero.RestoreItemAction{
+				modifyingActionGetter(func(item *unstructured.Unstructured) {
+					item.SetLabels(nil)
+				}),
+			},
+			want: []*test.APIResource{
+				test.Pods(
+					test.NewPod("ns-1", "pod-1")),
+			},
+		},
+		// TODO action that modifies namespace/name - what's the expected behavior?
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			for _, r := range tc.apiResources {
+				h.addItems(t, r)
+			}
+
+			// every restored item should have the restore and backup name labels, set
+			// them here so we don't have to do it in every test case definition above.
+			for _, resource := range tc.want {
+				for _, item := range resource.Items {
+					labels := item.GetLabels()
+					if labels == nil {
+						labels = make(map[string]string)
+					}
+
+					labels["velero.io/restore-name"] = tc.restore.Name
+					labels["velero.io/backup-name"] = tc.restore.Spec.BackupName
+
+					item.SetLabels(labels)
+				}
+			}
+
+			warnings, errs := h.restorer.Restore(
+				h.log,
+				tc.restore,
+				tc.backup,
+				nil, // volume snapshots
+				tc.tarball,
+				tc.actions,
+				nil, // snapshot location lister
+				nil, // volume snapshotter getter
+			)
+
+			assertEmptyResults(t, warnings, errs)
+			assertRestoredItems(t, h, tc.want)
+		})
+	}
+}
+
+// TestRestoreActionAdditionalItems runs restores with restore item actions that return additional items
+// to be restored, and verifies that that the correct set of items is created in the API. Verification is
+// done by looking at the namespaces/names of the items in the API; contents are not checked.
+func TestRestoreActionAdditionalItems(t *testing.T) {
+	tests := []struct {
+		name         string
+		restore      *velerov1api.Restore
+		backup       *velerov1api.Backup
+		tarball      io.Reader
+		apiResources []*test.APIResource
+		actions      []velero.RestoreItemAction
+		want         map[*test.APIResource][]string
+	}{
+		{
+			name:         "additional items that are already being restored are not restored twice",
+			restore:      defaultRestore().Restore(),
+			backup:       defaultBackup().Backup(),
+			tarball:      newTarWriter(t).addItems("pods", test.NewPod("ns-1", "pod-1"), test.NewPod("ns-2", "pod-2")).done(),
+			apiResources: []*test.APIResource{test.Pods()},
+			actions: []velero.RestoreItemAction{
+				&pluggableAction{
+					selector: velero.ResourceSelector{IncludedNamespaces: []string{"ns-1"}},
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: input.Item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.Pods, Namespace: "ns-2", Name: "pod-2"},
+							},
+						}, nil
+					},
+				},
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1", "ns-2/pod-2"},
+			},
+		},
+		// TODO the below test case fails, which seems like a bug
+		// {
+		// 	name:         "when using a restore namespace filter, additional items that are in a non-included namespace are not restored",
+		// 	restore:      defaultRestore().IncludedNamespaces("ns-1").Restore(),
+		// 	backup:       defaultBackup().Backup(),
+		// 	tarball:      newTarWriter(t).addItems("pods", test.NewPod("ns-1", "pod-1"), test.NewPod("ns-2", "pod-2")).done(),
+		// 	apiResources: []*test.APIResource{test.Pods()},
+		// 	actions: []velero.RestoreItemAction{
+		// 		&pluggableAction{
+		// 			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+		// 				return &velero.RestoreItemActionExecuteOutput{
+		// 					UpdatedItem: input.Item,
+		// 					AdditionalItems: []velero.ResourceIdentifier{
+		// 						{GroupResource: kuberesource.Pods, Namespace: "ns-2", Name: "pod-2"},
+		// 					},
+		// 				}, nil
+		// 			},
+		// 		},
+		// 	},
+		// 	want: map[*test.APIResource][]string{
+		// 		test.Pods(): {"ns-1/pod-1"},
+		// 	},
+		// },
+		{
+			name:    "when using a restore namespace filter, additional items that are cluster-scoped are restored",
+			restore: defaultRestore().IncludedNamespaces("ns-1").Restore(),
+			backup:  defaultBackup().Backup(),
+			tarball: newTarWriter(t).
+				addItems("pods", test.NewPod("ns-1", "pod-1")).
+				addItems("persistentvolumes", test.NewPV("pv-1")).
+				done(),
+			apiResources: []*test.APIResource{test.Pods(), test.PVs()},
+			actions: []velero.RestoreItemAction{
+				&pluggableAction{
+					executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+						return &velero.RestoreItemActionExecuteOutput{
+							UpdatedItem: input.Item,
+							AdditionalItems: []velero.ResourceIdentifier{
+								{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+							},
+						}, nil
+					},
+				},
+			},
+			want: map[*test.APIResource][]string{
+				test.Pods(): {"ns-1/pod-1"},
+				test.PVs():  {"/pv-1"},
+			},
+		},
+		// TODO the below test case fails, which seems like a bug
+		// {
+		// 	name:    "when using a restore resource filter, additional items that are non-included resources are not restored",
+		// 	restore: defaultRestore().IncludedResources("pods").Restore(),
+		// 	backup:  defaultBackup().Backup(),
+		// 	tarball: newTarWriter(t).
+		// 		addItems("pods", test.NewPod("ns-1", "pod-1")).
+		// 		addItems("persistentvolumes", test.NewPV("pv-1")).
+		// 		done(),
+		// 	apiResources: []*test.APIResource{test.Pods(), test.PVs()},
+		// 	actions: []velero.RestoreItemAction{
+		// 		&pluggableAction{
+		// 			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+		// 				return &velero.RestoreItemActionExecuteOutput{
+		// 					UpdatedItem: input.Item,
+		// 					AdditionalItems: []velero.ResourceIdentifier{
+		// 						{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+		// 					},
+		// 				}, nil
+		// 			},
+		// 		},
+		// 	},
+		// 	want: map[*test.APIResource][]string{
+		// 		test.Pods(): {"ns-1/pod-1"},
+		// 		test.PVs():  nil,
+		// 	},
+		// },
+		// TODO the below test case fails, which seems like a bug
+		// {
+		// 	name:    "when IncludeClusterResources=false, additional items that are cluster-scoped are not restored",
+		// 	restore: defaultRestore().IncludeClusterResources(false).Restore(),
+		// 	backup:  defaultBackup().Backup(),
+		// 	tarball: newTarWriter(t).
+		// 		addItems("pods", test.NewPod("ns-1", "pod-1")).
+		// 		addItems("persistentvolumes", test.NewPV("pv-1")).
+		// 		done(),
+		// 	apiResources: []*test.APIResource{test.Pods(), test.PVs()},
+		// 	actions: []velero.RestoreItemAction{
+		// 		&pluggableAction{
+		// 			executeFunc: func(input *velero.RestoreItemActionExecuteInput) (*velero.RestoreItemActionExecuteOutput, error) {
+		// 				return &velero.RestoreItemActionExecuteOutput{
+		// 					UpdatedItem: input.Item,
+		// 					AdditionalItems: []velero.ResourceIdentifier{
+		// 						{GroupResource: kuberesource.PersistentVolumes, Name: "pv-1"},
+		// 					},
+		// 				}, nil
+		// 			},
+		// 		},
+		// 	},
+		// 	want: map[*test.APIResource][]string{
+		// 		test.Pods(): {"ns-1/pod-1"},
+		// 		test.PVs():  nil,
+		// 	},
+		// },
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			for _, r := range tc.apiResources {
+				h.addItems(t, r)
+			}
+
+			warnings, errs := h.restorer.Restore(
+				h.log,
+				tc.restore,
+				tc.backup,
+				nil, // volume snapshots
+				tc.tarball,
+				tc.actions,
+				nil, // snapshot location lister
+				nil, // volume snapshotter getter
+			)
+
+			assertEmptyResults(t, warnings, errs)
+			assertAPIContents(t, h, tc.want)
+		})
+	}
+}
+
+// TestShouldRestore runs the ShouldRestore function for various permutations of
+// existing/nonexisting/being-deleted PVs, PVCs, and namespaces, and verifies the
+// result/error matches expectations.
+func TestShouldRestore(t *testing.T) {
+	tests := []struct {
+		name         string
+		pvName       string
+		apiResources []*test.APIResource
+		namespaces   []*corev1api.Namespace
+		want         bool
+		wantErr      error
+	}{
+		{
+			name:   "when PV is not found, result is true",
+			pvName: "pv-1",
+			want:   true,
+		},
+		{
+			name:   "when PV is found and has Phase=Released, result is false",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta: metav1.TypeMeta{
+						APIVersion: "v1",
+						Kind:       "PersistentVolume",
+					},
+					ObjectMeta: metav1.ObjectMeta{
+						Name: "pv-1",
+					},
+					Status: corev1api.PersistentVolumeStatus{
+						Phase: corev1api.VolumeReleased,
+					},
+				}),
+			},
+			want: false,
+		},
+		{
+			name:   "when PV is found and has associated PVC and namespace that aren't deleting, result is false",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+				test.PVCs(test.NewPVC("ns-1", "pvc-1")),
+			},
+			namespaces: []*corev1api.Namespace{test.NewNamespace("ns-1")},
+			want:       false,
+		},
+		{
+			name:   "when PV is found and has associated PVC that is deleting, result is false + timeout error",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+				test.PVCs(
+					test.NewPVC("ns-1", "pvc-1", test.WithDeletionTimestamp(time.Now())),
+				),
+			},
+			want:    false,
+			wantErr: errors.New("timed out waiting for the condition"),
+		},
+		{
+			name:   "when PV is found, has associated PVC that's not deleting, has associated NS that is terminating, result is false + timeout error",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+				test.PVCs(test.NewPVC("ns-1", "pvc-1")),
+			},
+			namespaces: []*corev1api.Namespace{
+				{
+					TypeMeta:   test.NewNamespace("").TypeMeta,
+					ObjectMeta: test.NewNamespace("ns-1").ObjectMeta,
+					Status: corev1api.NamespaceStatus{
+						Phase: corev1api.NamespaceTerminating,
+					},
+				},
+			},
+			want:    false,
+			wantErr: errors.New("timed out waiting for the condition"),
+		},
+		{
+			name:   "when PV is found, has associated PVC that's not deleting, has associated NS that has deletion timestamp, result is false + timeout error",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+				test.PVCs(test.NewPVC("ns-1", "pvc-1")),
+			},
+			namespaces: []*corev1api.Namespace{
+				test.NewNamespace("ns-1", test.WithDeletionTimestamp(time.Now())),
+			},
+			want:    false,
+			wantErr: errors.New("timed out waiting for the condition"),
+		},
+		{
+			name:   "when PV is found, associated PVC is not found, result is false + timeout error",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+			},
+			want:    false,
+			wantErr: errors.New("timed out waiting for the condition"),
+		},
+		{
+			name:   "when PV is found, has associated PVC, associated namespace not found, result is false + timeout error",
+			pvName: "pv-1",
+			apiResources: []*test.APIResource{
+				test.PVs(&corev1api.PersistentVolume{
+					TypeMeta:   test.NewPV("").TypeMeta,
+					ObjectMeta: test.NewPV("pv-1").ObjectMeta,
+					Spec: corev1api.PersistentVolumeSpec{
+						ClaimRef: &corev1api.ObjectReference{
+							Namespace: "ns-1",
+							Name:      "pvc-1",
+						},
+					},
+				}),
+				test.PVCs(test.NewPVC("ns-1", "pvc-1")),
+			},
+			want:    false,
+			wantErr: errors.New("timed out waiting for the condition"),
+		},
+	}
+
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			h := newHarness(t)
+
+			ctx := &context{
+				log:                        h.log,
+				dynamicFactory:             client.NewDynamicFactory(h.DynamicClient),
+				namespaceClient:            h.KubeClient.CoreV1().Namespaces(),
+				resourceTerminatingTimeout: time.Millisecond,
+			}
+
+			for _, resource := range tc.apiResources {
+				h.addItems(t, resource)
+			}
+
+			for _, ns := range tc.namespaces {
+				_, err := ctx.namespaceClient.Create(ns)
+				require.NoError(t, err)
+			}
+
+			pvClient, err := ctx.dynamicFactory.ClientForGroupVersionResource(
+				schema.GroupVersion{Group: "", Version: "v1"},
+				metav1.APIResource{Name: "persistentvolumes"},
+				"",
+			)
+			require.NoError(t, err)
+
+			res, err := ctx.shouldRestore(tc.pvName, pvClient)
+			assert.Equal(t, tc.want, res)
+			if tc.wantErr != nil {
+				if assert.NotNil(t, err, "expected a non-nil error") {
+					assert.EqualError(t, err, tc.wantErr.Error())
+				}
+			} else {
+				assert.Nil(t, err)
+			}
+		})
+	}
+}
+
+func assertRestoredItems(t *testing.T, h *harness, want []*test.APIResource) {
+	t.Helper()
+
+	for _, resource := range want {
+		resourceClient := h.DynamicClient.Resource(resource.GVR())
+		for _, item := range resource.Items {
+			var client dynamic.ResourceInterface
+			if item.GetNamespace() != "" {
+				client = resourceClient.Namespace(item.GetNamespace())
+			} else {
+				client = resourceClient
+			}
+
+			res, err := client.Get(item.GetName(), metav1.GetOptions{})
+			if !assert.NoError(t, err) {
+				continue
+			}
+
+			itemJSON, err := json.Marshal(item)
+			if !assert.NoError(t, err) {
+				continue
+			}
+
+			t.Logf("%v", string(itemJSON))
+
+			u := make(map[string]interface{})
+			if !assert.NoError(t, json.Unmarshal(itemJSON, &u)) {
+				continue
+			}
+			want := &unstructured.Unstructured{Object: u}
+
+			// These fields get non-nil zero values in the unstructured objects if they're
+			// empty in the structured objects. Remove them to make comparison easier.
+			unstructured.RemoveNestedField(want.Object, "metadata", "creationTimestamp")
+			unstructured.RemoveNestedField(want.Object, "status")
+
+			assert.Equal(t, want, res)
+		}
+	}
+}
+
 // assertResourceCreationOrder ensures that resources were created in the expected
 // order. Any resources *not* in resourcePriorities are required to come *after* all
 // resources in any order.
diff --git a/pkg/restore/restore_test.go b/pkg/restore/restore_test.go
index 6bcc42ea4..a2bb5dd11 100644
--- a/pkg/restore/restore_test.go
+++ b/pkg/restore/restore_test.go
@@ -19,7 +19,6 @@ package restore
 import (
 	"encoding/json"
 	"testing"
-	"time"
 
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
@@ -32,7 +31,6 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/util/sets"
-	"k8s.io/apimachinery/pkg/watch"
 	discoveryfake "k8s.io/client-go/discovery/fake"
 	kubefake "k8s.io/client-go/kubernetes/fake"
 	"k8s.io/client-go/kubernetes/scheme"
@@ -444,27 +442,6 @@ func (r *mockPVRestorer) executePVAction(obj *unstructured.Unstructured) (*unstr
 	return args.Get(0).(*unstructured.Unstructured), args.Error(1)
 }
 
-type mockWatch struct {
-	mock.Mock
-}
-
-func (w *mockWatch) Stop() {
-	w.Called()
-}
-
-func (w *mockWatch) ResultChan() <-chan watch.Event {
-	args := w.Called()
-	return args.Get(0).(chan watch.Event)
-}
-
-type fakeWatch struct{}
-
-func (w *fakeWatch) Stop() {}
-
-func (w *fakeWatch) ResultChan() <-chan watch.Event {
-	return make(chan watch.Event)
-}
-
 func TestResetMetadataAndStatus(t *testing.T) {
 	tests := []struct {
 		name        string
@@ -571,241 +548,6 @@ func TestIsCompleted(t *testing.T) {
 	}
 }
 
-func TestShouldRestore(t *testing.T) {
-	pv := `apiVersion: v1
-kind: PersistentVolume
-metadata:
-  annotations:
-    EXPORT_block: "\nEXPORT\n{\n\tExport_Id = 1;\n\tPath = /export/pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce;\n\tPseudo
-      = /export/pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce;\n\tAccess_Type = RW;\n\tSquash
-      = no_root_squash;\n\tSecType = sys;\n\tFilesystem_id = 1.1;\n\tFSAL {\n\t\tName
-      = VFS;\n\t}\n}\n"
-    Export_Id: "1"
-    Project_Id: "0"
-    Project_block: ""
-    Provisioner_Id: 5fdf4025-78a5-11e8-9ece-0242ac110004
-    kubernetes.io/createdby: nfs-dynamic-provisioner
-    pv.kubernetes.io/provisioned-by: example.com/nfs
-    volume.beta.kubernetes.io/mount-options: vers=4.1
-  creationTimestamp: 2018-06-25T18:27:35Z
-  finalizers:
-  - kubernetes.io/pv-protection
-  name: pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-  resourceVersion: "2576"
-  selfLink: /api/v1/persistentvolumes/pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-  uid: 6ecd24e4-78a5-11e8-a0d8-e2ad1e9734ce
-spec:
-  accessModes:
-  - ReadWriteMany
-  capacity:
-    storage: 1Mi
-  claimRef:
-    apiVersion: v1
-    kind: PersistentVolumeClaim
-    name: nfs
-    namespace: default
-    resourceVersion: "2565"
-    uid: 6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-  nfs:
-    path: /export/pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-    server: 10.103.235.254
-  storageClassName: example-nfs
-status:
-  phase: Bound`
-
-	pvc := `apiVersion: v1
-kind: PersistentVolumeClaim
-metadata:
-  annotations:
-    control-plane.alpha.kubernetes.io/leader: '{"holderIdentity":"5fdf5572-78a5-11e8-9ece-0242ac110004","leaseDurationSeconds":15,"acquireTime":"2018-06-25T18:27:35Z","renewTime":"2018-06-25T18:27:37Z","leaderTransitions":0}'
-    kubectl.kubernetes.io/last-applied-configuration: |
-      {"apiVersion":"v1","kind":"PersistentVolumeClaim","metadata":{"annotations":{},"name":"nfs","namespace":"default"},"spec":{"accessModes":["ReadWriteMany"],"resources":{"requests":{"storage":"1Mi"}},"storageClassName":"example-nfs"}}
-    pv.kubernetes.io/bind-completed: "yes"
-    pv.kubernetes.io/bound-by-controller: "yes"
-    volume.beta.kubernetes.io/storage-provisioner: example.com/nfs
-  creationTimestamp: 2018-06-25T18:27:28Z
-  finalizers:
-  - kubernetes.io/pvc-protection
-  name: nfs
-  namespace: default
-  resourceVersion: "2578"
-  selfLink: /api/v1/namespaces/default/persistentvolumeclaims/nfs
-  uid: 6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-spec:
-  accessModes:
-  - ReadWriteMany
-  resources:
-    requests:
-      storage: 1Mi
-  storageClassName: example-nfs
-  volumeName: pvc-6a74b5af-78a5-11e8-a0d8-e2ad1e9734ce
-status:
-  accessModes:
-  - ReadWriteMany
-  capacity:
-    storage: 1Mi
-  phase: Bound`
-
-	tests := []struct {
-		name              string
-		expectNSFound     bool
-		expectPVFound     bool
-		pvPhase           string
-		expectPVCFound    bool
-		expectPVCGet      bool
-		expectPVCDeleting bool
-		expectNSGet       bool
-		expectNSDeleting  bool
-		nsPhase           v1.NamespacePhase
-		expectedResult    bool
-	}{
-		{
-			name:           "pv not found, no associated pvc or namespace",
-			expectedResult: true,
-		},
-		{
-			name:           "pv found, phase released",
-			pvPhase:        string(v1.VolumeReleased),
-			expectPVFound:  true,
-			expectedResult: false,
-		},
-		{
-			name:           "pv found, has associated pvc and namespace that's aren't deleting",
-			expectPVFound:  true,
-			expectPVCGet:   true,
-			expectNSGet:    true,
-			expectPVCFound: true,
-			expectedResult: false,
-		},
-		{
-			name:              "pv found, has associated pvc that's deleting, don't look up namespace",
-			expectPVFound:     true,
-			expectPVCGet:      true,
-			expectPVCFound:    true,
-			expectPVCDeleting: true,
-			expectedResult:    false,
-		},
-		{
-			name:           "pv found, has associated pvc that's not deleting, has associated namespace that's terminating",
-			expectPVFound:  true,
-			expectPVCGet:   true,
-			expectPVCFound: true,
-			expectNSGet:    true,
-			expectNSFound:  true,
-			nsPhase:        v1.NamespaceTerminating,
-			expectedResult: false,
-		},
-		{
-			name:             "pv found, has associated pvc that's not deleting, has associated namespace that has deletion timestamp",
-			expectPVFound:    true,
-			expectPVCGet:     true,
-			expectPVCFound:   true,
-			expectNSGet:      true,
-			expectNSFound:    true,
-			expectNSDeleting: true,
-			expectedResult:   false,
-		},
-		{
-			name:           "pv found, associated pvc not found, namespace not queried",
-			expectPVFound:  true,
-			expectPVCGet:   true,
-			expectedResult: false,
-		},
-		{
-			name:           "pv found, associated pvc found, namespace not found",
-			expectPVFound:  true,
-			expectPVCGet:   true,
-			expectPVCFound: true,
-			expectNSGet:    true,
-			expectedResult: false,
-		},
-	}
-
-	for _, test := range tests {
-		t.Run(test.name, func(t *testing.T) {
-			dynamicFactory := &velerotest.FakeDynamicFactory{}
-			gv := schema.GroupVersion{Group: "", Version: "v1"}
-
-			pvClient := &velerotest.FakeDynamicClient{}
-			defer pvClient.AssertExpectations(t)
-
-			pvResource := metav1.APIResource{Name: "persistentvolumes", Namespaced: false}
-			dynamicFactory.On("ClientForGroupVersionResource", gv, pvResource, "").Return(pvClient, nil)
-
-			pvcClient := &velerotest.FakeDynamicClient{}
-			defer pvcClient.AssertExpectations(t)
-
-			pvcResource := metav1.APIResource{Name: "persistentvolumeclaims", Namespaced: true}
-			dynamicFactory.On("ClientForGroupVersionResource", gv, pvcResource, "default").Return(pvcClient, nil)
-
-			obj, _, err := scheme.Codecs.UniversalDecoder(v1.SchemeGroupVersion).Decode([]byte(pv), nil, &unstructured.Unstructured{})
-			pvObj := obj.(*unstructured.Unstructured)
-			require.NoError(t, err)
-
-			obj, _, err = scheme.Codecs.UniversalDecoder(v1.SchemeGroupVersion).Decode([]byte(pvc), nil, &unstructured.Unstructured{})
-			pvcObj := obj.(*unstructured.Unstructured)
-			require.NoError(t, err)
-
-			nsClient := &velerotest.FakeNamespaceClient{}
-			defer nsClient.AssertExpectations(t)
-			ns := newTestNamespace(pvcObj.GetNamespace()).Namespace
-
-			// Set up test expectations
-			if test.pvPhase != "" {
-				require.NoError(t, unstructured.SetNestedField(pvObj.Object, test.pvPhase, "status", "phase"))
-			}
-
-			if test.expectPVFound {
-				pvClient.On("Get", pvObj.GetName(), metav1.GetOptions{}).Return(pvObj, nil)
-			} else {
-				pvClient.On("Get", pvObj.GetName(), metav1.GetOptions{}).Return(&unstructured.Unstructured{}, k8serrors.NewNotFound(schema.GroupResource{Resource: "persistentvolumes"}, pvObj.GetName()))
-			}
-
-			if test.expectPVCDeleting {
-				pvcObj.SetDeletionTimestamp(&metav1.Time{Time: time.Now()})
-			}
-
-			// the pv needs to be found before moving on to look for pvc/namespace
-			// however, even if the pv is found, we may be testing the PV's phase and not expecting
-			// the pvc/namespace to be looked up
-			if test.expectPVCGet {
-				if test.expectPVCFound {
-					pvcClient.On("Get", pvcObj.GetName(), metav1.GetOptions{}).Return(pvcObj, nil)
-				} else {
-					pvcClient.On("Get", pvcObj.GetName(), metav1.GetOptions{}).Return(&unstructured.Unstructured{}, k8serrors.NewNotFound(schema.GroupResource{Resource: "persistentvolumeclaims"}, pvcObj.GetName()))
-				}
-			}
-
-			if test.nsPhase != "" {
-				ns.Status.Phase = test.nsPhase
-			}
-
-			if test.expectNSDeleting {
-				ns.SetDeletionTimestamp(&metav1.Time{Time: time.Now()})
-			}
-
-			if test.expectNSGet {
-				if test.expectNSFound {
-					nsClient.On("Get", pvcObj.GetNamespace(), mock.Anything).Return(ns, nil)
-				} else {
-					nsClient.On("Get", pvcObj.GetNamespace(), metav1.GetOptions{}).Return(&v1.Namespace{}, k8serrors.NewNotFound(schema.GroupResource{Resource: "namespaces"}, pvcObj.GetNamespace()))
-				}
-			}
-
-			ctx := &context{
-				dynamicFactory:             dynamicFactory,
-				log:                        velerotest.NewLogger(),
-				namespaceClient:            nsClient,
-				resourceTerminatingTimeout: 1 * time.Millisecond,
-			}
-
-			result, err := ctx.shouldRestore(pvObj.GetName(), pvClient)
-
-			assert.Equal(t, test.expectedResult, result)
-		})
-	}
-}
-
 func TestGetItemFilePath(t *testing.T) {
 	res := getItemFilePath("root", "resource", "", "item")
 	assert.Equal(t, "root/resources/resource/cluster/item.json", res)
diff --git a/pkg/test/resources.go b/pkg/test/resources.go
index f0c1dce32..002290f24 100644
--- a/pkg/test/resources.go
+++ b/pkg/test/resources.go
@@ -17,6 +17,8 @@ limitations under the License.
 package test
 
 import (
+	"time"
+
 	appsv1 "k8s.io/api/apps/v1"
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -310,3 +312,11 @@ func WithFinalizers(vals ...string) func(obj metav1.Object) {
 		obj.SetFinalizers(vals)
 	}
 }
+
+// WithDeletionTimestamp is a functional option that applies the specified
+// deletion timestamp to an object.
+func WithDeletionTimestamp(val time.Time) func(obj metav1.Object) {
+	return func(obj metav1.Object) {
+		obj.SetDeletionTimestamp(&metav1.Time{Time: val})
+	}
+}