vm snapshot test case

ocs_ci/helpers/cnv_helpers.py

@@ -176,7 +176,7 @@ def create_dv(
     namespace=constants.CNV_NAMESPACE,
 ):
     """
-    Create/Clones a DV using a specified data source
+    Creates a DV using a specified data source
 
     Args:
         access_mode (str): The access mode for the volume. Default is `constants.ACCESS_MODE_RWX`

ocs_ci/ocs/cnv/virtual_machine.py

@@ -24,6 +24,7 @@
 from ocs_ci.ocs.cnv.virtual_machine_instance import VirtualMachineInstance
 from ocs_ci.ocs import constants, ocp
 from ocs_ci.ocs.resources import pvc
+from ocs_ci.ocs.resources.pvc import PVC
 from ocs_ci.utility import templating
 from ocs_ci.utility.utils import TimeoutSampler
 from ocs_ci.ocs.exceptions import UsernameNotFoundException, CommandFailed
@@ -93,6 +94,7 @@ def create_vm_workload(
         access_mode=constants.ACCESS_MODE_RWX,
         pvc_size="30Gi",
         source_url=constants.CNV_CENTOS_SOURCE,
+        existing_pvc_obj=None,
         ssh=True,
         verify=True,
     ):
@@ -107,6 +109,7 @@ def create_vm_workload(
             sc_name (str): The name of the storage class to use. Default is `constants.DEFAULT_CNV_CEPH_RBD_SC`.
             pvc_size (str): The size of the PVC. Default is "30Gi".
             source_url (str): The URL of the vm registry image. Default is `constants.CNV_CENTOS_SOURCE`
+            existing_pvc_obj (obj, optional): PVC object to use existing pvc as a backend volume to VM
 
         """
         self.volume_interface = volume_interface
@@ -121,7 +124,13 @@ def create_vm_workload(
             self._add_ssh_key_to_vm(vm_data)
 
         if volume_interface == constants.VM_VOLUME_PVC:
-            self._create_vm_pvc(vm_data=vm_data)
+            if existing_pvc_obj:
+                vm_data["spec"]["template"]["spec"]["volumes"][0][
+                    "persistentVolumeClaim"
+                ] = {"claimName": existing_pvc_obj.name}
+                self.pvc_name = existing_pvc_obj.name
+            else:
+                self._create_vm_pvc(vm_data=vm_data)
         elif volume_interface == constants.VM_VOLUME_DV:
             self._create_vm_data_volume(vm_data=vm_data)
         elif volume_interface == constants.VM_VOLUME_DVT:
@@ -280,6 +289,20 @@ def get(self, out_yaml_format=True):
             resource_name=self._vm_name, out_yaml_format=out_yaml_format
         )
 
+    def get_vm_pvc_obj(self):
+        """
+        Retrieves VM PVC obj
+
+        Returns:
+            obj: PVC object
+
+        """
+        ocp_pvc_obj = OCP(kind=constants.PVC, namespace=self.namespace).get(
+            resource_name=self.pvc_name
+        )
+        pvc_obj = PVC(**ocp_pvc_obj)
+        return pvc_obj
+
     def get_os_username(self):
         """
         Retrieve the operating system username from the cloud-init data associated with the virtual machine
@@ -628,14 +651,21 @@ def delete(self):
         self.vm_ocp_obj.delete(resource_name=self._vm_name)
         self.vm_ocp_obj.wait_for_delete(resource_name=self._vm_name, timeout=180)
         if self.volume_interface == constants.VM_VOLUME_PVC:
-            self.pvc_obj.delete()
-            self.pvc_obj.ocp.wait_for_delete(
-                resource_name=self.pvc_obj.name, timeout=180
-            )
-            self.volumeimportsource_obj.delete()
+            # Deletes only when PVC & VIS obj exists
+            if self.pvc_obj:
+                self.pvc_obj.delete()
+                self.pvc_obj.ocp.wait_for_delete(
+                    resource_name=self.pvc_obj.name, timeout=180
+                )
+            if self.volumeimportsource_obj:
+                self.volumeimportsource_obj.delete()
         elif self.volume_interface == constants.VM_VOLUME_DV:
-            self.dv_obj.delete()
-            self.dv_obj.ocp.wait_for_delete(resource_name=self.dv_obj.name, timeout=180)
+            # Deletes only when DV obj exists
+            if self.dv_obj:
+                self.dv_obj.delete()
+                self.dv_obj.ocp.wait_for_delete(
+                    resource_name=self.dv_obj.name, timeout=180
+                )
         if self.ns_obj:
             self.ns_obj.delete_project(project_name=self.namespace)
 

tests/conftest.py

@@ -7056,6 +7056,7 @@ def factory(
         storageclass=constants.DEFAULT_CNV_CEPH_RBD_SC,
         pvc_size="30Gi",
         source_url=constants.CNV_CENTOS_SOURCE,
+        existing_pvc_obj=None,
         namespace=None,
     ):
         """
@@ -7065,6 +7066,7 @@ def factory(
             storageclass (str): The name of the storage class to use. Default is `constants.DEFAULT_CNV_CEPH_RBD_SC`.
             pvc_size (str): The size of the PVC. Default is "30Gi".
             source_url (str): The URL of the vm registry image. Default is `constants.CNV_CENTOS_SOURCE`.
+            existing_pvc_obj (obj, optional): PVC object to use existing pvc as a backend volume to VM
             namespace (str, optional): The namespace to create the vm on. Default, creates a unique namespace.
 
         Returns:
@@ -7079,6 +7081,7 @@ def factory(
             sc_name=storageclass,
             pvc_size=pvc_size,
             source_url=source_url,
+            existing_pvc_obj=existing_pvc_obj,
         )
         cnv_workloads.append(cnv_wl)
         return cnv_workloads
@@ -7088,7 +7091,8 @@ def teardown():
         Cleans up the CNV workloads
 
         """
-        for cnv_wl in cnv_workloads:
+        # Iterating from end so that restored VMs are deleted before source
+        for cnv_wl in cnv_workloads[::-1]:
             cnv_wl.delete()
 
     request.addfinalizer(teardown)

tests/functional/workloads/cnv/test_vm_cloning_ops.py → tests/functional/workloads/cnv/test_vm_snapshot_cloning_ops.py

@@ -62,3 +62,60 @@ def test_vm_clone(self, cnv_workload, clone_vm_workload, setup_cnv):
             ), f"Failed: MD5 comparison between source {vm_obj.name} and cloned {clone_obj.name} VMs"
             run_dd_io(vm_obj=clone_obj, file_path=file_paths[1])
             clone_obj.stop()
+
+    @workloads
+    @pytest.mark.polarion_id("OCS-6299")
+    def test_vm_snapshot_ops(
+        self, cnv_workload, snapshot_factory, snapshot_restore_factory, setup_cnv
+    ):
+        """
+        This test performs the VM PVC snapshot operations
+
+        Test steps:
+        1. Create VMs, add data(e.g., files) to all the VMs
+        2. Create a snapshot for a VM backed pvc
+        3. Restore the snapshot (to same access mode of the parent PVC and storage_class) by following the
+        documented procedure from ODF official docs
+        4. Create new vm using restored pvc Verify existing data of the VM are not changed.
+        5. Add further data(e.g., new file) to the VM
+        6. Repeat the above procedure for all the VMs in the system
+        7. Delete all the VMs created as part of this test
+        """
+        file_paths = ["/file.txt", "/new_file.txt"]
+        # TODO: Add multi_cnv fixture to configure VMs based on specifications
+        vm_obj = cnv_workload(
+            volume_interface=constants.VM_VOLUME_PVC,
+            source_url=constants.CNV_FEDORA_SOURCE,
+        )[0]
+        # Writing IO on source VM
+        source_csum = run_dd_io(vm_obj=vm_obj, file_path=file_paths[0], verify=True)
+        # Stopping VM before taking snapshot of the VM PVC
+        vm_obj.stop()
+        # Taking Snapshot of PVC
+        pvc_obj = vm_obj.get_vm_pvc_obj()
+        snap_obj = snapshot_factory(pvc_obj)
+        # Restore the snapshot
+        res_snap_obj = snapshot_restore_factory(
+            snapshot_obj=snap_obj,
+            storageclass=vm_obj.sc_name,
+            size=vm_obj.pvc_size,
+            volume_mode=snap_obj.parent_volume_mode,
+            access_mode=vm_obj.pvc_access_mode,
+            status=constants.STATUS_BOUND,
+            timeout=300,
+        )
+        # Create new VM using the restored PVC
+        res_vm_obj = cnv_workload(
+            volume_interface=constants.VM_VOLUME_PVC,
+            source_url=constants.CNV_FEDORA_SOURCE,
+            existing_pvc_obj=res_snap_obj,
+            namespace=vm_obj.namespace,
+        )[1]
+        # Write new file to VM
+        run_dd_io(vm_obj=res_vm_obj, file_path=file_paths[1], verify=True)
+        # Validate data integrity of file written before taking snapshot
+        res_csum = cal_md5sum_vm(vm_obj=res_vm_obj, file_path=file_paths[0])
+        assert (
+            source_csum == res_csum
+        ), f"Failed: MD5 comparison between source {vm_obj.name} and cloned {res_vm_obj.name} VMs"
+        res_vm_obj.stop()