Enhance StorageConsumer cr and class (#11621)

* Enhance StorageConsumer cr and class
* if version deco added

Signed-off-by: Daniel Osypenko <dosypenk@redhat.com>

Author: DanielOsypenko

ocs_ci/ocs/resources/storageconsumer.py

@@ -7,6 +7,7 @@
 from ocs_ci.framework import config
 from ocs_ci.ocs import constants, ocp
 from ocs_ci.ocs.resources.ocs import OCS
+from ocs_ci.ocs.version import if_version
 from ocs_ci.utility.utils import exec_cmd
 
 log = logging.getLogger(__name__)
@@ -19,6 +20,18 @@ class StorageConsumer:
 
     def __init__(self, consumer_name, consumer_context=None):
         """
+        Starting from ODF 4.19 (Converged) this CR has optional Spec fields:
+        StorageQuotaInGiB          int
+        StorageClasses             []
+        VolumeSnapshotClasses      []
+        VolumeGroupSnapshotClasses []
+
+        Starting from ODF 4.19 (Converged) this CR has optional Status fields:
+        Client                ClientStatus
+        RadosNamespace        RadosNamespaceStatus
+        CephFsSubVolumeGroup  SubVolumeGroupStatus
+        CephCsiClientProfiles [] ClientProfileStatus
+
         Args:
             consumer_name (string): name of the StorageConsumer resource
             consumer_context (int): index of cluster context. This is needed for
@@ -120,3 +133,205 @@ def get_heartbeat_cronjob(self):
                 if job["metadata"]["name"].endswith("status-reporter")
             ][0]
         return cronjob
+
+    @if_version(">4.18")
+    def get_client_status(self):
+        """
+        Get client status from storageconsumer resource and apply patch.
+
+        Returns:
+            dict: client status
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return self.ocp.get(resource_name=self.name).get("status").get("client")
+
+    @if_version(">4.18")
+    def get_rados_namespace_status(self):
+        """
+        Get rados namespace status from storageconsumer resource and apply patch.
+
+        Returns:
+            dict: rados namespace status
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return (
+                self.ocp.get(resource_name=self.name)
+                .get("status")
+                .get("radosNamespace")
+            )
+
+    @if_version(">4.18")
+    def get_cephfs_subvolume_group_status(self):
+        """
+        Get cephfs subvolume group status from storageconsumer resource and apply patch.
+
+        Returns:
+            dict: cephfs subvolume group status
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return (
+                self.ocp.get(resource_name=self.name)
+                .get("status")
+                .get("cephFsSubVolumeGroup")
+            )
+
+    @if_version(">4.18")
+    def get_ceph_csi_client_profiles(self):
+        """
+        Get ceph csi client profiles from storageconsumer resource and apply patch.
+
+        Returns:
+            dict: ceph csi client profiles
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return (
+                self.ocp.get(resource_name=self.name)
+                .get("status")
+                .get("cephCsiClientProfiles")
+            )
+
+    def get_storage_quota_in_gib(self):
+        """
+        Get storage quota in GiB from storageconsumer resource.
+
+        Returns:
+            int: storage quota in GiB
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return (
+                self.ocp.get(resource_name=self.name)
+                .get("spec")
+                .get("storageQuotaInGiB")
+            )
+
+    @if_version(">4.18")
+    def get_storage_classes(self):
+        """
+        Get storage classes from storageconsumer resource and apply patch.
+
+        Returns:
+            list: storage classes
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            return (
+                self.ocp.get(resource_name=self.name).get("spec").get("storageClasses")
+            )
+
+    def set_storage_quota_in_gib(self, quota):
+        """
+        Update storage quota in GiB in storageconsumer resource and apply patch.
+
+        Args:
+            quota (int): storage quota in GiB
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            patch_param = f'{{"spec": {{"storageQuotaInGiB": {quota}}}}}'
+            self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def add_custom_storage_class(self, storage_class):
+        """
+        Add storage class to storageconsumer resource and apply patch.
+
+        Args:
+            storage_class (string): storage class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            patch_param = f'{{"spec": {{"storageClasses": ["{storage_class}"]}}}}'
+            self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def remove_custom_storage_class(self, storage_class):
+        """
+        Remove storage class from storageconsumer resource and apply patch.
+
+        Args:
+            storage_class (string): storage class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            current_storage_classes = (
+                self.ocp.get(resource_name=self.name).get("spec").get("storageClasses")
+            )
+            if storage_class in current_storage_classes:
+                current_storage_classes.remove(storage_class)
+                patch_param = (
+                    f'{{"spec": {{"storageClasses": {current_storage_classes}}}}}'
+                )
+                self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def add_custom_volume_snapshot_class(self, snapshot_class):
+        """
+        Add volume snapshot class to storageconsumer resource and apply patch.
+
+        Args:
+            snapshot_class (string): volume snapshot class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            patch_param = (
+                f'{{"spec": {{"volumeSnapshotClasses": ["{snapshot_class}"]}}}}'
+            )
+            self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def remove_custom_volume_snapshot_class(self, snapshot_class):
+        """
+        Remove volume snapshot class from storageconsumer resource and apply patch.
+
+        Args:
+            snapshot_class (string): volume snapshot class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            current_snapshot_classes = (
+                self.ocp.get(resource_name=self.name)
+                .get("spec")
+                .get("volumeSnapshotClasses")
+            )
+            if snapshot_class in current_snapshot_classes:
+                current_snapshot_classes.remove(snapshot_class)
+                patch_param = f'{{"spec": {{"volumeSnapshotClasses": {current_snapshot_classes}}}}}'
+                self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def add_custom_volume_group_snapshot_class(self, group_snapshot_class):
+        """
+        Add volume group snapshot class to storageconsumer resource  and apply patch.
+
+        Args:
+            group_snapshot_class (string): volume group snapshot class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            patch_param = f'{{"spec": {{"volumeGroupSnapshotClasses": ["{group_snapshot_class}"]}}}}'
+            self.ocp.patch(resource_name=self.name, params=patch_param)
+
+    @if_version(">4.18")
+    def remove_custom_volume_group_snapshot_class(self, group_snapshot_class):
+        """
+        Remove volume group snapshot class from storageconsumer resource and apply patch.
+
+        Args:
+            group_snapshot_class (string): volume group snapshot class
+
+        """
+        with config.RunWithConfigContext(self.consumer_context):
+            current_group_snapshot_classes = (
+                self.ocp.get(resource_name=self.name)
+                .get("spec")
+                .get("volumeGroupSnapshotClasses")
+            )
+            if group_snapshot_class in current_group_snapshot_classes:
+                current_group_snapshot_classes.remove(group_snapshot_class)
+                patch_param = f'{{"spec": {{"volumeGroupSnapshotClasses": {current_group_snapshot_classes}}}}}'
+                self.ocp.patch(resource_name=self.name, params=patch_param)

ocs_ci/ocs/version.py

@@ -22,14 +22,15 @@
 import pprint
 import re
 import sys
+from functools import wraps
 
 from ocs_ci import framework
 from ocs_ci.framework import config
 from ocs_ci.ocs import constants, node, ocp
 from ocs_ci.ocs.ocp import OCP
 from ocs_ci.ocs.exceptions import CommandFailed
 from ocs_ci.utility import utils
-
+from ocs_ci.utility.utils import skipif_ocs_version
 
 logger = logging.getLogger(__name__)
 
@@ -275,3 +276,37 @@ def main():
         return
 
     report_ocs_version(cluster_version, image_dict, file_obj=sys.stdout)
+
+
+def if_version(expressions):
+    """
+    Decorator to skip the function if the OCS version does not match the required version
+
+    ! Important note: This decorator is silently skipping the function
+    if the OCS version does not match the required version. Only warning message will be added to the log.
+    If the test should fail when wrapped function was skipped, use this decorator in combination with pytest 'assert'.
+
+    Args:
+        expressions (str OR list): condition for which we need to check, eg:
+            A single expression string '>=4.2' OR
+            A list of expressions like ['<4.3', '>4.2'], ['<=4.3', '>=4.2']
+
+    Returns:
+         result of executing the function if the OCS version matches the required version
+
+    """
+
+    def decorator(func):
+        @wraps(func)
+        def wrapper(*args, **kwargs):
+            if skipif_ocs_version(expressions):
+                return func(*args, **kwargs)
+            else:
+                logger.warning(
+                    f"Skipping {func.__name__} because the OCS version "
+                    f"does not match the required version {expressions}"
+                )
+
+        return wrapper
+
+    return decorator