tests: add the defer fixture

to ease the resource management in some complex tests

Signed-off-by: Gaëtan Lehmann <gaetan.lehmann@vates.tech>

Author: glehmann

conftest.py

@@ -13,6 +13,7 @@
 import lib.config as global_config
 from lib import pxe
 from lib.common import (
+    Defer,
     callable_marker,
     DiskDevName,
     HostAddress,
@@ -760,3 +761,36 @@ def cifs_iso_sr(host, cifs_iso_device_config):
     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

@@ -367,3 +367,5 @@ def _param_clear(host, xe_prefix, uuid, param_name):
     """ Common implementation for param_clear. """
     args = {'uuid': uuid, 'param-name': param_name}
     host.xe(f'{xe_prefix}-param-clear', args)
+
+Defer = Callable[[Callable[[], object]], None]

tests/storage/ext/test_ext_sr.py

@@ -5,7 +5,7 @@
 import logging
 
 from lib.commands import SSHCommandFailed
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.fistpoint import FistPoint
 from lib.host import Host
 from lib.sr import SR
@@ -88,20 +88,20 @@ def test_snapshot(self, vm_on_ext_sr):
     @pytest.mark.small_vm
     @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
     def test_coalesce(
-        self, storage_test_vm: VM, vdi_on_ext_sr: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, vdi_on_ext_sr: VDI, vdi_op: CoalesceOperation, defer: Defer
     ):
-        coalesce_integrity(storage_test_vm, vdi_on_ext_sr, vdi_op, request)
+        coalesce_integrity(storage_test_vm, vdi_on_ext_sr, vdi_op, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_ext_sr: VM, compression: XVACompression, request: pytest.FixtureRequest):
-        xva_export_import(vm_on_ext_sr, compression, request)
+    def test_xva_export_import(self, vm_on_ext_sr: VM, compression: XVACompression, defer: Defer):
+        xva_export_import(vm_on_ext_sr, compression, defer)
 
     @pytest.mark.small_vm
     def test_vdi_export_import(
-        self, storage_test_vm: VM, ext_sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, ext_sr: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, ext_sr, image_format, request)
+        vdi_export_import(storage_test_vm, ext_sr, image_format, defer)
 
     # *** tests with reboots (longer tests).
 

tests/storage/lvm/test_lvm_sr.py

@@ -5,7 +5,7 @@
 import logging
 
 from lib.commands import SSHCommandFailed
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.fistpoint import FistPoint
 from lib.host import Host
 from lib.sr import SR
@@ -141,20 +141,20 @@ def test_failing_resize_on_inflate_after_setSizePhys(self, host, lvm_sr, vm_on_l
     @pytest.mark.small_vm
     @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
     def test_coalesce(
-        self, storage_test_vm: VM, vdi_on_lvm_sr: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, vdi_on_lvm_sr: VDI, vdi_op: CoalesceOperation, defer: Defer
     ):
-        coalesce_integrity(storage_test_vm, vdi_on_lvm_sr, vdi_op, request)
+        coalesce_integrity(storage_test_vm, vdi_on_lvm_sr, vdi_op, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_lvm_sr: VM, compression: XVACompression, request: pytest.FixtureRequest):
-        xva_export_import(vm_on_lvm_sr, compression, request)
+    def test_xva_export_import(self, vm_on_lvm_sr: VM, compression: XVACompression, defer: Defer):
+        xva_export_import(vm_on_lvm_sr, compression, defer)
 
     @pytest.mark.small_vm
     def test_vdi_export_import(
-        self, storage_test_vm: VM, lvm_sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, lvm_sr: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, lvm_sr, image_format, request)
+        vdi_export_import(storage_test_vm, lvm_sr, image_format, defer)
 
     # *** tests with reboots (longer tests).
 

tests/storage/lvmoiscsi/test_lvmoiscsi_sr.py

@@ -1,6 +1,6 @@
 import pytest
 
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -69,22 +69,22 @@ def test_coalesce(
         storage_test_vm: 'VM',
         vdi_on_lvmoiscsi_sr: 'VDI',
         vdi_op: CoalesceOperation,
-        request: pytest.FixtureRequest,
+        defer: Defer,
     ):
-        coalesce_integrity(storage_test_vm, vdi_on_lvmoiscsi_sr, vdi_op, request)
+        coalesce_integrity(storage_test_vm, vdi_on_lvmoiscsi_sr, vdi_op, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
     def test_xva_export_import(
-        self, vm_on_lvmoiscsi_sr: VM, compression: XVACompression, request: pytest.FixtureRequest
+        self, vm_on_lvmoiscsi_sr: VM, compression: XVACompression, defer: Defer
     ):
-        xva_export_import(vm_on_lvmoiscsi_sr, compression, request)
+        xva_export_import(vm_on_lvmoiscsi_sr, compression, defer)
 
     @pytest.mark.small_vm
     def test_vdi_export_import(
-        self, storage_test_vm: VM, lvmoiscsi_sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, lvmoiscsi_sr: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, lvmoiscsi_sr, image_format, request)
+        vdi_export_import(storage_test_vm, lvmoiscsi_sr, image_format, defer)
 
     # *** tests with reboots (longer tests).
 

tests/storage/nfs/test_nfs_sr.py

@@ -3,7 +3,7 @@
 import pytest
 
 from lib.commands import SSHCommandFailed
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -116,24 +116,24 @@ def test_snapshot(self, dispatch_nfs):
     @pytest.mark.parametrize('dispatch_nfs', ['vdi_on_nfs_sr', 'vdi_on_nfs4_sr'], indirect=True)
     @pytest.mark.parametrize('vdi_op', ['snapshot', 'clone'])
     def test_coalesce(
-        self, storage_test_vm: VM, dispatch_nfs: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, dispatch_nfs: VDI, vdi_op: CoalesceOperation, defer: Defer
     ):
-        coalesce_integrity(storage_test_vm, dispatch_nfs, vdi_op, request)
+        coalesce_integrity(storage_test_vm, dispatch_nfs, vdi_op, defer)
 
     @pytest.mark.small_vm
     # Make sure this fixture is called before the parametrized one
     @pytest.mark.usefixtures('vm_ref')
     @pytest.mark.parametrize('dispatch_nfs', ['vm_on_nfs_sr', 'vm_on_nfs4_sr'], indirect=True)
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, dispatch_nfs: VM, compression: XVACompression, request: pytest.FixtureRequest):
-        xva_export_import(dispatch_nfs, compression, request)
+    def test_xva_export_import(self, dispatch_nfs: VM, compression: XVACompression, defer: Defer):
+        xva_export_import(dispatch_nfs, compression, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize('dispatch_nfs', ['nfs_sr', 'nfs4_sr'], indirect=True)
     def test_vdi_export_import(
-        self, storage_test_vm: VM, dispatch_nfs: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, dispatch_nfs: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, dispatch_nfs, image_format, request)
+        vdi_export_import(storage_test_vm, dispatch_nfs, image_format, defer)
 
     # *** tests with reboots (longer tests).
 

tests/storage/storage.py

@@ -5,7 +5,7 @@
 import logging
 
 from lib.commands import SSHCommandFailed
-from lib.common import GiB, strtobool, wait_for, wait_for_not
+from lib.common import Defer, GiB, strtobool, wait_for, wait_for_not
 from lib.host import Host
 from lib.sr import SR
 from lib.vdi import VDI, ImageFormat
@@ -181,9 +181,9 @@ def install_randstream(vm: 'VM'):
 
 CoalesceOperation = Literal['snapshot', 'clone']
 
-def coalesce_integrity(vm: VM, vdi: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest):
+def coalesce_integrity(vm: VM, vdi: VDI, vdi_op: CoalesceOperation, defer: Defer):
     vbd = vm.connect_vdi(vdi)
-    request.addfinalizer(lambda: vm.disconnect_vdi(vdi))
+    defer(lambda: vm.disconnect_vdi(vdi))
 
     dev = f'/dev/{vbd.param_get("device")}'
     vm.ssh(f"randstream generate -v {dev}")
@@ -192,7 +192,7 @@ def coalesce_integrity(vm: VM, vdi: VDI, vdi_op: CoalesceOperation, request: pyt
     match vdi_op:
         case 'clone': new_vdi = vdi.clone()
         case 'snapshot': new_vdi = vdi.snapshot()
-    request.addfinalizer(lambda: new_vdi.destroy() if new_vdi is not None else None)
+    defer(lambda: new_vdi.destroy() if new_vdi is not None else None)
 
     vm.ssh(f"randstream generate -v --seed 1 --size 128Mi {dev}")
     vm.ssh(f"randstream validate -v --expected-checksum ad2ca9af {dev}")
@@ -201,7 +201,7 @@ def coalesce_integrity(vm: VM, vdi: VDI, vdi_op: CoalesceOperation, request: pyt
 
 XVACompression = Literal['none', 'gzip', 'zstd']
 
-def xva_export_import(vm: VM, compression: XVACompression, request: pytest.FixtureRequest):
+def xva_export_import(vm: VM, compression: XVACompression, defer: Defer):
     # The tests using this function are using specific fixtures to create the VM on the expected SR
     # In consequence, we can't use the storage_test_vm, so we have to start the VM explicitly and install randstream
     vm.start()
@@ -214,25 +214,25 @@ def xva_export_import(vm: VM, compression: XVACompression, request: pytest.Fixtu
     vm.shutdown(verify=True)
 
     xva_path = f'/tmp/{vm.uuid}.xva'
-    request.addfinalizer(lambda: vm.host.ssh(f'rm -f {xva_path}'))
+    defer(lambda: vm.host.ssh(f'rm -f {xva_path}'))
     vm.export(xva_path, compression)
     # check that the zero blocks are not part of the result. Most of the data is from the random stream, so
     # compression has little effect. We just check the result is between 500 and 700 MiB
     size_mb = int(vm.host.ssh(f'du -sm --apparent-size {xva_path}').split()[0])
     assert 500 < size_mb < 700, f"unexpected xva size: {size_mb}"
 
     imported_vm = vm.host.import_vm(xva_path, vm.vdis[0].sr.uuid)
-    request.addfinalizer(lambda: imported_vm.destroy())
+    defer(lambda: imported_vm.destroy())
     imported_vm.start()
     imported_vm.wait_for_vm_running_and_ssh_up()
     imported_vm.ssh("randstream validate -v --expected-checksum 24e905d6 /root/data")
 
-def vdi_export_import(vm: VM, sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest):
+def vdi_export_import(vm: VM, sr: SR, image_format: ImageFormat, defer: Defer):
     vdi_src = sr.create_vdi(image_format=image_format)
-    request.addfinalizer(lambda: vdi_src.destroy() if vdi_src is not None else None)
+    defer(lambda: vdi_src.destroy() if vdi_src is not None else None)
 
     vbd = vm.connect_vdi(vdi_src)
-    request.addfinalizer(lambda: vm.disconnect_vdi(vdi_src) if vdi_src is not None else None)
+    defer(lambda: vm.disconnect_vdi(vdi_src) if vdi_src is not None else None)
     dev = f'/dev/{vbd.param_get("device")}'
 
     # generate 2 blocks of data of 200MiB, at position 0 and at position 500MiB
@@ -244,7 +244,7 @@ def vdi_export_import(vm: VM, sr: SR, image_format: ImageFormat, request: pytest
     vm.disconnect_vdi(vdi_src)
 
     image_path = f'/tmp/{vdi_src.uuid}.{image_format}'
-    request.addfinalizer(lambda: vm.host.ssh(f'rm -f {image_path}'))
+    defer(lambda: vm.host.ssh(f'rm -f {image_path}'))
 
     vm.host.xe('vdi-export', {'uuid': vdi_src.uuid, 'filename': image_path, 'format': image_format})
     vdi_src = vdi_src.destroy()
@@ -253,11 +253,11 @@ def vdi_export_import(vm: VM, sr: SR, image_format: ImageFormat, request: pytest
     size_mb = int(vm.host.ssh(f'du -sm --apparent-size {image_path}').split()[0])
     assert 400 < size_mb < 410, f"unexpected image size: {size_mb}"
     vdi_dest = sr.create_vdi(image_format=image_format)
-    request.addfinalizer(lambda: vdi_dest.destroy())
+    defer(lambda: vdi_dest.destroy())
 
     vm.host.xe('vdi-import', {'uuid': vdi_dest.uuid, 'filename': image_path, 'format': image_format})
     vm.connect_vdi(vdi_dest, 'xvdb')
-    request.addfinalizer(lambda: vm.disconnect_vdi(vdi_dest))
+    defer(lambda: vm.disconnect_vdi(vdi_dest))
 
     vm.ssh(f"randstream validate -v --size 200MiB --expected-checksum c6310c52 {dev}")
     vm.ssh(f"randstream validate -v --position 500MiB --size 200MiB --expected-checksum 1cb4218e {dev}")

tests/storage/xfs/test_xfs_sr.py

@@ -6,7 +6,7 @@
 import time
 
 from lib.commands import SSHCommandFailed
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.host import Host
 from lib.sr import SR
 from lib.vdi import VDI
@@ -105,20 +105,20 @@ def test_snapshot(self, vm_on_xfs_sr):
     @pytest.mark.small_vm
     @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
     def test_coalesce(
-        self, storage_test_vm: VM, vdi_on_xfs_sr: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, vdi_on_xfs_sr: VDI, vdi_op: CoalesceOperation, defer: Defer
     ):
-        coalesce_integrity(storage_test_vm, vdi_on_xfs_sr, vdi_op, request)
+        coalesce_integrity(storage_test_vm, vdi_on_xfs_sr, vdi_op, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_xfs_sr: VM, compression: XVACompression, request: pytest.FixtureRequest):
-        xva_export_import(vm_on_xfs_sr, compression, request)
+    def test_xva_export_import(self, vm_on_xfs_sr: VM, compression: XVACompression, defer: Defer):
+        xva_export_import(vm_on_xfs_sr, compression, defer)
 
     @pytest.mark.small_vm
     def test_vdi_export_import(
-        self, storage_test_vm: VM, xfs_sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, xfs_sr: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, xfs_sr, image_format, request)
+        vdi_export_import(storage_test_vm, xfs_sr, image_format, defer)
 
     # *** tests with reboots (longer tests).
 

tests/storage/zfs/test_zfs_sr.py

@@ -6,7 +6,7 @@
 import time
 
 from lib.commands import SSHCommandFailed
-from lib.common import vm_image, wait_for
+from lib.common import Defer, vm_image, wait_for
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -107,20 +107,20 @@ def test_snapshot(self, vm_on_zfs_sr):
     @pytest.mark.small_vm
     @pytest.mark.parametrize("vdi_op", ["snapshot", "clone"])
     def test_coalesce(
-        self, storage_test_vm: VM, vdi_on_zfs_sr: VDI, vdi_op: CoalesceOperation, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, vdi_on_zfs_sr: VDI, vdi_op: CoalesceOperation, defer: Defer
     ):
-        coalesce_integrity(storage_test_vm, vdi_on_zfs_sr, vdi_op, request)
+        coalesce_integrity(storage_test_vm, vdi_on_zfs_sr, vdi_op, defer)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_zfs_sr: VM, compression: XVACompression, request: pytest.FixtureRequest):
-        xva_export_import(vm_on_zfs_sr, compression, request)
+    def test_xva_export_import(self, vm_on_zfs_sr: VM, compression: XVACompression, defer: Defer):
+        xva_export_import(vm_on_zfs_sr, compression, defer)
 
     @pytest.mark.small_vm
     def test_vdi_export_import(
-        self, storage_test_vm: VM, zfs_sr: SR, image_format: ImageFormat, request: pytest.FixtureRequest
+        self, storage_test_vm: VM, zfs_sr: SR, image_format: ImageFormat, defer: Defer
     ):
-        vdi_export_import(storage_test_vm, zfs_sr, image_format, request)
+        vdi_export_import(storage_test_vm, zfs_sr, image_format, defer)
 
     # *** tests with reboots (longer tests).