Merge pull request #400 from xcp-ng/gln/storage-typehints

Author: glehmann

tests/storage/cephfs/conftest.py

@@ -4,45 +4,52 @@
 
 from lib import config
 from lib.common import exec_nofail, raise_errors
+from lib.host import Host
+from lib.pool import Pool
+from lib.sr import SR
+from lib.vdi import VDI
+from lib.vm import VM
 
 # explicit import for package-scope fixtures
 from pkgfixtures import pool_with_saved_yum_state
 
+from typing import Generator
+
 @pytest.fixture(scope='package')
-def pool_without_ceph(host):
+def pool_without_ceph(host: Host) -> Generator[Pool, None, None]:
     for h in host.pool.hosts:
         assert not host.file_exists('/usr/sbin/mount.ceph'), \
             "mount.ceph must not be installed on the host at the beginning of the tests"
     yield host.pool
 
 @pytest.fixture(scope='package')
-def pool_with_ceph(pool_without_ceph, pool_with_saved_yum_state):
+def pool_with_ceph(pool_without_ceph: Pool, pool_with_saved_yum_state: Pool) -> Generator[Pool, None, None]:
     pool = pool_with_saved_yum_state
     for h in pool.hosts:
         h.yum_install(['centos-release-ceph-jewel'], enablerepo="base,extras")
         h.yum_install(['ceph-common'], enablerepo="base,extras")
     yield pool
 
 @pytest.fixture(scope='package')
-def cephfs_device_config():
+def cephfs_device_config() -> dict[str, str]:
     return config.sr_device_config("CEPHFS_DEVICE_CONFIG")
 
 @pytest.fixture(scope='package')
-def cephfs_sr(host, cephfs_device_config, pool_with_ceph):
+def cephfs_sr(host: Host, cephfs_device_config: dict[str, str], pool_with_ceph: Pool) -> Generator[SR, None, None]:
     """ A CephFS SR on first host. """
     sr = host.sr_create('cephfs', "CephFS-SR-test", cephfs_device_config, shared=True)
     yield sr
     # teardown
     sr.destroy()
 
 @pytest.fixture(scope='module')
-def vdi_on_cephfs_sr(cephfs_sr):
+def vdi_on_cephfs_sr(cephfs_sr: SR) -> Generator[VDI, None, None]:
     vdi = cephfs_sr.create_vdi('CephFS-VDI-test')
     yield vdi
     vdi.destroy()
 
 @pytest.fixture(scope='module')
-def vm_on_cephfs_sr(host, cephfs_sr, vm_ref):
+def vm_on_cephfs_sr(host: Host, cephfs_sr: SR, vm_ref: str) -> Generator[VM, None, None]:
     vm = host.import_vm(vm_ref, sr_uuid=cephfs_sr.uuid)
     yield vm
     # teardown

tests/storage/cephfs/test_cephfs_sr.py

@@ -5,6 +5,11 @@
 
 from lib.commands import SSHCommandFailed
 from lib.common import vm_image, wait_for
+from lib.host import Host
+from lib.pool import Pool
+from lib.sr import SR
+from lib.vdi import VDI
+from lib.vm import VM
 from tests.storage import vdi_is_open
 
 # Requirements:
@@ -19,7 +24,7 @@ class TestCephFSSRCreateDestroy:
     and VM import.
     """
 
-    def test_create_cephfs_sr_without_ceph(self, host, cephfs_device_config):
+    def test_create_cephfs_sr_without_ceph(self, host: Host, cephfs_device_config: dict[str, str]) -> None:
         # This test must be the first in the series in this module
         assert not host.file_exists('/usr/sbin/mount.ceph'), \
             "mount.ceph must not be installed on the host at the beginning of the tests"
@@ -32,7 +37,9 @@ def test_create_cephfs_sr_without_ceph(self, host, cephfs_device_config):
             sr.destroy()
             assert False, "SR creation should not have succeeded!"
 
-    def test_create_and_destroy_sr(self, host, cephfs_device_config, pool_with_ceph):
+    def test_create_and_destroy_sr(
+        self, host: Host, cephfs_device_config: dict[str, str], pool_with_ceph: Pool
+    ) -> None:
         # Create and destroy tested in the same test to leave the host as unchanged as possible
         sr = host.sr_create('cephfs', "CephFS-SR-test", cephfs_device_config, shared=True, verify=True)
         # import a VM in order to detect vm import issues here rather than in the vm_on_xfs_fixture used in
@@ -44,23 +51,23 @@ def test_create_and_destroy_sr(self, host, cephfs_device_config, pool_with_ceph)
 @pytest.mark.usefixtures("cephfs_sr")
 class TestCephFSSR:
     @pytest.mark.quicktest
-    def test_quicktest(self, cephfs_sr):
+    def test_quicktest(self, cephfs_sr: SR) -> None:
         cephfs_sr.run_quicktest()
 
-    def test_vdi_is_not_open(self, vdi_on_cephfs_sr):
+    def test_vdi_is_not_open(self, vdi_on_cephfs_sr: VDI) -> None:
         assert not vdi_is_open(vdi_on_cephfs_sr)
 
     @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_cephfs_sr):
+    def test_start_and_shutdown_VM(self, vm_on_cephfs_sr: VM) -> None:
         vm = vm_on_cephfs_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_cephfs_sr):
+    def test_snapshot(self, vm_on_cephfs_sr: VM) -> None:
         vm = vm_on_cephfs_sr
         vm.start()
         try:
@@ -73,7 +80,7 @@ def test_snapshot(self, vm_on_cephfs_sr):
 
     @pytest.mark.reboot
     @pytest.mark.small_vm
-    def test_reboot(self, vm_on_cephfs_sr, host, cephfs_sr):
+    def test_reboot(self, vm_on_cephfs_sr: VM, host: Host, cephfs_sr: SR) -> None:
         sr = cephfs_sr
         vm = vm_on_cephfs_sr
         host.reboot(verify=True)
@@ -84,7 +91,7 @@ def test_reboot(self, vm_on_cephfs_sr, host, cephfs_sr):
         vm.shutdown(verify=True)
 
     @pytest.mark.reboot # reboots the host
-    def test_ceph_missing(self, host, cephfs_sr):
+    def test_ceph_missing(self, host: Host, cephfs_sr: SR) -> None:
         sr = cephfs_sr
         ceph_installed = True
         try:

tests/storage/cephfs/test_cephfs_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,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_cephfs_sr, local_sr_on_hostB1):
+    def test_cold_crosspool_migration(
+        self, host: Host, hostB1: Host, vm_on_cephfs_sr: VM, local_sr_on_hostB1: SR
+    ) -> None:
         cold_migration_then_come_back(vm_on_cephfs_sr, host, hostB1, local_sr_on_hostB1)
 
-    def test_live_crosspool_migration(self, host, hostB1, vm_on_cephfs_sr, local_sr_on_hostB1):
+    def test_live_crosspool_migration(
+        self, host: Host, hostB1: Host, vm_on_cephfs_sr: VM, local_sr_on_hostB1: SR
+    ) -> None:
         live_storage_migration_then_come_back(vm_on_cephfs_sr, host, hostB1, local_sr_on_hostB1)

tests/storage/cephfs/test_cephfs_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,12 +18,16 @@
 @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_live_intrapool_shared_migration(self, host, hostA2, vm_on_cephfs_sr):
+    def test_live_intrapool_shared_migration(self, host: Host, hostA2: Host, vm_on_cephfs_sr: VM) -> None:
         sr = vm_on_cephfs_sr.get_sr()
         live_storage_migration_then_come_back(vm_on_cephfs_sr, host, hostA2, sr)
 
-    def test_cold_intrapool_migration(self, host, hostA2, vm_on_cephfs_sr, local_sr_on_hostA2):
+    def test_cold_intrapool_migration(
+        self, host: Host, hostA2: Host, vm_on_cephfs_sr: VM, local_sr_on_hostA2: SR
+    ) -> None:
         cold_migration_then_come_back(vm_on_cephfs_sr, host, hostA2, local_sr_on_hostA2)
 
-    def test_live_intrapool_migration(self, host, hostA2, vm_on_cephfs_sr, local_sr_on_hostA2):
+    def test_live_intrapool_migration(
+        self, host: Host, hostA2: Host, vm_on_cephfs_sr: VM, local_sr_on_hostA2: SR
+    ) -> None:
         live_storage_migration_then_come_back(vm_on_cephfs_sr, host, hostA2, local_sr_on_hostA2)

tests/storage/conftest.py

@@ -5,7 +5,9 @@
 from lib.vm import VM
 from tests.storage import install_randstream
 
-def pytest_collection_modifyitems(config, items):
+from typing import Any, Generator
+
+def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
     # modify ordering so that ext is always tested first,
     # before more complex storage drivers
     for item in reversed(list(items)):
@@ -14,6 +16,6 @@ def pytest_collection_modifyitems(config, items):
             items.insert(0, item)
 
 @pytest.fixture(scope='module')
-def storage_test_vm(running_unix_vm: VM):
+def storage_test_vm(running_unix_vm: VM) -> Generator[VM, None, None]:
     install_randstream(running_unix_vm)
     yield running_unix_vm

tests/storage/ext/conftest.py

@@ -6,15 +6,16 @@
 
 from lib.host import Host
 from lib.sr import SR
-from lib.vdi import ImageFormat
+from lib.vdi import VDI, ImageFormat
+from lib.vm import VM
 
-from typing import Generator
+from typing import Any, Generator
 
 @pytest.fixture(scope='package')
 def ext_sr(host: Host,
            unused_512B_disks: dict[Host, list[Host.BlockDeviceInfo]],
            image_format: ImageFormat
-           ) -> Generator[SR]:
+           ) -> Generator[SR, None, None]:
     """ An EXT SR on first host. """
     sr_disk = unused_512B_disks[host][0]["name"]
     sr = host.sr_create('ext', "EXT-local-SR-test",
@@ -25,13 +26,13 @@ def ext_sr(host: Host,
     sr.destroy()
 
 @pytest.fixture(scope='module')
-def vdi_on_ext_sr(ext_sr: SR):
+def vdi_on_ext_sr(ext_sr: SR) -> Generator[VDI, None, None]:
     vdi = ext_sr.create_vdi('EXT-local-VDI-test')
     yield vdi
     vdi.destroy()
 
 @pytest.fixture(scope='module')
-def vm_on_ext_sr(host, ext_sr, vm_ref):
+def vm_on_ext_sr(host: Host, ext_sr: SR, vm_ref: str) -> Generator[VM, None, None]:
     vm = host.import_vm(vm_ref, sr_uuid=ext_sr.uuid)
     yield vm
     # teardown

tests/storage/ext/test_ext_sr.py

@@ -32,7 +32,7 @@ class TestEXTSRCreateDestroy:
     and VM import.
     """
 
-    def test_create_sr_with_missing_device(self, host):
+    def test_create_sr_with_missing_device(self, host: Host) -> None:
         try_to_create_sr_with_missing_device('ext', 'EXT-local-SR-test', host)
 
     def test_create_and_destroy_sr(self, host: Host,
@@ -53,13 +53,13 @@ def test_create_and_destroy_sr(self, host: Host,
 @pytest.mark.usefixtures("ext_sr")
 class TestEXTSR:
     @pytest.mark.quicktest
-    def test_quicktest(self, ext_sr):
+    def test_quicktest(self, ext_sr: SR) -> None:
         ext_sr.run_quicktest()
 
-    def test_vdi_is_not_open(self, vdi_on_ext_sr):
+    def test_vdi_is_not_open(self, vdi_on_ext_sr: VDI) -> None:
         assert not vdi_is_open(vdi_on_ext_sr)
 
-    def test_vdi_image_format(self, vdi_on_ext_sr: VDI, image_format: ImageFormat):
+    def test_vdi_image_format(self, vdi_on_ext_sr: VDI, image_format: ImageFormat) -> None:
         fmt = vdi_on_ext_sr.get_image_format()
         # feature-detect: if the SM doesn't report image-format, skip this check
         if not fmt:
@@ -68,15 +68,15 @@ def test_vdi_image_format(self, vdi_on_ext_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_ext_sr):
+    def test_start_and_shutdown_VM(self, vm_on_ext_sr: VM) -> None:
         vm = vm_on_ext_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_ext_sr):
+    def test_snapshot(self, vm_on_ext_sr: VM) -> None:
         vm = vm_on_ext_sr
         vm.start()
         try:
@@ -87,23 +87,23 @@ 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):
+    def test_coalesce(self, storage_test_vm: VM, vdi_on_ext_sr: VDI, vdi_op: CoalesceOperation) -> None:
         coalesce_integrity(storage_test_vm, vdi_on_ext_sr, vdi_op)
 
     @pytest.mark.small_vm
     @pytest.mark.parametrize("compression", ["none", "gzip", "zstd"])
-    def test_xva_export_import(self, vm_on_ext_sr: VM, compression: XVACompression):
+    def test_xva_export_import(self, vm_on_ext_sr: VM, compression: XVACompression) -> None:
         xva_export_import(vm_on_ext_sr, compression)
 
     @pytest.mark.small_vm
-    def test_vdi_export_import(self, storage_test_vm: VM, ext_sr: SR, image_format: ImageFormat):
+    def test_vdi_export_import(self, storage_test_vm: VM, ext_sr: SR, image_format: ImageFormat) -> None:
         vdi_export_import(storage_test_vm, ext_sr, image_format)
 
     # *** tests with reboots (longer tests).
 
     @pytest.mark.small_vm
     @pytest.mark.big_vm
-    def test_blktap_activate_failure(self, vm_on_ext_sr):
+    def test_blktap_activate_failure(self, vm_on_ext_sr: VM) -> None:
         from lib.fistpoint import FistPoint
         vm = vm_on_ext_sr
         with FistPoint(vm.host, "blktap_activate_inject_failure"), pytest.raises(SSHCommandFailed):
@@ -112,7 +112,7 @@ def test_blktap_activate_failure(self, vm_on_ext_sr):
 
     @pytest.mark.small_vm
     @pytest.mark.big_vm
-    def test_resize(self, vm_on_ext_sr):
+    def test_resize(self, vm_on_ext_sr: VM) -> None:
         vm = vm_on_ext_sr
         vdi = VDI(vm.vdi_uuids()[0], host=vm.host)
         old_size = vdi.get_virtual_size()
@@ -124,7 +124,7 @@ def test_resize(self, vm_on_ext_sr):
 
     @pytest.mark.small_vm
     @pytest.mark.big_vm
-    def test_failing_resize(self, host, ext_sr, vm_on_ext_sr, exit_on_fistpoint):
+    def test_failing_resize(self, host: Host, ext_sr: SR, vm_on_ext_sr: VM, exit_on_fistpoint: None) -> None:
         vm = vm_on_ext_sr
         vdi = VDI(vm.vdi_uuids()[0], host=vm.host)
         old_size = vdi.get_virtual_size()
@@ -135,13 +135,13 @@ def test_failing_resize(self, host, ext_sr, vm_on_ext_sr, exit_on_fistpoint):
                 vdi.resize(new_size)
             except SSHCommandFailed:
                 logging.info(f"Launching SR scan for {ext_sr} after failure")
-                host.xe("sr-scan", {"uuid": ext_sr})
+                host.xe("sr-scan", {"uuid": ext_sr.uuid})
 
         assert vdi.get_virtual_size() == new_size
 
     @pytest.mark.reboot
     @pytest.mark.small_vm
-    def test_reboot(self, host, ext_sr, vm_on_ext_sr):
+    def test_reboot(self, host: Host, ext_sr: SR, vm_on_ext_sr: VM) -> None:
         sr = ext_sr
         vm = vm_on_ext_sr
         host.reboot(verify=True)

tests/storage/ext/test_ext_sr_crosspool_migration.py

@@ -1,5 +1,10 @@
+from __future__ import annotations
+
 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:
@@ -13,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_ext_sr, local_sr_on_hostB1):
+    def test_cold_crosspool_migration(self, host: Host, hostB1: Host, vm_on_ext_sr: VM, local_sr_on_hostB1: SR) -> None:
         cold_migration_then_come_back(vm_on_ext_sr, host, hostB1, local_sr_on_hostB1)
 
-    def test_live_crosspool_migration(self, host, hostB1, vm_on_ext_sr, local_sr_on_hostB1):
+    def test_live_crosspool_migration(self, host: Host, hostB1: Host, vm_on_ext_sr: VM, local_sr_on_hostB1: SR) -> None:
         live_storage_migration_then_come_back(vm_on_ext_sr, host, hostB1, local_sr_on_hostB1)

tests/storage/ext/test_ext_sr_intrapool_migration.py

@@ -1,5 +1,10 @@
+from __future__ import annotations
+
 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:
@@ -13,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_ext_sr, local_sr_on_hostA2):
+    def test_cold_intrapool_migration(self, host: Host, hostA2: Host, vm_on_ext_sr: VM, local_sr_on_hostA2: SR) -> None:
         cold_migration_then_come_back(vm_on_ext_sr, host, hostA2, local_sr_on_hostA2)
 
-    def test_live_intrapool_migration(self, host, hostA2, vm_on_ext_sr, local_sr_on_hostA2):
+    def test_live_intrapool_migration(self, host: Host, hostA2: Host, vm_on_ext_sr: VM, local_sr_on_hostA2: SR) -> None:
         live_storage_migration_then_come_back(vm_on_ext_sr, host, hostA2, local_sr_on_hostA2)