diff --git a/conftest.py b/conftest.py index 2f8178bca..dc8dacae4 100644 --- a/conftest.py +++ b/conftest.py @@ -13,6 +13,7 @@ import lib.config as global_config from lib import pxe from lib.common import ( + Defer, DiskDevName, HostAddress, callable_marker, @@ -766,3 +767,36 @@ def cifs_iso_sr(host: Host, cifs_iso_device_config: dict[str, Any]) -> Generator yield sr # teardown sr.forget() + +@pytest.fixture() +def defer(request: pytest.FixtureRequest) -> Defer: + """ + A Go-inspired cleanup fixture that registers functions to be executed + after the test completes. + + This fixture provides a functional alternative to 'yield' fixtures and + 'try...finally' blocks. It is particularly useful for managing resources + that must remain 'alive' during post-mortem debugging (e.g., --pdb), as + registered finalizers only execute after the debugger session exits. + + Execution Order: + Finalizers are executed in LIFO (Last-In, First-Out) order. The last + function deferred will be the first one executed during teardown. + + Usage: + def test_example(defer): + resource = create_resource() + defer(lambda: resource.cleanup()) + + # If an assertion fails here, 'resource' is still available + # for inspection in --pdb. + assert resource.is_valid() + + Args: + request: The internal pytest request object used to register finalizers. + + Returns: + The 'request.addfinalizer' method, allowing for immediate registration + of teardown logic. + """ + return request.addfinalizer diff --git a/lib/common.py b/lib/common.py index 40d47d44e..f7eac8cc0 100644 --- a/lib/common.py +++ b/lib/common.py @@ -44,6 +44,7 @@ HostAddress: TypeAlias = str DiskDevName: TypeAlias = str +Defer: TypeAlias = Callable[[Callable[[], object]], None] class PackageManagerEnum(Enum): UNKNOWN = 1 diff --git a/tests/install/conftest.py b/tests/install/conftest.py index bbc279b75..c8dfb6c40 100644 --- a/tests/install/conftest.py +++ b/tests/install/conftest.py @@ -11,7 +11,7 @@ from data import ARP_SERVER, ISO_IMAGES, ISO_IMAGES_BASE, ISO_IMAGES_CACHE, TEST_SSH_PUBKEY, TOOLS from lib import installer, pxe from lib.commands import local_cmd -from lib.common import callable_marker, url_download, wait_for +from lib.common import Defer, callable_marker, url_download, wait_for from lib.installer import AnswerFile from typing import TYPE_CHECKING, Any, Generator, Sequence @@ -271,7 +271,8 @@ def remastered_iso(installer_iso: dict[str, str | bool], answerfile: AnswerFile yield remastered_iso @pytest.fixture(scope='function') -def vm_booted_with_installer(host: Host, create_vms: list[VM], remastered_iso: str) -> Generator[VM, None, None]: +def vm_booted_with_installer(host: Host, create_vms: list[VM], remastered_iso: str, defer: Defer) \ + -> Generator[VM, None, None]: host_vm, = create_vms # one single VM iso = remastered_iso @@ -280,52 +281,39 @@ def vm_booted_with_installer(host: Host, create_vms: list[VM], remastered_iso: s assert mac_address is not None logging.info("Host VM has MAC %s", mac_address) - remote_iso = None - try: - remote_iso = host.pool.push_iso(iso) - host_vm.insert_cd(os.path.basename(remote_iso)) - - try: - host_vm.start() - wait_for(host_vm.is_running, "Wait for host VM running") - - # catch host-vm IP address - wait_for(lambda: pxe.arp_addresses_for(mac_address), - "Wait for DHCP server to see Host VM in ARP tables", - timeout_secs=10 * 60) - ips = pxe.arp_addresses_for(mac_address) - logging.info("Host VM has IPs %s", ips) - assert len(ips) == 1 - host_vm.ip = ips[0] - ip = host_vm.ip - assert ip is not None - - # host may not be up if ARP cache was filled - wait_for(lambda: local_cmd(["ping", "-c1", ip], check=False), - "Wait for host up", timeout_secs=10 * 60, retry_delay_secs=10) - wait_for(lambda: local_cmd(["nc", "-zw5", ip, "22"], check=False), - "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5) - - yield host_vm - - logging.info("Shutting down Host VM") - assert host_vm.ip is not None - installer.poweroff(host_vm.ip) - wait_for(host_vm.is_halted, "Wait for host VM halted") - - except Exception as e: - logging.critical("caught exception %s", e) - host_vm.shutdown(force=True) - raise - except KeyboardInterrupt: - logging.warning("keyboard interrupt") - host_vm.shutdown(force=True) - raise - - host_vm.eject_cd() - finally: - if remote_iso: - host.pool.remove_iso(remote_iso) + remote_iso = host.pool.push_iso(iso) + host_vm.insert_cd(os.path.basename(remote_iso)) + defer(lambda: host.pool.remove_iso(remote_iso)) + + host_vm.start() + defer(lambda: host_vm.shutdown(force=True)) + wait_for(host_vm.is_running, "Wait for host VM running") + + # catch host-vm IP address + wait_for(lambda: pxe.arp_addresses_for(mac_address), + "Wait for DHCP server to see Host VM in ARP tables", + timeout_secs=10 * 60) + ips = pxe.arp_addresses_for(mac_address) + logging.info("Host VM has IPs %s", ips) + assert len(ips) == 1 + host_vm.ip = ips[0] + ip = host_vm.ip + assert ip is not None + + # host may not be up if ARP cache was filled + wait_for(lambda: local_cmd(["ping", "-c1", ip], check=False), + "Wait for host up", timeout_secs=10 * 60, retry_delay_secs=10) + wait_for(lambda: local_cmd(["nc", "-zw5", ip, "22"], check=False), + "Wait for ssh up on host", timeout_secs=10 * 60, retry_delay_secs=5) + + yield host_vm + + logging.info("Shutting down Host VM") + assert host_vm.ip is not None + installer.poweroff(host_vm.ip) + wait_for(host_vm.is_halted, "Wait for host VM halted") + + host_vm.eject_cd() @pytest.fixture(scope='function') def xcpng_chained(request: pytest.FixtureRequest) -> None: diff --git a/tests/migration/test_cross_pool_migration.py b/tests/migration/test_cross_pool_migration.py index e88cbcbd7..6bf736eb0 100644 --- a/tests/migration/test_cross_pool_migration.py +++ b/tests/migration/test_cross_pool_migration.py @@ -2,22 +2,19 @@ import logging -from lib.common import wait_for, wait_for_not +from lib.common import Defer, wait_for, wait_for_not from lib.host import Host from lib.vm import VM @pytest.mark.multi_vms # run on a variety of VMs @pytest.mark.big_vm # and also on a really big VM ideally -def test_cross_pool_migration(hostB1: Host, imported_vm: VM) -> None: +def test_cross_pool_migration(hostB1: Host, imported_vm: VM, defer: Defer) -> None: vm = imported_vm.clone() - try: - vm.start() - vm.wait_for_os_booted() - vm.migrate(hostB1) - wait_for_not(vm.exists_on_previous_pool, "Wait for VM not on old pool anymore") - wait_for(vm.exists, "Wait for VM on new pool") - vm.wait_for_os_booted() - vm.shutdown(verify=True) - finally: - logging.info("Destroy VM %s" % vm.uuid) - vm.destroy() + defer(lambda: vm.destroy()) + vm.start() + vm.wait_for_os_booted() + vm.migrate(hostB1) + wait_for_not(vm.exists_on_previous_pool, "Wait for VM not on old pool anymore") + wait_for(vm.exists, "Wait for VM on new pool") + vm.wait_for_os_booted() + vm.shutdown(verify=True) diff --git a/tests/misc/test_basic_without_ssh.py b/tests/misc/test_basic_without_ssh.py index 34331bb07..279bfce0f 100644 --- a/tests/misc/test_basic_without_ssh.py +++ b/tests/misc/test_basic_without_ssh.py @@ -2,7 +2,7 @@ import logging -from lib.common import wait_for +from lib.common import Defer, wait_for from lib.host import Host from lib.sr import SR from lib.vm import VM @@ -59,25 +59,21 @@ def test_suspend(self, imported_vm: VM) -> None: vm.resume() vm.wait_for_os_booted() - def test_snapshot(self, imported_vm: VM) -> None: + def test_snapshot(self, imported_vm: VM, defer: Defer) -> None: vm = imported_vm snapshot = vm.snapshot() - try: - snapshot.revert() - vm.start() - vm.wait_for_os_booted() - finally: - snapshot.destroy(verify=True) + defer(lambda: snapshot.destroy(verify=True)) + snapshot.revert() + vm.start() + vm.wait_for_os_booted() - def test_checkpoint(self, imported_vm: VM) -> None: + def test_checkpoint(self, imported_vm: VM, defer: Defer) -> None: vm = imported_vm snapshot = vm.checkpoint() - try: - snapshot.revert() - vm.resume() - vm.wait_for_os_booted() - finally: - snapshot.destroy(verify=True) + defer(lambda: snapshot.destroy(verify=True)) + snapshot.revert() + vm.resume() + vm.wait_for_os_booted() # Live migration tests # We want to test storage migration (memory+disks) and live migration without storage migration (memory only). diff --git a/tests/misc/test_export.py b/tests/misc/test_export.py index 28756aad5..4983ffa21 100644 --- a/tests/misc/test_export.py +++ b/tests/misc/test_export.py @@ -2,6 +2,7 @@ import logging +from lib.common import Defer from lib.host import Host from lib.vm import VM @@ -13,8 +14,9 @@ # From --vm parameter: # - A VM to import and export -def export_test(host: Host, vm: VM, filepath: str, compress: Literal['none', 'gzip', 'zstd'] = 'none') -> None: +def export_test(host: Host, vm: VM, filepath: str, compress: Literal['none', 'gzip', 'zstd'], defer: Defer) -> None: vm.export(filepath, compress) + defer(lambda: host.ssh(f'rm -f {filepath}', check=False)) assert host.file_exists(filepath) def check_file_type(expected: str) -> None: @@ -29,29 +31,24 @@ def check_file_type(expected: str) -> None: else: assert False, 'Unsupported compress mode' - vm2 = None - try: - vm2 = host.import_vm(filepath) - vm2.start() - vm2.wait_for_os_booted() - vm2.shutdown(verify=True) - finally: - logging.info("Delete %s" % filepath) - host.ssh(f'rm -f {filepath}', check=False) - if vm2 is not None: - vm2.destroy() + vm2 = host.import_vm(filepath) + defer(lambda: vm2.destroy()) + vm2.start() + vm2.wait_for_os_booted() + vm2.shutdown(verify=True) @pytest.mark.small_vm # run on a small VM to test the functions @pytest.mark.big_vm # and also on a really big VM ideally to make sure it scales class TestExport: - def test_export_zstd(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM) -> None: + def test_export_zstd(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM, defer: Defer) -> None: filepath = formatted_and_mounted_ext4_disk + '/test-export-zstd.xva' - export_test(host, imported_vm, filepath, 'zstd') + export_test(host, imported_vm, filepath, 'zstd', defer) - def test_export_gzip(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM) -> None: + def test_export_gzip(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM, defer: Defer) -> None: filepath = formatted_and_mounted_ext4_disk + '/test-export-gzip.xva' - export_test(host, imported_vm, filepath, 'gzip') + export_test(host, imported_vm, filepath, 'gzip', defer) - def test_export_uncompressed(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM) -> None: + def test_export_uncompressed(self, host: Host, formatted_and_mounted_ext4_disk: str, imported_vm: VM, + defer: Defer) -> None: filepath = formatted_and_mounted_ext4_disk + '/test-export-uncompressed.xva' - export_test(host, imported_vm, filepath, 'none') + export_test(host, imported_vm, filepath, 'none', defer) diff --git a/tests/packages/bugtool/test_bugtool.py b/tests/packages/bugtool/test_bugtool.py index 872123b5a..6a0bd7f13 100644 --- a/tests/packages/bugtool/test_bugtool.py +++ b/tests/packages/bugtool/test_bugtool.py @@ -2,6 +2,7 @@ import subprocess +from lib.common import Defer from lib.host import Host # This smoke test runs xen-bugtool and verifies that the archive it generates @@ -17,35 +18,27 @@ def verify_contains(host: Host, archive: str, files: list[str]) -> None: class TestsBugtool: # Verify a minimal bugtool invocation that only queries certain capabilities - def test_bugtool_entries(self, host: Host) -> None: - filename = '' - try: - filename = host.ssh('xen-bugtool -y -s --entries=xenserver-logs,xenserver-databases,system-logs') - verify_contains(host, filename, - [ - "var/log/xensource.log", - "var/log/SMlog", - "xapi-db.xml", - ]) - finally: - if filename: - host.ssh(f'rm -f {filename}') + def test_bugtool_entries(self, host: Host, defer: Defer) -> None: + filename = host.ssh('xen-bugtool -y -s --entries=xenserver-logs,xenserver-databases,system-logs') + defer(lambda: host.ssh(f'rm -f {filename}')) + verify_contains(host, filename, + [ + "var/log/xensource.log", + "var/log/SMlog", + "xapi-db.xml", + ]) # Verify that a full xen-bugtool invocation contains the most essential files - def test_bugtool_all(self, host: Host) -> None: - filename = '' - try: - filename = host.ssh('xen-bugtool -y -s') - verify_contains(host, filename, - [ - "var/log/xensource.log", - "var/log/SMlog", - "xapi-db.xml", - "acpidump.out", - "etc/fstab", - "etc/xapi.conf", - "etc/xensource/pool.conf", - ]) - finally: - if filename: - host.ssh(f'rm -f {filename}') + def test_bugtool_all(self, host: Host, defer: Defer) -> None: + filename = host.ssh('xen-bugtool -y -s') + defer(lambda: host.ssh(f'rm -f {filename}')) + verify_contains(host, filename, + [ + "var/log/xensource.log", + "var/log/SMlog", + "xapi-db.xml", + "acpidump.out", + "etc/fstab", + "etc/xapi.conf", + "etc/xensource/pool.conf", + ]) diff --git a/tests/uefi_sb/test_varstored_cert_flow.py b/tests/uefi_sb/test_varstored_cert_flow.py index 853193e06..4c05ab7c3 100644 --- a/tests/uefi_sb/test_varstored_cert_flow.py +++ b/tests/uefi_sb/test_varstored_cert_flow.py @@ -2,7 +2,7 @@ import logging -from lib.common import wait_for +from lib.common import Defer, wait_for from lib.efi import EFIAuth from lib.host import Host from lib.snapshot import Snapshot @@ -70,40 +70,31 @@ def auto_revert_vm(self, uefi_vm_and_snapshot: tuple[VM, Snapshot]) -> Generator # Revert the VM, which has the interesting effect of also shutting it down instantly revert_vm_state(vm, snapshot) - def test_snapshot_revert_restores_certs(self, uefi_vm: VM) -> None: + def test_snapshot_revert_restores_certs(self, uefi_vm: VM, defer: Defer) -> None: vm = uefi_vm vm_auths = generate_keys(as_dict=True) vm.install_uefi_certs([vm_auths[key] for key in ['PK', 'KEK', 'db', 'dbx']]) snapshot = vm.snapshot() - try: - # clear all certs - vm.set_uefi_setup_mode() - snapshot.revert() - logging.info("Check that the VM certs were restored") - for key in ['PK', 'KEK', 'db', 'dbx']: - check_vm_cert_md5sum(vm, key, vm_auths[key].auth()) - finally: - snapshot.destroy() - - def test_vm_import_restores_certs(self, uefi_vm: VM, formatted_and_mounted_ext4_disk: str) -> None: + defer(lambda: snapshot.destroy()) + # clear all certs + vm.set_uefi_setup_mode() + snapshot.revert() + logging.info("Check that the VM certs were restored") + for key in ['PK', 'KEK', 'db', 'dbx']: + check_vm_cert_md5sum(vm, key, vm_auths[key].auth()) + + def test_vm_import_restores_certs(self, uefi_vm: VM, formatted_and_mounted_ext4_disk: str, defer: Defer) -> None: vm = uefi_vm vm_auths = generate_keys(as_dict=True) vm.install_uefi_certs([vm_auths[key] for key in ['PK', 'KEK', 'db', 'dbx']]) filepath = formatted_and_mounted_ext4_disk + '/test-export-with-uefi-certs.xva' vm.export(filepath, 'zstd') - vm2 = None - try: - vm2 = vm.host.import_vm(filepath) - logging.info("Check that the VM certs were imported with the VM") - for key in ['PK', 'KEK', 'db', 'dbx']: - check_vm_cert_md5sum(vm2, key, vm_auths[key].auth()) - finally: - try: - if vm2 is not None: - logging.info(f"Destroy VM {vm2.uuid}") - vm2.destroy(verify=True) - finally: - vm.host.ssh('rm -f {filepath}', check=False) + defer(lambda: vm.host.ssh('rm -f {filepath}', check=False)) + vm2 = vm.host.import_vm(filepath) + defer(lambda: vm2.destroy()) + logging.info("Check that the VM certs were imported with the VM") + for key in ['PK', 'KEK', 'db', 'dbx']: + check_vm_cert_md5sum(vm2, key, vm_auths[key].auth()) @pytest.mark.small_vm @pytest.mark.usefixtures("host_at_least_8_3") diff --git a/tests/xen/test_ring0.py b/tests/xen/test_ring0.py index f1bbcb674..03775c45c 100644 --- a/tests/xen/test_ring0.py +++ b/tests/xen/test_ring0.py @@ -4,6 +4,7 @@ import secrets import time +from lib.common import Defer from lib.host import Host from lib.vm import VM @@ -35,15 +36,13 @@ def host_without_livepatch_loaded(host: Host) -> Host: return host -def do_execute_xst(host: Host, modname: str, testname: str | None = None) -> None: +def do_execute_xst(host: Host, modname: str, defer: Defer, testname: str | None = None) -> None: if testname is None: testname = modname host.ssh(f"modprobe xst_{modname}") - try: - host.ssh(f"echo 1 > /sys/kernel/debug/xst/{testname}/run") - host.ssh(f"grep -q 'status: pass' /sys/kernel/debug/xst/{testname}/results") - finally: - host.ssh(f"modprobe -r xst_{modname}", check=False) + defer(lambda: host.ssh(f"modprobe -r xst_{modname}", check=False)) + host.ssh(f"echo 1 > /sys/kernel/debug/xst/{testname}/run") + host.ssh(f"grep -q 'status: pass' /sys/kernel/debug/xst/{testname}/results") @pytest.mark.reboot # host_with_ring0_tests @@ -52,53 +51,49 @@ class TestRing0Tests: def test_privcmd_restrict(self, host: Host) -> None: host.ssh("/usr/bin/privcmd-restrict_test") - def test_xst_alloc_balloon(self, host: Host) -> None: - do_execute_xst(host, "alloc_balloon") + def test_xst_alloc_balloon(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "alloc_balloon", defer) - def test_xst_big_module(self, host: Host) -> None: - do_execute_xst(host, "big_module") + def test_xst_big_module(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "big_module", defer) @pytest.mark.skip("may hang system with fifo evtchn") - def test_xst_evtchn_latency(self, host: Host) -> None: - do_execute_xst(host, "evtchn_latency", "evtchn_lat") + def test_xst_evtchn_latency(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "evtchn_latency", defer, testname="evtchn_lat") @pytest.mark.skip("only makes sense for 2l evtchn") - def test_xst_evtchn_limit(self, host: Host) -> None: - do_execute_xst(host, "evtchn_limit") + def test_xst_evtchn_limit(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "evtchn_limit", defer) - def test_xst_evtchn_stress(self, host: Host) -> None: - do_execute_xst(host, "evtchn_stress") + def test_xst_evtchn_stress(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "evtchn_stress", defer) @pytest.mark.skip("leaks event channels infinitely") - def test_xst_evtchn_unbind(self, host: Host) -> None: - do_execute_xst(host, "evtchn_unbind") + def test_xst_evtchn_unbind(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "evtchn_unbind", defer) - def test_xst_get_user_pages(self, host: Host) -> None: + def test_xst_get_user_pages(self, host: Host, defer: Defer) -> None: host.ssh("modprobe xst_get_user_pages") - try: - host.ssh("/usr/bin/gup_test") - finally: - host.ssh("modprobe -r xst_get_user_pages", check=False) + defer(lambda: host.ssh("modprobe -r xst_get_user_pages", check=False)) + host.ssh("/usr/bin/gup_test") - def test_xst_grant_copy_perf(self, host: Host) -> None: - do_execute_xst(host, "grant_copy_perf", "gntcpy_perf") + def test_xst_grant_copy_perf(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "grant_copy_perf", defer, testname="gntcpy_perf") @pytest.mark.small_vm - def test_xst_ioemu_msi(self, host: Host, running_unix_vm: VM) -> None: + def test_xst_ioemu_msi(self, host: Host, running_unix_vm: VM, defer: Defer) -> None: # TODO: validate MSI reception in guest vm = running_unix_vm domid = vm.param_get("dom-id") host.ssh("modprobe xst_ioemu_msi") - try: - host.ssh(f"echo {domid} > /sys/kernel/debug/xst/ioemu_msi/domid") - host.ssh("echo 1 > /sys/kernel/debug/xst/ioemu_msi/data") - host.ssh("echo 1 > /sys/kernel/debug/xst/ioemu_msi/run") - host.ssh("grep -q 'status: pass' /sys/kernel/debug/xst/ioemu_msi/results") - finally: - host.ssh("modprobe -r xst_ioemu_msi", check=False) + defer(lambda: host.ssh("modprobe -r xst_ioemu_msi", check=False)) + host.ssh(f"echo {domid} > /sys/kernel/debug/xst/ioemu_msi/domid") + host.ssh("echo 1 > /sys/kernel/debug/xst/ioemu_msi/data") + host.ssh("echo 1 > /sys/kernel/debug/xst/ioemu_msi/run") + host.ssh("grep -q 'status: pass' /sys/kernel/debug/xst/ioemu_msi/results") @pytest.mark.usefixtures("host_at_least_8_3") - def test_xst_livepatch(self, host_without_livepatch_loaded: Host) -> None: + def test_xst_livepatch(self, host_without_livepatch_loaded: Host, defer: Defer) -> None: """ This test loads a `livepatch_testee` module, and triggers the test function `test_function_default` by writing to @@ -111,52 +106,48 @@ def test_xst_livepatch(self, host_without_livepatch_loaded: Host) -> None: around). So don't test that. """ host = host_without_livepatch_loaded - try: - host.ssh("modprobe livepatch_testee") - - marker = secrets.token_hex() - logging.debug(f"using pre-patch marker {marker}") - host.ssh(f"echo {marker} > /dev/kmsg") - host.ssh("echo 1 > /proc/livepatch_testee/cmd") - host.ssh(f"dmesg | grep -A 9999 {marker} | grep -q test_function_default_old") - - host.ssh("modprobe livepatch_tester") - - marker = secrets.token_hex() - logging.debug(f"using post-patch marker {marker}") - host.ssh(f"echo {marker} > /dev/kmsg") - host.ssh("echo 1 > /proc/livepatch_testee/cmd") - host.ssh(f"dmesg | grep -A 9999 {marker} | grep -q test_function_default_new") - finally: - host.ssh("modprobe -r livepatch_testee", check=False) - - def test_xst_memory_leak(self, host: Host) -> None: + host.ssh("modprobe livepatch_testee") + defer(lambda: host.ssh("modprobe -r livepatch_testee", check=False)) + + marker = secrets.token_hex() + logging.debug(f"using pre-patch marker {marker}") + host.ssh(f"echo {marker} > /dev/kmsg") + host.ssh("echo 1 > /proc/livepatch_testee/cmd") + host.ssh(f"dmesg | grep -A 9999 {marker} | grep -q test_function_default_old") + + host.ssh("modprobe livepatch_tester") + + marker = secrets.token_hex() + logging.debug(f"using post-patch marker {marker}") + host.ssh(f"echo {marker} > /dev/kmsg") + host.ssh("echo 1 > /proc/livepatch_testee/cmd") + host.ssh(f"dmesg | grep -A 9999 {marker} | grep -q test_function_default_new") + + def test_xst_memory_leak(self, host: Host, defer: Defer) -> None: if not host.file_exists("/sys/kernel/debug/kmemleak"): pytest.skip("CONFIG_DEBUG_KMEMLEAK is not set") host.ssh("modprobe xst_memory_leak") + defer(lambda: host.ssh("modprobe -r xst_memory_leak", check=False)) - try: - host.ssh("echo clear > /sys/kernel/debug/kmemleak") - host.ssh("echo 1 > /sys/kernel/debug/xst/memleak/run") - host.ssh("modprobe -r xst_memory_leak") - host.ssh("echo scan > /sys/kernel/debug/kmemleak") - # scan twice with a delay inbetween, otherwise the leak may not show up - time.sleep(5) - host.ssh("echo scan > /sys/kernel/debug/kmemleak") - host.ssh("grep -q unreferenced /sys/kernel/debug/kmemleak") - finally: - host.ssh("modprobe -r xst_memory_leak", check=False) + host.ssh("echo clear > /sys/kernel/debug/kmemleak") + host.ssh("echo 1 > /sys/kernel/debug/xst/memleak/run") + host.ssh("modprobe -r xst_memory_leak") + host.ssh("echo scan > /sys/kernel/debug/kmemleak") + # scan twice with a delay inbetween, otherwise the leak may not show up + time.sleep(5) + host.ssh("echo scan > /sys/kernel/debug/kmemleak") + host.ssh("grep -q unreferenced /sys/kernel/debug/kmemleak") - def test_xst_pte_set_clear_flags(self, host: Host) -> None: - do_execute_xst(host, "pte_set_clear_flags") + def test_xst_pte_set_clear_flags(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "pte_set_clear_flags", defer) - def test_xst_ptwr_xchg(self, host: Host) -> None: - do_execute_xst(host, "ptwr_xchg") + def test_xst_ptwr_xchg(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "ptwr_xchg", defer) - def test_xst_set_memory_uc(self, host: Host) -> None: - do_execute_xst(host, "set_memory_uc") + def test_xst_set_memory_uc(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "set_memory_uc", defer) @pytest.mark.skip("crashes the host, disabled by default") - def test_xst_soft_lockup(self, host: Host) -> None: - do_execute_xst(host, "soft_lockup") + def test_xst_soft_lockup(self, host: Host, defer: Defer) -> None: + do_execute_xst(host, "soft_lockup", defer)