hotplug_virtio_mem: forbid virtio_mem unplug for QEMU < 8.1.0

The unplug support for virtio_mem was introduced in RHEL 9.4
with QEMU 8.1, corrects the test to only do this action when
the proper QEMU is used. Also applies the black code formatter.

Signed-off-by: mcasquer <[email protected]>

qemu/tests/hotplug_virtio_mem.py

@@ -1,10 +1,15 @@
+import re
 import time
 
+from virttest import utils_qemu
 from virttest import error_context
+from virttest import utils_misc
 
 from virttest.qemu_monitor import QMPCmdError
 from virttest.utils_misc import normalize_data_size
 from virttest.utils_test.qemu import MemoryHotplugTest
+from virttest.utils_version import VersionInterval
+
 from provider import virtio_mem_utils
 
 
@@ -17,6 +22,10 @@ def run(test, params, env):
     3) Check virtio-mem device
     4) Resize virtio-mem device twice
     5) Check virtio-mem device
+    6) If QEMU allows, try to unplug
+    7) Check the expected error message
+    8) Resize virtio-mem device to zero
+    9) Unplug successfully the virtio-mem device
 
     :param test: QEMU test object
     :param params: Dictionary with the test parameters
@@ -26,6 +35,10 @@ def run(test, params, env):
     timeout = params.get_numeric("login_timeout", 240)
     threshold = params.get_numeric("threshold", target_type=float)
     error_msg = params.get("error_msg")
+    qemu_path = utils_misc.get_qemu_binary(params)
+    qemu_version = utils_qemu.get_qemu_version(qemu_path)
+    match = re.search(r"[0-9]+\.[0-9]+\.[0-9]+(\-[0-9]+)?", str(qemu_version))
+    host_qemu = match.group(0)
     vm = env.get_vm(params["main_vm"])
     vm.verify_alive()
     session = vm.wait_for_login(timeout=timeout)
@@ -37,36 +50,41 @@ def run(test, params, env):
     device_id = vmem_dev.get_qid()
     requested_size_vmem_test = params.get("requested-size_test_%s" % target_mem)
 
-    node_id = int(vmem_dev.get_param('node'))
-    req_size = vmem_dev.get_param('requested-size')
-    initial_req_size = str(int(float(normalize_data_size(req_size, 'B'))))
+    node_id = int(vmem_dev.get_param("node"))
+    req_size = vmem_dev.get_param("requested-size")
+    initial_req_size = str(int(float(normalize_data_size(req_size, "B"))))
 
-    virtio_mem_utils.check_memory_devices(device_id, initial_req_size,
-                                          threshold, vm, test)
-    virtio_mem_utils.check_numa_plugged_mem(node_id, initial_req_size,
-                                            threshold, vm, test)
+    virtio_mem_utils.check_memory_devices(
+        device_id, initial_req_size, threshold, vm, test
+    )
+    virtio_mem_utils.check_numa_plugged_mem(
+        node_id, initial_req_size, threshold, vm, test
+    )
     for requested_size in requested_size_vmem_test.split():
-        req_size_normalized = int(float(normalize_data_size(requested_size,
-                                  'B')))
-        vm.monitor.qom_set(device_id, "requested-size",
-                           req_size_normalized)
+        req_size_normalized = int(float(normalize_data_size(requested_size, "B")))
+        vm.monitor.qom_set(device_id, "requested-size", req_size_normalized)
         time.sleep(30)
-        virtio_mem_utils.check_memory_devices(device_id, requested_size,
-                                              threshold, vm, test)
-        virtio_mem_utils.check_numa_plugged_mem(node_id, requested_size,
-                                                threshold, vm, test)
-    try:
-        hotplug_test.unplug_memory(vm, target_mem)
-    except QMPCmdError as e:
-        if error_msg not in str(e.data):
-            test.fail("Unexpected error message: %s" % str(e.data))
-        test.log.info(error_msg)
-    else:
-        test.fail("%s shouldn't have been unplugged! 'size' is greater than 0"
-                  % target_mem)
+        virtio_mem_utils.check_memory_devices(
+            device_id, requested_size, threshold, vm, test
+        )
+        virtio_mem_utils.check_numa_plugged_mem(
+            node_id, requested_size, threshold, vm, test
+        )
+    if host_qemu in VersionInterval('[8.1.0,)'):
+        try:
+            hotplug_test.unplug_memory(vm, target_mem)
+        except QMPCmdError as e:
+            if error_msg not in str(e.data):
+                test.fail("Unexpected error message: %s" % str(e.data))
+            test.log.info(error_msg)
+        else:
+            test.fail(
+                "%s shouldn't have been unplugged! 'size' is greater than 0"
+                % target_mem
+            )
 
-    vm.monitor.qom_set(device_id, "requested-size", 0)
-    time.sleep(10)
-    hotplug_test.unplug_memory(vm, target_mem)
+        vm.monitor.qom_set(device_id, "requested-size", 0)
+        time.sleep(10)
+        hotplug_test.unplug_memory(vm, target_mem)
 
     session.close()