tests/storage/zfsvol: add type hints

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

Author: glehmann

tests/storage/zfsvol/conftest.py

@@ -6,19 +6,23 @@
 
 from lib.host import Host
 from lib.sr import SR
+from lib.vdi import VDI
+from lib.vm import VM
 
 # Explicitly import package-scoped fixtures (see explanation in pkgfixtures.py)
 from pkgfixtures import host_with_saved_yum_state_toolstack_restart, sr_disk_wiped
 
+from typing import Generator
+
 @pytest.fixture(scope='package')
-def host_with_zfsvol(host_with_saved_yum_state_toolstack_restart):
+def host_with_zfsvol(host_with_saved_yum_state_toolstack_restart: Host) -> Generator[Host, None, None]:
     host = host_with_saved_yum_state_toolstack_restart
     host.yum_install(['xcp-ng-xapi-storage-volume-zfsvol'])
     host.restart_toolstack(verify=True)
     yield host
 
 @pytest.fixture(scope='package')
-def zfsvol_sr(host: Host, sr_disk_wiped, host_with_zfsvol):
+def zfsvol_sr(host: Host, sr_disk_wiped: str, host_with_zfsvol: Host) -> Generator[SR, None, None]:
     """ A ZFS Volume SR on first host. """
     device = '/dev/' + sr_disk_wiped
     sr = host.sr_create('zfs-vol', "ZFS-local-SR-test", {'device': device})
@@ -28,13 +32,13 @@ def zfsvol_sr(host: Host, sr_disk_wiped, host_with_zfsvol):
     host.ssh(f'wipefs -a {device}')
 
 @pytest.fixture(scope='module')
-def vdi_on_zfsvol_sr(zfsvol_sr: SR):
+def vdi_on_zfsvol_sr(zfsvol_sr: SR) -> Generator[VDI, None, None]:
     vdi = zfsvol_sr.create_vdi('ZFS-local-VDI-test')
     yield vdi
     vdi.destroy()
 
 @pytest.fixture(scope='module')
-def vm_on_zfsvol_sr(host, zfsvol_sr, vm_ref):
+def vm_on_zfsvol_sr(host: Host, zfsvol_sr: SR, vm_ref: str) -> Generator[VM, None, None]:
     vm = host.import_vm(vm_ref, sr_uuid=zfsvol_sr.uuid)
     yield vm
     # teardown

tests/storage/zfsvol/test_zfsvol_sr.py

@@ -5,6 +5,7 @@
 import logging
 
 from lib.common import vm_image, wait_for
+from lib.host import Host
 from lib.sr import SR
 from lib.vdi import VDI
 from lib.vm import VM
@@ -16,14 +17,15 @@
 
 pytestmark = pytest.mark.usefixtures("host_at_least_8_3")
 
+
 class TestZfsvolSRCreateDestroy:
     """
     Tests that do not use fixtures that setup the SR or import VMs,
     because they precisely need to test SR creation and destruction,
     and VM import.
     """
 
-    def test_create_and_destroy_sr(self, sr_disk_wiped, host_with_zfsvol):
+    def test_create_and_destroy_sr(self, sr_disk_wiped: str, host_with_zfsvol: Host) -> None:
         host = host_with_zfsvol
         # Create and destroy tested in the same test to leave the host as unchanged as possible
         sr = host.sr_create('zfs-vol', "ZFS-local-SR-test", {'device': '/dev/' + sr_disk_wiped}, verify=True)
@@ -38,12 +40,12 @@ class TestZfsvolVm:
 
     @pytest.mark.xfail
     @pytest.mark.quicktest
-    def test_quicktest(self, zfsvol_sr):
+    def test_quicktest(self, zfsvol_sr: SR) -> None:
         zfsvol_sr.run_quicktest()
 
     @pytest.mark.small_vm # run with a small VM to test the features
     @pytest.mark.big_vm # and ideally with a big VM to test it scales
-    def test_start_and_shutdown_VM(self, vm_on_zfsvol_sr):
+    def test_start_and_shutdown_VM(self, vm_on_zfsvol_sr: VM) -> None:
         vm = vm_on_zfsvol_sr
         vm.start()
         vm.wait_for_os_booted()
@@ -52,7 +54,7 @@ def test_start_and_shutdown_VM(self, vm_on_zfsvol_sr):
     @pytest.mark.xfail # needs support for destroying snapshots
     @pytest.mark.small_vm
     @pytest.mark.big_vm
-    def test_snapshot(self, vm_on_zfsvol_sr):
+    def test_snapshot(self, vm_on_zfsvol_sr: VM) -> None:
         vm = vm_on_zfsvol_sr
         vm.start()
         try:
@@ -64,16 +66,16 @@ def test_snapshot(self, vm_on_zfsvol_sr):
     @pytest.mark.small_vm
     @pytest.mark.parametrize("vdi_op", ["snapshot"])  # "clone" requires a snapshot
     @pytest.mark.skip("zfsvol doesn't provide vhd-parent")
-    def test_coalesce(self, storage_test_vm: VM, vdi_on_zfsvol_sr: VDI, vdi_op: CoalesceOperation):
+    def test_coalesce(self, storage_test_vm: VM, vdi_on_zfsvol_sr: VDI, vdi_op: CoalesceOperation) -> None:
         coalesce_integrity(storage_test_vm, vdi_on_zfsvol_sr, vdi_op)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_zfsvol_sr: VM, compression: XVACompression):
+    def test_xva_export_import(self, vm_on_zfsvol_sr: VM, compression: XVACompression) -> None:
         xva_export_import(vm_on_zfsvol_sr, compression)
 
     @pytest.mark.small_vm
-    def test_vdi_export_import(self, storage_test_vm: VM, zfsvol_sr: SR, image_format: ImageFormat):
+    def test_vdi_export_import(self, storage_test_vm: VM, zfsvol_sr: SR, image_format: ImageFormat) -> None:
         vm = storage_test_vm
         sr = zfsvol_sr
         vdi: VDI | None = sr.create_vdi(image_format=image_format)
@@ -112,7 +114,7 @@ def test_vdi_export_import(self, storage_test_vm: VM, zfsvol_sr: SR, image_forma
 
     @pytest.mark.reboot
     @pytest.mark.small_vm
-    def test_reboot(self, vm_on_zfsvol_sr, host, zfsvol_sr):
+    def test_reboot(self, vm_on_zfsvol_sr: VM, host: Host, zfsvol_sr: SR) -> None:
         sr = zfsvol_sr
         vm = vm_on_zfsvol_sr
         host.reboot(verify=True)

tests/storage/zfsvol/test_zfsvol_sr_crosspool_migration.py

@@ -1,5 +1,8 @@
 import pytest
 
+from lib.host import Host
+from lib.sr import SR
+from lib.vm import VM
 from tests.storage import cold_migration_then_come_back, live_storage_migration_then_come_back
 
 # Requirements:
@@ -16,8 +19,12 @@
 @pytest.mark.big_vm # and ideally with a big VM to test it scales
 @pytest.mark.usefixtures("hostB1", "local_sr_on_hostB1")
 class Test:
-    def test_cold_crosspool_migration(self, host, hostB1, vm_on_zfsvol_sr, local_sr_on_hostB1):
+    def test_cold_crosspool_migration(
+        self, host: Host, hostB1: Host, vm_on_zfsvol_sr: VM, local_sr_on_hostB1: SR
+    ) -> None:
         cold_migration_then_come_back(vm_on_zfsvol_sr, host, hostB1, local_sr_on_hostB1)
 
-    def test_live_crosspool_migration(self, host, hostB1, vm_on_zfsvol_sr, local_sr_on_hostB1):
+    def test_live_crosspool_migration(
+        self, host: Host, hostB1: Host, vm_on_zfsvol_sr: VM, local_sr_on_hostB1: SR
+    ) -> None:
         live_storage_migration_then_come_back(vm_on_zfsvol_sr, host, hostB1, local_sr_on_hostB1)

tests/storage/zfsvol/test_zfsvol_sr_intrapool_migration.py

@@ -1,5 +1,8 @@
 import pytest
 
+from lib.host import Host
+from lib.sr import SR
+from lib.vm import VM
 from tests.storage import cold_migration_then_come_back, live_storage_migration_then_come_back
 
 # Requirements:
@@ -16,8 +19,12 @@
 @pytest.mark.big_vm # and ideally with a big VM to test it scales
 @pytest.mark.usefixtures("hostA2", "local_sr_on_hostA2")
 class Test:
-    def test_cold_intrapool_migration(self, host, hostA2, vm_on_zfsvol_sr, local_sr_on_hostA2):
+    def test_cold_intrapool_migration(
+        self, host: Host, hostA2: Host, vm_on_zfsvol_sr: VM, local_sr_on_hostA2: SR
+    ) -> None:
         cold_migration_then_come_back(vm_on_zfsvol_sr, host, hostA2, local_sr_on_hostA2)
 
-    def test_live_intrapool_migration(self, host, hostA2, vm_on_zfsvol_sr, local_sr_on_hostA2):
+    def test_live_intrapool_migration(
+        self, host: Host, hostA2: Host, vm_on_zfsvol_sr: VM, local_sr_on_hostA2: SR
+    ) -> None:
         live_storage_migration_then_come_back(vm_on_zfsvol_sr, host, hostA2, local_sr_on_hostA2)