Merge pull request #1 from Kamoppl/kamilg/update_TF

Kamilg/updatetf

connection/base_executor.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -35,6 +36,9 @@ class BaseExecutor:
     def wait_for_connection(self, timeout: timedelta = None):
         pass
 
+    def resolve_ip_address(self):
+        return "127.0.0.1"
+
     def run(self, command, timeout: timedelta = timedelta(minutes=30)):
         if TestRun.dut and TestRun.dut.env:
             command = f"{TestRun.dut.env} && {command}"
@@ -60,7 +64,7 @@ class BaseExecutor:
 
     def check_if_process_exists(self, pid: int):
         output = self.run(f"ps aux | awk '{{print $2 }}' | grep ^{pid}$", timedelta(seconds=10))
-        return True if output.exit_code == 0 else False
+        return output.exit_code == 0
 
     def kill_process(self, pid: int):
         # TERM signal should be used in preference to the KILL signal, since a

connection/local_executor.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -7,25 +8,40 @@ import subprocess
 from datetime import timedelta
 
 from connection.base_executor import BaseExecutor
+from core.test_run import TestRun
 from test_utils.output import Output
 
 
 class LocalExecutor(BaseExecutor):
     def _execute(self, command, timeout):
+        bash_path = TestRun.config.get("bash_path", "/bin/bash")
+
         completed_process = subprocess.run(
             command,
             shell=True,
+            executable=bash_path,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            timeout=timeout.total_seconds())
+            timeout=timeout.total_seconds(),
+        )
 
-        return Output(completed_process.stdout,
+        return Output(
-                      completed_process.stderr,
+            completed_process.stdout, completed_process.stderr, completed_process.returncode
-                      completed_process.returncode)
+        )
 
-    def _rsync(self, src, dst, delete=False, symlinks=False, checksum=False, exclude_list=[],
+    def _rsync(
-               timeout: timedelta = timedelta(seconds=90), dut_to_controller=False):
+        self,
+        src,
+        dst,
+        delete=False,
+        symlinks=False,
+        checksum=False,
+        exclude_list=[],
+        timeout: timedelta = timedelta(seconds=90),
+        dut_to_controller=False,
+    ):
         options = []
+        bash_path = TestRun.config.get("bash_path", "/bin/bash")
 
         if delete:
             options.append("--delete")
@@ -40,9 +56,11 @@ class LocalExecutor(BaseExecutor):
         completed_process = subprocess.run(
             f'rsync -r {src} {dst} {" ".join(options)}',
             shell=True,
+            executable=bash_path,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-            timeout=timeout.total_seconds())
+            timeout=timeout.total_seconds(),
+        )
 
         if completed_process.returncode:
             raise Exception(f"rsync failed:\n{completed_process}")

connection/ssh_executor.py

@@ -1,64 +1,127 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
-
+import os
+import re
 import socket
 import subprocess
+from datetime import timedelta, datetime
+
 import paramiko
 
-from datetime import timedelta, datetime
 from connection.base_executor import BaseExecutor
-from core.test_run import TestRun
+from core.test_run import TestRun, Blocked
 from test_utils.output import Output
 
 
 class SshExecutor(BaseExecutor):
-    def __init__(self, ip, username, port=22):
+    def __init__(self, host, username, port=22):
-        self.ip = ip
+        self.host = host
         self.user = username
         self.port = port
         self.ssh = paramiko.SSHClient()
+        self.ssh_config = None
         self._check_config_for_reboot_timeout()
 
     def __del__(self):
         self.ssh.close()
 
-    def connect(self, user=None, port=None,
+    def connect(self, user=None, port=None, timeout: timedelta = timedelta(seconds=30)):
-                timeout: timedelta = timedelta(seconds=30)):
+        hostname = self.host
         user = user or self.user
         port = port or self.port
         self.ssh.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+        config, sock, key_filename = None, None, None
+        # search for 'host' in SSH config
         try:
-            self.ssh.connect(self.ip, username=user,
+            path = os.path.expanduser("~/.ssh/config")
-                             port=port, timeout=timeout.total_seconds(),
+            config = paramiko.SSHConfig.from_path(path)
-                             banner_timeout=timeout.total_seconds())
+        except FileNotFoundError:
+            pass
+
+        if config is not None:
+            target = config.lookup(self.host)
+            hostname = target["hostname"]
+            key_filename = target.get("identityfile", None)
+            user = target.get("user", user)
+            port = target.get("port", port)
+            if target.get("proxyjump", None) is not None:
+                proxy = config.lookup(target["proxyjump"])
+                jump = paramiko.SSHClient()
+                jump.set_missing_host_key_policy(paramiko.AutoAddPolicy())
+                try:
+                    jump.connect(
+                        proxy["hostname"],
+                        username=proxy["user"],
+                        port=int(proxy.get("port", 22)),
+                        key_filename=proxy.get("identityfile", None),
+                    )
+                    transport = jump.get_transport()
+                    local_addr = (proxy["hostname"], int(proxy.get("port", 22)))
+                    dest_addr = (hostname, port)
+                    sock = transport.open_channel("direct-tcpip", dest_addr, local_addr)
+                except Exception as e:
+                    raise ConnectionError(
+                        f"An exception of type '{type(e)}' occurred while trying to "
+                        f"connect to proxy '{proxy['hostname']}'.\n {e}"
+                    )
+
+        if user is None:
+            TestRun.block("There is no user given in config.")
+
+        try:
+            self.ssh.connect(
+                hostname,
+                username=user,
+                port=port,
+                timeout=timeout.total_seconds(),
+                banner_timeout=timeout.total_seconds(),
+                sock=sock,
+                key_filename=key_filename,
+            )
+            self.ssh_config = config
         except paramiko.AuthenticationException as e:
             raise paramiko.AuthenticationException(
                 f"Authentication exception occurred while trying to connect to DUT. "
-                f"Please check your SSH key-based authentication.\n{e}")
+                f"Please check your SSH key-based authentication.\n{e}"
+            )
         except (paramiko.SSHException, socket.timeout) as e:
-            raise ConnectionError(f"An exception of type '{type(e)}' occurred while trying to "
+            raise ConnectionError(
-                                  f"connect to {self.ip}.\n {e}")
+                f"An exception of type '{type(e)}' occurred while trying to "
+                f"connect to {hostname}.\n {e}"
+            )
 
     def disconnect(self):
         try:
             self.ssh.close()
         except Exception:
-            raise Exception(f"An exception occurred while trying to disconnect from {self.ip}")
+            raise Exception(f"An exception occurred while trying to disconnect from {self.host}")
 
     def _execute(self, command, timeout):
         try:
-            (stdin, stdout, stderr) = self.ssh.exec_command(command,
+            (stdin, stdout, stderr) = self.ssh.exec_command(
-                                                            timeout=timeout.total_seconds())
+                command, timeout=timeout.total_seconds()
+            )
         except paramiko.SSHException as e:
-            raise ConnectionError(f"An exception occurred while executing command '{command}' on"
+            raise ConnectionError(
-                                  f" {self.ip}\n{e}")
+                f"An exception occurred while executing command '{command}' on" f" {self.host}\n{e}"
+            )
 
         return Output(stdout.read(), stderr.read(), stdout.channel.recv_exit_status())
 
-    def _rsync(self, src, dst, delete=False, symlinks=False, checksum=False, exclude_list=[],
+    def _rsync(
-               timeout: timedelta = timedelta(seconds=90), dut_to_controller=False):
+        self,
+        src,
+        dst,
+        delete=False,
+        symlinks=False,
+        checksum=False,
+        exclude_list=[],
+        timeout: timedelta = timedelta(seconds=90),
+        dut_to_controller=False,
+    ):
         options = []
 
         if delete:
@@ -71,21 +134,29 @@ class SshExecutor(BaseExecutor):
         for exclude in exclude_list:
             options.append(f"--exclude {exclude}")
 
-        src_to_dst = f"{self.user}@{self.ip}:{src} {dst} " if dut_to_controller else\
+        src_to_dst = (
-                     f"{src} {self.user}@{self.ip}:{dst} "
+            f"{self.user}@{self.host}:{src} {dst} "
+            if dut_to_controller
+            else f"{src} {self.user}@{self.host}:{dst} "
+        )
 
         try:
             completed_process = subprocess.run(
                 f'rsync -r -e "ssh -p {self.port} '
                 f'-o UserKnownHostsFile=/dev/null -o StrictHostKeyChecking=no" '
-                + src_to_dst + f'{" ".join(options)}',
+                + src_to_dst
+                + f'{" ".join(options)}',
                 shell=True,
+                executable="/bin/bash",
                 stdout=subprocess.PIPE,
                 stderr=subprocess.PIPE,
-                timeout=timeout.total_seconds())
+                timeout=timeout.total_seconds(),
+            )
         except Exception as e:
-            TestRun.LOGGER.exception(f"Exception occurred during rsync process. "
+            TestRun.LOGGER.exception(
-                                     f"Please check your SSH key-based authentication.\n{e}")
+                f"Exception occurred during rsync process. "
+                f"Please check your SSH key-based authentication.\n{e}"
+            )
 
         if completed_process.returncode:
             raise Exception(f"rsync failed:\n{completed_process}")
@@ -107,12 +178,13 @@ class SshExecutor(BaseExecutor):
     def reboot(self):
         self.run("reboot")
         self.wait_for_connection_loss()
-        self.wait_for_connection(timedelta(seconds=self.reboot_timeout)) \
+        self.wait_for_connection(
-            if self.reboot_timeout is not None else self.wait_for_connection()
+            timedelta(seconds=self.reboot_timeout)
+        ) if self.reboot_timeout is not None else self.wait_for_connection()
 
     def is_active(self):
         try:
-            self.ssh.exec_command('', timeout=5)
+            self.ssh.exec_command("", timeout=5)
             return True
         except Exception:
             return False
@@ -124,7 +196,7 @@ class SshExecutor(BaseExecutor):
                 try:
                     self.connect()
                     return
-                except paramiko.AuthenticationException:
+                except (paramiko.AuthenticationException, Blocked):
                     raise
                 except Exception:
                     continue
@@ -140,3 +212,46 @@ class SshExecutor(BaseExecutor):
                 except Exception:
                     return
             raise ConnectionError("Timeout occurred before ssh connection loss")
+
+    def resolve_ip_address(self):
+        user, hostname, port = self.user, self.host, self.port
+        key_file = None
+        pattern = rb"^Authenticated to.+\[(\d+\.\d+\.\d+\.\d+)].*$"
+        param, command = " -v", "''"
+        try:
+            if self.ssh_config:
+                host = self.ssh_config.lookup(self.host)
+                if re.fullmatch(r"^\d+\.\d+\.\d+\.\d+$", host["hostname"]):
+                    return host["hostname"]
+
+                if host.get("proxyjump", None) is not None:
+                    proxy = self.ssh_config.lookup(host["proxyjump"])
+
+                    user = proxy.get("user", user)
+                    hostname = proxy["hostname"]
+                    port = proxy.get("port", port)
+                    key_file = proxy.get("identityfile", key_file)
+                    command = f"nslookup {host['hostname']}"
+                    pattern = rb"^Address:\s+(\d+\.\d+\.\d+\.\d+)\s*$"
+                    param = ""
+                else:
+                    user = host.get("user", user)
+                    port = host.get("port", port)
+                    key_file = host.get("identityfile", key_file)
+            user_str = f"{user}@"
+            identity_str = f" -i {os.path.abspath(key_file[0])}" if key_file else ""
+
+            completed_process = subprocess.run(
+                f"ssh{identity_str} -p {port}{param} {user_str}{hostname} {command}",
+                shell=True,
+                executable="/bin/bash",
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                timeout=30,
+            )
+            matches = re.findall(
+                pattern, completed_process.stdout + completed_process.stderr, re.MULTILINE
+            )
+            return matches[-1].decode("utf-8")
+        except:
+            return None

core/pair_testing.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -31,6 +32,7 @@ import random
 
 from core.test_run import TestRun
 
+
 def testcase_id(param_set):
     if len(param_set.values) == 1:
         return param_set.values[0]
@@ -77,7 +79,6 @@ def register_testcases(metafunc, argnames, argvals):
     """
     from _pytest.python import CallSpec2, _find_parametrized_scope
     from _pytest.mark import ParameterSet
-    from _pytest.fixtures import scope2index
 
     parameter_sets = [ParameterSet(values=val, marks=[], id=None) for val in argvals]
     metafunc._validate_if_using_arg_names(argnames, False)
@@ -86,21 +87,20 @@ def register_testcases(metafunc, argnames, argvals):
 
     ids = [testcase_id(param_set) for param_set in parameter_sets]
 
-    scope = _find_parametrized_scope(argnames, metafunc._arg2fixturedefs, False)
+    scope_ = _find_parametrized_scope(argnames=argnames, arg2fixturedefs=metafunc._arg2fixturedefs,
-    scopenum = scope2index(scope, descr=f"parametrizex() call in {metafunc.function.__name__}")
+                                      indirect=False)
 
     calls = []
-    for callspec in metafunc._calls or [CallSpec2(metafunc)]:
+    for callspec in metafunc._calls or [CallSpec2()]:
         for param_index, (param_id, param_set) in enumerate(zip(ids, parameter_sets)):
-            newcallspec = callspec.copy()
+            newcallspec = callspec.setmulti(
-            newcallspec.setmulti2(
+                valtypes=arg_value_types,
-                arg_value_types,
+                argnames=argnames,
-                argnames,
+                valset=param_set.values,
-                param_set.values,
+                id=str(param_id),
-                param_id,
+                marks=param_set.marks,
-                param_set.marks,
+                scope=scope_,
-                scopenum,
+                param_index=param_index,
-                param_index,
             )
             calls.append(newcallspec)
 

core/test_run_utils.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -127,29 +128,30 @@ TestRun.__setup_disks = __setup_disks
 
 @classmethod
 def __presetup(cls):
-    cls.plugin_manager = PluginManager(cls.item, cls.config)
-    cls.plugin_manager.hook_pre_setup()
 
     if cls.config['type'] == 'ssh':
         try:
             IP(cls.config['ip'])
+            cls.config['host'] = cls.config['ip']
         except ValueError:
             TestRun.block("IP address from config is in invalid format.")
+        except KeyError:
+            if 'host' not in cls.config:
+                TestRun.block("No IP address or host defined in config")
 
         port = cls.config.get('port', 22)
 
-        if 'user' in cls.config:
+        cls.executor = SshExecutor(
-            cls.executor = SshExecutor(
+            cls.config['host'],
-                cls.config['ip'],
+            cls.config.get('user', None),
-                cls.config['user'],
+            port
-                port
+        )
-            )
-        else:
-            TestRun.block("There is no user given in config.")
     elif cls.config['type'] == 'local':
         cls.executor = LocalExecutor()
     else:
         TestRun.block("Execution type (local/ssh) is missing in DUT config!")
+    cls.plugin_manager = PluginManager(cls.item, cls.config)
+    cls.plugin_manager.hook_pre_setup()
 
 
 TestRun.presetup = __presetup
@@ -170,6 +172,7 @@ def __setup(cls):
     except Exception as ex:
         raise Exception(f"Failed to setup DUT instance:\n"
                         f"{str(ex)}\n{traceback.format_exc()}")
+    cls.dut.ip = cls.dut.ip or cls.executor.resolve_ip_address()
     cls.__setup_disks()
 
     TestRun.LOGGER.info(f"Re-seeding random number generator with seed: {cls.random_seed}")

internal_plugins/power_control_libvirt/__init__.py

@@ -1,31 +1,42 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
+
 from datetime import timedelta
 
 from connection.local_executor import LocalExecutor
 from connection.ssh_executor import SshExecutor
 from core.test_run import TestRun
 
+DEFAULT_REBOOT_TIMEOUT = 60
+
 
 class PowerControlPlugin:
     def __init__(self, params, config):
         print("Power Control LibVirt Plugin initialization")
         try:
-            self.ip = config['ip']
+            self.host = config["host"]
-            self.user = config['user']
+            self.user = config["user"]
-        except Exception:
+            self.connection_type = config["connection_type"]
-            raise Exception("Missing fields in config! ('ip' and 'user' required)")
+            self.port = config.get("port", 22)
+
+        except AttributeError:
+            raise (
+                "Missing fields in config! ('host','user','connection_type','vm_name' "
+                "are required fields)"
+            )
 
     def pre_setup(self):
         print("Power Control LibVirt Plugin pre setup")
-        if self.config['connection_type'] == 'ssh':
+        if self.connection_type == "ssh":
             self.executor = SshExecutor(
-                self.ip,
+                self.host,
                 self.user,
-                self.config.get('port', 22)
+                self.port,
             )
+            self.executor.connect()
         else:
             self.executor = LocalExecutor()
 
@@ -36,13 +47,21 @@ class PowerControlPlugin:
         pass
 
     def power_cycle(self):
-        self.executor.run(f"virsh reset {self.config['domain']}")
+        self.executor.run_expect_success(f"sudo virsh reset {TestRun.dut.virsh['vm_name']}")
-        TestRun.executor.wait_for_connection_loss()
+        TestRun.executor.disconnect()
-        timeout = TestRun.config.get('reboot_timeout')
+        TestRun.executor.wait_for_connection(timedelta(seconds=TestRun.dut.virsh["reboot_timeout"]))
-        if timeout:
+
-            TestRun.executor.wait_for_connection(timedelta(seconds=int(timeout)))
+    def check_if_vm_exists(self, vm_name) -> bool:
-        else:
+        return self.executor.run(f"sudo virsh list|grep -w {vm_name}").exit_code == 0
-            TestRun.executor.wait_for_connection()
+
+    def parse_virsh_config(self, vm_name, reboot_timeout=DEFAULT_REBOOT_TIMEOUT) -> dict | None:
+        if not self.check_if_vm_exists(vm_name=vm_name):
+            raise ValueError(f"Virsh power plugin error: couldn't find VM {vm_name} on host "
+                             f"{self.host}")
+        return {
+            "vm_name": vm_name,
+            "reboot_timeout": reboot_timeout,
+        }
 
 
 plugin_class = PowerControlPlugin

internal_plugins/vdbench/__init__.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -39,9 +40,9 @@ class Vdbench:
 
         fs_utils.create_directory(self.working_dir)
         TestRun.LOGGER.info("Copying vdbench to working dir.")
-        fs_utils.copy(posixpath.join(self.source_dir, "*"), self.working_dir,
+        fs_utils.copy(
-                      True, True)
+            source=self.source_dir, destination=self.working_dir, force=True, recursive=True
-        pass
+        )
 
     def teardown(self):
         pass

storage_devices/device.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 import posixpath

storage_devices/disk.py

@@ -173,10 +173,18 @@ class Disk(Device):
                     disk_type: DiskType,
                     serial_number,
                     block_size):
-        if disk_type is DiskType.nand or disk_type is DiskType.optane:
+
-            return NvmeDisk(path, disk_type, serial_number, block_size)
+        resolved_disk_type = None
-        else:
+        for resolved_disk_type in [NvmeDisk, SataDisk, VirtioDisk]:
-            return SataDisk(path, disk_type, serial_number, block_size)
+            try:
+                resolved_disk_type.get_unplug_path()
+            except:
+                continue
+
+        if resolved_disk_type is None:
+            raise Exception(f"Unrecognized device type for {path}")
+
+        return resolved_disk_type(path, disk_type, serial_number, block_size)
 
 
 class NvmeDisk(Disk):
@@ -204,6 +212,20 @@ class NvmeDisk(Disk):
     def get_lba_format_in_use(self):
         return nvme_cli.get_lba_format_in_use(self)
 
+    @classmethod
+    def get_unplug_path(cls, device_id):
+        base = f"/sys/block/{device_id}/device"
+        for suffix in ["/remove", "/device/remove"]:
+            try:
+                output = fs_utils.ls_item(base + suffix)
+                fs_utils.parse_ls_output(output)[0]
+            except:
+                continue
+
+            return base + suffix
+
+        raise Exception(f"Couldn't create unplug path for {device_id}")
+
 
 class SataDisk(Disk):
     plug_all_command = "for i in $(find -H /sys/devices/ -path '*/scsi_host/*/scan' -type f); " \
@@ -213,9 +235,10 @@ class SataDisk(Disk):
         Disk.__init__(self, path, disk_type, serial_number, block_size)
         self.plug_command = SataDisk.plug_all_command
         self.unplug_command = \
-            f"echo 1 > {self.get_sysfs_properties(self.get_device_id()).full_path}/device/delete"
+            f"echo 1 > {self.get_unplug_path(self.get_device_id())}"
 
-    def get_sysfs_properties(self, device_id):
+    @classmethod
+    def get_unplug_path(cls, device_id):
         ls_command = f"$(find -H /sys/devices/ -name {device_id} -type d)"
         output = fs_utils.ls_item(f"{ls_command}")
         sysfs_addr = fs_utils.parse_ls_output(output)[0]
@@ -223,15 +246,44 @@ class SataDisk(Disk):
             raise Exception(f"Failed to find sysfs address: ls -l {ls_command}")
         dirs = sysfs_addr.full_path.split('/')
         scsi_address = dirs[-3]
-        matches = re.search(
+        try:
-            r"^(?P<controller>\d+)[-:](?P<port>\d+)[-:](?P<target>\d+)[-:](?P<lun>\d+)$",
+            re.search(
-            scsi_address)
+                r"^\d+[-:]\d+[-:]\d+[-:]\d+$",
-        controller_id = matches["controller"]
+                scsi_address)
-        port_id = matches["port"]
+        except:
-        target_id = matches["target"]
+            raise Exception(f"Failed to find controller for {device_id}")
-        lun = matches["lun"]
+
+        return sysfs_addr.full_path + "/device/delete"
+
+
+class VirtioDisk(Disk):
+    plug_all_command = "echo 1 > /sys/bus/pci/rescan"
+
+    def __init__(self, path, disk_type, serial_number, block_size):
+        Disk.__init__(self, path, disk_type, serial_number, block_size)
+        self.plug_command = VirtioDisk.plug_all_command
+        self.unplug_command = \
+            f"echo 1 > {self.get_unplug_path(self.get_device_id())}"
+
+    @classmethod
+    def get_unplug_path(cls, device_id):
+        ls_command = f"$(find -H /sys/devices/ -name {device_id} -type d)"
+        output = fs_utils.ls_item(f"{ls_command}")
+        sysfs_addr = fs_utils.parse_ls_output(output)[0]
+        if not sysfs_addr:
+            raise Exception(f"Failed to find sysfs address: ls -l {ls_command}")
+
+        dirs = sysfs_addr.full_path.split("/")
+
+        for i, path_component in enumerate(dirs[::-1]):
+            # Search for scsi address in sysfs path
+            matches = re.search(
+                r"^\d+:\d+:\d+.\d+$",
+                path_component)
+            if matches:
+                break
+        else:
+            raise Exception(f"Failed to find controller for {device_id}")
+
+        return "/".join(dirs[:-i]) + "/remove"
 
-        host_path = "/".join(itertools.takewhile(lambda x: not x.startswith("host"), dirs))
-        self.plug_command = f"echo '{port_id} {target_id} {lun}' > " \
-            f"{host_path}/host{controller_id}/scsi_host/host{controller_id}/scan"
-        return sysfs_addr

test_tools/blktrace.py

@@ -1,8 +1,11 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
+
 import math
+import time
 
 from aenum import IntFlag, Enum
 
@@ -133,6 +136,10 @@ class BlkTrace:
         drop_caches(DropCachesMode.ALL)
 
         TestRun.executor.run_expect_success(f"kill -s SIGINT {self.blktrace_pid}")
+
+        time.sleep(3)
+        if TestRun.executor.check_if_process_exists(self.blktrace_pid):
+            TestRun.fail("blktrace monitoring for device is still active")
         self.blktrace_pid = -1
 
         # dummy command for swallowing output of killed command

test_tools/disk_utils.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -340,9 +341,13 @@ def wipe_filesystem(device, force=True):
 def check_if_device_supports_trim(device):
     if device.get_device_id().startswith("nvme"):
         return True
+    command_output = TestRun.executor.run(f'hdparm -I {device.path} | grep "TRIM supported"')
+    if command_output.exit_code == 0:
+        return True
     command_output = TestRun.executor.run(
-        f'hdparm -I {device.path} | grep "TRIM supported"')
+        f"lsblk -dn {device.path} -o DISC-MAX | grep -o \'[0-9]\\+\'"
-    return command_output.exit_code == 0
+    )
+    return int(command_output.stdout) > 0
 
 
 def get_device_filesystem_type(device_id):
@@ -375,7 +380,13 @@ def _is_by_id_path(path: str):
 
 def _is_dev_path_whitelisted(path: str):
     """check if given path is whitelisted"""
-    whitelisted_paths = [r"cas\d+-\d+", r"/dev/dm-\d+"]
+    whitelisted_paths = [
+        r"/dev/ram\d+",
+        r"/nullb\d+",
+        r"/dev/drbd\d+",
+        r"cas\d+-\d+",
+        r"/dev/dm-\d+"
+    ]
 
     for whitelisted_path in whitelisted_paths:
         if re.search(whitelisted_path, path) is not None:

test_tools/fs_utils.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2022 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -99,6 +100,10 @@ def check_if_regular_file_exists(path):
     return TestRun.executor.run(f"test -f \"{path}\"").exit_code == 0
 
 
+def check_if_special_block_exist(path):
+    return TestRun.executor.run(f"test -b \"{path}\"").exit_code == 0
+
+
 def check_if_symlink_exists(path):
     return TestRun.executor.run(f"test -L \"{path}\"").exit_code == 0
 
@@ -111,7 +116,7 @@ def copy(source: str,
     cmd = f"cp{' --force' if force else ''}" \
           f"{' --recursive' if recursive else ''}" \
           f"{' --dereference' if dereference else ''} " \
-          f"\"{source}\" \"{destination}\""
+          f"{source} {destination}"
     return TestRun.executor.run_expect_success(cmd)
 
 
@@ -266,7 +271,7 @@ def uncompress_archive(file, destination=None):
 def ls(path, options=''):
     default_options = "-lA --time-style=+'%Y-%m-%d %H:%M:%S'"
     output = TestRun.executor.run(
-        f"ls {default_options} {options} \"{path}\"")
+        f"ls {default_options} {options} {path}")
     return output.stdout
 
 
@@ -308,7 +313,6 @@ def parse_ls_output(ls_output, dir_path=''):
         from test_utils.filesystem.file import File, FsItem
         from test_utils.filesystem.directory import Directory
         from test_utils.filesystem.symlink import Symlink
-
         if file_type == '-':
             fs_item = File(full_path)
         elif file_type == 'd':

test_utils/disk_finder.py

@@ -39,7 +39,7 @@ def get_block_devices_list():
     block_devices = []
 
     for dev in devices:
-        if ('sd' in dev or 'nvme' in dev) and dev not in os_disks:
+        if any([prefix in dev for prefix in ["sd", "nvme", "vd"]]) and dev not in os_disks:
             block_devices.append(dev)
 
     return block_devices

test_utils/dmesg.py

@@ -0,0 +1,15 @@
+#
+# Copyright(c) 2024 Huawei Technologies Co., Ltd.
+# SPDX-License-Identifier: BSD-3-Clause
+#
+
+from core.test_run import TestRun
+from test_utils.output import Output
+
+
+def get_dmesg() -> str:
+    return TestRun.executor.run("dmesg").stdout
+
+
+def clear_dmesg() -> Output:
+    return TestRun.executor.run("dmesg -C")

test_utils/dut.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2023-2024 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -9,27 +10,43 @@ from storage_devices.disk import Disk, DiskType
 class Dut:
     def __init__(self, dut_info):
         self.config = dut_info
-        self.disks = []
+        self.disks = [
-        for disk_info in dut_info.get('disks', []):
+            Disk.create_disk(
-            self.disks.append(Disk.create_disk(disk_info['path'],
+                disk_info["path"],
-                                               DiskType[disk_info['type']],
+                DiskType[disk_info["type"]],
-                                               disk_info['serial'],
+                disk_info["serial"],
-                                               disk_info['blocksize']))
+                disk_info["blocksize"],
+                )
+            for disk_info in dut_info.get("disks", [])
+        ]
+
         self.disks.sort(key=lambda disk: disk.disk_type, reverse=True)
 
-        self.ipmi = dut_info['ipmi'] if 'ipmi' in dut_info else None
+        self.ipmi = dut_info.get("ipmi")
-        self.spider = dut_info['spider'] if 'spider' in dut_info else None
+        self.spider = dut_info.get("spider")
-        self.wps = dut_info['wps'] if 'wps' in dut_info else None
+        self.wps = dut_info.get("wps")
-        self.env = dut_info['env'] if 'env' in dut_info else None
+        self.env = dut_info.get("env")
-        self.ip = dut_info['ip'] if 'ip' in dut_info else "127.0.0.1"
+        self.ip = dut_info.get("ip")
+        self.virsh = self.__parse_virsh_config(dut_info)
 
     def __str__(self):
-        dut_str = f'ip: {self.ip}\n'
+        dut_str = f"ip: {self.ip}\n"
-        dut_str += f'ipmi: {self.ipmi["ip"]}\n' if self.ipmi is not None else ''
+        dut_str += f'ipmi: {self.ipmi["ip"]}\n' if self.ipmi is not None else ""
-        dut_str += f'spider: {self.spider["ip"]}\n' if self.spider is not None else ''
+        dut_str += f'spider: {self.spider["ip"]}\n' if self.spider is not None else ""
-        dut_str += f'wps: {self.wps["ip"]} port: {self.wps["port"]}\n' \
+        dut_str += (
-            if self.wps is not None else ''
+            f'wps: {self.wps["ip"]} port: {self.wps["port"]}\n' if self.wps is not None else ""
-        dut_str += f'disks:\n'
+        )
+        dut_str += (
+            f'virsh.vm_name: {self.virsh["vm_name"]}\n'
+            if (self.virsh is not None)
+            else ""
+        )
+        dut_str += (
+            f'virsh.reboot_timeout: {self.virsh["reboot_timeout"]}\n'
+            if (self.virsh is not None)
+            else ""
+        )
+        dut_str += f"disks:\n"
         for disk in self.disks:
             dut_str += f"\t{disk}"
         dut_str += "\n"
@@ -41,3 +58,17 @@ class Dut:
             if d.disk_type == disk_type:
                 ret_list.append(d)
         return ret_list
+
+    @staticmethod
+    def __parse_virsh_config(dut_info) -> dict | None:
+        from core.test_run import TestRun
+        if "power_control" not in TestRun.plugin_manager.req_plugins.keys():
+            return None
+        try:
+            virsh_controller = TestRun.plugin_manager.get_plugin("power_control")
+            return virsh_controller.parse_virsh_config(
+                vm_name=dut_info["vm_name"], reboot_timeout=dut_info.get("reboot_timeout")
+            )
+        except NameError:
+            return None
+

test_utils/filesystem/symlink.py

@@ -1,5 +1,6 @@
 #
 # Copyright(c) 2019-2021 Intel Corporation
+# Copyright(c) 2023 Huawei Technologies Co., Ltd.
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
@@ -9,6 +10,7 @@ from test_tools.fs_utils import (
     create_directory,
     check_if_symlink_exists,
     check_if_directory_exists,
+    check_if_special_block_exist
 )
 from test_utils.filesystem.file import File
 
@@ -76,6 +78,11 @@ class Symlink(File):
         elif not create:
             raise FileNotFoundError("Requested symlink does not exist.")
 
+        is_special_block = check_if_special_block_exist(link_path)
+        if is_special_block:
+            if not target or readlink(link_path) == readlink(target):
+                return cls(link_path)
+
         is_dir = check_if_directory_exists(link_path)
         if is_dir:
             raise IsADirectoryError(

test_utils/os_utils.py

@@ -199,7 +199,7 @@ def defaultize_memory_affecting_functions():
     TestRun.executor.run_expect_success("swapon --all")
 
 
-def get_free_memory():
+def get_mem_free():
     """Returns free amount of memory in bytes"""
     output = TestRun.executor.run_expect_success("free -b")
     output = output.stdout.splitlines()