Add e2e test case for DeleteVolume success on backend volume deletion

If a backend volume is deleted, DeleteVolume call for the same should
succeed, detecting the image is missing and delete the related OMaps.

This commit adds a test case to ensure this is occuring correctly.

Updates #474

Signed-off-by: ShyamsundarR <srangana@redhat.com>
(cherry picked from commit b93ed21fe8)

e2e/cephfs.go

@@ -165,6 +165,30 @@ var _ = Describe("cephfs", func() {
 						Fail(err.Error())
 					}
 				})
+
+				By("creating a PVC, deleting backing subvolume, and checking successful PV deletion", func() {
+					pvc, err := loadPVC(pvcPath)
+					if pvc == nil {
+						Fail(err.Error())
+					}
+					pvc.Namespace = f.UniqueName
+
+					err = createPVCAndvalidatePV(f.ClientSet, pvc, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+
+					err = deleteBackingCephFSVolume(f, pvc)
+					if err != nil {
+						Fail(err.Error())
+					}
+
+					err = deletePVCAndValidatePV(f.ClientSet, pvc, deployTimeout)
+					if err != nil {
+						Fail(err.Error())
+					}
+				})
+
 				By("Resize PVC and check application directory size", func() {
 					v, err := f.ClientSet.Discovery().ServerVersion()
 					if err != nil {

e2e/utils.go

@@ -557,7 +557,7 @@ func validatePVCAndAppBinding(pvcPath, appPath string, f *framework.Framework) {
 	}
 }
 
-func getRBDImageSpec(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
+func getImageIDFromPVC(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
 	c := f.ClientSet.CoreV1()
 	pvc, err := c.PersistentVolumeClaims(pvcNamespace).Get(pvcName, metav1.GetOptions{})
 	if err != nil {
@@ -571,9 +571,27 @@ func getRBDImageSpec(pvcNamespace, pvcName string, f *framework.Framework) (stri
 
 	imageIDRegex := regexp.MustCompile(`(\w+\-?){5}$`)
 	imageID := imageIDRegex.FindString(pv.Spec.CSI.VolumeHandle)
+
+	return imageID, nil
+}
+func getRBDImageSpec(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
+	imageID, err := getImageIDFromPVC(pvcNamespace, pvcName, f)
+	if err != nil {
+		return "", err
+	}
+
 	return fmt.Sprintf("replicapool/csi-vol-%s", imageID), nil
 }
 
+func getCephFSVolumeName(pvcNamespace, pvcName string, f *framework.Framework) (string, error) {
+	imageID, err := getImageIDFromPVC(pvcNamespace, pvcName, f)
+	if err != nil {
+		return "", err
+	}
+
+	return fmt.Sprintf("csi-vol-%s", imageID), nil
+}
+
 func getImageMeta(rbdImageSpec, metaKey string, f *framework.Framework) (string, error) {
 	cmd := fmt.Sprintf("rbd image-meta get %s %s", rbdImageSpec, metaKey)
 	opt := metav1.ListOptions{
@@ -775,6 +793,24 @@ func validateNormalUserPVCAccess(pvcPath string, f *framework.Framework) {
 // 	})
 // }
 
+func deleteBackingCephFSVolume(f *framework.Framework, pvc *v1.PersistentVolumeClaim) error {
+	volname, err := getCephFSVolumeName(pvc.Namespace, pvc.Name, f)
+	if err != nil {
+		return err
+	}
+
+	opt := metav1.ListOptions{
+		LabelSelector: "app=rook-ceph-tools",
+	}
+	_, stdErr := execCommandInPod(f, "ceph fs subvolume rm myfs "+volname+" csi", rookNS, &opt)
+	Expect(stdErr).Should(BeEmpty())
+
+	if stdErr != "" {
+		return fmt.Errorf("error deleting backing volume %s", volname)
+	}
+	return nil
+}
+
 func listRBDImages(f *framework.Framework) []string {
 	opt := metav1.ListOptions{
 		LabelSelector: "app=rook-ceph-tools",