Small fixes in security tests

Signed-off-by: Katarzyna Treder <katarzyna.treder@h-partners.com>

test/functional/api/cas/casadm.py

@@ -297,9 +297,7 @@ def get_param_promotion_nhit(
 def set_cache_mode(
     cache_mode: CacheMode, cache_id: int, flush: bool = None, shortcut: bool = False
 ) -> Output:
-    flush_cache = None
-    if flush is not None:
-        flush_cache = "yes" if flush else "no"
+    flush_cache = "yes" if flush else "no"
     output = TestRun.executor.run(
         set_cache_mode_cmd(
             cache_mode=cache_mode.name.lower(),

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_add_core_cache_id.py

@@ -20,6 +20,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     get_fuzz_config,
     run_cmd_and_validate,
@@ -68,6 +69,9 @@ def test_fuzzy_add_core_cache_id(
             )
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = [str(cache.cache_id).encode("ascii")]
         fuzz_config = get_fuzz_config("cache_id.yml")

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_add_core_core_id.py

@@ -24,6 +24,7 @@
     get_fuzz_config,
     run_cmd_and_validate,
 )
+from test_utils.os_utils import Udev
 
 core_id_min = 0
 core_id_max = 4095
@@ -68,6 +69,9 @@ def test_fuzzy_add_core_core_id(
             )
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         fuzz_config = get_fuzz_config("core_id.yml")
         PeachFuzzer.generate_config(fuzz_config)

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_add_core_device.py

@@ -20,6 +20,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     run_cmd_and_validate,
     get_device_fuzz_config,
@@ -67,6 +68,9 @@ def test_fuzzy_add_core_device(
             )
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = [
             disk.path

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_script_add_core_try_add_cache_id.py

@@ -9,6 +9,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskTypeSet, DiskType
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     run_cmd_and_validate,
     get_fuzz_config,
@@ -41,6 +42,9 @@ def test_fuzzy_script_add_core_try_add_cache_id():
             command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     for index, cmd in TestRun.iteration(
         enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
     ):

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_script_add_core_try_add_core_id.py

@@ -9,6 +9,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskTypeSet, DiskType
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     run_cmd_and_validate,
     get_fuzz_config,
@@ -43,6 +44,9 @@ def test_fuzzy_script_add_core_try_add_core_id():
             command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     for index, cmd in TestRun.iteration(
         enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
     ):

test/functional/tests/security/fuzzy/kernel/fuzzy_add_core/test_fuzzy_script_add_core_try_add_device.py

@@ -7,6 +7,7 @@
 from api.cas.cli import script_try_add_cmd, remove_detached_cmd
 from core.test_run import TestRun
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     run_cmd_and_validate,
     get_device_fuzz_config,
@@ -36,6 +37,9 @@ def test_fuzzy_script_add_core_try_add_core_device():
             command_template=base_cmd, count=TestRun.usr.fuzzy_iter_count
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     for index, cmd in TestRun.iteration(
         enumerate(commands), f"Run command {TestRun.usr.fuzzy_iter_count} times"
     ):

test/functional/tests/security/fuzzy/kernel/fuzzy_start/test_fuzzy_start_cache_flags.py

@@ -19,6 +19,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from test_utils.size import Unit, Size
 from tests.security.fuzzy.kernel.common.common import (
     get_fuzz_config,
@@ -49,6 +50,9 @@ def test_fuzzy_start_cache_flags(cache_mode, cache_line_size, unaligned_io, use_
         cache_disk = TestRun.disks["cache"]
         cache_disk.create_partitions([Size(400, Unit.MebiByte)])
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Start and stop cache"):
         # Reload kernel modules
         cache = casadm.start_cache(
@@ -93,7 +97,7 @@ def test_fuzzy_start_cache_flags(cache_mode, cache_line_size, unaligned_io, use_
                     any_alphanumeric_pattern = r"\w+"
                     base_cmd = re.sub(
                         pattern=f"{incompatible_param} {any_alphanumeric_pattern}",
-                        sub="",
+                        repl="",
                         string=base_cmd,
                     )
             base_cmd = f"{base_cmd.strip()} {param}"

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_remove_core_cache_id.py

@@ -18,6 +18,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -56,6 +57,9 @@ def test_fuzzy_remove_core_cache_id(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = [str(core.cache_id).encode("ascii")]
         PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_remove_core_core_id.py

@@ -18,6 +18,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -56,6 +57,9 @@ def test_fuzzy_remove_core_core_id(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = [str(core.core_id).encode("ascii")]
         PeachFuzzer.generate_config(get_fuzz_config("core_id.yml"))

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_remove_core_flag.py

@@ -17,6 +17,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -55,6 +56,9 @@ def test_fuzzy_remove_core_flag(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = ["", "-f", "--force"]
         valid_values = [v.encode("ascii") for v in valid_values]

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_remove_inactive_cache_id.py

@@ -20,6 +20,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -58,6 +59,9 @@ def test_fuzzy_remove_inactive_cache_id(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Create init config from running configuration."):
         InitConfig.create_init_config_from_running_configuration()
 
@@ -93,3 +97,6 @@ def test_fuzzy_remove_inactive_cache_id(
                     cache.stop(no_data_flush=True)
                     core_disk.unplug()
                     casadm.load_cache(device=cache_disk.partitions[0])
+
+    with TestRun.step("Plug core device"):
+        core_disk.plug_all()

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_remove_inactive_core_id.py

@@ -19,6 +19,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -57,6 +58,9 @@ def test_fuzzy_remove_inactive_core_id(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Create init config from running configuration."):
         InitConfig.create_init_config_from_running_configuration()
 
@@ -92,3 +96,6 @@ def test_fuzzy_remove_inactive_core_id(
                     cache.stop(no_data_flush=True)
                     core_disk.unplug()
                     casadm.load_cache(device=cache_disk.partitions[0])
+
+    with TestRun.step("Plug core device"):
+        core_disk.plug_all()

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_stop_cache_cache_id.py

@@ -18,6 +18,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -56,6 +57,9 @@ def test_fuzzy_stop_cache_cache_id(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = [str(core.cache_id).encode("ascii")]
         PeachFuzzer.generate_config(get_fuzz_config("cache_id.yml"))

test/functional/tests/security/fuzzy/kernel/fuzzy_stop_remove/test_fuzzy_stop_cache_flag.py

@@ -18,6 +18,7 @@
 from core.test_run import TestRun
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools.peach_fuzzer.peach_fuzzer import PeachFuzzer
+from test_utils.os_utils import Udev
 from tests.security.fuzzy.kernel.common.common import (
     prepare_cas_instance,
     get_fuzz_config,
@@ -56,6 +57,9 @@ def test_fuzzy_stop_cache_flag(
             cleaning_policy=cleaning_policy,
         )
 
+    with TestRun.step("Disable udev"):
+        Udev.disable()
+
     with TestRun.step("Prepare PeachFuzzer"):
         valid_values = ["", "-n", "--no-data-flush"]
         valid_values = [v.encode("ascii") for v in valid_values]

test/functional/tests/security/fuzzy/kernel/fuzzy_with_io/fuzzy_get_param/test_fuzzy_get_param_name.py

@@ -87,7 +87,7 @@ def test_fuzzy_get_param_name(cache_mode, cache_line_size, unaligned_io, use_io_
             if param == str(ParamName.seq_cutoff):
                 cmd += f" --core-id {core.core_id}"
 
-            cmd = base_cmd.replace("{param}", param)
+            cmd = cmd.replace("{param}", param)
 
             run_cmd_and_validate(
                 cmd=get_cmd(cmd, param.encode("ascii")),

test/functional/tests/security/test_security_user.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -13,6 +14,7 @@
 from storage_devices.disk import DiskType, DiskTypeSet, DiskTypeLowerThan
 from test_tools import fs_utils
 from test_tools.disk_utils import Filesystem
+from test_utils.os_utils import create_user, check_if_user_exists
 from test_utils.output import CmdException
 from test_utils.size import Size, Unit
 
@@ -70,9 +72,7 @@ def test_user_cli():
         casadm.stop_all_caches()
 
     with TestRun.step("Add non-root user account."):
-        TestRun.executor.run(f"useradd -N -r -l {user_name}")
-        user_home_dir = fs_utils.parse_ls_output(fs_utils.ls_item(f"/home/{user_name}"))[0]
-        user_home_dir.chmod_numerical(777, True)
+        add_user()
 
     with TestRun.step("Try to start cache."):
         try:
@@ -95,7 +95,7 @@ def test_user_cli():
 
     with TestRun.step("Try to set cache mode."):
         try:
-            output = run_as_other_user(cli.set_cache_mode_cmd(CacheMode.WB,
+            output = run_as_other_user(cli.set_cache_mode_cmd(CacheMode.WB.name.lower(),
                                                               str(cache.cache_id)), user_name)
             if output.exit_code == 0:
                 TestRun.LOGGER.error("Setting cache mode should fail!")
@@ -130,7 +130,7 @@ def test_user_cli():
 
     with TestRun.step("Try to list caches."):
         try:
-            output = run_as_other_user(cli.list_cmd(), user_name)
+            output = run_as_other_user(cli.list_caches_cmd(), user_name)
             if output.exit_code == 0:
                 TestRun.LOGGER.error("Listing caches should fail!")
         except CmdException:
@@ -222,7 +222,7 @@ def test_user_cli():
     with TestRun.step("Try to load IO class configuration."):
         try:
             output = run_as_other_user(cli.load_io_classes_cmd(
-                str(cache.cache_id), io_conf_copy), user_name)
+                str(cache.cache_id), io_conf_copy.full_path), user_name)
             if output.exit_code == 0:
                 TestRun.LOGGER.error("Loading IO class configuration should fail!")
         except CmdException:
@@ -273,7 +273,7 @@ def test_user_cli():
 
     with TestRun.step("Try to list caches with 'sudo'."):
         try:
-            run_as_other_user(cli.list_cmd(), user_name, True)
+            run_as_other_user(cli.list_caches_cmd(), user_name, True)
         except CmdException:
             TestRun.LOGGER.error("Non-root sudoer user should be able to list caches.")
 
@@ -339,7 +339,7 @@ def test_user_cli():
 
     with TestRun.step("Try to load IO class configuration with 'sudo'."):
         try:
-            run_as_other_user(cli.load_io_classes_cmd(str(cache.cache_id), io_conf_copy),
+            run_as_other_user(cli.load_io_classes_cmd(str(cache.cache_id), io_conf_copy.full_path),
                               user_name, True)
         except CmdException:
             TestRun.LOGGER.error("Non-root sudoer user should be able to "
@@ -420,9 +420,7 @@ def test_user_service():
         core.unmount()
 
     with TestRun.step("Add non-root user account."):
-        TestRun.executor.run(f"useradd -N -r -l {user_name}")
-        user_home_dir = fs_utils.parse_ls_output(fs_utils.ls_item(f"/home/{user_name}"))[0]
-        user_home_dir.chmod_numerical(777, True)
+        add_user()
 
     with TestRun.step("Try to stop OpenCAS service."):
         try:
@@ -492,3 +490,9 @@ def run_as_other_user(command, user: str, sudo: bool = False):
     if output.exit_code != 0 or output.stderr is not "":
         raise CmdException("Must be run as root.", output)
     return output
+
+
+def add_user():
+    create_user(user_name, additional_params=["N", "r", "l"])
+    if not check_if_user_exists(user_name):
+        raise Exception("Failed to create user.")