Test Automation for encryption dashboard summary

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

Removed constructor from the class ObjectDetails.

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

Fixed black Issue

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

Created a encryption UI maping variables in the constnts.py file

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

[4.18] Test bucket replication with versioning - utility functions and happy path validation (#10957)

Signed-off-by: Sagi Hirshfeld <[email protected]>

Fix AMQ setup by cloning the latest compatible branch of strimzi-kafka-operator (#11082)

Signed-off-by: Sagi Hirshfeld <[email protected]>

Created a EncryptionModule class and moved the encryption summary check functionality in same class

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

ocs_ci/ocs/amq.py

@@ -79,7 +79,13 @@ def _clone_amq(self):
         """
         try:
             log.info(f"cloning amq in {self.dir}")
-            git_clone_cmd = f"git clone {self.repo} "
+
+            # Further setup in this class assumes the use of zookeeper, before
+            # the following change was made to strimzi-kafka-operator:
+            # https://github.com/strimzi/strimzi-kafka-operator/pull/10982
+            last_compatible_branch = "release-0.44.x"
+            git_clone_cmd = f"git clone --branch {last_compatible_branch} {self.repo} "
+
             run(git_clone_cmd, shell=True, cwd=self.dir, check=True)
             self.amq_dir = "strimzi-kafka-operator/packaging/install/cluster-operator/"
             self.amq_kafka_pers_yaml = (

ocs_ci/ocs/bucket_utils.py

@@ -2921,3 +2921,114 @@ def create_s3client_from_assume_role_creds(mcg_obj, assume_role_creds):
         aws_session_token=assumed_session_token,
     )
     return assumed_s3_resource.meta.client
+
+
+def put_bucket_versioning_via_awscli(
+    mcg_obj, awscli_pod, bucket_name, status="Enabled"
+):
+    """
+    Put bucket versioning using AWS CLI
+
+    Args:
+        mcg_obj (MCG): MCG object
+        awscli_pod (Pod): Pod object where AWS CLI is installed
+        bucket_name (str): Name of the bucket
+        status (str): Status of the versioning
+
+    """
+    awscli_pod.exec_cmd_on_pod(
+        command=craft_s3_command(
+            f"put-bucket-versioning --bucket {bucket_name} --versioning-configuration Status={status}",
+            mcg_obj=mcg_obj,
+            api=True,
+        )
+    )
+
+
+def upload_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key, amount=1, size="1M"):
+    """
+    Upload multiple random data versions to a given object key and return their ETag values
+
+    Args:
+        mcg_obj (MCG): MCG object
+        awscli_pod (Pod): Pod object where AWS CLI is installed
+        bucket_name (str): Name of the bucket
+        obj_key (str): Object key
+        amount (int): Number of versions to create
+        size (str): Size of the object. I.E 1M
+
+    Returns:
+        list: List of ETag values of versions in latest to oldest order
+    """
+    file_dir = os.path.join("/tmp", str(uuid4()))
+    awscli_pod.exec_cmd_on_pod(f"mkdir {file_dir}")
+
+    etags = []
+
+    for i in range(amount):
+        file_path = os.path.join(file_dir, f"{obj_key}_{i}")
+        awscli_pod.exec_cmd_on_pod(
+            command=f"dd if=/dev/urandom of={file_path} bs={size} count=1"
+        )
+        # Use debug and redirect it to stdout to get
+        # the uploaded object's ETag in the response
+        resp = awscli_pod.exec_cmd_on_pod(
+            command=craft_s3_command(
+                f"cp {file_path} s3://{bucket_name}/{obj_key} --debug 2>&1",
+                mcg_obj=mcg_obj,
+            ),
+            out_yaml_format=False,
+        )
+
+        # Parse the ETag from the response
+        # Filter the line containing the JSON
+        line = next(filter(lambda line: "ETag" in line, resp.splitlines()))
+        json_start = line.index("{")
+        json_str = line[json_start:]
+
+        # Fix quotes to read as dict
+        json_str = json_str.replace("'", '"')
+        json_str = json_str.replace('""', '"')
+
+        # Convert to dict and extract the ETag
+        new_etag = json.loads(json_str).get("ETag")
+
+        # Later versions should precede the older ones
+        # this achieves the expected order of versions
+        # we'd get from list-object-versions
+        etags.insert(0, new_etag)
+
+    return etags
+
+
+def get_obj_versions(mcg_obj, awscli_pod, bucket_name, obj_key):
+    """
+    Get object versions using AWS CLI
+
+    Args:
+        mcg_obj (MCG): MCG object
+        awscli_pod (Pod): Pod object where AWS CLI is installed
+        bucket_name (str): Name of the bucket
+        obj_key (str): Object key
+
+    Returns:
+        list: List of dictionaries containing the versions data
+    """
+    resp = awscli_pod.exec_cmd_on_pod(
+        command=craft_s3_command(
+            f"list-object-versions --bucket {bucket_name} --prefix {obj_key}",
+            mcg_obj=mcg_obj,
+            api=True,
+        ),
+        out_yaml_format=False,
+    )
+
+    versions_dicts = []
+    if resp and "Versions" in resp:
+        versions_dicts = json.loads(resp).get("Versions")
+
+        # Remove quotes from the ETag values for easier usage
+        for d in versions_dicts:
+            d["ETag"] = d["ETag"].strip('"')
+
+    return versions_dicts

ocs_ci/ocs/constants.py

@@ -3117,3 +3117,11 @@
 MACHINE_POOL_ACTIONS = [CREATE, EDIT, DELETE]
 # MDR multicluster roles
 MDR_ROLES = ["ActiveACM", "PassiveACM", "PrimaryODF", "SecondaryODF"]
+
+ENCRYPTION_DASHBOARD_CONTEXT_MAP = {
+    "Cluster-wide encryption": "cluster_wide_encryption",
+    "Storage class encryption": "storageclass_encryption",
+    "In-transit encryption": "intransit_encryption",
+    "Block storage": "block_storage",
+    "Object storage": "object_storage",
+}

ocs_ci/ocs/resources/mcg_replication_policy.py

@@ -99,3 +99,20 @@ def to_dict(self):
         dict["log_replication_info"]["endpoint_type"] = "AZURE"
 
         return dict
+
+
+class ReplicationPolicyWithVersioning(McgReplicationPolicy):
+    """
+    A class to handle the MCG bucket replication policy JSON structure with versioning.
+
+    """
+
+    def __init__(self, target_bucket, sync_versions=True, prefix=""):
+        super().__init__(target_bucket, prefix)
+        self.sync_versions = sync_versions
+
+    def to_dict(self):
+        dict = super().to_dict()
+        dict["rules"][0]["sync_versions"] = self.sync_versions
+
+        return dict

ocs_ci/ocs/ui/helpers_ui.py

@@ -8,6 +8,7 @@
 from ocs_ci.ocs.ui.base_ui import login_ui, close_browser
 from ocs_ci.ocs.ui.add_replace_device_ui import AddReplaceDeviceUI
 from ocs_ci.ocs.resources.storage_cluster import get_deviceset_count, get_osd_size
+from ocs_ci.ocs.exceptions import ResourceNotFoundError
 
 logger = logging.getLogger(__name__)
 
@@ -217,3 +218,27 @@ def is_ui_deployment():
         return True
 
     return False
+
+
+def extract_encryption_status(root_element, svg_path):
+    """Function to extract encryption status from an SVG element
+
+    Args:
+        root_element (str): Dom root element
+        svg_element (str): svg element path
+
+    Returns:
+        bool: if encryption status is enable for given element return True otherwise False.
+
+    Raises:
+        ResourceNotFoundError: If given resource is not found.
+    """
+    try:
+        svg_element = root_element.find_element(By.CSS_SELECTOR, svg_path)
+        if svg_element and svg_element.tag_name == "svg":
+            if svg_element.get_attribute("data-test") == "success-icon":
+                return True
+            else:
+                return False
+    except Exception as e:
+        raise ResourceNotFoundError(f"Given SVG element is not Found: {e}")

ocs_ci/ocs/ui/page_objects/encryption_module.py

@@ -0,0 +1,138 @@
+from ocs_ci.ocs.ui.helpers_ui import logger, extract_encryption_status
+from ocs_ci.ocs.constants import ENCRYPTION_DASHBOARD_CONTEXT_MAP
+from ocs_ci.ocs.ui.page_objects.page_navigator import PageNavigator
+
+
+class EncryptionModule(PageNavigator):
+    def _get_encryption_summary(self, context_key):
+        """
+        Generic method to collect encryption summary based on the context.
+
+        Args:
+            context_key (str): Key to determine the validation location.
+
+        Returns:
+            dict: Encryption summary for the given context.
+        """
+        encryption_summary = {
+            "object_storage": {"status": None, "kms": ""},
+            "cluster_wide_encryption": {"status": None, "kms": ""},
+            "storageclass_encryption": {"status": None, "kms": ""},
+            "intransit_encryption": {"status": None},
+        }
+
+        logger.info(f"Getting Encryption Summary for context: {context_key}")
+
+        # Open the encryption summary popup
+        self.do_click(
+            self.validation_loc["encryption_summary"][context_key]["enabled"],
+            enable_screenshot=True,
+        )
+
+        self.page_has_loaded(
+            module_loc=self.validation_loc["encryption_summary"][context_key][
+                "encryption_content_data"
+            ]
+        )
+
+        # Get elements for text and root
+        encryption_content_location = self.validation_loc["encryption_summary"][
+            context_key
+        ]["encryption_content_data"]
+        encryption_summary_text = self.get_element_text(encryption_content_location)
+        root_elements = self.get_elements(encryption_content_location)
+
+        if not root_elements:
+            raise ValueError("Error getting root web element")
+        root_element = root_elements[0]
+
+        # Process encryption summary text
+        current_context = None
+        for line in encryption_summary_text.split("\n"):
+            line = line.strip()
+            if line in ENCRYPTION_DASHBOARD_CONTEXT_MAP:
+                current_context = ENCRYPTION_DASHBOARD_CONTEXT_MAP[line]
+                continue
+
+            if (
+                current_context
+                in [
+                    "object_storage",
+                    "cluster_wide_encryption",
+                    "storageclass_encryption",
+                ]
+                and "External Key Management Service" in line
+            ):
+                encryption_summary[current_context]["kms"] = line.split(":")[-1].strip()
+                encryption_summary[current_context]["status"] = (
+                    extract_encryption_status(
+                        root_element,
+                        self._get_svg_selector(context_key, current_context),
+                    )
+                )
+            elif current_context == "intransit_encryption":
+                encryption_summary[current_context]["status"] = (
+                    extract_encryption_status(
+                        root_element,
+                        self._get_svg_selector(context_key, current_context),
+                    )
+                )
+
+        logger.info(f"Encryption Summary for {context_key}: {encryption_summary}")
+
+        # Close the popup
+        logger.info("Closing the popup")
+        self.do_click(
+            self.validation_loc["encryption_summary"][context_key]["close"],
+            enable_screenshot=True,
+        )
+
+        return encryption_summary
+
+    def _get_svg_selector(self, context_key, current_context):
+        """
+        Get the appropriate SVG selector for extracting encryption status.
+
+        Args:
+            context_key (str): The context key.
+            current_context (str): The current encryption context.
+
+        Returns:
+            str: SVG selector path.
+        """
+        selectors = {
+            "object_storage": {
+                "object_storage": "div.pf-v5-l-flex:nth-child(1) > div:nth-child(2) > svg",
+                "intransit_encryption": "div.pf-v5-l-flex:nth-child(4) > div:nth-child(2) > svg",
+            },
+            "file_and_block": {
+                "cluster_wide_encryption": (
+                    "div.pf-m-align-items-center:nth-child(1) > "
+                    "div:nth-child(2) > svg:nth-child(1)"
+                ),
+                "storageclass_encryption": (
+                    "div.pf-v5-l-flex:nth-child(6) > "
+                    "div:nth-child(2) > svg:nth-child(1)"
+                ),
+                "intransit_encryption": "div.pf-v5-l-flex:nth-child(10) > div:nth-child(2) > svg",
+            },
+        }
+        return selectors.get(context_key, {}).get(current_context, "")
+
+    def get_object_encryption_summary(self):
+        """
+        Retrieve the encryption summary for the object details page.
+
+        Returns:
+            dict: Encryption summary on object details page.
+        """
+        return self._get_encryption_summary("object_storage")
+
+    def get_block_file_encryption_summary(self):
+        """
+        Retrieve the encryption summary for the block and file page.
+
+        Returns:
+            dict: Encryption summary on block and file page.
+        """
+        return self._get_encryption_summary("file_and_block")

ocs_ci/ocs/ui/page_objects/storage_system_details.py

@@ -2,11 +2,13 @@
 from ocs_ci.ocs.ui.base_ui import logger, BaseUI
 from ocs_ci.ocs.ui.page_objects.storage_system_tab import StorageSystemTab
 from ocs_ci.utility import version
+from ocs_ci.ocs.ui.page_objects.encryption_module import EncryptionModule
 
 
-class StorageSystemDetails(StorageSystemTab):
+class StorageSystemDetails(StorageSystemTab, EncryptionModule):
     def __init__(self):
         StorageSystemTab.__init__(self)
+        EncryptionModule.__init__(self)
 
     def nav_details_overview(self):
         logger.info("Click on Overview tab")
@@ -33,6 +35,8 @@ def nav_details_object(self):
         else:
             self.do_click(self.validation_loc["object"], enable_screenshot=True)
 
+        return self
+
     def nav_block_and_file(self):
         """
         Accessible only at StorageSystems / StorageSystem details / Overview