Adding test automation for reclaimspace disable operation

Signed-off-by: Parag Kamble <[email protected]>

ocs_ci/helpers/helpers.py

@@ -5582,3 +5582,117 @@ def apply_custom_taint_and_toleration(taint_label="xyz"):
             )
             for pod_obj in pod_list:
                 pod_obj.delete(wait=False)
+
+
+def get_reclaimspacecronjob_for_pvc(pvc_obj):
+    """
+    Retrieve the ReclaimSpaceCronJob object associated with a given PVC.
+
+    Args:
+        pvc_obj (object): PersistentVolumeClaim (PVC) object.
+
+    Returns:
+        object: OCP object representing the ReclaimSpaceCronJob associated with the PVC.
+
+    Raises:
+        ValueError: If the PVC does not have the required annotation for ReclaimSpaceCronJob.
+    """
+    # Reload PVC object if annotations are missing
+    if "annotations" not in pvc_obj.data["metadata"]:
+        pvc_obj.reload()
+
+    # Retrieve the CronJob name from annotations
+    cron_job_name = pvc_obj.data["metadata"]["annotations"].get(
+        "reclaimspace.csiaddons.openshift.io/cronjob"
+    )
+    if not cron_job_name:
+        logger.error(f"PVC '{pvc_obj.name}' lacks annotation for reclaimspace cronjob.")
+        raise ValueError("PVC has no annotation for reclaimspace cronjob")
+
+    logger.info(f"Found ReclaimSpaceCronJob '{cron_job_name}' for PVC '{pvc_obj.name}'")
+
+    # Create and return the CronJob object
+    return OCP(
+        kind=constants.RECLAIMSPACECRONJOB,
+        namespace=pvc_obj.namespace,
+        resource_name=cron_job_name,
+    )
+
+
+def change_reclaimspacecronjob_state_for_pvc(pvc_objs, suspend=True):
+    """
+    Enable or disable the ReclaimSpace operation for the PVC's ReclaimSpaceCronJob.
+
+    Args:
+        pvc_objs (list): List of PersistentVolumeClaim (PVC) objects.
+        suspend (bool): If True, disables ReclaimSpace; if False, enables ReclaimSpace.
+
+    Returns:
+        bool: True if the operation was successfully applied to all PVCs.
+    """
+    action = "Disabling" if suspend else "Enabling"
+
+    for pvc_obj in pvc_objs:
+        logger.info(f"{action} ReclaimSpace operation for PVC '{pvc_obj.name}'")
+
+        # Retrieve the associated CronJob object
+        cron_obj = get_reclaimspacecronjob_for_pvc(pvc_obj)
+
+        # Update the annotation state
+        state_value = "unmanaged" if suspend else "managed"
+        cron_obj.annotate(f"csiaddons.openshift.io/state={state_value}", overwrite=True)
+        logger.debug(
+            f"Annotation 'csiaddons.openshift.io/state' set to '{state_value}' for PVC '{pvc_obj.name}'"
+        )
+
+        # Patch the 'suspend' state in the CronJob spec
+        if suspend:
+            suspend_patch = '[{"op": "add", "path": "/spec/suspend", "value": true}]'
+            logger.info(
+                f"'suspend' set to True in ReclaimSpaceCronJob for PVC '{pvc_obj.name}'"
+            )
+        else:
+            suspend_patch = '[{"op": "remove", "path": "/spec/suspend"}]'
+            logger.info(
+                f"'suspend' removed from ReclaimSpaceCronJob for PVC '{pvc_obj.name}'"
+            )
+
+        cron_obj.patch(params=suspend_patch, format_type="json")
+
+    return True
+
+
+def verify_reclaimspacecronjob_suspend_state_for_pvc(pvc_obj):
+    """
+    Verify the suspend state of the ReclaimSpaceCronJob associated with the given PVC.
+
+    Args:
+        pvc_obj (object): PersistentVolumeClaim (PVC) object.
+
+    Returns:
+        bool: True if the suspend state is True and the state annotation is 'unmanaged', False otherwise.
+    """
+    # Retrieve the ReclaimSpaceCronJob object for the PVC
+    reclaimspace_cronjob = get_reclaimspacecronjob_for_pvc(pvc_obj)
+
+    # Extract and log the suspend state
+    suspend_state = reclaimspace_cronjob.data["spec"].get("suspend", False)
+    logger.info(
+        f"ReclaimSpaceCronJob suspend state for PVC '{pvc_obj.name}' is '{suspend_state}'"
+    )
+
+    # Extract and log the state annotation
+    state_annotation = reclaimspace_cronjob.data["metadata"]["annotations"].get(
+        "csiaddons.openshift.io/state"
+    )
+    logger.info(
+        f"Annotation 'csiaddons.openshift.io/state' is '{state_annotation}' for PVC '{pvc_obj.name}'"
+    )
+
+    # Verify the suspend state and annotation
+    if suspend_state and state_annotation == "unmanaged":
+        logger.info(f"ReclaimSpace operation is disabled for PVC '{pvc_obj.name}'")
+        return True
+
+    logger.info(f"ReclaimSpace operation is enabled for PVC '{pvc_obj.name}'")
+    return False

tests/functional/pv/space_reclaim/test_disable_reclaimspace_operation.py

@@ -0,0 +1,123 @@
+import logging
+import pytest
+from ocs_ci.framework.pytest_customization.marks import green_squad
+from ocs_ci.ocs import constants
+from ocs_ci.framework.testlib import tier1
+from ocs_ci.helpers.helpers import (
+    change_reclaimspacecronjob_state_for_pvc,
+    get_rbd_image_info,
+    create_pods,
+    verify_reclaimspacecronjob_suspend_state_for_pvc,
+)
+from ocs_ci.ocs.exceptions import UnexpectedBehaviour
+from ocs_ci.utility.retry import retry
+from ocs_ci.ocs.resources.pod import delete_pods
+
+log = logging.getLogger(__name__)
+
+
+@green_squad
+@tier1
+class TestDisableReclaimSpaceOperation:
+    @pytest.fixture(autouse=True)
+    def setup(self, storageclass_factory, multi_pvc_factory):
+        """Setup the test environment by creating StorageClass and PVCs."""
+        reclaimspace_annotations = {
+            "reclaimspace.csiaddons.openshift.io/schedule": "* * * * *"
+        }
+        self.sc_obj = storageclass_factory(
+            interface=constants.CEPHBLOCKPOOL, annotations=reclaimspace_annotations
+        )
+        self.pvc_objs = multi_pvc_factory(
+            size=5,
+            num_of_pvc=3,
+            storageclass=self.sc_obj,
+            access_modes=[
+                f"{constants.ACCESS_MODE_RWX}-Block",
+                f"{constants.ACCESS_MODE_RWO}-Block",
+            ],
+            wait_each=True,
+        )
+
+    @retry(UnexpectedBehaviour, tries=3, delay=10)
+    def wait_till_expected_image_size(self, pvc_obj, expected_size):
+        """Wait until the RBD image size matches the expected size."""
+        rbd_image_name = pvc_obj.get_rbd_image_name
+        image_info = get_rbd_image_info(constants.DEFAULT_CEPHBLOCKPOOL, rbd_image_name)
+        image_size = image_info.get("used_size_gib")
+        if image_size != expected_size:
+            raise UnexpectedBehaviour(
+                f"RBD image {rbd_image_name} size mismatch: {image_size}GiB, expected {expected_size}GiB"
+            )
+        log.info(f"RBD Image {rbd_image_name} is size of {image_size}GiB")
+        return True
+
+    def execute_reclaimspace_test(self, pod_factory, suspend_state):
+        """Test reclaim space operation for PVCs with pods."""
+        # Create and attach pods to PVCs
+        pod_objs = create_pods(
+            self.pvc_objs,
+            pod_factory,
+            constants.CEPHBLOCKPOOL,
+            pods_for_rwx=1,
+            status=constants.STATUS_RUNNING,
+        )
+
+        # Write data to block devices
+        actual_data_written = 1.0  # 1 GiB
+        for pod_obj in pod_objs:
+            storage_path = pod_obj.get_storage_path("block")
+            log.info(f"Writing {actual_data_written}GiB of data to the block device")
+            pod_obj.exec_cmd_on_pod(
+                f"dd if=/dev/zero of={storage_path} bs=1M count=1024 oflag=direct > /dev/null 2>&1 &",
+                shell=True,
+            )
+
+        # Validate RBD image sizes after data writes
+        for pvc_obj in self.pvc_objs:
+            self.wait_till_expected_image_size(pvc_obj, actual_data_written)
+
+        # Delete pods
+        delete_pods(pod_objs, wait=True)
+
+        # Verify RBD image sizes after pods are deleted
+        expected_volume_size = actual_data_written if suspend_state else 0.0
+        for pvc_obj in self.pvc_objs:
+            self.wait_till_expected_image_size(pvc_obj, expected_volume_size)
+
+    @pytest.mark.polarion_id("OCS-6279")
+    def test_disable_reclaimspace_operation(self, pod_factory):
+        """Test to verify disabling and enabling reclaim space operation.
+
+        Steps:
+            1. Create a RBD PVC with diferent access modes (RWO, RWX)
+            2. Run a pod and attach a PVC to the pod.
+            3. Disable reclaimspace for all PVC by editing reclaimspacecronjob.
+            4. Verify ReclaimSpace Operation is disabled for the PVC.
+            5. re-enable reclaimspace operation for the PVC.
+            6. Verify reclaimspace Operation is enabled for the PVC.
+        """
+
+        log.info("Disabling reclaim space operation for all PVCs.")
+        change_reclaimspacecronjob_state_for_pvc(self.pvc_objs, suspend=True)
+
+        log.info("Verifying ReclaimSpaceCronJob suspend state (suspend=true).")
+        for pvc_obj in self.pvc_objs:
+            assert verify_reclaimspacecronjob_suspend_state_for_pvc(
+                pvc_obj
+            ), f"Reclaimspace cronjob is not suspended for PVC: {pvc_obj.name}"
+
+        log.info("Validating ReclaimSpace operation is disabled.")
+        self.execute_reclaimspace_test(pod_factory, suspend_state=True)
+
+        log.info("Re-enabling reclaim space cronjob for all PVCs.")
+        change_reclaimspacecronjob_state_for_pvc(self.pvc_objs, suspend=False)
+
+        log.info("Verifying ReclaimSpaceCronJob suspend state (suspend=false).")
+        for pvc_obj in self.pvc_objs:
+            assert not verify_reclaimspacecronjob_suspend_state_for_pvc(
+                pvc_obj
+            ), f"Reclaimspace cronjob is still suspended for PVC: {pvc_obj.name}"
+
+        log.info("Validating ReclaimSpace operation is enabled.")
+        self.execute_reclaimspace_test(pod_factory, suspend_state=False)