Updated in-transit encryption test automation for enable/disable operations.

tests/functional/encryption/test_intransit_encryption_sanity.py

@@ -1,74 +1,138 @@
 import logging
 import pytest
-
+import time
 from ocs_ci.ocs.resources.storage_cluster import (
-    in_transit_encryption_verification,
     set_in_transit_encryption,
     get_in_transit_encryption_config_state,
 )
 from ocs_ci.framework.pytest_customization.marks import (
     tier1,
     skipif_ocs_version,
     green_squad,
-    cloud_platform_required,
 )
 from ocs_ci.framework import config
+from ocs_ci.ocs import constants
+from ocs_ci.helpers.helpers import create_pods
+from concurrent.futures import ThreadPoolExecutor
 
 log = logging.getLogger(__name__)
 
 
 @green_squad
-@cloud_platform_required
+@skipif_ocs_version("<4.18")
 class TestInTransitEncryptionSanity:
     @pytest.fixture(autouse=True)
     def set_encryption_at_teardown(self, request):
+        """
+        Fixture to restore encryption state and clean up resources after the test.
+        """
+
         def teardown():
-            if config.ENV_DATA.get("in_transit_encryption"):
-                set_in_transit_encryption()
-            else:
-                set_in_transit_encryption(enabled=False)
+            initial_state = config.ENV_DATA.get("in_transit_encryption", False)
+            set_in_transit_encryption(enabled=initial_state)
+
+            # Deleting pods if any exist
+            for pod_obj in getattr(self, "all_pods", []):
+                pod_obj.delete()
 
         request.addfinalizer(teardown)
 
+    def toggle_intransit_encryption_state(self):
+        """
+        Toggles the in-transit encryption state on the cluster.
+        """
+        current_state = get_in_transit_encryption_config_state()
+        new_state = not current_state
+
+        log.info(
+            f"Toggling in-transit encryption from "
+            f"{'enabled' if current_state else 'disabled'} to "
+            f"{'enabled' if new_state else 'disabled'}."
+        )
+
+        result = set_in_transit_encryption(enabled=new_state)
+        assert result, "Failed to toggle in-transit encryption state."
+        log.info(
+            f"In-transit encryption is now {'enabled' if new_state else 'disabled'}."
+        )
+        return result
+
     @tier1
-    @skipif_ocs_version("<4.13")
     @pytest.mark.polarion_id("OCS-4861")
-    def test_intransit_encryption_enable_disable_statetransition(self):
+    def test_intransit_encryption_enable_disable_statetransition(
+        self, multi_pvc_factory, pod_factory
+    ):
         """
-        The test does the following:
-        1. Enable in-transit Encryption if not Enabled.
-        2. Verify in-transit Encryption is Enable on setup.
-        3. Disable Encryption
-        4. Verify in-transit encryption configuration is removed.
-        5. Enable encryption Again and verify it.
-        6. Verify in-transit encryption config is exists.
+        Test to validate in-transit encryption enable-disable state transitions.
 
+        Steps:
+        1. Create a cephfs, rpd pvcs with different access mode.
+        2. Change in-transit Encryption state.
+        3. Create a pods and attach the PVC to it.
+        4. Start IO from  All pods.
+        5. During the IO running on the pod toggle intransit encryption state.
         """
-        if not get_in_transit_encryption_config_state():
-            if config.ENV_DATA.get("in_transit_encryption"):
-                pytest.fail("In-transit encryption is not enabled on the setup")
-            else:
-                set_in_transit_encryption()
-
-        log.info("Verifying the in-transit encryption is enable on setup.")
-        assert in_transit_encryption_verification()
-
-        log.info("Disabling the in-transit encryption.")
-        set_in_transit_encryption(enabled=False)
-
-        # Verify that encryption is actually disabled by checking that a ValueError is raised.
-        log.info("Verifying the in-transit encryption is disabled.")
-        with pytest.raises(ValueError):
-            assert (
-                not in_transit_encryption_verification()
-            ), "In-transit Encryption was expected to be disabled, but it's enabled in the setup."
-
-        if config.ENV_DATA.get("in_transit_encryption"):
-            log.info("Re-enabling in-transit encryption.")
-            set_in_transit_encryption()
-
-            # Verify that encryption is enabled again after re-enabling it
-            log.info(
-                "Verifying the in-transit encryption config after enabling the cluster."
+        size = 5
+        access_modes = {
+            constants.CEPHBLOCKPOOL: [
+                f"{constants.ACCESS_MODE_RWO}-Block",
+                f"{constants.ACCESS_MODE_RWX}-Block",
+            ],
+            constants.CEPHFILESYSTEM: [
+                constants.ACCESS_MODE_RWO,
+                constants.ACCESS_MODE_RWX,
+            ],
+        }
+
+        # Create PVCs for CephBlockPool and CephFS
+        pvc_objects = {
+            interface: multi_pvc_factory(
+                interface=interface,
+                access_modes=modes,
+                size=size,
+                num_of_pvc=2,
             )
-            assert in_transit_encryption_verification()
+            for interface, modes in access_modes.items()
+        }
+
+        for interface, pvcs in pvc_objects.items():
+            assert pvcs, f"Failed to create PVCs for {interface}."
+
+        # Toggle encryption state
+        assert (
+            self.toggle_intransit_encryption_state()
+        ), "Failed to change in-transit encryption state."
+
+        # Create pods for each interface
+        self.all_pods = []
+        for interface, pvcs in pvc_objects.items():
+            pods = create_pods(
+                pvc_objs=pvcs,
+                pod_factory=pod_factory,
+                interface=interface,
+                pods_for_rwx=2,  # Create 2 pods for each RWX PVC
+                status=constants.STATUS_RUNNING,
+            )
+            assert pods, f"Failed to create pods for {interface}."
+            self.all_pods.extend(pods)
+
+        # Perform I/O on all pods using ThreadPoolExecutor
+        with ThreadPoolExecutor() as executor:
+            futures = [
+                executor.submit(
+                    pod_obj.run_io, storage_type="fs", size="1G", runtime=60
+                )
+                for pod_obj in self.all_pods
+            ]
+
+            # Toggle encryption state during I/O operations
+            for _ in range(2):
+                log.info("Toggling encryption state during I/O.")
+                assert (
+                    self.toggle_intransit_encryption_state()
+                ), "Failed to change in-transit encryption state."
+                time.sleep(5)
+
+            # Wait for I/O operations to complete
+            for future in futures:
+                future.result()