Avoid destroying/unloading the resources before entering the debugger

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

lib/common.py

@@ -44,6 +44,7 @@
 
 HostAddress: TypeAlias = str
 DiskDevName: TypeAlias = str
+Defer: TypeAlias = Callable[[Callable[[], object]], None]
 
 class PackageManagerEnum(Enum):
     UNKNOWN = 1

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:

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)

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).

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)

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",
+                        ])