diff --git a/ci/ci-test.sh b/ci/ci-test.sh
index 1fdf2986e..2e557e5fd 100755
--- a/ci/ci-test.sh
+++ b/ci/ci-test.sh
@@ -7,7 +7,7 @@ SNAP_CLASS=deploy/sample/zfssnapclass.yaml
 export OPENEBS_NAMESPACE=${OPENEBS_NAMESPACE:-openebs}
 TEST_DIR="$SCRIPT_DIR"/../tests
 
-CRDS_TO_DELETE_ON_CLEANUP="zfsrestores.zfs.openebs.io zfssnapshots.zfs.openebs.io zfsvolumes.zfs.openebs.io zfsbackups.zfs.openebs.io zfsnodes.zfs.openebs.io"
+CRDS_TO_DELETE_ON_CLEANUP="zfsrestores.zfs.openebs.io zfssnapshots.zfs.openebs.io zfsvolumes.zfs.openebs.io zfsbackups.zfs.openebs.io zfsnodes.zfs.openebs.io volumesnapshotclasses.snapshot.storage.k8s.io volumesnapshotcontents.snapshot.storage.k8s.io volumesnapshots.snapshot.storage.k8s.io"
 
 help() {
   cat <<EOF >&2
diff --git a/pkg/builder/volbuilder/build.go b/pkg/builder/volbuilder/build.go
index d1913e4f3..9b4415289 100644
--- a/pkg/builder/volbuilder/build.go
+++ b/pkg/builder/volbuilder/build.go
@@ -21,6 +21,8 @@ import (
 	apis "github.com/openebs/zfs-localpv/pkg/apis/openebs.io/zfs/v1"
 )
 
+const MarkForDeletionAnnotation string = "openebs.io/mark-for-deletion"
+
 // Builder is the builder object for ZFSVolume
 type Builder struct {
 	volume *ZFSVolume
@@ -154,6 +156,15 @@ func (b *Builder) WithVolBlockSize(bs string) *Builder {
 	return b
 }
 
+// WithAnnotation sets the annotation of ZFSVolume
+func (b *Builder) WithAnnotation() *Builder {
+	if b.volume.Object.Annotations == nil {
+		b.volume.Object.Annotations = make(map[string]string)
+	}
+	b.volume.Object.Annotations[MarkForDeletionAnnotation] = "true"
+	return b
+}
+
 // WithVolumeType sets if ZFSVolume needs to be thin provisioned
 func (b *Builder) WithVolumeType(vtype string) *Builder {
 	b.volume.Object.Spec.VolumeType = vtype
diff --git a/pkg/driver/controller.go b/pkg/driver/controller.go
index 990eee458..1a84d0557 100644
--- a/pkg/driver/controller.go
+++ b/pkg/driver/controller.go
@@ -503,14 +503,14 @@ func (cs *controller) DeleteVolume(
 	defer unlock()
 
 	// verify if the volume has already been deleted
-	vol, err := zfs.GetVolume(volumeID)
+	vol, err := zfs.GetZFSVolume(volumeID)
 	if vol != nil && vol.DeletionTimestamp != nil {
-		goto deleteResponse
+		return csipayload.NewDeleteVolumeResponseBuilder().Build(), nil
 	}
 
 	if err != nil {
 		if k8serror.IsNotFound(err) {
-			goto deleteResponse
+			return csipayload.NewDeleteVolumeResponseBuilder().Build(), nil
 		}
 		return nil, errors.Wrapf(
 			err,
@@ -524,19 +524,45 @@ func (cs *controller) DeleteVolume(
 		return nil, status.Error(codes.Internal, "can not delete, volume creation is in progress")
 	}
 
-	// Delete the corresponding ZV CR
-	err = zfs.DeleteVolume(volumeID)
+	// Fetch the list of snapshot for the given volume
+	snapList, err := zfs.GetSnapshotForVolume(volumeID)
 	if err != nil {
-		return nil, errors.Wrapf(
-			err,
-			"failed to handle delete volume request for {%s}",
-			volumeID,
+		return nil, status.Errorf(
+			codes.NotFound,
+			"failed to handle delete volume request for {%s}, "+
+				"validation failed checking for snapshots. Error: %s",
+			req.VolumeId,
+			err.Error(),
 		)
 	}
 
+	// Delete the corresponding ZV CR only if there are no snapshots present for the volume
+	if len(snapList.Items) == 0 {
+		err = zfs.DeleteVolume(volumeID)
+		if err != nil {
+			return nil, errors.Wrapf(
+				err,
+				"failed to handle delete volume request for {%s}",
+				volumeID,
+			)
+		}
+	} else {
+		// add annotation to the volume to indicate that it is eligible for deletion
+		// once all the snapshots are deleted and the reclaim policy is not Retain
+		// this volume will be deleted
+		err = zfs.MarkForDeletion(volumeID)
+		if err != nil {
+			return nil, errors.Wrapf(
+				err,
+				"failed to annotate volume on deletion request for {%s}",
+				volumeID,
+			)
+		}
+
+	}
+
 	sendEventOrIgnore("", volumeID, vol.Spec.Capacity, analytics.VolumeDeprovision)
 
-deleteResponse:
 	return csipayload.NewDeleteVolumeResponseBuilder().Build(), nil
 }
 
@@ -815,8 +841,39 @@ func (cs *controller) DeleteSnapshot(
 		// should succeed when an invalid snapshot id is used
 		return &csi.DeleteSnapshotResponse{}, nil
 	}
+	volumeID := snapshotID[0]
 	unlock := cs.volumeLock.LockVolumeWithSnapshot(snapshotID[0], snapshotID[1])
 	defer unlock()
+
+	// verify if the snapshot has already been deleted
+	snap, err := zfs.GetZFSSnapshot(snapshotID[1])
+	if snap != nil && snap.DeletionTimestamp != nil {
+		return csipayload.NewDeleteSnapshotResponseBuilder().Build(), nil
+	}
+
+	if err != nil {
+		if k8serror.IsNotFound(err) {
+			return csipayload.NewDeleteSnapshotResponseBuilder().Build(), nil
+		}
+		return nil, errors.Wrapf(
+			err,
+			"failed to get snapshot for {%s}",
+			snapshotID[1],
+		)
+	}
+
+	// Fetch the list of snapshot for the given volume
+	snapList, err := zfs.GetSnapshotForVolume(volumeID)
+	if err != nil {
+		return nil, status.Errorf(
+			codes.FailedPrecondition,
+			"failed to handle delete snapshot request for {%s}, "+
+				"validation failed checking for snapshot list for volume. Error: %s",
+			volumeID,
+			err.Error(),
+		)
+	}
+
 	if err := zfs.DeleteSnapshot(snapshotID[1]); err != nil {
 		return nil, status.Errorf(
 			codes.Internal,
@@ -825,7 +882,33 @@ func (cs *controller) DeleteSnapshot(
 			err.Error(),
 		)
 	}
-	return &csi.DeleteSnapshotResponse{}, nil
+
+	eligibleForDeletion, err := zfs.IsVolumeEligibleForDeletion(volumeID)
+	if err != nil {
+		return nil, status.Errorf(
+			codes.FailedPrecondition,
+			"failed to handle delete snapshot request for {%s}, "+
+				"validation failed checking for eligible for deletion. Error: %s",
+			volumeID,
+			err.Error(),
+		)
+	}
+
+	// Delete the corresponding ZV CR only if this is the last snapshot
+	// for the volume and the corresponding pvc is deleted
+	if len(snapList.Items) == 1 && eligibleForDeletion {
+		err = zfs.DeleteVolume(volumeID)
+		if err != nil {
+			return nil, errors.Wrapf(
+				err,
+				"failed to handle delete volume request for {%s}",
+				volumeID,
+			)
+		}
+		klog.Infof("volume %s deleted after the deletion of last snapshot %s ", volumeID, snapshotID[1])
+	}
+
+	return csipayload.NewDeleteSnapshotResponseBuilder().Build(), nil
 }
 
 // ListSnapshots lists all snapshots for the
diff --git a/pkg/response/delete.go b/pkg/response/delete.go
index 29a3d4015..18b7b244b 100644
--- a/pkg/response/delete.go
+++ b/pkg/response/delete.go
@@ -39,3 +39,23 @@ func NewDeleteVolumeResponseBuilder() *DeleteVolumeResponseBuilder {
 func (b *DeleteVolumeResponseBuilder) Build() *csi.DeleteVolumeResponse {
 	return b.response
 }
+
+// DeleteSnapshotResponseBuilder helps building an
+// instance of csi DeleteSnapshotResponse
+type DeleteSnapshotResponseBuilder struct {
+	response *csi.DeleteSnapshotResponse
+}
+
+// NewDeleteSnapshotResponseBuilder returns a new
+// instance of DeleteSnapshotResponseBuilder
+func NewDeleteSnapshotResponseBuilder() *DeleteSnapshotResponseBuilder {
+	return &DeleteSnapshotResponseBuilder{
+		response: &csi.DeleteSnapshotResponse{},
+	}
+}
+
+// Build returns the constructed instance
+// of csi DeleteSnapshotResponse
+func (b *DeleteSnapshotResponseBuilder) Build() *csi.DeleteSnapshotResponse {
+	return b.response
+}
diff --git a/pkg/zfs/volume.go b/pkg/zfs/volume.go
index 446e4a667..0d356c20b 100644
--- a/pkg/zfs/volume.go
+++ b/pkg/zfs/volume.go
@@ -27,6 +27,8 @@ import (
 	"github.com/openebs/zfs-localpv/pkg/builder/restorebuilder"
 	"github.com/openebs/zfs-localpv/pkg/builder/snapbuilder"
 	"github.com/openebs/zfs-localpv/pkg/builder/volbuilder"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/klog/v2"
 )
@@ -62,6 +64,8 @@ const (
 	OpenEBSCasTypeKey string = "openebs.io/cas-type"
 	// ZFSCasTypeName for the name of the cas-type
 	ZFSCasTypeName string = "localpv-zfs"
+	// MarkForDeletionAnnotation is the annotation key to mark the volume for deletion
+	MarkForDeletionAnnotation string = "openebs.io/marked-for-deletion"
 )
 
 var (
@@ -213,13 +217,6 @@ func DeleteSnapshot(snapname string) (err error) {
 	return
 }
 
-// GetVolume the corresponding ZFSVolume CR
-func GetVolume(volumeID string) (*apis.ZFSVolume, error) {
-	return volbuilder.NewKubeclient().
-		WithNamespace(OpenEBSNamespace).
-		Get(volumeID, metav1.GetOptions{})
-}
-
 // DeleteVolume deletes the corresponding ZFSVol CR
 func DeleteVolume(volumeID string) (err error) {
 	err = volbuilder.NewKubeclient().WithNamespace(OpenEBSNamespace).Delete(volumeID)
@@ -251,6 +248,18 @@ func GetZFSVolume(volumeID string) (*apis.ZFSVolume, error) {
 	return vol, err
 }
 
+// UpdateZFSVolumeAnnotation updtates the ZFSVolume CR with the given annotation
+func UpdateZFSVolumeAnnotation(vol *apis.ZFSVolume) error {
+	newVol, err := volbuilder.BuildFrom(vol).
+		WithAnnotation().Build()
+
+	if err != nil {
+		return err
+	}
+	_, err = volbuilder.NewKubeclient().WithNamespace(OpenEBSNamespace).Update(newVol)
+	return err
+}
+
 // GetZFSVolumeState returns ZFSVolume OwnerNode and State for
 // the given volume. CreateVolume request may call it again and
 // again until volume is "Ready".
@@ -441,3 +450,46 @@ func IsVolumeReady(vol *apis.ZFSVolume) bool {
 
 	return false
 }
+
+// GetSnapshotForVolume fetches all the snapshots for the given volume
+func GetSnapshotForVolume(volumeID string) (*apis.ZFSSnapshotList, error) {
+	listOptions := metav1.ListOptions{
+		LabelSelector: ZFSVolKey + "=" + volumeID,
+	}
+	snapList, err := snapbuilder.NewKubeclient().WithNamespace(OpenEBSNamespace).List(listOptions)
+	return snapList, err
+}
+
+// MarkForDeletion marks the volume for deletion by adding the annotation
+func MarkForDeletion(volumeName string) error {
+	zv, err := GetZFSVolume(volumeName)
+	if err != nil {
+		klog.Errorf("failed to get ZV %s: %v", volumeName, err)
+		return err
+	}
+
+	err = UpdateZFSVolumeAnnotation(zv)
+	if err != nil {
+		klog.Errorf("Failed to annotate the ZV with marked for deletion %s: %v", volumeName, err)
+		return err
+	}
+	return nil
+}
+
+// IsVolumeEligibleForDeletion checks if the volume can be deleted or not
+func IsVolumeEligibleForDeletion(volumeName string) (bool, error) {
+
+	zfsVol, err := GetZFSVolume(volumeName)
+	if err != nil {
+		return false, status.Errorf(
+			codes.Internal,
+			"failed to get ZFSVolume %s: %v",
+			volumeName,
+			err,
+		)
+	}
+	if zfsVol.Annotations[MarkForDeletionAnnotation] == "true" {
+		return true, nil
+	}
+	return false, nil
+}
diff --git a/tests/provision_test.go b/tests/provision_test.go
index 35246080a..ad1170fbb 100644
--- a/tests/provision_test.go
+++ b/tests/provision_test.go
@@ -41,7 +41,7 @@ func exhaustiveVolumeTests(parameters map[string]string) {
 	snapshotAndCloneCreate()
 	// btrfs does not support online resize
 	if fstype != "btrfs" {
-		By("Resizing the PVC", resizeAndVerifyPVC)
+		By("Resizing the PVC", func() { resizeAndVerifyPVC(pvcNameFS) })
 	}
 	snapshotAndCloneCleanUp()
 	cleanUp()
@@ -51,8 +51,8 @@ func exhaustiveVolumeTests(parameters map[string]string) {
 func create(parameters map[string]string) {
 	By("####### Creating the storage class : " + parameters["fstype"] + " #######")
 	createFstypeStorageClass(parameters)
-	By("creating and verifying PVC bound status", createAndVerifyPVC)
-	By("Creating and deploying app pod", createDeployVerifyApp)
+	By("creating and verifying PVC bound status", func() { createAndVerifyPVC(pvcNameFS) })
+	By("Creating and deploying app pod", func() { createDeployVerifyApp(appNameFS, pvcNameFS) })
 	By("verifying ZFSVolume object", VerifyZFSVolume)
 	By("verifying storage class parameters")
 	VerifyStorageClassParams(parameters)
@@ -60,53 +60,94 @@ func create(parameters map[string]string) {
 
 // Creates the snapshot/clone resources
 func snapshotAndCloneCreate() {
-	createSnapshot(pvcName, snapName)
-	verifySnapshotCreated(snapName)
-	createClone(clonePvcName, snapName, scObj.Name)
-	By("Creating and deploying clone app pod", createDeployVerifyCloneApp)
+	createSnapshot(pvcNameFS, snapNameFS)
+	verifySnapshotCreated(snapNameFS)
+	createClone(clonePvcNameFS, snapNameFS, scObj.Name)
+	By("Creating and deploying clone app pod", func() { createDeployVerifyCloneApp(cloneAppNameFS, clonePvcNameFS) })
 }
 
 // Removes the snapshot/clone resources
 func snapshotAndCloneCleanUp() {
-	deleteAppDeployment(cloneAppName)
-	deletePVC(clonePvcName)
-	deleteSnapshot(pvcName, snapName)
+	deleteAppDeployment(cloneAppNameFS)
+	deletePVC(clonePvcNameFS)
+	deleteSnapshot(pvcNameFS, snapNameFS)
 }
 
 // Removes the resources
 func cleanUp() {
-	deleteAppDeployment(appName)
-	deletePVC(pvcName)
+	deleteAppDeployment(appNameFS)
+	deletePVC(pvcNameFS)
 	By("Deleting storage class", deleteStorageClass)
 }
 
 func blockVolCreationTest() {
 	By("Creating default storage class", createStorageClass)
-	By("creating and verifying PVC bound status", createAndVerifyBlockPVC)
+	By("creating and verifying PVC bound status", func() { createAndVerifyPVC(pvcNameBlock) })
 
-	By("Creating and deploying app pod", createDeployVerifyBlockApp)
+	By("Creating and deploying app pod", func() { createDeployVerifyApp(appNameBlock, pvcNameBlock) })
 	By("verifying ZFSVolume object", VerifyZFSVolume)
 	By("verifying ZFSVolume property change", VerifyZFSVolumePropEdit)
-	By("Deleting application deployment")
 
-	createSnapshot(pvcName, snapName)
-	verifySnapshotCreated(snapName)
-	createClone(clonePvcName, snapName, scObj.Name)
-	By("Creating and deploying clone app pod", createDeployVerifyCloneApp)
+	createSnapshot(pvcNameBlock, snapNameBlock)
+	verifySnapshotCreated(snapNameBlock)
+	createClone(clonePvcNameBlock, snapNameBlock, scObj.Name)
+	By("Creating and deploying clone app pod", func() { createDeployVerifyCloneApp(cloneAppNameBlock, clonePvcNameBlock) })
 
-	By("Deleting clone and main application deployment")
-	deleteAppDeployment(cloneAppName)
-	deleteAppDeployment(appName)
+	By("Deleting main application deployment")
+	deleteAppDeployment(appNameBlock)
 
-	By("Deleting snapshot, main pvc and clone pvc")
-	deletePVC(clonePvcName)
-	deleteSnapshot(pvcName, snapName)
-	deletePVC(pvcName)
+	zvName := getZVName(pvcNameBlock)
+	By("Deleting main pvc")
+	deletePVC(pvcNameBlock)
 
+	By("Verifying ZFSVolume object after pvc deletion when snapshot is present", VerifyZFSVolume)
+
+	By("Deleting clone application deployment")
+	deleteAppDeployment(cloneAppNameBlock)
+
+	By("Deleting snapshot and clone pvc")
+
+	deletePVC(clonePvcNameBlock)
+	By("Verifying that ZV is present after pvc deletion ", func() { IsZVPresentConsistently(zvName) })
+	deleteSnapshot(pvcNameBlock, snapNameBlock)
+	By("Verifying that ZV is deleted after snapshot deletion ", func() { IsZVDeletedEventually(zvName) })
+
+	By("Deleting storage class", deleteStorageClass)
+}
+
+func blockVolCreationWithReclaimRetainTest() {
+	By("Creating storage class retain reclaim policy", createStorageClassWithReclaimPolicy)
+	By("creating and verifying PVC bound status", func() { createAndVerifyPVC(pvcNameBlock) })
+
+	By("Creating and deploying app pod", func() { createDeployVerifyApp(appNameBlock, pvcNameBlock) })
+	By("verifying ZFSVolume object", VerifyZFSVolume)
+
+	createSnapshot(pvcNameBlock, snapNameBlock)
+	verifySnapshotCreated(snapNameBlock)
+
+	By("Deleting main application deployment")
+	deleteAppDeployment(appNameBlock)
+
+	zvName := getZVName(pvcNameBlock)
+	By("Deleting main pvc")
+	deletePVC(pvcNameBlock)
+
+	By("Verifying ZFSVolume object after pvc deletion when snapshot is present", VerifyZFSVolume)
+
+	By("Verifying that ZV is present after pvc deletion ", func() { IsZVPresentConsistently(zvName) })
+	By("Deleting snapshot")
+	deleteSnapshot(pvcNameBlock, snapNameBlock)
+	By("Verifying that ZV is present after pvc deletion ", func() { IsZVPresentConsistently(zvName) })
+	deletePV(zvName)
+	By("Verifying that ZV is present after pvc deletion ", func() { IsZVPresentConsistently(zvName) })
+	By("Deleting the ZV for cleanup ", func() { DeleteZV(zvName) })
 	By("Deleting storage class", deleteStorageClass)
+
 }
 
 func volumeCreationTest() {
 	By("Running volume creation test", fsVolCreationTest)
 	By("Running block volume creation test", blockVolCreationTest)
+	By("Running block volume creation test with retain reclaim policy ", blockVolCreationWithReclaimRetainTest)
+
 }
diff --git a/tests/pv/kubernetes.go b/tests/pv/kubernetes.go
new file mode 100644
index 000000000..31aff113c
--- /dev/null
+++ b/tests/pv/kubernetes.go
@@ -0,0 +1,155 @@
+package pv
+
+import (
+	"context"
+	"strings"
+
+	"github.com/openebs/lib-csi/pkg/common/errors"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	client "github.com/openebs/lib-csi/pkg/common/kubernetes/client"
+	"k8s.io/client-go/kubernetes"
+)
+
+// getClientsetFn is a typed function that
+// abstracts fetching of clientset
+type getClientsetFn func() (clientset *kubernetes.Clientset, err error)
+
+// getClientsetFromPathFn is a typed function that
+// abstracts fetching of clientset from kubeConfigPath
+type getClientsetForPathFn func(kubeConfigPath string) (clientset *kubernetes.Clientset, err error)
+
+// getpvcFn is a typed function that
+// abstracts fetching of pvc
+type getFn func(cli *kubernetes.Clientset, name string, opts metav1.GetOptions) (*corev1.PersistentVolume, error)
+
+// deleteFn is a typed function that abstracts
+// deletion of pvcs
+type deleteFn func(cli *kubernetes.Clientset, name string, deleteOpts *metav1.DeleteOptions) error
+
+// Kubeclient enables kubernetes API operations
+// on pvc instance
+type Kubeclient struct {
+	// clientset refers to pvc clientset
+	// that will be responsible to
+	// make kubernetes API calls
+	clientset *kubernetes.Clientset
+
+	// kubeconfig path to get kubernetes clientset
+	kubeConfigPath string
+
+	// functions useful during mocking
+	getClientset        getClientsetFn
+	getClientsetForPath getClientsetForPathFn
+	get                 getFn
+	del                 deleteFn
+}
+
+// KubeclientBuildOption abstracts creating an
+// instance of kubeclient
+type KubeclientBuildOption func(*Kubeclient)
+
+// withDefaults sets the default options
+// of kubeclient instance
+func (k *Kubeclient) withDefaults() {
+	if k.getClientset == nil {
+		k.getClientset = func() (clients *kubernetes.Clientset, err error) {
+			return client.New().Clientset()
+		}
+	}
+
+	if k.getClientsetForPath == nil {
+		k.getClientsetForPath = func(kubeConfigPath string) (clients *kubernetes.Clientset, err error) {
+			return client.New(client.WithKubeConfigPath(kubeConfigPath)).Clientset()
+		}
+	}
+
+	if k.get == nil {
+		k.get = func(cli *kubernetes.Clientset, name string, opts metav1.GetOptions) (*corev1.PersistentVolume, error) {
+			return cli.CoreV1().PersistentVolumes().Get(context.TODO(), name, opts)
+		}
+	}
+
+	if k.del == nil {
+		k.del = func(cli *kubernetes.Clientset, name string, deleteOpts *metav1.DeleteOptions) error {
+			return cli.CoreV1().PersistentVolumes().Delete(context.TODO(), name, *deleteOpts)
+		}
+	}
+
+}
+
+// WithClientSet sets the kubernetes client against
+// the kubeclient instance
+func WithClientSet(c *kubernetes.Clientset) KubeclientBuildOption {
+	return func(k *Kubeclient) {
+		k.clientset = c
+	}
+}
+
+// WithKubeConfigPath sets the kubeConfig path
+// against client instance
+func WithKubeConfigPath(path string) KubeclientBuildOption {
+	return func(k *Kubeclient) {
+		k.kubeConfigPath = path
+	}
+}
+
+// NewKubeClient returns a new instance of kubeclient meant for
+// pvc operations
+func NewKubeClient(opts ...KubeclientBuildOption) *Kubeclient {
+	k := &Kubeclient{}
+	for _, o := range opts {
+		o(k)
+	}
+	k.withDefaults()
+	return k
+}
+
+func (k *Kubeclient) getClientsetForPathOrDirect() (*kubernetes.Clientset, error) {
+	if k.kubeConfigPath != "" {
+		return k.getClientsetForPath(k.kubeConfigPath)
+	}
+	return k.getClientset()
+}
+
+// getClientsetOrCached returns either a new instance
+// of kubernetes client or its cached copy
+func (k *Kubeclient) getClientsetOrCached() (*kubernetes.Clientset, error) {
+	if k.clientset != nil {
+		return k.clientset, nil
+	}
+
+	cs, err := k.getClientsetForPathOrDirect()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get clientset")
+	}
+	k.clientset = cs
+	return k.clientset, nil
+}
+
+// Get returns a pvc resource
+// instances present in kubernetes cluster
+func (k *Kubeclient) Get(name string, opts metav1.GetOptions) (*corev1.PersistentVolume, error) {
+	if strings.TrimSpace(name) == "" {
+		return nil, errors.New("failed to get pvc: missing pv name")
+	}
+	cli, err := k.getClientsetOrCached()
+	if err != nil {
+		return nil, errors.Wrapf(err, "failed to get pv {%s}", name)
+	}
+	return k.get(cli, name, opts)
+}
+
+// Delete deletes a pvc instance from the
+// kubecrnetes cluster
+func (k *Kubeclient) Delete(name string, deleteOpts *metav1.DeleteOptions) error {
+	if strings.TrimSpace(name) == "" {
+		return errors.New("failed to delete pv: missing pv name")
+	}
+	cli, err := k.getClientsetOrCached()
+	if err != nil {
+		return errors.Wrapf(err, "failed to delete pv {%s}", name)
+	}
+	return k.del(cli, name, deleteOpts)
+}
diff --git a/tests/sc/build.go b/tests/sc/build.go
index 07386a936..2484e4fa9 100644
--- a/tests/sc/build.go
+++ b/tests/sc/build.go
@@ -18,6 +18,7 @@ package sc
 
 import (
 	"github.com/openebs/lib-csi/pkg/common/errors"
+	corev1 "k8s.io/api/core/v1"
 	storagev1 "k8s.io/api/storage/v1"
 )
 
@@ -89,6 +90,12 @@ func (b *Builder) WithProvisioner(provisioner string) *Builder {
 	return b
 }
 
+// WithReclaimPolicy sets the ReclaimPolicy field of storageclass with provided argument.
+func (b *Builder) WithReclaimPolicy(reclaimPolicy *corev1.PersistentVolumeReclaimPolicy) *Builder {
+	b.sc.object.ReclaimPolicy = reclaimPolicy
+	return b
+}
+
 // WithVolumeExpansion sets the AllowedVolumeExpansion field of storageclass with provided argument.
 func (b *Builder) WithVolumeExpansion(expansionAllowed bool) *Builder {
 	b.sc.object.AllowVolumeExpansion = &expansionAllowed
diff --git a/tests/suite_test.go b/tests/suite_test.go
index 16dfd82f7..7e91d6d83 100644
--- a/tests/suite_test.go
+++ b/tests/suite_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/openebs/zfs-localpv/pkg/builder/volbuilder"
 	"github.com/openebs/zfs-localpv/tests/deploy"
 	"github.com/openebs/zfs-localpv/tests/pod"
+	"github.com/openebs/zfs-localpv/tests/pv"
 	"github.com/openebs/zfs-localpv/tests/pvc"
 	"github.com/openebs/zfs-localpv/tests/sc"
 	appsv1 "k8s.io/api/apps/v1"
@@ -41,18 +42,30 @@ const (
 )
 
 var (
-	ZFSClient      *volbuilder.Kubeclient
-	SCClient       *sc.Kubeclient
-	PVCClient      *pvc.Kubeclient
-	DeployClient   *deploy.Kubeclient
-	PodClient      *pod.KubeClient
-	scName         = "zfspv-sc"
-	ZFSProvisioner = "zfs.csi.openebs.io"
-	pvcName        = "zfspv-pvc"
-	snapName       = "zfspv-snap"
-	appName        = "busybox-zfspv"
-	clonePvcName   = "zfspv-pvc-clone"
-	cloneAppName   = "busybox-zfspv-clone"
+	ZFSClient           *volbuilder.Kubeclient
+	SCClient            *sc.Kubeclient
+	PVCClient           *pvc.Kubeclient
+	PVClient            *pv.Kubeclient
+	DeployClient        *deploy.Kubeclient
+	PodClient           *pod.KubeClient
+	scName              = "zfspv-sc"
+	ZFSProvisioner      = "zfs.csi.openebs.io"
+	RetainReclaimPolicy = corev1.PersistentVolumeReclaimPolicy(corev1.PersistentVolumeReclaimRetain)
+
+	pvcNameFS    = "zfspv-pvc-fs"
+	pvcNameBlock = "zfspv-pvc-block"
+
+	appNameFS    = "busybox-zfspv-fs"
+	appNameBlock = "busybox-zfspv-block"
+
+	snapNameFS    = "zfspv-snap-fs"
+	snapNameBlock = "zfspv-snap-block"
+
+	clonePvcNameFS    = "zfspv-pvc-clone-fs"
+	clonePvcNameBlock = "zfspv-pvc-clone-block"
+
+	cloneAppNameFS    = "busybox-zfspv-clone-fs"
+	cloneAppNameBlock = "busybox-zfspv-clone-block"
 
 	scObj              *storagev1.StorageClass
 	deployObj          *appsv1.Deployment
@@ -76,6 +89,7 @@ func init() {
 	}
 	SCClient = sc.NewKubeClient(sc.WithKubeConfigPath(KubeConfigPath))
 	PVCClient = pvc.NewKubeClient(pvc.WithKubeConfigPath(KubeConfigPath))
+	PVClient = pv.NewKubeClient(pv.WithKubeConfigPath(KubeConfigPath))
 	DeployClient = deploy.NewKubeClient(deploy.WithKubeConfigPath(KubeConfigPath))
 	PodClient = pod.NewKubeClient(pod.WithKubeConfigPath(KubeConfigPath))
 	ZFSClient = volbuilder.NewKubeclient(volbuilder.WithKubeConfigPath(KubeConfigPath))
diff --git a/tests/utils.go b/tests/utils.go
index 3bd459e9e..686ce4e19 100644
--- a/tests/utils.go
+++ b/tests/utils.go
@@ -117,7 +117,7 @@ func GetVolumeProperty(vol *apis.ZFSVolume, prop string) (string, error) {
 // else returns false
 func IsPVCDeletedEventually(pvcName string) bool {
 	return gomega.Eventually(func() bool {
-		_, err := PVCClient.
+		_, err := PVCClient.WithNamespace(OpenEBSNamespace).
 			Get(pvcName, metav1.GetOptions{})
 		return k8serrors.IsNotFound(err)
 	},
@@ -125,6 +125,32 @@ func IsPVCDeletedEventually(pvcName string) bool {
 		Should(gomega.BeTrue())
 }
 
+// IsPVDeletedEventually tries to get the deleted pv
+// and returns true if pv is not found
+// else returns false
+func IsPVDeletedEventually(pvName string) bool {
+	return gomega.Eventually(func() bool {
+		_, err := PVClient.
+			Get(pvName, metav1.GetOptions{})
+		return k8serrors.IsNotFound(err)
+	},
+		120, 10).
+		Should(gomega.BeTrue())
+}
+
+// IsZVDeletedEventually tries to get the deleted zv
+// and returns true if zv is not found
+// else returns false
+func IsZVDeletedEventually(zvName string) bool {
+	return gomega.Eventually(func() bool {
+		_, err := ZFSClient.WithNamespace(OpenEBSNamespace).
+			Get(zvName, metav1.GetOptions{})
+		return k8serrors.IsNotFound(err)
+	},
+		120, 10).
+		Should(gomega.BeTrue())
+}
+
 // VerifyStorageClassParams verifies the volume properties set at creation time
 func VerifyStorageClassParams(property map[string]string) {
 	vol, err := ZFSClient.WithNamespace(OpenEBSNamespace).
@@ -197,6 +223,28 @@ func createFstypeStorageClass(addons map[string]string) {
 	gomega.Expect(err).To(gomega.BeNil(), "while creating a ext4 storageclass {%s}", scName)
 }
 
+func createStorageClassWithReclaimPolicy() {
+	var (
+		err error
+	)
+
+	parameters := map[string]string{
+		"poolname": POOLNAME,
+	}
+
+	ginkgo.By("building a default storage class")
+	scObj, err = sc.NewBuilder().
+		WithGenerateName(scName).
+		WithParametersNew(parameters).
+		WithProvisioner(ZFSProvisioner).
+		WithReclaimPolicy(&RetainReclaimPolicy).Build()
+	gomega.Expect(err).ShouldNot(gomega.HaveOccurred(),
+		"while building default storageclass obj with prefix {%s}", scName)
+
+	scObj, err = SCClient.Create(scObj)
+	gomega.Expect(err).To(gomega.BeNil(), "while creating a default storageclass {%s}", scName)
+}
+
 func createStorageClass() {
 	var (
 		err error
@@ -349,10 +397,9 @@ func deleteStorageClass() {
 		"while deleting zfs storageclass {%s}", scObj.Name)
 }
 
-func createAndVerifyPVC() {
+func createAndVerifyPVC(pvcName string) {
 	var (
-		err     error
-		pvcName = "zfspv-pvc"
+		err error
 	)
 	ginkgo.By("building a pvc")
 	pvcObj, err = pvc.NewBuilder().
@@ -361,53 +408,11 @@ func createAndVerifyPVC() {
 		WithStorageClass(scObj.Name).
 		WithAccessModes(accessModes).
 		WithCapacity(capacity).Build()
-	gomega.Expect(err).ShouldNot(
-		gomega.HaveOccurred(),
-		"while building pvc {%s} in namespace {%s}",
-		pvcName,
-		OpenEBSNamespace,
-	)
 
-	ginkgo.By("creating above pvc")
-	pvcObj, err = PVCClient.WithNamespace(OpenEBSNamespace).Create(pvcObj)
-	gomega.Expect(err).To(
-		gomega.BeNil(),
-		"while creating pvc {%s} in namespace {%s}",
-		pvcName,
-		OpenEBSNamespace,
-	)
-
-	ginkgo.By("verifying pvc status as bound")
-
-	status := IsPVCBoundEventually(pvcName)
-	gomega.Expect(status).To(gomega.Equal(true),
-		"while checking status equal to bound")
-
-	pvcObj, err = PVCClient.WithNamespace(OpenEBSNamespace).Get(pvcObj.Name, metav1.GetOptions{})
-	gomega.Expect(err).To(
-		gomega.BeNil(),
-		"while retrieving pvc {%s} in namespace {%s}",
-		pvcName,
-		OpenEBSNamespace,
-	)
-}
-
-func createAndVerifyBlockPVC() {
-	var (
-		err     error
-		pvcName = "zfspv-pvc"
-	)
-
-	volmode := corev1.PersistentVolumeBlock
-
-	ginkgo.By("building a pvc")
-	pvcObj, err = pvc.NewBuilder().
-		WithName(pvcName).
-		WithNamespace(OpenEBSNamespace).
-		WithStorageClass(scObj.Name).
-		WithAccessModes(accessModes).
-		WithVolumeMode(&volmode).
-		WithCapacity(capacity).Build()
+	if pvcName == "zfspv-pvc-block" {
+		volmode := corev1.PersistentVolumeBlock
+		pvcObj.Spec.VolumeMode = &volmode
+	}
 	gomega.Expect(err).ShouldNot(
 		gomega.HaveOccurred(),
 		"while building pvc {%s} in namespace {%s}",
@@ -439,10 +444,9 @@ func createAndVerifyBlockPVC() {
 	)
 }
 
-func resizeAndVerifyPVC() {
+func resizeAndVerifyPVC(pvcName string) {
 	var (
-		err     error
-		pvcName = "zfspv-pvc"
+		err error
 	)
 	ginkgo.By("updating the pvc with new size")
 	pvcObj, err = PVCClient.WithNamespace(OpenEBSNamespace).Get(pvcObj.Name, metav1.GetOptions{})
@@ -476,13 +480,17 @@ func resizeAndVerifyPVC() {
 		OpenEBSNamespace,
 	)
 }
-func createDeployVerifyApp() {
+func createDeployVerifyApp(appName, pvcName string) {
 	ginkgo.By("creating and deploying app pod")
-	createAndDeployAppPod(appName, pvcName)
+	if pvcName == "zfspv-pvc-block" || pvcName == "pvc-name-for-del-test" {
+		createAndDeployBlockAppPod(appName, pvcName)
+	} else {
+		createAndDeployAppPod(appName, pvcName)
+	}
 	ginkgo.By("verifying app pod is running", func() { verifyAppPodRunning(appName) })
 }
 
-func createDeployVerifyCloneApp() {
+func createDeployVerifyCloneApp(cloneAppName, clonePvcName string) {
 	ginkgo.By("creating and deploying app pod")
 	createAndDeployAppPod(cloneAppName, clonePvcName)
 	ginkgo.By("verifying app pod is running", func() { verifyAppPodRunning(cloneAppName) })
@@ -545,7 +553,7 @@ func createAndDeployAppPod(appname string, pvcname string) {
 	)
 }
 
-func createAndDeployBlockAppPod() {
+func createAndDeployBlockAppPod(appName, pvcName string) {
 	var err error
 	labels := map[string]string{
 		"app":     "busybox",
@@ -585,7 +593,7 @@ func createAndDeployBlockAppPod() {
 				WithVolumeBuilders(
 					k8svolume.NewBuilder().
 						WithName("datavol1").
-						WithPVCSource(pvcObj.Name),
+						WithPVCSource(pvcName),
 				).
 				WithTerminationGracePeriodSeconds(5),
 		).
@@ -602,11 +610,6 @@ func createAndDeployBlockAppPod() {
 	)
 }
 
-func createDeployVerifyBlockApp() {
-	ginkgo.By("creating and deploying app pod", createAndDeployBlockAppPod)
-	ginkgo.By("verifying app pod is running", func() { verifyAppPodRunning(appName) })
-}
-
 func verifyAppPodRunning(appname string) {
 	var err error
 	gomega.Eventually(func() bool {
@@ -643,6 +646,32 @@ func deletePVC(pvcname string) {
 	gomega.Expect(status).To(gomega.Equal(true), "while trying to get deleted pvc")
 }
 
+func deletePV(pvName string) {
+	err := PVClient.Delete(pvName, &metav1.DeleteOptions{})
+	gomega.Expect(err).To(
+		gomega.BeNil(),
+		"while deleting pv {%s} ",
+		pvName,
+	)
+	ginkgo.By("verifying deleted pv")
+	status := IsPVDeletedEventually(pvName)
+	gomega.Expect(status).To(gomega.Equal(true), "while trying to get deleted pv")
+}
+
+// DeleteZV deletes the zv
+func DeleteZV(zvName string) {
+	err := ZFSClient.WithNamespace(OpenEBSNamespace).Delete(zvName)
+	gomega.Expect(err).To(
+		gomega.BeNil(),
+		"while deleting zv {%s} in namespace {%s}",
+		zvName,
+		OpenEBSNamespace,
+	)
+	ginkgo.By("verifying deleted zv")
+	status := IsZVDeletedEventually(zvName)
+	gomega.Expect(status).To(gomega.Equal(true), "while trying to get deleted zv")
+}
+
 func getStoragClassParams() []map[string]string {
 	return []map[string]string{
 		{
@@ -713,3 +742,21 @@ func getStoragClassParams() []map[string]string {
 		},
 	}
 }
+
+// IsZVPresentConsistently checks if the zfs volume is present or not consistently
+func IsZVPresentConsistently(zvName string) bool {
+	return gomega.Consistently(func() bool {
+		volume, err := ZFSClient.WithNamespace(OpenEBSNamespace).Get(zvName, metav1.GetOptions{})
+		gomega.Expect(err).ShouldNot(gomega.HaveOccurred())
+		return volume.Name == zvName
+	},
+		60, 5). // Check consistency for 60 seconds, polling every 5 seconds
+		Should(gomega.BeTrue())
+}
+
+// getZVName return the zv name
+func getZVName(pvcName string) string {
+	pvcObj, _ = PVCClient.WithNamespace(OpenEBSNamespace).Get(pvcName, metav1.GetOptions{})
+	return pvcObj.Spec.VolumeName
+
+}