test_add:test_fault_injection_core_in_raid

Signed-off-by: Kamil Gierszewski <[email protected]>

test/functional/api/cas/cli_messages.py

@@ -252,6 +252,15 @@
     r"Option '--cache-line-size \(-x\)' is not allowed"
 ]
 
+partition_not_suitable_for_array = [
+    r'\S+ is not suitable for this array'
+]
+
+device_or_resource_busy = [
+    r"cannot open \S+: Device or resource busy"
+]
+
+
 def check_stderr_msg(output: Output, expected_messages, negate=False):
     return __check_string_msg(output.stderr, expected_messages, negate)
 

test/functional/tests/fault_injection/test_fault_injection_core_in_raid.py

@@ -0,0 +1,70 @@
+#
+# Copyright(c) 2022 Intel Corporation
+# SPDX-License-Identifier: BSD-3-Clause-Clear
+#
+import re
+import pytest
+from api.cas import casadm, cli_messages
+from api.cas.cache_config import CacheMode
+from core.test_run import TestRun
+from storage_devices.disk import DiskType, DiskTypeSet
+from storage_devices.raid import Raid, RaidConfiguration, MetadataVariant, Level
+from test_utils.size import Size, Unit
+from api.cas.cli_messages import (
+    partition_not_suitable_for_array,
+    device_or_resource_busy
+)
+
+expected_msg_1 = partition_not_suitable_for_array[0]
+expected_msg_2 = device_or_resource_busy[0]
+
+
+@pytest.mark.parametrizex("cache_mode", CacheMode)
+@pytest.mark.require_disk("cache", DiskTypeSet([DiskType.optane, DiskType.nand]))
+@pytest.mark.require_disk("core", DiskTypeSet([DiskType.sata, DiskType.hdd]))
+@pytest.mark.require_disk("core2", DiskTypeSet([DiskType.sata, DiskType.hdd]))
+def test_fault_injection_core_in_raid(cache_mode):
+    """
+        title: Test if OpenCAS rejects using core device to build SW RAID.
+        description: |
+          Test if OpenCAS handles properly attempting of use core device to build SW RAID.
+        pass_criteria:
+          - Expected to reject RAID creation with proper warning.
+    """
+    with TestRun.step("Prepare CAS device."):
+        cache_disk = TestRun.disks['cache']
+        first_core_disk = TestRun.disks['core']
+        second_core_disk = TestRun.disks['core2']
+        cache_disk.create_partitions([Size(2, Unit.GibiByte)])
+        first_core_disk.create_partitions([Size(2, Unit.GibiByte)])
+        second_core_disk.create_partitions([Size(2, Unit.GibiByte)])
+        cache_dev = cache_disk.partitions[0]
+        first_core_dev = first_core_disk.partitions[0]
+        second_core_dev = second_core_disk.partitions[0]
+
+    with TestRun.step("Start cas and add core."):
+        cache = casadm.start_cache(cache_dev, cache_mode, force=True)
+        casadm.add_core(cache, first_core_dev)
+
+    with TestRun.step("Attempt to use core device to build SW RAID."):
+        config = RaidConfiguration(
+            level=Level.Raid1,
+            metadata=MetadataVariant.Legacy,
+            number_of_devices=2)
+
+        try:
+            Raid.create(config, [first_core_dev, second_core_dev])
+            TestRun.fail(f"RAID created successfully. Expected otherwise.")
+        except Exception as ex:
+            output = ex.output
+
+    with TestRun.step("Looking for any of 2 expected messages."):
+
+        if re.search(expected_msg_1, output.stderr) or re.search(
+                expected_msg_2, output.stderr):
+            TestRun.LOGGER.info("RAID not created. Found expected warning in exception message.")
+        else:
+            TestRun.LOGGER.error(f"RAID not created but warning message not as expected.\n"
+                                 f"Actual: '{output}'.\n"
+                                 f"Expected: '{expected_msg_1}' or '{expected_msg_2}'.")
+