[WIP] Adding test automation for reclaimspace disable operation

ocs_ci/helpers/helpers.py

@@ -5392,3 +5392,83 @@ def verify_performance_profile_change(perf_profile):
     ), f"Performance profile is not updated successfully to {perf_profile}"
     logger.info(f"Performance profile successfully got updated to {perf_profile} mode")
     return True
+
+
+def get_reclaimspacecronjob_for_pvc(pvc_obj):
+    """
+    Retrieve the ReclaimSpaceCronJob object for a given PVC.
+
+    Args:
+        pvc_obj (object): PersistentVolumeClaim (PVC) object.
+
+    Returns:
+        object: The ReclaimSpaceCronJob object associated with the PVC.
+
+    Raises:
+        ValueError: If the PVC lacks the required annotation for ReclaimSpaceCronJob.
+    """
+    # Extract CronJob name from PVC annotations
+    if "annotations" not in pvc_obj.data["metadata"]:
+        pvc_obj.reload()
+
+    cron_job_name = pvc_obj.data["metadata"]["annotations"].get(
+        "reclaimspace.csiaddons.openshift.io/cronjob"
+    )
+
+    if not cron_job_name:
+        logger.error("PVC has no 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}'")
+    pvc_namespace = pvc_obj.namespace
+
+    # Create and return the CronJob object
+    cron_obj = OCP(
+        kind=constants.RECLAIMSPACECRONJOB,
+        namespace=pvc_namespace,
+        resource_name=cron_job_name,
+    )
+    logger.debug(
+        f"ReclaimSpaceCronJob object created for '{cron_job_name}' in namespace '{pvc_namespace}'"
+    )
+
+    return cron_obj
+
+
+def change_reclaimspacecronjob_state_for_pvc(pvc_obj, enable=True):
+    """
+    Enable or disable the ReclaimSpace operation for a given PVC's ReclaimSpaceCronJob.
+
+    Args:
+        pvc_obj (object): PersistentVolumeClaim (PVC) object.
+        enable (bool): True to enable, False to disable the ReclaimSpace operation.
+
+    Returns:
+        bool: True if the operation was successfully set.
+    """
+    action = "Enabling" if enable else "Disabling"
+    logger.info(f"{action} ReclaimSpace operation for PVC '{pvc_obj.name}'")
+
+    # Retrieve CronJob object for the PVC
+    cron_obj = get_reclaimspacecronjob_for_pvc(pvc_obj)
+
+    # Set annotation state based on enable parameter
+    state_value = "managed" if enable else "unmanaged"
+    annotate_state = f"csiaddons.openshift.io/state={state_value}"
+
+    cron_obj.annotate(annotate_state, overwrite=True)
+    logger.debug(
+        f"Updated annotation 'csiaddons.openshift.io/state' to '{state_value}'"
+    )
+
+    # Add or remove 'suspend' based on enable parameter
+    if enable:
+        suspend_patch = '[{"op": "remove", "path": "/spec/suspend"}]'
+        logger.info("Removed 'suspend' from ReclaimSpaceCronJob spec")
+    else:
+        suspend_patch = '[{"op": "add", "path": "/spec/suspend", "value": true}]'
+        logger.info("Set 'suspend' to True in ReclaimSpaceCronJob spec")
+
+    cron_obj.patch(params=suspend_patch, format_type="json")
+
+    return True

tests/functional/pv/space_reclaim/test_disable_reclaimspace_operation.py

@@ -0,0 +1,76 @@
+import logging
+
+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_reclaimspacecronjob_for_pvc,
+)
+
+
+log = logging.getLogger(__name__)
+
+
+@green_squad
+@tier1
+class TestDisableReclaimSpaceOperation:
+    def test_disable_reclaimspace_operation(self, multi_pvc_factory, pod_factory):
+        """Test to verify reclaim space Disable operation for the PVC.
+
+        Steps:
+            1.	Create a PVC from the CephBlockPool and attach it to a pod.
+            2.	To disable the reclaimspace operation for the PVC, set suspend = true and change the state annotation
+                to unmanaged.
+            3.	Verify that the reclaimspace operation is disabled for the PVC by checking that suspend = true is set.
+            4.	Re-enable the reclaimspace cronjob by removing suspend = true and changing the state annotation back
+                to managed for each PVC.
+            5.	Verify that all reclaimspace cronjobs are enabled.
+
+        """
+
+        # Create a PVC's with volume block mode
+        pvc_objs = multi_pvc_factory(
+            size=5,
+            num_of_pvc=3,
+            access_modes=[f"{constants.ACCESS_MODE_RWO}-Block"],
+            wait_each=True,
+        )
+
+        # Create POds
+        self.pod_objs = []
+
+        # Create pods
+        for pvc in pvc_objs:
+            pod_obj = pod_factory(
+                pvc=pvc,
+                status=constants.STATUS_RUNNING,
+                raw_block_pv=True,
+            )
+            self.pod_objs.append(pod_obj)
+
+        # Disable Reclaimspace operation for the PVC
+        log.info("Changing reclaimspace cronjob status to disable for all PVC object.")
+        for pvc in pvc_objs:
+            change_reclaimspacecronjob_state_for_pvc(pvc, enable=False)
+
+        # Verify Reclaimnspace cronjob is set as disable.
+        for pvc in pvc_objs:
+            reclaimspace_cronjob = get_reclaimspacecronjob_for_pvc(pvc)
+            suspend_state = reclaimspace_cronjob.data["spec"].get("suspend", False)
+
+            assert suspend_state, "reclaimspacecronjob state is not in suspened state"
+
+        # ReEnable the reclaimspace Operation for PVC
+        log.info("Changing reclaimspace cronjob status to Enable for all PVC object.")
+        for pvc in pvc_objs:
+            change_reclaimspacecronjob_state_for_pvc(pvc, enable=True)
+
+        # Verify Reclaimnspace cronjob is enable for the PVC
+        for pvc in pvc_objs:
+            reclaimspace_cronjob = get_reclaimspacecronjob_for_pvc(pvc)
+            suspend_state = reclaimspace_cronjob.data["spec"].get("suspend", False)
+
+            assert (
+                not suspend_state
+            ), "reclaimspace cronjob state is still in suspened state"