tests/storage/zfs: add type hints

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

Author: glehmann

tests/storage/zfs/conftest.py

@@ -4,25 +4,24 @@
 
 import logging
 
+from lib.host import Host
 from lib.sr import SR
+from lib.vdi import VDI, ImageFormat
+from lib.vm import VM
 
 # Explicitly import package-scoped fixtures (see explanation in pkgfixtures.py)
-from lib.vdi import ImageFormat
 from pkgfixtures import host_with_saved_yum_state, sr_disk_wiped
 
-from typing import TYPE_CHECKING, Generator
-
-if TYPE_CHECKING:
-    from lib.host import Host
-    from lib.sr import SR
+from typing import Generator
 
 POOL_NAME = 'pool0'
 POOL_PATH = '/' + POOL_NAME
 
 @pytest.fixture(scope='package')
-def host_without_zfs(host):
+def host_without_zfs(host: Host) -> Generator[Host, None, None]:
     assert not host.file_exists('/usr/sbin/zpool'), \
         "zfs must not be installed on the host at the beginning of the tests"
+    yield host
 
 # NOTE: @pytest.mark.usefixtures does not parametrize this fixture.
 # To recreate host_with_zfs for each image_format value, accept
@@ -32,21 +31,21 @@ def host_without_zfs(host):
 def host_with_zfs(host_without_zfs: Host,
                   host_with_saved_yum_state: Host,
                   image_format: ImageFormat
-                  ) -> Generator[Host]:
+                  ) -> Generator[Host, None, None]:
     host = host_with_saved_yum_state
     host.yum_install(['zfs'])
     host.ssh('modprobe zfs')
     yield host
 
 @pytest.fixture(scope='package')
-def zpool_vol0(sr_disk_wiped, host_with_zfs: Host):
+def zpool_vol0(sr_disk_wiped: str, host_with_zfs: Host) -> Generator[None, None, None]:
     host_with_zfs.ssh(f'zpool create -f {POOL_NAME} /dev/{sr_disk_wiped}')
     yield
     # teardown
     host_with_zfs.ssh(f'zpool destroy {POOL_NAME}')
 
 @pytest.fixture(scope='package')
-def zfs_sr(host: Host, image_format: ImageFormat, zpool_vol0: None) -> Generator[SR]:
+def zfs_sr(host: Host, image_format: ImageFormat, zpool_vol0: None) -> Generator[SR, None, None]:
     """ A ZFS SR on first host. """
     sr = host.sr_create('zfs', "ZFS-local-SR-test", {
         'location': POOL_PATH,
@@ -57,13 +56,13 @@ def zfs_sr(host: Host, image_format: ImageFormat, zpool_vol0: None) -> Generator
     sr.destroy()
 
 @pytest.fixture(scope='module')
-def vdi_on_zfs_sr(zfs_sr: SR):
+def vdi_on_zfs_sr(zfs_sr: SR) -> Generator[VDI, None, None]:
     vdi = zfs_sr.create_vdi('ZFS-local-VDI-test')
     yield vdi
     vdi.destroy()
 
 @pytest.fixture(scope='module')
-def vm_on_zfs_sr(host, zfs_sr, vm_ref):
+def vm_on_zfs_sr(host: Host, zfs_sr: SR, vm_ref: str) -> Generator[VM, None, None]:
     vm = host.import_vm(vm_ref, sr_uuid=zfs_sr.uuid)
     yield vm
     # teardown

tests/storage/zfs/test_zfs_sr.py

@@ -7,6 +7,7 @@
 
 from lib.commands import SSHCommandFailed
 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
@@ -20,12 +21,6 @@
     xva_export_import,
 )
 
-from typing import TYPE_CHECKING
-
-if TYPE_CHECKING:
-    from lib.host import Host
-    from lib.vdi import VDI
-
 from .conftest import POOL_NAME, POOL_PATH
 
 # Requirements:
@@ -72,13 +67,13 @@ def test_create_and_destroy_sr(self, host: Host, image_format: ImageFormat) -> N
 @pytest.mark.usefixtures("zpool_vol0")
 class TestZFSSR:
     @pytest.mark.quicktest
-    def test_quicktest(self, zfs_sr):
+    def test_quicktest(self, zfs_sr: SR) -> None:
         zfs_sr.run_quicktest()
 
-    def test_vdi_is_not_open(self, vdi_on_zfs_sr):
+    def test_vdi_is_not_open(self, vdi_on_zfs_sr: VDI) -> None:
         assert not vdi_is_open(vdi_on_zfs_sr)
 
-    def test_vdi_image_format(self, vdi_on_zfs_sr: VDI, image_format: ImageFormat):
+    def test_vdi_image_format(self, vdi_on_zfs_sr: VDI, image_format: ImageFormat) -> None:
         fmt = vdi_on_zfs_sr.get_image_format()
         # feature-detect: if the SM doesn't report image-format, skip this check
         if not fmt:
@@ -87,15 +82,15 @@ def test_vdi_image_format(self, vdi_on_zfs_sr: VDI, image_format: ImageFormat):
 
     @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_zfs_sr):
+    def test_start_and_shutdown_VM(self, vm_on_zfs_sr: VM) -> None:
         vm = vm_on_zfs_sr
         vm.start()
         vm.wait_for_os_booted()
         vm.shutdown(verify=True)
 
     @pytest.mark.small_vm
     @pytest.mark.big_vm
-    def test_snapshot(self, vm_on_zfs_sr):
+    def test_snapshot(self, vm_on_zfs_sr: VM) -> None:
         vm = vm_on_zfs_sr
         vm.start()
         try:
@@ -106,23 +101,23 @@ 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):
+    def test_coalesce(self, storage_test_vm: VM, vdi_on_zfs_sr: VDI, vdi_op: CoalesceOperation) -> None:
         coalesce_integrity(storage_test_vm, vdi_on_zfs_sr, vdi_op)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_zfs_sr: VM, compression: XVACompression):
+    def test_xva_export_import(self, vm_on_zfs_sr: VM, compression: XVACompression) -> None:
         xva_export_import(vm_on_zfs_sr, compression)
 
     @pytest.mark.small_vm
-    def test_vdi_export_import(self, storage_test_vm: VM, zfs_sr: SR, image_format: ImageFormat):
+    def test_vdi_export_import(self, storage_test_vm: VM, zfs_sr: SR, image_format: ImageFormat) -> None:
         vdi_export_import(storage_test_vm, zfs_sr, image_format)
 
     # *** tests with reboots (longer tests).
 
     @pytest.mark.reboot
     @pytest.mark.small_vm
-    def test_reboot(self, vm_on_zfs_sr, host, zfs_sr):
+    def test_reboot(self, vm_on_zfs_sr: VM, host: Host, zfs_sr: SR) -> None:
         sr = zfs_sr
         vm = vm_on_zfs_sr
         host.reboot(verify=True)
@@ -133,7 +128,7 @@ def test_reboot(self, vm_on_zfs_sr, host, zfs_sr):
         vm.shutdown(verify=True)
 
     @pytest.mark.reboot
-    def test_zfs_missing(self, host: Host, zfs_sr):
+    def test_zfs_missing(self, host: Host, zfs_sr: SR) -> None:
         sr = zfs_sr
         zfs_installed = True
         try:
@@ -161,7 +156,7 @@ def test_zfs_missing(self, host: Host, zfs_sr):
                 host.ssh('modprobe zfs')
 
     @pytest.mark.reboot
-    def test_zfs_unmounted(self, host: Host, zfs_sr):
+    def test_zfs_unmounted(self, host: Host, zfs_sr: SR) -> None:
         sr = zfs_sr
         zpool_imported = True
         try:

tests/storage/zfs/test_zfs_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:
@@ -15,8 +18,8 @@
 @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_zfs_sr, local_sr_on_hostB1):
+    def test_cold_crosspool_migration(self, host: Host, hostB1: Host, vm_on_zfs_sr: VM, local_sr_on_hostB1: SR) -> None:
         cold_migration_then_come_back(vm_on_zfs_sr, host, hostB1, local_sr_on_hostB1)
 
-    def test_live_crosspool_migration(self, host, hostB1, vm_on_zfs_sr, local_sr_on_hostB1):
+    def test_live_crosspool_migration(self, host: Host, hostB1: Host, vm_on_zfs_sr: VM, local_sr_on_hostB1: SR) -> None:
         live_storage_migration_then_come_back(vm_on_zfs_sr, host, hostB1, local_sr_on_hostB1)

tests/storage/zfs/test_zfs_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:
@@ -15,8 +18,8 @@
 @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_zfs_sr, local_sr_on_hostA2):
+    def test_cold_intrapool_migration(self, host: Host, hostA2: Host, vm_on_zfs_sr: VM, local_sr_on_hostA2: SR) -> None:
         cold_migration_then_come_back(vm_on_zfs_sr, host, hostA2, local_sr_on_hostA2)
 
-    def test_live_intrapool_migration(self, host, hostA2, vm_on_zfs_sr, local_sr_on_hostA2):
+    def test_live_intrapool_migration(self, host: Host, hostA2: Host, vm_on_zfs_sr: VM, local_sr_on_hostA2: SR) -> None:
         live_storage_migration_then_come_back(vm_on_zfs_sr, host, hostA2, local_sr_on_hostA2)