Merge pull request #403 from xcp-ng/gln/xen-guest-tools-typehints

Author: glehmann

tests/guest_tools/unix/test_guest_tools_unix.py

@@ -1,9 +1,13 @@
+from __future__ import annotations
+
 import pytest
 
 import logging
 import time
 
 from lib.common import PackageManagerEnum, wait_for
+from lib.host import Host
+from lib.sr import SR
 from lib.vm import VM
 
 # Requirements:
@@ -14,30 +18,32 @@
 # - A VM to import, supported by the Linux/install.sh script of the guest tools ISO
 #   (without this flag you get an alpine, and that is not suitable)
 
+
 class State:
-    def __init__(self):
-        self.tools_version = None
-        self.vm_distro = None
+    def __init__(self) -> None:
+        self.tools_version: str | None = None
+        self.vm_distro: str | None = None
+
 
 @pytest.mark.multi_vms
 @pytest.mark.usefixtures("unix_vm")
 class TestGuestToolsUnix:
     @pytest.fixture(scope='class')
-    def state(self):
+    def state(self) -> State:
         return State()
 
-    def _check_tools_version(self, vm, tools_version):
+    def _check_tools_version(self, vm: VM, tools_version: str | None) -> None:
         logging.info("Check that the detected tools version is '%s'" % tools_version)
         detected_version = vm.tools_version()
         assert detected_version == tools_version
 
-    def _check_os_info(self, vm, vm_distro):
+    def _check_os_info(self, vm: VM, vm_distro: str | None) -> None:
         logging.info("Check that the detected distro is '%s'" % vm_distro)
         detected_distro = vm.distro()
         assert detected_distro == vm_distro
 
     @pytest.fixture(scope="class", autouse=True)
-    def vm_install(self, running_vm: VM, state):
+    def vm_install(self, running_vm: VM, state: State) -> None:
         vm = running_vm
 
         # skip test for some unixes
@@ -90,18 +96,18 @@ def vm_install(self, running_vm: VM, state):
         wait_for(lambda: vm.ssh_with_result('pgrep -f xe-daemon').returncode == 0,
                  "Wait for xe-daemon running")
 
-    def test_check_tools(self, running_vm, state):
+    def test_check_tools(self, running_vm: VM, state: State) -> None:
         vm = running_vm
         self._check_tools_version(vm, state.tools_version)
         self._check_os_info(vm, state.vm_distro)
 
-    def test_check_tools_after_reboot(self, running_vm, state):
+    def test_check_tools_after_reboot(self, running_vm: VM, state: State) -> None:
         vm = running_vm
         vm.reboot(verify=True)
         self._check_tools_version(vm, state.tools_version)
         self._check_os_info(vm, state.vm_distro)
 
-    def test_xenstore(self, running_vm: VM):
+    def test_xenstore(self, running_vm: VM) -> None:
         logging.info("Testing various xenstore commands from the guest")
         vm = running_vm
         vm.ssh('xenstore-ls')
@@ -113,14 +119,15 @@ def test_xenstore(self, running_vm: VM):
         vm.ssh('xenstore-rm data/test-xcp-ng')
         assert vm.ssh_with_result('xenstore-exists data/test-xcp-ng').returncode != 0
 
-    def test_clean_shutdown(self, running_vm):
+    def test_clean_shutdown(self, running_vm: VM) -> None:
         vm = running_vm
         vm.shutdown(verify=True)
         # restore VM state
         vm.start()
         vm.wait_for_vm_running_and_ssh_up()
 
-    def test_storage_migration(self, running_vm, host, hostA2, local_sr_on_hostA2, state):
+    def test_storage_migration(self, running_vm: VM, host: Host, hostA2: Host,
+                               local_sr_on_hostA2: SR, state: State) -> None:
         vm = running_vm
         # migrate to default SR on hostA2
         vm.migrate(hostA2, local_sr_on_hostA2)

tests/guest_tools/win/conftest.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 import logging
@@ -18,7 +20,7 @@
 from lib.windows.guest_tools import install_guest_tools
 from lib.windows.other_tools import install_other_drivers
 
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, Generator, Tuple
 
 @pytest.fixture(scope="module")
 def running_windows_vm_without_tools(imported_vm: VM) -> VM:
@@ -36,7 +38,9 @@ def running_windows_vm_without_tools(imported_vm: VM) -> VM:
 
 
 @pytest.fixture(scope="module")
-def unsealed_windows_vm_and_snapshot(running_windows_vm_without_tools: VM):
+def unsealed_windows_vm_and_snapshot(
+    running_windows_vm_without_tools: VM
+) -> Generator[Tuple[VM, Snapshot], None, None]:
     """Unseal VM and get its IP, then shut it down. Cache the unsealed state in a snapshot to save time."""
     vm = running_windows_vm_without_tools
     vm_shutdown_without_tools(vm)
@@ -46,7 +50,7 @@ def unsealed_windows_vm_and_snapshot(running_windows_vm_without_tools: VM):
 
 
 @pytest.fixture
-def running_unsealed_windows_vm(unsealed_windows_vm_and_snapshot: Tuple[VM, Snapshot]):
+def running_unsealed_windows_vm(unsealed_windows_vm_and_snapshot: Tuple[VM, Snapshot]) -> Generator[VM, None, None]:
     vm, snapshot = unsealed_windows_vm_and_snapshot
     vm.start()
     wait_for_vm_running_and_ssh_up_without_tools(vm)
@@ -55,7 +59,9 @@ def running_unsealed_windows_vm(unsealed_windows_vm_and_snapshot: Tuple[VM, Snap
 
 
 @pytest.fixture(scope="class")
-def vm_install_test_tools_per_test_class(unsealed_windows_vm_and_snapshot, guest_tools_iso: Dict[str, Any]):
+def vm_install_test_tools_per_test_class(
+    unsealed_windows_vm_and_snapshot: Tuple[VM, Snapshot], guest_tools_iso: Dict[str, Any]
+) -> Generator[VM, None, None]:
     vm, snapshot = unsealed_windows_vm_and_snapshot
     vm.start()
     wait_for_vm_running_and_ssh_up_without_tools(vm)
@@ -66,7 +72,7 @@ def vm_install_test_tools_per_test_class(unsealed_windows_vm_and_snapshot, guest
 
 
 @pytest.fixture
-def vm_install_test_tools_no_reboot(running_unsealed_windows_vm: VM, guest_tools_iso: Dict[str, Any]):
+def vm_install_test_tools_no_reboot(running_unsealed_windows_vm: VM, guest_tools_iso: Dict[str, Any]) -> VM:
     install_guest_tools(running_unsealed_windows_vm, guest_tools_iso, PowerAction.Nothing)
     return running_unsealed_windows_vm
 
@@ -76,12 +82,14 @@ def vm_install_test_tools_no_reboot(running_unsealed_windows_vm: VM, guest_tools
     ids=list(WIN_GUEST_TOOLS_ISOS.keys()),
     params=list(WIN_GUEST_TOOLS_ISOS.values()),
 )
-def guest_tools_iso(host: Host, request: pytest.FixtureRequest, nfs_iso_sr: SR):
+def guest_tools_iso(
+    host: Host, request: pytest.FixtureRequest, nfs_iso_sr: SR
+) -> Generator[Dict[str, Any], None, None]:
     yield from iso_create(host, nfs_iso_sr, request.param)
 
 
 @pytest.fixture(scope="module")
-def other_tools_iso(host: Host, nfs_iso_sr: SR):
+def other_tools_iso(host: Host, nfs_iso_sr: SR) -> Generator[Dict[str, Any], None, None]:
     yield from iso_create(host, nfs_iso_sr, OTHER_GUEST_TOOLS_ISO)
 
 
@@ -90,7 +98,7 @@ def vm_install_other_drivers(
     unsealed_windows_vm_and_snapshot: Tuple[VM, Snapshot],
     other_tools_iso: Dict[str, Any],
     request: pytest.FixtureRequest,
-):
+) -> Generator[Tuple[VM, Dict[str, Any]], None, None]:
     vm, snapshot = unsealed_windows_vm_and_snapshot
     param = request.param
     install_other_drivers(vm, other_tools_iso["name"], param)

tests/guest_tools/win/test_guest_tools_win.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 import logging
@@ -70,10 +72,10 @@
 @pytest.mark.multi_vms
 @pytest.mark.usefixtures("windows_vm")
 class TestGuestToolsWindows:
-    def test_drivers_detected(self, vm_install_test_tools_per_test_class: VM):
+    def test_drivers_detected(self, vm_install_test_tools_per_test_class: VM) -> None:
         pass
 
-    def test_vif_replug(self, vm_install_test_tools_per_test_class: VM):
+    def test_vif_replug(self, vm_install_test_tools_per_test_class: VM) -> None:
         vm = vm_install_test_tools_per_test_class
         for _iter in range(3):
             vifs = vm.vifs()
@@ -87,7 +89,7 @@ def test_vif_replug(self, vm_install_test_tools_per_test_class: VM):
                 vif.plug()
             wait_for(vm.is_ssh_up, "Wait for SSH up")
 
-    def test_rss(self, vm_install_test_tools_per_test_class: VM):
+    def test_rss(self, vm_install_test_tools_per_test_class: VM) -> None:
         """
         Receive-side scaling is known to be broken on some driver versions.
 
@@ -98,7 +100,7 @@ def test_rss(self, vm_install_test_tools_per_test_class: VM):
         for vif in vifs:
             assert vif_has_rss(vif)
 
-    def test_reporting_after_xeniface_disable(self, vm_install_test_tools_per_test_class: VM):
+    def test_reporting_after_xeniface_disable(self, vm_install_test_tools_per_test_class: VM) -> None:
         vm = vm_install_test_tools_per_test_class
         for _iter in range(3):
             logging.info("Disable Xeniface")
@@ -109,7 +111,7 @@ def test_reporting_after_xeniface_disable(self, vm_install_test_tools_per_test_c
             check_vm_distro(vm)
             check_vm_clipboard(vm)
 
-    def test_reporting_after_suspend(self, vm_install_test_tools_per_test_class: VM):
+    def test_reporting_after_suspend(self, vm_install_test_tools_per_test_class: VM) -> None:
         vm = vm_install_test_tools_per_test_class
         for _iter in range(3):
             vm.suspend(verify=True)
@@ -118,7 +120,7 @@ def test_reporting_after_suspend(self, vm_install_test_tools_per_test_class: VM)
             check_vm_distro(vm)
             check_vm_clipboard(vm)
 
-    def test_xenvbd_unmap(self, vm_install_test_tools_per_test_class: VM):
+    def test_xenvbd_unmap(self, vm_install_test_tools_per_test_class: VM) -> None:
         """Xenvbd must always advertise unmap to allow migration between backends with different discard support."""
         vm = vm_install_test_tools_per_test_class
         trim_supported = strtobool(
@@ -127,7 +129,7 @@ def test_xenvbd_unmap(self, vm_install_test_tools_per_test_class: VM):
         )
         assert trim_supported
 
-    def test_xenvbd_ssd(self, vm_install_test_tools_per_test_class: VM):
+    def test_xenvbd_ssd(self, vm_install_test_tools_per_test_class: VM) -> None:
         """Xenvbd must always advertise as SSD to avoid unnecessary defragging by Windows."""
         vm = vm_install_test_tools_per_test_class
         is_ssd = strtobool(
@@ -144,7 +146,7 @@ def test_xenvbd_ssd(self, vm_install_test_tools_per_test_class: VM):
 @pytest.mark.multi_vms
 @pytest.mark.usefixtures("windows_vm")
 class TestGuestToolsWindowsDestructive:
-    def test_uninstall_tools(self, vm_install_test_tools_no_reboot: VM):
+    def test_uninstall_tools(self, vm_install_test_tools_no_reboot: VM) -> None:
         vm = vm_install_test_tools_no_reboot
         vm_shutdown_without_tools(vm)
         vm.start()
@@ -157,15 +159,15 @@ def test_uninstall_tools(self, vm_install_test_tools_no_reboot: VM):
         assert vm.are_windows_tools_uninstalled()
         check_vm_dns(vm)
 
-    def test_uninstall_tools_early(self, vm_install_test_tools_no_reboot: VM):
+    def test_uninstall_tools_early(self, vm_install_test_tools_no_reboot: VM) -> None:
         vm = vm_install_test_tools_no_reboot
         logging.info("Uninstall Windows PV drivers before rebooting")
         uninstall_guest_tools(vm, action=PowerAction.Reboot)
         assert vm.are_windows_tools_uninstalled()
 
     def test_install_with_other_tools(
         self, vm_install_other_drivers: Tuple[VM, dict[str, Any]], guest_tools_iso: dict[str, Any]
-    ):
+    ) -> None:
         vm, param = vm_install_other_drivers
         if param["upgradable"]:
             install_guest_tools(vm, guest_tools_iso, PowerAction.Reboot, check=False)
@@ -175,15 +177,15 @@ def test_install_with_other_tools(
             assert exitcode == ERROR_INSTALL_FAILURE
 
     @pytest.mark.usefixtures("uefi_vm")
-    def test_uefi_vm_suspend_refused_without_tools(self, running_unsealed_windows_vm: VM):
+    def test_uefi_vm_suspend_refused_without_tools(self, running_unsealed_windows_vm: VM) -> None:
         vm = running_unsealed_windows_vm
         with pytest.raises(SSHCommandFailed, match="lacks the feature"):
             vm.suspend()
         wait_for_vm_running_and_ssh_up_without_tools(vm)
 
     # Test of the unplug rework, where the driver must remain activated even if the device ID changes.
     # Also serves as a "close-enough" test of vendor device toggling.
-    def test_toggle_device_id(self, running_unsealed_windows_vm: VM, guest_tools_iso: dict[str, Any]):
+    def test_toggle_device_id(self, running_unsealed_windows_vm: VM, guest_tools_iso: dict[str, Any]) -> None:
         vm = running_unsealed_windows_vm
         assert vm.param_get("platform", "device_id") == "0002"
         install_guest_tools(vm, guest_tools_iso, PowerAction.Shutdown, check=False)

tests/guest_tools/win/test_xenclean.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import pytest
 
 import logging
@@ -41,7 +43,7 @@ def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: Literal[True]
     ...
 
 
-def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: bool = False):
+def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: bool = False) -> str | None:
     """
     Run XenClean from the provided guest tools.
 
@@ -80,10 +82,12 @@ def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: bool = False)
         onboarding_phase = ONBOARDING_PHASES[int(exitcode)]
         logging.info(f"Onboarding phase: {onboarding_phase}")
         return onboarding_phase
+    else:
+        return None
 
 
 @pytest.fixture(scope="module")
-def onboarding_guest_tools_iso(guest_tools_iso):
+def onboarding_guest_tools_iso(guest_tools_iso: Dict[str, Any]) -> Dict[str, Any]:
     if not guest_tools_iso.get("onboard_family"):
         pytest.skip("Onboarding info not declared in data.py")
     return guest_tools_iso
@@ -92,24 +96,30 @@ def onboarding_guest_tools_iso(guest_tools_iso):
 @pytest.mark.multi_vms
 @pytest.mark.usefixtures("windows_vm")
 class TestXenClean:
-    def test_xenclean_without_tools(self, running_unsealed_windows_vm: VM, guest_tools_iso):
+    def test_xenclean_without_tools(
+        self, running_unsealed_windows_vm: VM, guest_tools_iso: Dict[str, Any]
+    ) -> None:
         vm = running_unsealed_windows_vm
         logging.info("XenClean with empty VM")
         run_xenclean(vm, guest_tools_iso)
         assert vm.are_windows_tools_uninstalled()
 
-    def test_xenclean_onboard_without_tools(self, running_unsealed_windows_vm: VM, onboarding_guest_tools_iso):
+    def test_xenclean_onboard_without_tools(self, running_unsealed_windows_vm: VM,
+                                            onboarding_guest_tools_iso: Dict[str, Any]) -> None:
         vm = running_unsealed_windows_vm
         logging.info("XenClean onboard with empty VM")
         assert run_xenclean(vm, onboarding_guest_tools_iso, onboard=True) == "ReadyForOnboard"
 
-    def test_xenclean_with_test_tools_early(self, vm_install_test_tools_no_reboot: VM, guest_tools_iso):
+    def test_xenclean_with_test_tools_early(
+        self, vm_install_test_tools_no_reboot: VM, guest_tools_iso: Dict[str, Any]
+    ) -> None:
         vm = vm_install_test_tools_no_reboot
         logging.info("XenClean with test tools (without reboot)")
         run_xenclean(vm, guest_tools_iso)
         assert vm.are_windows_tools_uninstalled()
 
-    def test_xenclean_with_test_tools(self, vm_install_test_tools_no_reboot: VM, guest_tools_iso):
+    def test_xenclean_with_test_tools(self, vm_install_test_tools_no_reboot: VM,
+                                      guest_tools_iso: Dict[str, Any]) -> None:
         vm = vm_install_test_tools_no_reboot
         vm.reboot()
         # HACK: In some cases, vm.reboot(verify=False) followed by vm.insert_cd() (as called by run_xenclean)
@@ -123,7 +133,8 @@ def test_xenclean_with_test_tools(self, vm_install_test_tools_no_reboot: VM, gue
         assert vm.are_windows_tools_uninstalled()
         check_vm_dns(vm)
 
-    def test_xenclean_onboard_with_test_tools(self, vm_install_test_tools_no_reboot: VM, onboarding_guest_tools_iso):
+    def test_xenclean_onboard_with_test_tools(self, vm_install_test_tools_no_reboot: VM,
+                                              onboarding_guest_tools_iso: Dict[str, Any]) -> None:
         vm = vm_install_test_tools_no_reboot
         vm.reboot()
         wait_for_vm_running_and_ssh_up_without_tools(vm)
@@ -133,7 +144,9 @@ def test_xenclean_onboard_with_test_tools(self, vm_install_test_tools_no_reboot:
         logging.info("Check tools still working")
         assert vm.are_windows_tools_working()
 
-    def test_xenclean_with_other_tools(self, vm_install_other_drivers: Tuple[VM, Dict], guest_tools_iso):
+    def test_xenclean_with_other_tools(
+        self, vm_install_other_drivers: Tuple[VM, Dict[str, Any]], guest_tools_iso: Dict[str, Any]
+    ) -> None:
         vm, param = vm_install_other_drivers
         if param.get("vendor_device"):
             pytest.skip("Skipping XenClean with vendor device present")
@@ -146,8 +159,8 @@ def test_xenclean_with_other_tools(self, vm_install_other_drivers: Tuple[VM, Dic
         check_vm_dns(vm)
 
     def test_xenclean_onboard_with_other_tools(
-        self, vm_install_other_drivers: Tuple[VM, Dict], onboarding_guest_tools_iso
-    ):
+        self, vm_install_other_drivers: Tuple[VM, Dict[str, Any]], onboarding_guest_tools_iso: Dict[str, Any]
+    ) -> None:
         vm, param = vm_install_other_drivers
         onboarding_phase = param.get("onboarding_phase")
         if not param.get("onboarding_phase"):