power_control_libvirt: Actually use libvirt

internal_plugins/power_control_libvirt/__init__.py

@@ -1,74 +1,46 @@
 #
 # Copyright(c) 2020-2021 Intel Corporation
 # Copyright(c) 2023-2025 Huawei Technologies Co., Ltd.
+# Copyright(c) 2026 Unvertical
 # SPDX-License-Identifier: BSD-3-Clause
 #
 
+import libvirt
+
 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.host = config["host"]
-            self.user = config["user"]
-            self.connection_type = config["connection_type"]
-            self.port = config.get("port", 22)
+            self.url = config["url"]
+            self.vm_name = config["vm_name"]
+            self.reboot_timeout = config.get("reboot_timeout", 60)
 
         except AttributeError:
             raise (
-                "Missing fields in config! ('host','user','connection_type','vm_name' "
-                "are required fields)"
+                "Missing fields in config! ('url','vm_name' are required fields)"
             )
 
     def pre_setup(self):
         print("Power Control LibVirt Plugin pre setup")
-        if self.connection_type == "ssh":
-            self.executor = SshExecutor(
-                self.host,
-                self.user,
-                self.port,
-            )
-            self.executor.connect()
-        else:
-            self.executor = LocalExecutor()
+        self.conn = libvirt.open(self.url)
+        self.domain = self.conn.lookupByName(self.vm_name)
 
     def post_setup(self):
         pass
 
     def teardown(self):
         pass
 
-    def power_cycle(self, wait_for_connection: bool = False, delay_until_reboot: int = 0) -> None:
-        self.executor.run_expect_success(f"sudo virsh destroy {TestRun.dut.virsh['vm_name']}")
+    def power_cycle(self, wait_for_connection: bool = False) -> None:
+        self.domain.reset()
         TestRun.executor.disconnect()
-        self.executor.run_expect_success(
-            f"(sleep {delay_until_reboot} && sudo virsh start {TestRun.dut.virsh['vm_name']}) &"
-        )
         if wait_for_connection:
-            TestRun.executor.wait_for_connection(
-                timedelta(seconds=TestRun.dut.virsh["reboot_timeout"])
-            )
-
-    def check_if_vm_exists(self, vm_name) -> bool:
-        return self.executor.run(f"sudo virsh list|grep -w {vm_name}").exit_code == 0
-
-    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 {self.host}"
-            )
-        return {
-            "vm_name": vm_name,
-            "reboot_timeout": reboot_timeout,
-        }
+            TestRun.executor.wait_for_connection(timedelta(seconds=self.reboot_timeout))
 
 
 plugin_class = PowerControlPlugin

requirements.txt

@@ -0,0 +1 @@
+libvirt-python>=12.0.0

test_utils/dut.py

@@ -28,7 +28,6 @@ def __init__(self, dut_info):
         self.wps = dut_info.get("wps")
         self.env = dut_info.get("env")
         self.ip = dut_info.get("ip")
-        self.virsh = self.__parse_virsh_config(dut_info)
 
     def __str__(self):
         dut_str = f"ip: {self.ip}\n"
@@ -37,16 +36,6 @@ def __str__(self):
         dut_str += (
             f'wps: {self.wps["ip"]} port: {self.wps["port"]}\n' if self.wps is not None else ""
         )
-        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 += "disks:\n"
         for disk in self.disks:
             dut_str += f"\t{disk}"
@@ -59,17 +48,3 @@ def get_disks_of_type(self, disk_type: DiskType):
             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
-