From ed6d92f30daf6b5ab36b20118d4eb373300f5a47 Mon Sep 17 00:00:00 2001 From: hholoubk Date: Wed, 24 Sep 2025 09:41:14 +0200 Subject: [PATCH] LIBVIRTAT-22036: Add VM shutdown test to iommu_device_lifecycle This change adds new test to verify virsh shutdown command is working as expected for VM with vIOMMU devices. Also added the optional possibility to start the VM again after the shutdown and parametrize the shutdown_timeout. Improving test coverage for RHEL-109504. --- .../sriov/vIOMMU/iommu_device_lifecycle.cfg | 3 + .../sriov/vIOMMU/iommu_device_lifecycle.py | 167 +++++++++++------- 2 files changed, 108 insertions(+), 62 deletions(-) diff --git a/libvirt/tests/cfg/sriov/vIOMMU/iommu_device_lifecycle.cfg b/libvirt/tests/cfg/sriov/vIOMMU/iommu_device_lifecycle.cfg index 1950bda758..0567c24530 100644 --- a/libvirt/tests/cfg/sriov/vIOMMU/iommu_device_lifecycle.cfg +++ b/libvirt/tests/cfg/sriov/vIOMMU/iommu_device_lifecycle.cfg @@ -41,3 +41,6 @@ - reboot_many_times: loop_time = 5 - reset: + - shutdown: + start_after_shutdown = yes + shutdown_timeout = 60 diff --git a/libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py b/libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py index 4d77baf66b..a52d0f1b68 100644 --- a/libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py +++ b/libvirt/tests/src/sriov/vIOMMU/iommu_device_lifecycle.py @@ -11,17 +11,109 @@ from provider.sriov import sriov_base from provider.viommu import viommu_base from provider.sriov import check_points as sriov_check_points +from provider.vm_lifecycle import lifecycle_base VIRSH_ARGS = {'debug': True, 'ignore_status': False} +def check_network_access(test, vm, session, need_sriov, ping_dest): + """ + Check network access in VM + """ + if need_sriov: + sriov_check_points.check_vm_network_accessed(session, ping_dest=ping_dest) + else: + s, o = utils_net.ping(ping_dest, count=5, timeout=10, session=session) + if s: + test.fail(f"Failed to ping {ping_dest}! status: {s}, output: {o}") + + +def test_save_suspend(test, vm, params, save_path, test_scenario): + """ + Test save/restore or suspend/resume scenario + """ + pid_ping, upsince = save_base.pre_save_setup(vm, serial=True) + + if test_scenario == "save_restore": + test.log.info("TEST_STEP: Save and restore the VM.") + virsh.save(vm.name, save_path, **VIRSH_ARGS) + virsh.restore(save_path, **VIRSH_ARGS) + elif test_scenario == "suspend_resume": + test.log.info("TEST_STEP: Suspend and resume the VM.") + virsh.suspend(vm.name, **VIRSH_ARGS) + virsh.resume(vm.name, **VIRSH_ARGS) + + save_base.post_save_check(vm, pid_ping, upsince, serial=True) + + +def test_reboot_reset_shutdown(test, vm, params, need_sriov, ping_dest, test_scenario): + """ + Test reboot or reset scenario + """ + # Common parameters + login_timeout = int(params.get('login_timeout', 240)) + + # Setup serial console + vm.cleanup_serial_console() + vm.create_serial_console() + + if test_scenario == "shutdown": + lifecycle_base.test_shutdown(test, vm, params) + if params.get("start_after_shutdown", "no") == "yes": + test.log.info("TEST_STEP: Starting VM after shutdown") + vm.start() + else: + return + elif test_scenario == "reset": + lifecycle_base.test_reset(test, vm, login_timeout) + else: # reboot_many_times + for _ in range(int(params.get('loop_time', '5'))): + lifecycle_base.test_reset(test, vm, login_timeout) + + # Final login to check network access + session = vm.wait_for_serial_login(timeout=login_timeout) + check_network_access(test, vm, session, need_sriov, ping_dest) + session.close() + + +def setup_vm_xml(test, vm, test_obj, need_sriov=False, cleanup_ifaces=True): + """ + Setup VM XML with required configurations + """ + test.log.info("TEST_SETUP: Update VM XML.") + # Setup IOMMU test environment + test_obj.setup_iommu_test(iommu_dict=eval(test_obj.params.get('iommu_dict', '{}')), + cleanup_ifaces=cleanup_ifaces) + test_obj.prepare_controller() + + # Configure devices + for dev in ["disk", "video"]: + dev_dict = eval(test_obj.params.get('%s_dict' % dev, '{}')) + if dev == "disk": + dev_dict = test_obj.update_disk_addr(dev_dict) + libvirt_vmxml.modify_vm_device( + vm_xml.VMXML.new_from_dumpxml(vm.name), dev, dev_dict) + + # Configure network interfaces + if need_sriov: + sriov_test_obj = sriov_base.SRIOVTest(vm, test, test_obj.params) + iface_dicts = sriov_test_obj.parse_iface_dict() + test.log.debug(iface_dicts) + test_obj.params["iface_dict"] = str(sriov_test_obj.parse_iface_dict()) + + iface_dict = test_obj.parse_iface_dict() + if cleanup_ifaces: + libvirt_vmxml.modify_vm_device( + vm_xml.VMXML.new_from_dumpxml(vm.name), + "interface", iface_dict) + + def run(test, params, env): """ Test lifecycle for vm with vIOMMU enabled with different devices. """ cleanup_ifaces = "yes" == params.get("cleanup_ifaces", "yes") ping_dest = params.get('ping_dest', "127.0.0.1") - iommu_dict = eval(params.get('iommu_dict', '{}')) test_scenario = params.get("test_scenario") need_sriov = "yes" == params.get("need_sriov", "no") @@ -31,73 +123,24 @@ def run(test, params, env): rand_id = utils_misc.generate_random_string(3) save_path = f'/var/tmp/{vm_name}_{rand_id}.save' test_obj = viommu_base.VIOMMUTest(vm, test, params) - if need_sriov: - sroiv_test_obj = sriov_base.SRIOVTest(vm, test, params) + try: - test.log.info("TEST_SETUP: Update VM XML.") - test_obj.setup_iommu_test(iommu_dict=iommu_dict, - cleanup_ifaces=cleanup_ifaces) - test_obj.prepare_controller() - for dev in ["disk", "video"]: - dev_dict = eval(params.get('%s_dict' % dev, '{}')) - if dev == "disk": - dev_dict = test_obj.update_disk_addr(dev_dict) - libvirt_vmxml.modify_vm_device( - vm_xml.VMXML.new_from_dumpxml(vm.name), dev, dev_dict) - if need_sriov: - iface_dicts = sroiv_test_obj.parse_iface_dict() - test.log.debug(iface_dicts) - test_obj.params["iface_dict"] = str(sroiv_test_obj.parse_iface_dict()) - iface_dict = test_obj.parse_iface_dict() - if cleanup_ifaces: - libvirt_vmxml.modify_vm_device( - vm_xml.VMXML.new_from_dumpxml(vm.name), - "interface", iface_dict) + # Setup VM XML configuration + setup_vm_xml(test, vm, test_obj, need_sriov, cleanup_ifaces) + # Start the VM test.log.info("TEST_STEP: Start the VM.") vm.start() test.log.debug(vm_xml.VMXML.new_from_dumpxml(vm.name)) - if test_scenario in ["reboot_many_times", "reset"]: - vm.cleanup_serial_console() - vm.create_serial_console() - if test_scenario == "reset": - test.log.info("TEST_STEP: Reset the VM.") - session = vm.wait_for_serial_login( - timeout=int(params.get('login_timeout'))) - virsh.reset(vm.name, **VIRSH_ARGS) - _match, _text = session.read_until_last_line_matches( - [r"[Ll]ogin:\s*"], timeout=240, internal_timeout=0.5) - session.close() - else: - for _ in range(int(params.get('loop_time', '5'))): - test.log.info("TEST_STEP: Reboot the VM.") - session = vm.wait_for_serial_login( - timeout=int(params.get('login_timeout'))) - session.sendline(params.get("reboot_command")) - _match, _text = session.read_until_last_line_matches( - [r"[Ll]ogin:\s*"], timeout=240, internal_timeout=0.5) - session.close() - - session = vm.wait_for_serial_login( - timeout=int(params.get('login_timeout'))) - - if need_sriov: - sriov_check_points.check_vm_network_accessed(session, ping_dest=ping_dest) - else: - s, o = utils_net.ping(ping_dest, count=5, timeout=10, session=session) - if s: - test.fail("Failed to ping %s! status: %s, output: %s." % (ping_dest, s, o)) + + # Execute the appropriate test scenario + if test_scenario in ["reboot_many_times", "reset", "shutdown"]: + test_reboot_reset_shutdown(test, vm, params, need_sriov, ping_dest, test_scenario) + elif test_scenario in ["save_restore", "suspend_resume"]: + test_save_suspend(test, vm, params, save_path, test_scenario) else: - pid_ping, upsince = save_base.pre_save_setup(vm, serial=True) - if test_scenario == "save_restore": - test.log.info("TEST_STEP: Save and restore the VM.") - virsh.save(vm.name, save_path, **VIRSH_ARGS) - virsh.restore(save_path, **VIRSH_ARGS) - elif test_scenario == "suspend_resume": - test.log.info("TEST_STEP: Suspend and resume the VM.") - virsh.suspend(vm.name, **VIRSH_ARGS) - virsh.resume(vm.name, **VIRSH_ARGS) - save_base.post_save_check(vm, pid_ping, upsince, serial=True) + test.log.warning(f"Unknown scenario: {test_scenario}") + finally: test_obj.teardown_iommu_test() if os.path.exists(save_path):