Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
- virtual_network.qemu_test.hotunplug_nic_with_operation:
type = hotunplug_nic_with_operation
start_vm = yes
take_regular_screendumps = no
create_vm_libvirt = "yes"
kill_vm_libvirt = "yes"
repeat_times = 1
login_timeout = 360
shutdown_command = "shutdown -h now"
ovmf:
kill_vm_libvirt_options = --nvram
RHEL.7:
make_change = yes
variants:
- with_shutdown:
variants:
- after_unplug:
sub_type_after_unplug = shutdown
shutdown_method = shell
check_image = yes
- with_reboot:
sub_type_after_unplug = reboot
reboot_method = shell
variants:
- nic_virtio_net:
nic_model_nic1 = virtio
- nic_e1000e_net:
nic_model_nic1 = e1000e
102 changes: 102 additions & 0 deletions libvirt/tests/src/virtual_network/qemu/hotunplug_nic_with_operation.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,102 @@
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
#
# Copyright Redhat
#
# SPDX-License-Identifier: GPL-2.0
# Author: Transferred from tp-qemu nic_hotplug test case
#
# ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
from virttest import virsh
from virttest import utils_misc
from virttest import utils_net

from virttest.libvirt_xml import vm_xml
from virttest.utils_test import libvirt

from provider.virtual_network import network_base

virsh_opt = {"debug": True, "ignore_status": False}


def run_sub_test(test, vm, params, plug_tag):
"""
Run subtest before/after hotplug/unplug device.

:param test: test object
:param vm: VM object
:param params: test parameters
:param plug_tag: identify when to run subtest, ex, after_plug
:return: whether vm was successfully shut-down if needed
"""
sub_type = params.get("sub_type_%s" % plug_tag)
login_timeout = params.get_numeric("login_timeout", 360)
vm.cleanup_serial_console()
vm.create_serial_console()
session = vm.wait_for_serial_login(timeout=login_timeout)
shutdown_method = params.get("shutdown_method", "shell")
shutdown_command = params.get("shutdown_command")
if sub_type == "reboot":
test.log.info("Running sub test '%s' %s", sub_type, plug_tag)
if params.get("reboot_method"):
vm.reboot(session, params["reboot_method"], 0, login_timeout, serial=True)
elif sub_type == "shutdown":
test.log.info("Running sub test '%s' %s", sub_type, plug_tag)
if shutdown_method == "shell":
session.sendline(shutdown_command)
if not utils_misc.wait_for(lambda: vm.state() == "shut off", 120):
test.fail("VM failed to shutdown within timeout")

Comment on lines +21 to +48
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Fix run_sub_test contract: always close session and return a boolean (shutdown outcome).
Right now the function leaks the serial session (Line 35) and never returns anything despite the docstring (Line 29), and shutdown_command may be None (Line 37) but is used (Line 45).

 def run_sub_test(test, vm, params, plug_tag):
@@
-    session = vm.wait_for_serial_login(timeout=login_timeout)
+    session = None
+    try:
+        session = vm.wait_for_serial_login(timeout=login_timeout)
         shutdown_method = params.get("shutdown_method", "shell")
         shutdown_command = params.get("shutdown_command")
         if sub_type == "reboot":
             test.log.info("Running sub test '%s' %s", sub_type, plug_tag)
             if params.get("reboot_method"):
                 vm.reboot(session, params["reboot_method"], 0, login_timeout, serial=True)
+            return False
         elif sub_type == "shutdown":
             test.log.info("Running sub test '%s' %s", sub_type, plug_tag)
             if shutdown_method == "shell":
-                session.sendline(shutdown_command)
+                if not shutdown_command:
+                    test.error("shutdown_command is required when shutdown_method=shell")
+                session.sendline(shutdown_command)
             if not utils_misc.wait_for(lambda: vm.state() == "shut off", 120):
                 test.fail("VM failed to shutdown within timeout")
+            return True
+        return False
+    finally:
+        if session:
+            session.close()

<!-- fingerprinting:phantom:poseidon:falcon -->

<!-- This is an auto-generated comment by CodeRabbit -->


def run(test, params, env):
"""
Test hotplug of NIC devices with shutdown/reboot operations.

Transferred from tp-qemu nic_hotplug.one_pci.nic_virtio.with_shutdown/with_reboot
and adapted to tp-libvirt structure and function usage.

Test procedure:
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For this case's test steps should be:

1. Boot a guest with nic device
2. check it connectivity 
3. Pause/resume vm and verify connectivity again
4. Hot unplug nic device
5. Perform shutdown or reboot

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sure, Updated

1) Boot a guest with interface
2) Check connectivity
3) Pause/resume VM and verify connectivity
4) Hotunplug the interface
5) Perform shutdown or reboot operation based on test variant

:param test: QEMU test object
:param params: Dictionary with the test parameters
:param env: Dictionary with test environment
"""
vm_name = params.get("main_vm")
vm = env.get_vm(vm_name)
login_timeout = params.get_numeric("login_timeout", 360)
repeat_times = params.get_numeric("repeat_times", 1)
ping_params = {'vm_ping_host': 'pass'}
ips = {'host_ip': utils_net.get_host_ip_address(params)}

def run_test():
"""
Execute the main interface hotplug with shutdown/reboot test.
"""
for iteration in range(repeat_times):
test.log.info("TEST_STEP 1:Start test iteration %s, guest xml", iteration + 1)
vmxml = vm_xml.VMXML.new_from_dumpxml(vm_name)
test.log.debug("Prepare guest xml:%s\n", vmxml)
iface = libvirt.get_vm_device(vmxml, "interface", index=-1)[0]

Comment on lines +81 to +84
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Guard NIC lookup: get_vm_device(...)[0] can throw if no interface is found.
Fail fast with a clearer error to avoid an opaque IndexError.

             vmxml = vm_xml.VMXML.new_from_dumpxml(vm_name)
             test.log.debug("Prepare guest xml:%s\n", vmxml)
-            iface = libvirt.get_vm_device(vmxml, "interface", index=-1)[0]
+            ifaces = libvirt.get_vm_device(vmxml, "interface", index=-1)
+            if not ifaces:
+                test.fail("No interface device found in VM XML")
+            iface = ifaces[0]

<details>
<summary>🤖 Prompt for AI Agents</summary>

In libvirt/tests/src/virtual_network/qemu/hotunplug_nic_with_operation.py around
lines 81 to 84, the code directly indexes the result of
libvirt.get_vm_device(...)[0] which raises an opaque IndexError when no
interface is found; change it to first assign the result to a variable (e.g.,
ifaces = libvirt.get_vm_device(vmxml, "interface", index=-1)), check if not
ifaces and call test.fail("No interface device found in VM XML") to fail fast
with a clear message, then set iface = ifaces[0]; keep the existing vmxml debug
logging.


</details>

<!-- fingerprinting:phantom:poseidon:falcon -->

<!-- This is an auto-generated comment by CodeRabbit -->

test.log.info("TEST_STEP 2: Ping host from guest")
session = vm.wait_for_login(timeout=login_timeout)
network_base.ping_check(ping_params, ips, session, force_ipv4=True)

test.log.info("TEST_STEP 3: Pause vm and resume and check connectivity again")
virsh.suspend(vm_name, **virsh_opt)
virsh.resume(vm_name, **virsh_opt)
network_base.ping_check(ping_params, ips, session, force_ipv4=True)

test.log.info("TEST_STEP 4: Ping host from guest after resume")
network_base.ping_check(ping_params, ips, session, force_ipv4=True)
session.close()

Comment on lines +89 to +97
Copy link

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟡 Minor

Remove the duplicate post-resume ping check (it runs twice).
Connectivity is checked at Line 92 and again at Lines 94-95 with the same args/session.

             test.log.info("TEST_STEP 3: Pause vm and resume and check connectivity again")
             virsh.suspend(vm_name, **virsh_opt)
             virsh.resume(vm_name, **virsh_opt)
             network_base.ping_check(ping_params, ips, session, force_ipv4=True)
-
-            test.log.info("TEST_STEP 4: Ping host from guest after resume")
-            network_base.ping_check(ping_params, ips, session, force_ipv4=True)
             session.close()

<details>
<summary>🤖 Prompt for AI Agents</summary>

In libvirt/tests/src/virtual_network/qemu/hotunplug_nic_with_operation.py around
lines 89 to 97, there is a duplicated post-resume connectivity check: remove the
second ping_check call and its preceding log ("TEST_STEP 4...") so connectivity
is only verified once after resume; ensure session.close() remains and, if
necessary, renumber or adjust test step comments to reflect the removed
duplicate.


</details>

<!-- fingerprinting:phantom:poseidon:falcon -->

<!-- This is an auto-generated comment by CodeRabbit -->

test.log.info("TEST_STEP 5: Hot-unplug the interface")
virsh.detach_device(vm_name, iface.xml, wait_for_event=True, **virsh_opt)
run_sub_test(test, vm, params, "after_unplug")

run_test()