test-framework: update plugins to work with TIO project + fix plugins load

Signed-off-by: Kamil Gierszewski <kamil.gierszewski@huawei.com>

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,8 +128,6 @@ 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:
@@ -151,6 +150,8 @@ def __presetup(cls):
         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

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

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 None
+        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
+