Merge pull request #469 from xcp-ng/dnt/windows

Add Windows tools onboarding tests

Author: dinhngtu

data.py-dist

@@ -72,6 +72,8 @@ WIN_GUEST_TOOLS_ISOS = {
         "xenclean_path": "package\\XenClean\\x64\\Invoke-XenClean.ps1",
         # ISO-relative path of root cert file to be installed before guest tools (optional)
         "testsign_cert": "testsign\\XCP-ng_Test_Signer.crt",
+        # What's the onboard family of our tools? This is equal to the WinPV VENDOR_NAME value
+        "onboard_family": "XCP-ng",
     },
     # Add more guest tool ISOs here as needed
 }
@@ -100,6 +102,8 @@ OTHER_GUEST_TOOLS = {
 
         # Can we upgrade automatically from this guest tool to our tools?
         "upgradable": True,
+        # What is the expected onboarding phase after running XenClean when this tool is installed? (optional)
+        "onboarding_phase": "see test_xenclean.py ONBOARDING_PHASES",
     },
     "vendor": {
         "vendor_device": True,

lib/windows/__init__.py

@@ -51,12 +51,28 @@ def try_get_and_store_vm_ip_serial(vm: VM, timeout: int):
     return True
 
 
+def vm_shutdown_without_tools(vm: VM):
+    if vm.is_running():
+        vm.ssh(WINDOWS_SHUTDOWN_COMMAND)
+        wait_for(vm.is_halted, "Wait for VM halted")
+
+
 def wait_for_vm_running_and_ssh_up_without_tools(vm: VM):
     wait_for(vm.is_running, "Wait for VM running")
     wait_for(vm.is_ssh_up, "Wait for SSH up")
 
 
 def enable_testsign(vm: VM, rootcert: Union[str, None]):
+    assert vm.is_running()
+
+    if strtobool(vm.param_get("platform", "secureboot")):
+        logging.info("Disable secure boot on test image")
+
+        vm_shutdown_without_tools(vm)
+        vm.param_set('platform', False, key='secureboot')
+        vm.start()
+        wait_for_vm_running_and_ssh_up_without_tools(vm)
+
     if rootcert is not None:
         vm.execute_powershell_script(
             f"""certutil -addstore -f Root '{rootcert}';
@@ -100,9 +116,7 @@ def insert_cd_safe(vm: VM, vdi_name: str, cd_path="D:/", retries=2):
             return
         except TimeoutError:
             logging.warning(f"Waiting for CD at {cd_path} failed, retrying by rebooting VM")
-            # There might be no VM tools so use SSH instead.
-            vm.ssh(WINDOWS_SHUTDOWN_COMMAND)
-            wait_for(vm.is_halted, "Wait for VM halted")
+            vm_shutdown_without_tools(vm)
 
     raise TimeoutError(f"Waiting for CD at {cd_path} failed")
 

tests/guest_tools/win/conftest.py

@@ -9,10 +9,10 @@
 from lib.sr import SR
 from lib.vm import VM
 from lib.windows import (
-    WINDOWS_SHUTDOWN_COMMAND,
     PowerAction,
     iso_create,
     try_get_and_store_vm_ip_serial,
+    vm_shutdown_without_tools,
     wait_for_vm_running_and_ssh_up_without_tools,
 )
 from lib.windows.guest_tools import install_guest_tools
@@ -39,9 +39,7 @@ def running_windows_vm_without_tools(imported_vm: VM) -> VM:
 def unsealed_windows_vm_and_snapshot(running_windows_vm_without_tools: VM):
     """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 is not usable yet (there's no tools).
-    vm.ssh(WINDOWS_SHUTDOWN_COMMAND)
-    wait_for(vm.is_halted, "Shutdown VM")
+    vm_shutdown_without_tools(vm)
     snapshot = vm.snapshot()
     yield vm, snapshot
     snapshot.destroy(verify=True)

tests/guest_tools/win/test_guest_tools_win.py

@@ -7,13 +7,13 @@
 from lib.common import strtobool, wait_for
 from lib.vm import VM
 from lib.windows import (
-    WINDOWS_SHUTDOWN_COMMAND,
     PowerAction,
     check_vm_clipboard,
     check_vm_distro,
     check_vm_dns,
     set_vm_dns,
     vif_has_rss,
+    vm_shutdown_without_tools,
     wait_for_vm_running_and_ssh_up_without_tools,
 )
 from lib.windows.guest_tools import (
@@ -146,9 +146,7 @@ def test_xenvbd_ssd(self, vm_install_test_tools_per_test_class: VM):
 class TestGuestToolsWindowsDestructive:
     def test_uninstall_tools(self, vm_install_test_tools_no_reboot: VM):
         vm = vm_install_test_tools_no_reboot
-        vm.ssh(WINDOWS_SHUTDOWN_COMMAND)
-        wait_for(vm.is_halted, "Shutdown VM")
-
+        vm_shutdown_without_tools(vm)
         vm.start()
         wait_for_vm_running_and_ssh_up_without_tools(vm)
 

tests/guest_tools/win/test_xenclean.py

@@ -14,20 +14,56 @@
     wait_for_vm_xenvif_offboard,
 )
 
-from typing import Any, Dict, Tuple
+from typing import Any, Dict, Literal, Tuple, overload
 
 # Test uninstallation of other drivers using the XenClean program.
 
+ONBOARDING_PHASES = {
+    0: "CleaningSucceeded",
+    1: "Error",
+    2: "UserCanceled",
+    3: "RebootPending",
+    64: "ReadyForOnboard",
+    65: "AlreadyOnboarded",
+    66: "OnboardDenied",
+}
 
-def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any]):
+ONBOARD_EXIT_CODE_FILE = "C:\\onboard.txt"
+
+
+@overload
+def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: Literal[False] = False) -> None:  #
+    ...
+
+
+@overload
+def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: Literal[True]) -> str:  #
+    ...
+
+
+def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any], onboard: bool = False):
+    """
+    Run XenClean from the provided guest tools.
+
+    A note on guest tools onboarding with XenClean:
+    Onboarding is the transition from one guest tool to another, typically driven externally by repeatedly running
+    XenClean. XenClean will exit with one of the exit codes documented in ONBOARDING_PHASES.
+    """
     insert_cd_safe(vm, guest_tools_iso["name"])
 
     logging.info("Run XenClean")
     xenclean_path = PureWindowsPath("D:\\") / guest_tools_iso["xenclean_path"]
     if guest_tools_iso["xenclean_path"].lower().endswith(".ps1"):
+        assert not onboard, "Onboarding not supported in older versions"
         xenclean_cmd = f"Set-Location C:\\; {xenclean_path} -NoReboot -Confirm:$false; {WINDOWS_SHUTDOWN_COMMAND}"
     else:
-        xenclean_cmd = f"Set-Location C:\\; {xenclean_path} -noReboot -noConfirm; {WINDOWS_SHUTDOWN_COMMAND}"
+        xenclean_cmd = f"Set-Location C:\\; {xenclean_path} -noReboot -noConfirm"
+        if onboard:
+            onboard_family = guest_tools_iso["onboard_family"]
+            xenclean_cmd += f" -onboard {onboard_family}; Set-Content {ONBOARD_EXIT_CODE_FILE} $LASTEXITCODE -Force"
+        else:
+            xenclean_cmd += "; if ($LASTEXITCODE -ne 0) {{throw}}"
+        xenclean_cmd += f"; {WINDOWS_SHUTDOWN_COMMAND}"
     vm.start_background_powershell(xenclean_cmd)
 
     # XenClean sometimes takes a bit long due to all the calls to the uninstallers. We need an extended timeout.
@@ -37,6 +73,20 @@ def run_xenclean(vm: VM, guest_tools_iso: Dict[str, Any]):
     vm.start()
     wait_for_vm_running_and_ssh_up_without_tools(vm)
     wait_for_vm_xenvif_offboard(vm)
+    if onboard:
+        exitcode = vm.execute_powershell_script(f"Get-Content {ONBOARD_EXIT_CODE_FILE} -ErrorAction SilentlyContinue")
+        logging.debug(f"Onboarding exit code: {exitcode}")
+        assert exitcode, "Expected exit code string"
+        onboarding_phase = ONBOARDING_PHASES[int(exitcode)]
+        logging.info(f"Onboarding phase: {onboarding_phase}")
+        return onboarding_phase
+
+
+@pytest.fixture(scope="module")
+def onboarding_guest_tools_iso(guest_tools_iso):
+    if not guest_tools_iso.get("onboard_family"):
+        pytest.skip("Onboarding info not declared in data.py")
+    return guest_tools_iso
 
 
 @pytest.mark.multi_vms
@@ -48,6 +98,11 @@ def test_xenclean_without_tools(self, running_unsealed_windows_vm: VM, guest_too
         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):
+        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):
         vm = vm_install_test_tools_no_reboot
         logging.info("XenClean with test tools (without reboot)")
@@ -68,6 +123,16 @@ 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):
+        vm = vm_install_test_tools_no_reboot
+        vm.reboot()
+        wait_for_vm_running_and_ssh_up_without_tools(vm)
+
+        logging.info("XenClean onboard with test tools")
+        assert run_xenclean(vm, onboarding_guest_tools_iso, onboard=True) == "AlreadyOnboarded"
+        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):
         vm, param = vm_install_other_drivers
         if param.get("vendor_device"):
@@ -79,3 +144,14 @@ def test_xenclean_with_other_tools(self, vm_install_other_drivers: Tuple[VM, Dic
         logging.info("Check tools uninstalled")
         assert vm.are_windows_tools_uninstalled()
         check_vm_dns(vm)
+
+    def test_xenclean_onboard_with_other_tools(
+        self, vm_install_other_drivers: Tuple[VM, Dict], onboarding_guest_tools_iso
+    ):
+        vm, param = vm_install_other_drivers
+        onboarding_phase = param.get("onboarding_phase")
+        if not param.get("onboarding_phase"):
+            pytest.skip("Skipping XenClean on other tools with no defined onboarding phase")
+
+        logging.info("XenClean onboard with other tools")
+        assert run_xenclean(vm, onboarding_guest_tools_iso, onboard=True) == onboarding_phase