CherryPicked: [4.20] [Storage] Test using Windows VMs should have vTPM - test_hotplug.py

libs/infra/images.py

@@ -64,14 +64,15 @@ class Windows:
     WIN2022_ISO_IMG: str | None = None
     WIN2025_ISO_IMG: str | None = None
     DIR: str = f"{BASE_IMAGES_DIR}/windows-images"
-    DOCKER_IMAGE_DIR = "docker/kubevirt-common-instancetypes"
+    DOCKER_IMAGE_DIR: str = "docker-local/kubevirt-common-instancetypes"
     UEFI_WIN_DIR: str = f"{DIR}/uefi"
     HA_DIR: str = f"{DIR}/HA-images"
     ISO_BASE_DIR = f"{DIR}/install_iso"
     ISO_WIN10_DIR: str = f"{ISO_BASE_DIR}/win10"
     ISO_WIN11_DIR: str = f"{ISO_BASE_DIR}/win11"
     ISO_WIN2022_DIR: str = f"{ISO_BASE_DIR}/win2022"
     ISO_WIN2025_DIR: str = f"{ISO_BASE_DIR}/win2025"
+    CONTAINER_DISK_DV_SIZE: str = "40Gi"
     DEFAULT_DV_SIZE: str = "70Gi"
     DEFAULT_MEMORY_SIZE: str = "8Gi"
     DEFAULT_MEMORY_SIZE_WSL: str = "12Gi"

tests/storage/test_hotplug.py

@@ -10,14 +10,13 @@
 from ocp_resources.kubevirt import KubeVirt
 from ocp_resources.storage_profile import StorageProfile
 
-from tests.os_params import WINDOWS_LATEST, WINDOWS_LATEST_LABELS
+from tests.utils import create_windows2022_dv_template_from_registry, create_windows2022_vm_with_vtpm
 from utilities.constants import HOTPLUG_DISK_SERIAL
 from utilities.hco import ResourceEditorValidateHCOReconcile
 from utilities.storage import (
     assert_disk_serial,
     assert_hotplugvolume_nonexist_optional_restart,
     create_dv,
-    data_volume,
     virtctl_volume,
     wait_for_vm_volume_ready,
 )
@@ -26,8 +25,6 @@
     fedora_vm_body,
     migrate_vm_and_verify,
     running_vm,
-    vm_instance_from_template,
-    wait_for_windows_vm,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -73,51 +70,37 @@ def hotplug_volume_windows_scope_class(
 
 
 @pytest.fixture(scope="class")
-def vm_instance_from_template_multi_storage_scope_class(
-    request,
+def windows_dv_from_registry_scope_class(
     unprivileged_client,
     namespace,
-    data_volume_multi_storage_scope_class,
-    cpu_for_migration,
-):
-    """Calls vm_instance_from_template contextmanager
-
-    Creates a VM from template and starts it (if requested).
-    """
-    with vm_instance_from_template(
-        request=request,
-        unprivileged_client=unprivileged_client,
-        namespace=namespace,
-        existing_data_volume=data_volume_multi_storage_scope_class,
-        vm_cpu_model=cpu_for_migration if request.param.get("set_vm_common_cpu") else None,
-    ) as vm:
-        yield vm
-
-
-@pytest.fixture(scope="class")
-def started_windows_vm_scope_class(
-    request,
-    vm_instance_from_template_multi_storage_scope_class,
+    storage_class_matrix__class__,
 ):
-    wait_for_windows_vm(
-        vm=vm_instance_from_template_multi_storage_scope_class,
-        version=request.param["os_version"],
-    )
+    """Creates a Windows 2022 DataVolume from registry container disk."""
+    with create_windows2022_dv_template_from_registry(
+        dv_name="dv-windows-2022-hotplug",
+        namespace=namespace.name,
+        client=unprivileged_client,
+        storage_class=next(iter(storage_class_matrix__class__)),
+    ) as dv_dict:
+        yield dv_dict
 
 
 @pytest.fixture(scope="class")
-def data_volume_multi_storage_scope_class(
-    request,
+def vm_instance_from_template_multi_storage_scope_class(
+    unprivileged_client,
     namespace,
-    storage_class_matrix__class__,
-    schedulable_nodes,
+    modern_cpu_for_migration,
+    windows_dv_from_registry_scope_class,
 ):
-    yield from data_volume(
-        request=request,
-        namespace=namespace,
-        storage_class_matrix=storage_class_matrix__class__,
-        schedulable_nodes=schedulable_nodes,
-    )
+    """Creates a Windows 2022 VM with vTPM from registry container disk."""
+    with create_windows2022_vm_with_vtpm(
+        dv_template=windows_dv_from_registry_scope_class,
+        namespace=namespace.name,
+        client=unprivileged_client,
+        vm_name="vm-win-2022-hotplug",
+        cpu_model=modern_cpu_for_migration,
+    ) as vm:
+        yield vm
 
 
 @pytest.fixture(scope="class")
@@ -241,38 +224,23 @@ def test_hotplug_volume_with_serial_and_persist_migrate(
 
 
 @pytest.mark.parametrize(
-    "data_volume_multi_storage_scope_class,"
-    "vm_instance_from_template_multi_storage_scope_class,"
-    "started_windows_vm_scope_class,"
     "hotplug_volume_windows_scope_class",
     [
         pytest.param(
-            {
-                "dv_name": "dv-windows",
-                "image": WINDOWS_LATEST.get("image_path"),
-                "dv_size": WINDOWS_LATEST.get("dv_size"),
-            },
-            {
-                "vm_name": f"vm-win-{WINDOWS_LATEST.get('os_version')}",
-                "template_labels": WINDOWS_LATEST_LABELS,
-            },
-            {"os_version": WINDOWS_LATEST.get("os_version")},
             {"persist": True, "serial": HOTPLUG_DISK_SERIAL},
         ),
     ],
     indirect=True,
 )
+@pytest.mark.usefixtures("hotplug_volume_windows_scope_class")
 @pytest.mark.tier3
 class TestHotPlugWindows:
     @pytest.mark.polarion("CNV-6525")
     @pytest.mark.dependency(name="test_windows_hotplug")
     def test_windows_hotplug(
         self,
         blank_disk_dv_multi_storage_scope_class,
-        data_volume_multi_storage_scope_class,
         vm_instance_from_template_multi_storage_scope_class,
-        started_windows_vm_scope_class,
-        hotplug_volume_windows_scope_class,
     ):
         wait_for_vm_volume_ready(vm=vm_instance_from_template_multi_storage_scope_class)
         assert_disk_serial(
@@ -290,10 +258,7 @@ def test_windows_hotplug_migrate(
         self,
         unprivileged_client,
         blank_disk_dv_multi_storage_scope_class,
-        data_volume_multi_storage_scope_class,
         vm_instance_from_template_multi_storage_scope_class,
-        started_windows_vm_scope_class,
-        hotplug_volume_windows_scope_class,
     ):
         if is_dv_migratable(dv=blank_disk_dv_multi_storage_scope_class):
             migrate_vm_and_verify(

tests/utils.py

@@ -5,9 +5,10 @@
 import re
 import shlex
 import tarfile
+from collections.abc import Generator
 from contextlib import contextmanager
 from io import BytesIO
-from typing import Generator, Optional
+from typing import Optional
 
 import bitmath
 import requests
@@ -19,13 +20,16 @@
 from ocp_resources.kubevirt import KubeVirt
 from ocp_resources.resource import ResourceEditor
 from ocp_resources.virtual_machine import VirtualMachine
+from ocp_resources.virtual_machine_cluster_instancetype import VirtualMachineClusterInstancetype
+from ocp_resources.virtual_machine_cluster_preference import VirtualMachineClusterPreference
 from ocp_resources.virtual_machine_instance_migration import VirtualMachineInstanceMigration
 from pyhelper_utils.shell import run_ssh_commands
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler, retry
 
 from utilities.constants import (
     DISK_SERIAL,
     HCO_DEFAULT_CPU_MODEL_KEY,
+    OS_FLAVOR_WIN_CONTAINER_DISK,
     RHSM_SECRET_NAME,
     TCP_TIMEOUT_30SEC,
     TIMEOUT_1MIN,
@@ -37,16 +41,22 @@
     TIMEOUT_10SEC,
     TIMEOUT_15SEC,
     TIMEOUT_30MIN,
+    U1_LARGE,
+    WIN_2K22,
+    WINDOWS_2K22_PREFERENCE,
     Images,
 )
 from utilities.hco import ResourceEditorValidateHCOReconcile
 from utilities.infra import (
     ExecCommandOnPod,
+    cleanup_artifactory_secret_and_config_map,
     get_artifactory_config_map,
     get_artifactory_header,
     get_artifactory_secret,
     get_http_image_url,
+    get_test_artifact_server_url,
 )
+from utilities.os_utils import get_windows_container_disk_path
 from utilities.virt import (
     VirtualMachineForTests,
     fedora_vm_body,
@@ -56,6 +66,7 @@
     wait_for_migration_finished,
     wait_for_ssh_connectivity,
     wait_for_updated_kv_value,
+    wait_for_windows_vm,
 )
 
 NUM_TEST_VMS = 3
@@ -639,3 +650,85 @@ def _get_wsl2_running_status():
     except TimeoutExpiredError:
         LOGGER.error("WSL2 guest is not running in the VM!")
         raise
+
+
+@contextmanager
+def create_windows2022_dv_template_from_registry(
+    dv_name: str,
+    namespace: str,
+    client: DynamicClient,
+    storage_class: str,
+) -> Generator[dict]:
+    """
+    Creates a Windows Server 2022 DataVolume from registry container disk.
+
+    Args:
+        dv_name: Name for the DataVolume
+        namespace: Kubernetes namespace
+        client: Kubernetes client
+        storage_class: Storage class name
+
+    Yields:
+        dict: DataVolume template dictionary with metadata and spec
+    """
+
+    artifactory_secret = get_artifactory_secret(namespace=namespace)
+    artifactory_config_map = get_artifactory_config_map(namespace=namespace)
+
+    dv = DataVolume(
+        name=dv_name,
+        namespace=namespace,
+        storage_class=storage_class,
+        source="registry",
+        url=f"{get_test_artifact_server_url(schema='registry')}/{get_windows_container_disk_path(os_value=WIN_2K22)}",
+        size=Images.Windows.CONTAINER_DISK_DV_SIZE,
+        client=client,
+        api_name="storage",
+        secret=artifactory_secret,
+        cert_configmap=artifactory_config_map.name,
+    )
+    dv.to_dict()
+
+    try:
+        yield {"metadata": dv.res["metadata"], "spec": dv.res["spec"]}
+    finally:
+        cleanup_artifactory_secret_and_config_map(
+            artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map
+        )
+
+
+@contextmanager
+def create_windows2022_vm_with_vtpm(
+    dv_template: dict,
+    namespace: str,
+    client: DynamicClient,
+    vm_name: str,
+    cpu_model: str | None,
+) -> Generator[VirtualMachineForTests]:
+    """
+    Creates a Windows Server 2022 VM with vTPM from registry container disk.
+
+    Args:
+        dv_dict: DataVolume template dictionary with metadata and spec
+        namespace: Kubernetes namespace
+        client: Kubernetes client
+        vm_name: Name for the VirtualMachine
+        cpu_model: CPU model specification (can be None)
+
+    Yields:
+        VirtualMachineForTests: Running Windows 2022 VM with vTPM
+    """
+
+    with VirtualMachineForTests(
+        name=vm_name,
+        namespace=namespace,
+        client=client,
+        os_flavor=OS_FLAVOR_WIN_CONTAINER_DISK,
+        vm_instance_type=VirtualMachineClusterInstancetype(name=U1_LARGE, client=client),
+        vm_preference=VirtualMachineClusterPreference(name=WINDOWS_2K22_PREFERENCE, client=client),
+        data_volume_template=dv_template,
+        cpu_model=cpu_model,
+    ) as vm:
+        running_vm(vm=vm)
+        wait_for_windows_vm(vm=vm, version="2022")
+        yield vm

utilities/constants.py

@@ -45,6 +45,7 @@
 #  OS constants
 OS_FLAVOR_CIRROS = "cirros"
 OS_FLAVOR_WINDOWS = "win"
+OS_FLAVOR_WIN_CONTAINER_DISK = "win-container-disk"
 OS_FLAVOR_RHEL = "rhel"
 OS_FLAVOR_FEDORA = "fedora"
 
@@ -768,6 +769,7 @@ class NamespacesNames:
 RHEL9_PREFERENCE = "rhel.9"
 RHEL10_PREFERENCE = "rhel.10"
 U1_SMALL = "u1.small"
+U1_LARGE = "u1.large"
 PROMETHEUS_K8S = "prometheus-k8s"
 INSTANCE_TYPE_STR = "instance_type"
 U1_MEDIUM_STR = "u1.medium"
@@ -871,6 +873,10 @@ class NamespacesNames:
 WIN_11 = "win11"
 WIN_2K25 = "win2k25"
 WIN_2K22 = "win2k22"
+
+# Windows VirtualMachine preferences
+WINDOWS_11_PREFERENCE = "windows.11"
+WINDOWS_2K22_PREFERENCE = "windows.2k22"
 WIN_2K16 = "win2k16"
 WIN_2K19 = "win2k19"
 

utilities/unittests/test_os_utils.py

@@ -13,9 +13,44 @@
     WINDOWS_OS_MAPPING,
     generate_linux_instance_type_os_matrix,
     generate_os_matrix_dict,
+    get_windows_container_disk_path,
 )
 
 
+class TestGetWindowsContainerDiskPath:
+    """Test cases for get_windows_container_disk_path function"""
+
+    EXPECTED_DIR = "docker-local/kubevirt-common-instancetypes"
+
+    @pytest.mark.parametrize(
+        "os_value, expected_suffix",
+        [
+            pytest.param("win10", "windows10-container-disk:4.99", id="win10"),
+            pytest.param("win11", "windows11-container-disk:4.99", id="win11"),
+            pytest.param("win2k19", "windows2k19-container-disk:4.99", id="win2k19"),
+            pytest.param("win2k22", "windows2k22-container-disk:4.99", id="win2k22"),
+            pytest.param("win2k25", "windows2k25-container-disk:4.99", id="win2k25"),
+        ],
+    )
+    def test_get_windows_container_disk_path(self, os_value: str, expected_suffix: str):
+        """Test generating container disk path for valid Windows OS values"""
+        result = get_windows_container_disk_path(os_value=os_value)
+
+        assert result == f"{self.EXPECTED_DIR}/{expected_suffix}"
+
+    @pytest.mark.parametrize(
+        "os_value",
+        [
+            pytest.param("linux", id="linux"),
+            pytest.param("rhel9", id="rhel"),
+        ],
+    )
+    def test_get_windows_container_disk_path_invalid_os(self, os_value: str):
+        """Test error when os_value doesn't start with 'win'"""
+        with pytest.raises(ValueError, match="os_value must start with 'win'"):
+            get_windows_container_disk_path(os_value=os_value)
+
+
 class TestGenerateOsMatrixDict:
     """Test cases for generate_os_matrix_dict function"""