OS tools refactor:

- make udev, runlevel, memory and wget separate files
- rename os_utils to os_tools and move it to test_tools
- remove ModuleRemoveMethod and set modprobe as default
- fix regex in is_kernel_module_loaded method
- remove get_sys_block_path
- move get_block_device_names method to disk tools

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

internal_plugins/scsi_debug/__init__.py

@@ -6,8 +6,8 @@ from time import sleep
 
 from core.test_run_utils import TestRun
 from storage_devices.device import Device
-from test_utils import os_utils
 from connection.utils.output import CmdException
+from test_tools.os_tools import load_kernel_module, is_kernel_module_loaded, unload_kernel_module
 
 
 class ScsiDebug:
@@ -24,7 +24,7 @@ class ScsiDebug:
     def reload(self):
         self.teardown()
         sleep(1)
-        load_output = os_utils.load_kernel_module(self.module_name, self.params)
+        load_output = load_kernel_module(self.module_name, self.params)
         if load_output.exit_code != 0:
             raise CmdException(f"Failed to load {self.module_name} module", load_output)
         TestRun.LOGGER.info(f"{self.module_name} loaded successfully.")
@@ -32,8 +32,8 @@ class ScsiDebug:
         TestRun.scsi_debug_devices = Device.get_scsi_debug_devices()
 
     def teardown(self):
-        if os_utils.is_kernel_module_loaded(self.module_name):
-            os_utils.unload_kernel_module(self.module_name)
+        if is_kernel_module_loaded(self.module_name):
+            unload_kernel_module(self.module_name)
 
 
 plugin_class = ScsiDebug

storage_devices/nullblk.py

@@ -6,10 +6,9 @@
 from core.test_run import TestRun
 from storage_devices.device import Device
 from test_tools.fs_utils import ls, parse_ls_output
-from test_utils.os_utils import (
+from test_tools.os_tools import (
     unload_kernel_module,
     is_kernel_module_loaded,
-    ModuleRemoveMethod,
     reload_kernel_module,
 )
 
@@ -37,7 +36,7 @@ class NullBlk(Device):
         if not is_kernel_module_loaded(cls._module):
             return
         TestRun.LOGGER.info("Removing null_blk ")
-        unload_kernel_module(module_name=cls._module, unload_method=ModuleRemoveMethod.modprobe)
+        unload_kernel_module(module_name=cls._module)
 
     @classmethod
     def list(cls):

storage_devices/ramdisk.py

@@ -10,7 +10,7 @@ from storage_devices.device import Device
 from test_tools import disk_utils
 from test_tools.fs_utils import ls, parse_ls_output
 from test_utils.filesystem.symlink import Symlink
-from test_utils.os_utils import reload_kernel_module, unload_kernel_module, is_kernel_module_loaded
+from test_tools.os_tools import reload_kernel_module, unload_kernel_module, is_kernel_module_loaded
 from types.size import Size, Unit
 
 

test_tools/blktrace.py

@@ -14,7 +14,7 @@ from datetime import timedelta
 from core.test_run import TestRun
 from storage_devices.device import Device
 from test_utils.filesystem.directory import Directory
-from test_utils.os_utils import is_mounted, drop_caches, DropCachesMode
+from test_tools.os_tools import is_mounted, drop_caches, DropCachesMode
 from types.size import Size, Unit
 
 DEBUGFS_MOUNT_POINT = "/sys/kernel/debug"

test_tools/disk_utils.py

@@ -8,6 +8,7 @@ import posixpath
 import re
 import time
 from enum import Enum
+from typing import List
 
 from core.test_run import TestRun
 from test_tools import fs_utils
@@ -266,7 +267,7 @@ def get_first_partition_offset(device, aligned: bool):
 
 
 def remove_partitions(device):
-    from test_utils.os_utils import Udev
+    from test_tools.udev import Udev
     if device.is_mounted():
         device.unmount()
 
@@ -406,3 +407,13 @@ def validate_dev_path(path: str):
         return path
 
     raise ValueError(f'By-id device link {path} is broken.')
+
+
+def get_block_device_names_list(exclude_list: List[int] = None) -> List[str]:
+    cmd = "lsblk -lo NAME"
+    if exclude_list is not None:
+        cmd += f" -e {','.join(str(type_id) for type_id in exclude_list)}"
+    devices = TestRun.executor.run_expect_success(cmd).stdout
+    devices_list = devices.splitlines()
+    devices_list.sort()
+    return devices_list

test_tools/fio/fio.py

@@ -10,9 +10,9 @@ import uuid
 from packaging.version import Version
 import test_tools.fio.fio_param
 import test_tools.fs_utils
+import test_tools.wget
 from core.test_run import TestRun
 from test_tools import fs_utils
-from test_utils import os_utils
 from connection.utils.output import CmdException
 
 
@@ -50,7 +50,7 @@ class Fio:
 
     def install(self):
         fio_url = f"http://brick.kernel.dk/snaps/fio-{self.min_fio_version}.tar.bz2"
-        fio_package = os_utils.download_file(fio_url)
+        fio_package = test_tools.wget.download_file(fio_url)
         fs_utils.uncompress_archive(fio_package)
         TestRun.executor.run_expect_success(
             f"cd {fio_package.parent_dir}/fio-{self.min_fio_version}"

test_tools/initramfs.py

@@ -4,7 +4,7 @@
 #
 
 from core.test_run import TestRun
-from test_utils.os_utils import get_distro, Distro
+from test_tools.os_tools import get_distro, Distro
 
 
 def update():

test_tools/mdadm.py

@@ -8,7 +8,7 @@ import re
 
 from core.test_run import TestRun
 from types.size import Unit
-from test_utils.os_utils import Udev
+from test_tools.udev import Udev
 
 
 class Mdadm:

test_tools/memory.py

@@ -0,0 +1,103 @@
+import math
+
+from connection.utils.output import CmdException
+from core.test_run import TestRun
+from test_tools.dd import Dd
+from test_tools.fs_utils import check_if_directory_exists, create_directory
+from test_tools.os_tools import OvercommitMemoryMode, drop_caches, DropCachesMode, \
+    MEMORY_MOUNT_POINT, is_mounted
+from types.size import Size, Unit
+
+
+def disable_memory_affecting_functions():
+    """Disables system functions affecting memory"""
+    # Don't allow sshd to be killed in case of out-of-memory:
+    TestRun.executor.run(
+        "echo '-1000' > /proc/`cat /var/run/sshd.pid`/oom_score_adj"
+    )
+    TestRun.executor.run(
+        "echo -17 > /proc/`cat /var/run/sshd.pid`/oom_adj"
+    )  # deprecated
+    TestRun.executor.run_expect_success(
+        f"echo {OvercommitMemoryMode.NEVER.value} > /proc/sys/vm/overcommit_memory"
+    )
+    TestRun.executor.run_expect_success("echo '100' > /proc/sys/vm/overcommit_ratio")
+    TestRun.executor.run_expect_success(
+        "echo '64      64      32' > /proc/sys/vm/lowmem_reserve_ratio"
+    )
+    TestRun.executor.run_expect_success("swapoff --all")
+    drop_caches(DropCachesMode.SLAB)
+
+
+def defaultize_memory_affecting_functions():
+    """Sets default values to system functions affecting memory"""
+    TestRun.executor.run_expect_success(
+        f"echo {OvercommitMemoryMode.DEFAULT.value} > /proc/sys/vm/overcommit_memory"
+    )
+    TestRun.executor.run_expect_success("echo 50 > /proc/sys/vm/overcommit_ratio")
+    TestRun.executor.run_expect_success(
+        "echo '256     256     32' > /proc/sys/vm/lowmem_reserve_ratio"
+    )
+    TestRun.executor.run_expect_success("swapon --all")
+
+
+def get_mem_free():
+    """Returns free amount of memory in bytes"""
+    output = TestRun.executor.run_expect_success("free -b")
+    output = output.stdout.splitlines()
+    for line in output:
+        if 'free' in line:
+            index = line.split().index('free') + 1  # 1st row has 1 element less than following rows
+        if 'Mem' in line:
+            mem_line = line.split()
+
+    return Size(int(mem_line[index]))
+
+
+def get_mem_available():
+    """Returns amount of available memory from /proc/meminfo"""
+    cmd = "cat /proc/meminfo | grep MemAvailable | awk '{ print $2 }'"
+    mem_available = TestRun.executor.run(cmd).stdout
+
+    return Size(int(mem_available), Unit.KibiByte)
+
+
+def get_module_mem_footprint(module_name):
+    """Returns allocated size of specific module's metadata from /proc/vmallocinfo"""
+    cmd = f"cat /proc/vmallocinfo | grep {module_name} | awk '{{ print $2 }}' "
+    output_lines = TestRun.executor.run(cmd).stdout.splitlines()
+    memory_used = 0
+    for line in output_lines:
+        memory_used += int(line)
+
+    return Size(memory_used)
+
+
+def allocate_memory(size: Size):
+    """Allocates given amount of memory"""
+    mount_ramfs()
+    TestRun.LOGGER.info(f"Allocating {size.get_value(Unit.MiB):0.2f} MiB of memory.")
+    bs = Size(1, Unit.Blocks512)
+    dd = (
+        Dd()
+        .block_size(bs)
+        .count(math.ceil(size / bs))
+        .input("/dev/zero")
+        .output(f"{MEMORY_MOUNT_POINT}/data")
+    )
+    output = dd.run()
+    if output.exit_code != 0:
+        raise CmdException("Allocating memory failed.", output)
+
+
+def mount_ramfs():
+    """Mounts ramfs to enable allocating memory space"""
+    if not check_if_directory_exists(MEMORY_MOUNT_POINT):
+        create_directory(MEMORY_MOUNT_POINT)
+    if not is_mounted(MEMORY_MOUNT_POINT):
+        TestRun.executor.run_expect_success(f"mount -t ramfs ramfs {MEMORY_MOUNT_POINT}")
+
+
+def unmount_ramfs():
+    """Unmounts ramfs and releases whole space allocated by it in memory"""
+    TestRun.executor.run_expect_success(f"umount {MEMORY_MOUNT_POINT}")

test_tools/os_tools.py

@@ -0,0 +1,236 @@
+#
+# Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+import posixpath
+import re
+import time
+
+from datetime import timedelta
+from enum import IntFlag, Enum, StrEnum
+from packaging import version
+
+from core.test_run import TestRun
+from storage_devices.device import Device
+from test_tools.disk_utils import get_sysfs_path
+from test_tools.fs_utils import check_if_file_exists
+from test_utils.filesystem.file import File
+from connection.utils.retry import Retry
+
+DEBUGFS_MOUNT_POINT = "/sys/kernel/debug"
+MEMORY_MOUNT_POINT = "/mnt/memspace"
+
+
+class Distro(StrEnum):
+    UBUNTU = "ubuntu"
+    DEBIAN = "debian"
+    REDHAT = "rhel"
+    OPENEULER = "openeuler"
+    CENTOS = "centos"
+
+
+class DropCachesMode(IntFlag):
+    PAGECACHE = 1
+    SLAB = 2
+    ALL = PAGECACHE | SLAB
+
+
+class OvercommitMemoryMode(Enum):
+    DEFAULT = 0
+    ALWAYS = 1
+    NEVER = 2
+
+
+class SystemManagerType(Enum):
+    sysv = 0
+    systemd = 1
+
+
+def get_distro():
+    output = TestRun.executor.run(
+        "cat /etc/os-release | grep -e \"^ID=\" | awk -F= '{print$2}' | tr -d '\"'"
+    ).stdout.lower()
+
+    try:
+        return Distro(output)
+    except ValueError:
+        raise ValueError(f"Could not resolve distro name. Command output: {output}")
+
+
+def drop_caches(level: DropCachesMode = DropCachesMode.ALL):
+    TestRun.executor.run_expect_success(
+        f"echo {level.value} > /proc/sys/vm/drop_caches")
+
+
+def get_number_of_processors_from_cpuinfo():
+    """Returns number of processors (count) which are listed out in /proc/cpuinfo"""
+    cmd = f"cat /proc/cpuinfo | grep processor | wc -l"
+    output = TestRun.executor.run(cmd).stdout
+
+    return int(output)
+
+
+def get_number_of_processes(process_name):
+    cmd = f"ps aux | grep {process_name} | grep -v grep | wc -l"
+    output = TestRun.executor.run(cmd).stdout
+
+    return int(output)
+
+
+def get_kernel_version():
+    version_string = TestRun.executor.run_expect_success("uname -r").stdout
+    version_string = version_string.split('-')[0]
+    return version.Version(version_string)
+
+
+def is_kernel_module_loaded(module_name):
+    output = TestRun.executor.run(f"lsmod | grep ^{module_name}$")
+    return output.exit_code == 0
+
+
+def load_kernel_module(module_name, module_args: {str, str}=None):
+    cmd = f"modprobe {module_name}"
+    if module_args is not None:
+        for key, value in module_args.items():
+            cmd += f" {key}={value}"
+    return TestRun.executor.run(cmd)
+
+
+def unload_kernel_module(module_name):
+    cmd = f"modprobe -r {module_name}"
+    return TestRun.executor.run_expect_success(cmd)
+
+
+def get_kernel_module_parameter(module_name, parameter):
+    param_file_path = f"/sys/module/{module_name}/parameters/{parameter}"
+    if not check_if_file_exists(param_file_path):
+        raise FileNotFoundError(f"File {param_file_path} does not exist!")
+    return File(param_file_path).read()
+
+
+def is_mounted(path: str):
+    if path is None or path.isspace():
+        raise Exception("Checked path cannot be empty")
+    command = f"mount | grep --fixed-strings '{path.rstrip('/')} '"
+    return TestRun.executor.run(command).exit_code == 0
+
+
+def mount_debugfs():
+    if not is_mounted(DEBUGFS_MOUNT_POINT):
+        TestRun.executor.run_expect_success(f"mount -t debugfs none {DEBUGFS_MOUNT_POINT}")
+
+
+def reload_kernel_module(module_name, module_args: {str, str}=None):
+    if is_kernel_module_loaded(module_name):
+        unload_kernel_module(module_name)
+
+    Retry.run_while_false(
+        lambda: load_kernel_module(module_name, module_args).exit_code == 0,
+        timeout=timedelta(seconds=5)
+    )
+
+
+def get_module_path(module_name):
+    cmd = f"modinfo {module_name}"
+
+    # module path is in second column of first line of `modinfo` output
+    module_info = TestRun.executor.run_expect_success(cmd).stdout
+    module_path = module_info.splitlines()[0].split()[1]
+
+    return module_path
+
+
+def get_executable_path(exec_name):
+    cmd = f"which {exec_name}"
+
+    path = TestRun.executor.run_expect_success(cmd).stdout
+
+    return path
+
+
+def kill_all_io(graceful=True):
+    if graceful:
+        # TERM signal should be used in preference to the KILL signal, since a
+        # process may install a handler for the TERM signal in order to perform
+        # clean-up steps before terminating in an orderly fashion.
+        TestRun.executor.run("killall -q --signal TERM dd fio blktrace")
+        time.sleep(3)
+    TestRun.executor.run("killall -q --signal TERM dd fio blktrace")
+    time.sleep(3)
+    TestRun.executor.run("killall -q --signal KILL dd fio blktrace")
+    TestRun.executor.run("kill -9 `ps aux | grep -i vdbench.* | awk '{ print $2 }'`")
+
+    if TestRun.executor.run("pgrep -x dd").exit_code == 0:
+        raise Exception(f"Failed to stop dd!")
+    if TestRun.executor.run("pgrep -x fio").exit_code == 0:
+        raise Exception(f"Failed to stop fio!")
+    if TestRun.executor.run("pgrep -x blktrace").exit_code == 0:
+        raise Exception(f"Failed to stop blktrace!")
+    if TestRun.executor.run("pgrep vdbench").exit_code == 0:
+        raise Exception(f"Failed to stop vdbench!")
+
+
+def sync():
+    TestRun.executor.run_expect_success("sync")
+
+
+def get_dut_cpu_number():
+    return int(TestRun.executor.run_expect_success("nproc").stdout)
+
+
+def get_dut_cpu_physical_cores():
+    """ Get list of CPU numbers that don't share physical cores """
+    output = TestRun.executor.run_expect_success("lscpu --all --parse").stdout
+
+    core_list = []
+    visited_phys_cores = []
+    for line in output.split("\n"):
+        if "#" in line:
+            continue
+
+        cpu_no, phys_core_no = line.split(",")[:2]
+        if phys_core_no not in visited_phys_cores:
+            core_list.append(cpu_no)
+            visited_phys_cores.append(phys_core_no)
+
+    return core_list
+
+
+def set_wbt_lat(device: Device, value: int):
+    if value < 0:
+        raise ValueError("Write back latency can't be negative number")
+
+    wbt_lat_config_path = posixpath.join(
+        get_sysfs_path(device.get_device_id()), "queue/wbt_lat_usec"
+    )
+
+    return TestRun.executor.run_expect_success(f"echo {value} > {wbt_lat_config_path}")
+
+
+def get_wbt_lat(device: Device):
+    wbt_lat_config_path = posixpath.join(
+        get_sysfs_path(device.get_device_id()), "queue/wbt_lat_usec"
+    )
+
+    return int(TestRun.executor.run_expect_success(f"cat {wbt_lat_config_path}").stdout)
+
+
+def get_cores_ids_range(numa_node: int):
+    output = TestRun.executor.run_expect_success(f"lscpu --all --parse").stdout
+    parse_output = re.findall(r'(\d+),(\d+),(?:\d+),(\d+),,', output, re.I)
+
+    return [element[0] for element in parse_output if int(element[2]) == numa_node]
+
+
+def create_user(username, additional_params=None):
+    command = "useradd "
+    if additional_params:
+        command += "".join([f"-{p} " for p in additional_params])
+    command += username
+    return TestRun.executor.run_expect_success(command)
+
+
+def check_if_user_exists(username):
+    return TestRun.executor.run(f"id {username}").exit_code == 0

test_tools/peach_fuzzer/peach_fuzzer.py

@@ -12,10 +12,10 @@ import tempfile
 import lxml.etree as etree
 from collections import namedtuple
 
+import test_tools.wget
 from core.test_run import TestRun
 from test_tools import fs_utils
 from test_tools.fs_utils import create_directory, check_if_file_exists, write_file
-from test_utils import os_utils
 
 
 class PeachFuzzer:
@@ -155,7 +155,7 @@ class PeachFuzzer:
         Install Peach Fuzzer on the DUT
         """
         create_directory(cls.base_dir, True)
-        peach_archive = os_utils.download_file(
+        peach_archive = test_tools.wget.download_file(
             cls.peach_fuzzer_3_0_url, destination_dir=cls.base_dir
         )
         TestRun.executor.run_expect_success(

test_tools/runlevel.py

@@ -0,0 +1,109 @@
+#
+# Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+from enum import IntEnum
+
+from core.test_run import TestRun
+from test_tools.os_tools import SystemManagerType
+
+
+class Runlevel(IntEnum):
+    """
+        Halt the system.
+        SysV Runlevel: 0
+        systemd Target: runlevel0.target, poweroff.target
+    """
+    runlevel0 = 0
+    poweroff = runlevel0
+
+    """
+        Single user mode.
+        SysV Runlevel: 1, s, single
+        systemd Target: runlevel1.target, rescue.target
+    """
+    runlevel1 = 1
+    rescue = runlevel1
+
+    """
+        User-defined/Site-specific runlevels. By default, identical to 3.
+        SysV Runlevel: 2, 4
+        systemd Target: runlevel2.target, runlevel4.target, multi-user.target
+    """
+    runlevel2 = 2
+
+    """
+        Multi-user, non-graphical. Users can usually login via multiple consoles or via the network.
+        SysV Runlevel: 3
+        systemd Target: runlevel3.target, multi-user.target
+    """
+    runlevel3 = 3
+    multi_user = runlevel3
+
+    """
+        Multi-user, graphical. Usually has all the services of runlevel 3 plus a graphical login.
+        SysV Runlevel: 5
+        systemd Target: runlevel5.target, graphical.target
+    """
+    runlevel5 = 5
+    graphical = runlevel5
+
+    """
+        Reboot
+        SysV Runlevel: 6
+        systemd Target: runlevel6.target, reboot.target
+    """
+    runlevel6 = 6
+    reboot = runlevel6
+
+    """
+        Emergency shell
+        SysV Runlevel: emergency
+        systemd Target: emergency.target
+    """
+    runlevel7 = 7
+    emergency = runlevel7
+
+
+def get_system_manager():
+    output = TestRun.executor.run_expect_success("ps -p 1").stdout
+    type = output.split('\n')[1].split()[3]
+    if type == "init":
+        return SystemManagerType.sysv
+    elif type == "systemd":
+        return SystemManagerType.systemd
+    raise Exception(f"Unknown system manager type ({type}).")
+
+
+def change_runlevel(runlevel: Runlevel):
+    if runlevel == get_runlevel():
+        return
+    if Runlevel.runlevel0 < runlevel < Runlevel.runlevel6:
+        system_manager = get_system_manager()
+        if system_manager == SystemManagerType.systemd:
+            TestRun.executor.run_expect_success(f"systemctl set-default {runlevel.name}.target")
+        else:
+            TestRun.executor.run_expect_success(
+                f"sed -i 's/^.*id:.*$/id:{runlevel.value}:initdefault: /' /etc/inittab")
+            TestRun.executor.run_expect_success(f"init {runlevel.value}")
+
+
+def get_runlevel():
+    system_manager = get_system_manager()
+    if system_manager == SystemManagerType.systemd:
+        result = TestRun.executor.run_expect_success("systemctl get-default")
+        try:
+            name = result.stdout.split(".")[0].replace("-", "_")
+            return Runlevel[name]
+        except Exception:
+            raise Exception(f"Cannot parse '{result.output}' to runlevel.")
+    else:
+        result = TestRun.executor.run_expect_success("runlevel")
+        try:
+            split_output = result.stdout.split()
+            runlevel = Runlevel(int(split_output[1]))
+            return runlevel
+        except Exception:
+            raise Exception(f"Cannot parse '{result.output}' to runlevel.")

test_tools/systemctl.py

@@ -23,3 +23,13 @@ def reload_daemon():
 
 def restart_service(name):
     TestRun.executor.run_expect_success(f"systemctl restart {name}")
+
+
+def get_service_path(unit_name):
+    cmd = f"systemctl cat {unit_name}"
+
+    # path is in second column of first line of output
+    info = TestRun.executor.run_expect_success(cmd).stdout
+    path = info.splitlines()[0].split()[1]
+
+    return path

test_tools/udev.py

@@ -0,0 +1,27 @@
+#
+# Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+from core.test_run import TestRun
+
+
+class Udev(object):
+    @staticmethod
+    def enable():
+        TestRun.LOGGER.info("Enabling udev")
+        TestRun.executor.run_expect_success("udevadm control --start-exec-queue")
+
+    @staticmethod
+    def disable():
+        TestRun.LOGGER.info("Disabling udev")
+        TestRun.executor.run_expect_success("udevadm control --stop-exec-queue")
+
+    @staticmethod
+    def trigger():
+        TestRun.executor.run_expect_success("udevadm trigger")
+
+    @staticmethod
+    def settle():
+        TestRun.executor.run_expect_success("udevadm settle")

test_tools/wget.py

@@ -0,0 +1,17 @@
+#
+# Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+from core.test_run import TestRun
+from test_utils.filesystem.file import File
+
+
+def download_file(url, destination_dir="/tmp"):
+    # TODO use wget module instead
+    command = ("wget --tries=3 --timeout=5 --continue --quiet "
+               f"--directory-prefix={destination_dir} {url}")
+    TestRun.executor.run_expect_success(command)
+    path = f"{destination_dir.rstrip('/')}/{File.get_name(url)}"
+    return File(path)

test_utils/disk_finder.py

@@ -2,13 +2,13 @@
 # Copyright(c) 2019-2021 Intel Corporation
 # SPDX-License-Identifier: BSD-3-Clause
 #
-
+import os
 import posixpath
 
 from core.test_run import TestRun
 from test_tools import disk_utils
+from test_tools.disk_utils import get_sysfs_path
 from test_tools.fs_utils import check_if_file_exists, readlink
-from test_utils import os_utils
 from connection.utils.output import CmdException
 
 
@@ -141,14 +141,13 @@ def get_system_disks():
     system_device = TestRun.executor.run_expect_success('mount | grep " / "').stdout.split()[0]
     readlink_output = readlink(system_device)
     device_name = readlink_output.split('/')[-1]
-    sys_block_path = os_utils.get_sys_block_path()
     used_device_names = __get_slaves(device_name)
     if not used_device_names:
         used_device_names = [device_name]
     disk_names = []
     for device_name in used_device_names:
-        if check_if_file_exists(f'{sys_block_path}/{device_name}/partition'):
-            parent_device = readlink(f'{sys_block_path}/{device_name}/..').split('/')[-1]
+        if check_if_file_exists(os.path.join(get_sysfs_path(device_name), "partition")):
+            parent_device = readlink(os.path.join(get_sysfs_path(device_name), "..")).split('/')[-1]
             disk_names.append(parent_device)
         else:
             disk_names.append(device_name)
@@ -159,7 +158,7 @@ def get_system_disks():
 def __get_slaves(device_name: str):
     try:
         device_names = TestRun.executor.run_expect_success(
-            f'ls {os_utils.get_sys_block_path()}/{device_name}/slaves').stdout.splitlines()
+            f"ls {os.path.join(get_sysfs_path(device_name), "slaves")}").stdout.splitlines()
     except CmdException as e:
         if "No such file or directory" not in e.output.stderr:
             raise

test_utils/os_utils.py

@@ -1,496 +0,0 @@
-#
-# Copyright(c) 2019-2022 Intel Corporation
-# Copyright(c) 2024 Huawei Technologies Co., Ltd.
-# SPDX-License-Identifier: BSD-3-Clause
-#
-
-import math
-import posixpath
-import re
-import time
-
-from datetime import timedelta
-from enum import IntFlag, Enum, IntEnum, StrEnum
-from packaging import version
-from typing import List
-
-from core.test_run import TestRun
-from storage_devices.device import Device
-from test_tools.dd import Dd
-from test_tools.disk_utils import get_sysfs_path
-from test_tools.fs_utils import check_if_directory_exists, create_directory, check_if_file_exists
-from test_utils.filesystem.file import File
-from connection.utils.output import CmdException
-from connection.utils.retry import Retry
-from types.size import Size, Unit
-
-DEBUGFS_MOUNT_POINT = "/sys/kernel/debug"
-MEMORY_MOUNT_POINT = "/mnt/memspace"
-
-
-class Distro(StrEnum):
-    UBUNTU = "ubuntu"
-    DEBIAN = "debian"
-    REDHAT = "rhel"
-    OPENEULER = "openeuler"
-    CENTOS = "centos"
-
-
-class DropCachesMode(IntFlag):
-    PAGECACHE = 1
-    SLAB = 2
-    ALL = PAGECACHE | SLAB
-
-
-class OvercommitMemoryMode(Enum):
-    DEFAULT = 0
-    ALWAYS = 1
-    NEVER = 2
-
-
-class Runlevel(IntEnum):
-    """
-        Halt the system.
-        SysV Runlevel: 0
-        systemd Target: runlevel0.target, poweroff.target
-    """
-    runlevel0 = 0
-    poweroff = runlevel0
-
-    """
-        Single user mode.
-        SysV Runlevel: 1, s, single
-        systemd Target: runlevel1.target, rescue.target
-    """
-    runlevel1 = 1
-    rescue = runlevel1
-
-    """
-        User-defined/Site-specific runlevels. By default, identical to 3.
-        SysV Runlevel: 2, 4
-        systemd Target: runlevel2.target, runlevel4.target, multi-user.target
-    """
-    runlevel2 = 2
-
-    """
-        Multi-user, non-graphical. Users can usually login via multiple consoles or via the network.
-        SysV Runlevel: 3
-        systemd Target: runlevel3.target, multi-user.target
-    """
-    runlevel3 = 3
-    multi_user = runlevel3
-
-    """
-        Multi-user, graphical. Usually has all the services of runlevel 3 plus a graphical login.
-        SysV Runlevel: 5
-        systemd Target: runlevel5.target, graphical.target
-    """
-    runlevel5 = 5
-    graphical = runlevel5
-
-    """
-        Reboot
-        SysV Runlevel: 6
-        systemd Target: runlevel6.target, reboot.target
-    """
-    runlevel6 = 6
-    reboot = runlevel6
-
-    """
-        Emergency shell
-        SysV Runlevel: emergency
-        systemd Target: emergency.target
-    """
-    runlevel7 = 7
-    emergency = runlevel7
-
-
-class SystemManagerType(Enum):
-    sysv = 0
-    systemd = 1
-
-
-def get_distro():
-    output = TestRun.executor.run(
-        "cat /etc/os-release | grep -e \"^ID=\" | awk -F= '{print$2}' | tr -d '\"'"
-    ).stdout.lower()
-
-    try:
-        return Distro(output)
-    except ValueError:
-        raise ValueError(f"Could not resolve distro name. Command output: {output}")
-
-
-def get_system_manager():
-    output = TestRun.executor.run_expect_success("ps -p 1").stdout
-    type = output.split('\n')[1].split()[3]
-    if type == "init":
-        return SystemManagerType.sysv
-    elif type == "systemd":
-        return SystemManagerType.systemd
-    raise Exception(f"Unknown system manager type ({type}).")
-
-
-def change_runlevel(runlevel: Runlevel):
-    if runlevel == get_runlevel():
-        return
-    if Runlevel.runlevel0 < runlevel < Runlevel.runlevel6:
-        system_manager = get_system_manager()
-        if system_manager == SystemManagerType.systemd:
-            TestRun.executor.run_expect_success(f"systemctl set-default {runlevel.name}.target")
-        else:
-            TestRun.executor.run_expect_success(
-                f"sed -i 's/^.*id:.*$/id:{runlevel.value}:initdefault: /' /etc/inittab")
-            TestRun.executor.run_expect_success(f"init {runlevel.value}")
-
-
-def get_runlevel():
-    system_manager = get_system_manager()
-    if system_manager == SystemManagerType.systemd:
-        result = TestRun.executor.run_expect_success("systemctl get-default")
-        try:
-            name = result.stdout.split(".")[0].replace("-", "_")
-            return Runlevel[name]
-        except Exception:
-            raise Exception(f"Cannot parse '{result.output}' to runlevel.")
-    else:
-        result = TestRun.executor.run_expect_success("runlevel")
-        try:
-            split_output = result.stdout.split()
-            runlevel = Runlevel(int(split_output[1]))
-            return runlevel
-        except Exception:
-            raise Exception(f"Cannot parse '{result.output}' to runlevel.")
-
-
-class Udev(object):
-    @staticmethod
-    def enable():
-        TestRun.LOGGER.info("Enabling udev")
-        TestRun.executor.run_expect_success("udevadm control --start-exec-queue")
-
-    @staticmethod
-    def disable():
-        TestRun.LOGGER.info("Disabling udev")
-        TestRun.executor.run_expect_success("udevadm control --stop-exec-queue")
-
-    @staticmethod
-    def trigger():
-        TestRun.executor.run_expect_success("udevadm trigger")
-
-    @staticmethod
-    def settle():
-        TestRun.executor.run_expect_success("udevadm settle")
-
-
-def drop_caches(level: DropCachesMode = DropCachesMode.ALL):
-    TestRun.executor.run_expect_success(
-        f"echo {level.value} > /proc/sys/vm/drop_caches")
-
-
-def disable_memory_affecting_functions():
-    """Disables system functions affecting memory"""
-    # Don't allow sshd to be killed in case of out-of-memory:
-    TestRun.executor.run(
-        "echo '-1000' > /proc/`cat /var/run/sshd.pid`/oom_score_adj"
-    )
-    TestRun.executor.run(
-        "echo -17 > /proc/`cat /var/run/sshd.pid`/oom_adj"
-    )  # deprecated
-    TestRun.executor.run_expect_success(
-        f"echo {OvercommitMemoryMode.NEVER.value} > /proc/sys/vm/overcommit_memory"
-    )
-    TestRun.executor.run_expect_success("echo '100' > /proc/sys/vm/overcommit_ratio")
-    TestRun.executor.run_expect_success(
-        "echo '64      64      32' > /proc/sys/vm/lowmem_reserve_ratio"
-    )
-    TestRun.executor.run_expect_success("swapoff --all")
-    drop_caches(DropCachesMode.SLAB)
-
-
-def defaultize_memory_affecting_functions():
-    """Sets default values to system functions affecting memory"""
-    TestRun.executor.run_expect_success(
-        f"echo {OvercommitMemoryMode.DEFAULT.value} > /proc/sys/vm/overcommit_memory"
-    )
-    TestRun.executor.run_expect_success("echo 50 > /proc/sys/vm/overcommit_ratio")
-    TestRun.executor.run_expect_success(
-        "echo '256     256     32' > /proc/sys/vm/lowmem_reserve_ratio"
-    )
-    TestRun.executor.run_expect_success("swapon --all")
-
-
-def get_mem_free():
-    """Returns free amount of memory in bytes"""
-    output = TestRun.executor.run_expect_success("free -b")
-    output = output.stdout.splitlines()
-    for line in output:
-        if 'free' in line:
-            index = line.split().index('free') + 1  # 1st row has 1 element less than following rows
-        if 'Mem' in line:
-            mem_line = line.split()
-
-    return Size(int(mem_line[index]))
-
-
-def get_mem_available():
-    """Returns amount of available memory from /proc/meminfo"""
-    cmd = "cat /proc/meminfo | grep MemAvailable | awk '{ print $2 }'"
-    mem_available = TestRun.executor.run(cmd).stdout
-
-    return Size(int(mem_available), Unit.KibiByte)
-
-
-def get_module_mem_footprint(module_name):
-    """Returns allocated size of specific module's metadata from /proc/vmallocinfo"""
-    cmd = f"cat /proc/vmallocinfo | grep {module_name} | awk '{{ print $2 }}' "
-    output_lines = TestRun.executor.run(cmd).stdout.splitlines()
-    memory_used = 0
-    for line in output_lines:
-        memory_used += int(line)
-
-    return Size(memory_used)
-
-
-def allocate_memory(size: Size):
-    """Allocates given amount of memory"""
-    mount_ramfs()
-    TestRun.LOGGER.info(f"Allocating {size.get_value(Unit.MiB):0.2f} MiB of memory.")
-    bs = Size(1, Unit.Blocks512)
-    dd = (
-        Dd()
-        .block_size(bs)
-        .count(math.ceil(size / bs))
-        .input("/dev/zero")
-        .output(f"{MEMORY_MOUNT_POINT}/data")
-    )
-    output = dd.run()
-    if output.exit_code != 0:
-        raise CmdException("Allocating memory failed.", output)
-
-
-def get_number_of_processors_from_cpuinfo():
-    """Returns number of processors (count) which are listed out in /proc/cpuinfo"""
-    cmd = f"cat /proc/cpuinfo | grep processor | wc -l"
-    output = TestRun.executor.run(cmd).stdout
-
-    return int(output)
-
-
-def get_number_of_processes(process_name):
-    cmd = f"ps aux | grep {process_name} | grep -v grep | wc -l"
-    output = TestRun.executor.run(cmd).stdout
-
-    return int(output)
-
-
-def mount_ramfs():
-    """Mounts ramfs to enable allocating memory space"""
-    if not check_if_directory_exists(MEMORY_MOUNT_POINT):
-        create_directory(MEMORY_MOUNT_POINT)
-    if not is_mounted(MEMORY_MOUNT_POINT):
-        TestRun.executor.run_expect_success(f"mount -t ramfs ramfs {MEMORY_MOUNT_POINT}")
-
-
-def unmount_ramfs():
-    """Unmounts ramfs and releases whole space allocated by it in memory"""
-    TestRun.executor.run_expect_success(f"umount {MEMORY_MOUNT_POINT}")
-
-
-def download_file(url, destination_dir="/tmp"):
-    # TODO use wget module instead
-    command = ("wget --tries=3 --timeout=5 --continue --quiet "
-               f"--directory-prefix={destination_dir} {url}")
-    TestRun.executor.run_expect_success(command)
-    path = f"{destination_dir.rstrip('/')}/{File.get_name(url)}"
-    return File(path)
-
-
-def get_kernel_version():
-    version_string = TestRun.executor.run_expect_success("uname -r").stdout
-    version_string = version_string.split('-')[0]
-    return version.Version(version_string)
-
-
-class ModuleRemoveMethod(Enum):
-    rmmod = "rmmod"
-    modprobe = "modprobe -r"
-
-
-def is_kernel_module_loaded(module_name):
-    output = TestRun.executor.run(f"lsmod | grep ^{module_name}")
-    return output.exit_code == 0
-
-
-def get_sys_block_path():
-    sys_block = "/sys/class/block"
-    if not check_if_directory_exists(sys_block):
-        sys_block = "/sys/block"
-    return sys_block
-
-
-def load_kernel_module(module_name, module_args: {str, str}=None):
-    cmd = f"modprobe {module_name}"
-    if module_args is not None:
-        for key, value in module_args.items():
-            cmd += f" {key}={value}"
-    return TestRun.executor.run(cmd)
-
-
-def unload_kernel_module(module_name, unload_method: ModuleRemoveMethod = ModuleRemoveMethod.rmmod):
-    cmd = f"{unload_method.value} {module_name}"
-    return TestRun.executor.run_expect_success(cmd)
-
-
-def get_kernel_module_parameter(module_name, parameter):
-    param_file_path = f"/sys/module/{module_name}/parameters/{parameter}"
-    if not check_if_file_exists(param_file_path):
-        raise FileNotFoundError(f"File {param_file_path} does not exist!")
-    return File(param_file_path).read()
-
-
-def is_mounted(path: str):
-    if path is None or path.isspace():
-        raise Exception("Checked path cannot be empty")
-    command = f"mount | grep --fixed-strings '{path.rstrip('/')} '"
-    return TestRun.executor.run(command).exit_code == 0
-
-
-def mount_debugfs():
-    if not is_mounted(DEBUGFS_MOUNT_POINT):
-        TestRun.executor.run_expect_success(f"mount -t debugfs none {DEBUGFS_MOUNT_POINT}")
-
-
-def reload_kernel_module(module_name, module_args: {str, str}=None,
-                         unload_method: ModuleRemoveMethod = ModuleRemoveMethod.rmmod):
-    if is_kernel_module_loaded(module_name):
-        unload_kernel_module(module_name, unload_method)
-
-    Retry.run_while_false(
-        lambda: load_kernel_module(module_name, module_args).exit_code == 0,
-        timeout=timedelta(seconds=5)
-    )
-
-
-def get_module_path(module_name):
-    cmd = f"modinfo {module_name}"
-
-    # module path is in second column of first line of `modinfo` output
-    module_info = TestRun.executor.run_expect_success(cmd).stdout
-    module_path = module_info.splitlines()[0].split()[1]
-
-    return module_path
-
-
-def get_executable_path(exec_name):
-    cmd = f"which {exec_name}"
-
-    path = TestRun.executor.run_expect_success(cmd).stdout
-
-    return path
-
-
-def get_udev_service_path(unit_name):
-    cmd = f"systemctl cat {unit_name}"
-
-    # path is in second column of first line of output
-    info = TestRun.executor.run_expect_success(cmd).stdout
-    path = info.splitlines()[0].split()[1]
-
-    return path
-
-
-def kill_all_io(graceful=True):
-    if graceful:
-        # TERM signal should be used in preference to the KILL signal, since a
-        # process may install a handler for the TERM signal in order to perform
-        # clean-up steps before terminating in an orderly fashion.
-        TestRun.executor.run("killall -q --signal TERM dd fio blktrace")
-        time.sleep(3)
-    TestRun.executor.run("killall -q --signal TERM dd fio blktrace")
-    time.sleep(3)
-    TestRun.executor.run("killall -q --signal KILL dd fio blktrace")
-    TestRun.executor.run("kill -9 `ps aux | grep -i vdbench.* | awk '{ print $2 }'`")
-
-    if TestRun.executor.run("pgrep -x dd").exit_code == 0:
-        raise Exception(f"Failed to stop dd!")
-    if TestRun.executor.run("pgrep -x fio").exit_code == 0:
-        raise Exception(f"Failed to stop fio!")
-    if TestRun.executor.run("pgrep -x blktrace").exit_code == 0:
-        raise Exception(f"Failed to stop blktrace!")
-    if TestRun.executor.run("pgrep vdbench").exit_code == 0:
-        raise Exception(f"Failed to stop vdbench!")
-
-
-def sync():
-    TestRun.executor.run_expect_success("sync")
-
-
-def get_dut_cpu_number():
-    return int(TestRun.executor.run_expect_success("nproc").stdout)
-
-
-def get_dut_cpu_physical_cores():
-    """ Get list of CPU numbers that don't share physical cores """
-    output = TestRun.executor.run_expect_success("lscpu --all --parse").stdout
-
-    core_list = []
-    visited_phys_cores = []
-    for line in output.split("\n"):
-        if "#" in line:
-            continue
-
-        cpu_no, phys_core_no = line.split(",")[:2]
-        if phys_core_no not in visited_phys_cores:
-            core_list.append(cpu_no)
-            visited_phys_cores.append(phys_core_no)
-
-    return core_list
-
-
-def set_wbt_lat(device: Device, value: int):
-    if value < 0:
-        raise ValueError("Write back latency can't be negative number")
-
-    wbt_lat_config_path = posixpath.join(
-        get_sysfs_path(device.get_device_id()), "queue/wbt_lat_usec"
-    )
-
-    return TestRun.executor.run_expect_success(f"echo {value} > {wbt_lat_config_path}")
-
-
-def get_wbt_lat(device: Device):
-    wbt_lat_config_path = posixpath.join(
-        get_sysfs_path(device.get_device_id()), "queue/wbt_lat_usec"
-    )
-
-    return int(TestRun.executor.run_expect_success(f"cat {wbt_lat_config_path}").stdout)
-
-
-def get_cores_ids_range(numa_node: int):
-    output = TestRun.executor.run_expect_success(f"lscpu --all --parse").stdout
-    parse_output = re.findall(r'(\d+),(\d+),(?:\d+),(\d+),,', output, re.I)
-
-    return [element[0] for element in parse_output if int(element[2]) == numa_node]
-
-
-def create_user(username, additional_params=None):
-    command = "useradd "
-    if additional_params:
-        command += "".join([f"-{p} " for p in additional_params])
-    command += username
-    return TestRun.executor.run_expect_success(command)
-
-
-def check_if_user_exists(username):
-    return TestRun.executor.run(f"id {username}").exit_code == 0
-
-
-def get_block_device_names_list(exclude_list: List[int] = None) -> List[str]:
-    cmd = "lsblk -lo NAME"
-    if exclude_list is not None:
-        cmd += f" -e {','.join(str(type_id) for type_id in exclude_list)}"
-    devices = TestRun.executor.run_expect_success(cmd).stdout
-    devices_list = devices.splitlines()
-    devices_list.sort()
-    return devices_list