WIP: [Storage] Windows tests should all use vTPM - module CDI Clone

tests/storage/cdi_clone/conftest.py

@@ -1,8 +1,11 @@
 import pytest
 from ocp_resources.datavolume import DataVolume
 
+from tests.storage.cdi_clone.constants import WINDOWS_CLONE_TIMEOUT
 from tests.storage.constants import QUAY_FEDORA_CONTAINER_IMAGE
-from utilities.constants import REGISTRY_STR, Images
+from utilities.artifactory import get_artifactory_config_map, get_artifactory_secret, get_test_artifact_server_url
+from utilities.constants import REGISTRY_STR, WIN_2K22, Images
+from utilities.os_utils import get_windows_container_disk_path
 from utilities.storage import create_dv, data_volume
 
 
@@ -59,3 +62,47 @@ def fedora_dv_with_block_volume_mode(
     ) as dv:
         dv.wait_for_dv_success()
         yield dv
+
+
+@pytest.fixture(scope="function")
+def source_dv_windows_registry(
+    unprivileged_client,
+    namespace,
+    storage_class_name_scope_function,
+):
+    """Fixture that creates a Windows 2022 DataVolume from registry."""
+    secret = get_artifactory_secret(namespace=namespace.name)
+    cert = get_artifactory_config_map(namespace=namespace.name)
+
+    with create_dv(
+        client=unprivileged_client,
+        dv_name=f"dv-source-{WIN_2K22}-registry",
+        namespace=namespace.name,
+        source="registry",
+        size=Images.Windows.CONTAINER_DISK_DV_SIZE,
+        storage_class=storage_class_name_scope_function,
+        url=f"{get_test_artifact_server_url(schema='registry')}/{get_windows_container_disk_path(os_value=WIN_2K22)}",
+        secret=secret,
+        cert_configmap=cert.name,
+    ) as dv:
+        dv.wait_for_dv_success(timeout=WINDOWS_CLONE_TIMEOUT)
+        yield dv
+
+
+@pytest.fixture(scope="function")
+def cloned_windows_dv_from_registry(
+    unprivileged_client,
+    source_dv_windows_registry,
+):
+    """Fixture that creates a cloned DataVolume from registry source."""
+    with create_dv(
+        client=unprivileged_client,
+        source="pvc",
+        dv_name=f"dv-target-{WIN_2K22}-vtpm",
+        namespace=source_dv_windows_registry.namespace,
+        size=source_dv_windows_registry.size,
+        source_pvc=source_dv_windows_registry.name,
+        storage_class=source_dv_windows_registry.storage_class,
+    ) as cdv:
+        cdv.wait_for_dv_success(timeout=WINDOWS_CLONE_TIMEOUT)
+        yield cdv

tests/storage/cdi_clone/constants.py

@@ -0,0 +1,5 @@
+"""Constants for CDI clone tests."""
+
+from utilities.constants import TIMEOUT_40MIN
+
+WINDOWS_CLONE_TIMEOUT = TIMEOUT_40MIN

tests/storage/cdi_clone/test_clone.py

@@ -4,18 +4,32 @@
 
 import pytest
 from ocp_resources.datavolume import DataVolume
+from ocp_resources.virtual_machine_cluster_instancetype import (
+    VirtualMachineClusterInstancetype,
+)
+from ocp_resources.virtual_machine_cluster_preference import (
+    VirtualMachineClusterPreference,
+)
 
 from tests.os_params import FEDORA_LATEST, WINDOWS_11, WINDOWS_11_TEMPLATE_LABELS
+from tests.storage.cdi_clone.constants import WINDOWS_CLONE_TIMEOUT
 from tests.storage.utils import (
     assert_pvc_snapshot_clone_annotation,
     assert_use_populator,
     create_windows_vm_validate_guest_agent_info,
+    validate_os_info_vmi_vs_windows_os,
+    verify_vtpm_in_windows_vm,
+    wait_for_windows_vm,
 )
 from utilities.constants import (
     OS_FLAVOR_FEDORA,
+    OS_FLAVOR_WIN_CONTAINER_DISK,
     OS_FLAVOR_WINDOWS,
     TIMEOUT_1MIN,
     TIMEOUT_40MIN,
+    U1_LARGE,
+    WIN_2K22,
+    WINDOWS_2K22_PREFERENCE,
     Images,
 )
 from utilities.storage import (
@@ -32,8 +46,6 @@
     running_vm,
 )
 
-WINDOWS_CLONE_TIMEOUT = TIMEOUT_40MIN
-
 
 def create_vm_from_clone_dv_template(
     vm_name,
@@ -187,6 +199,34 @@ def test_successful_vm_from_cloned_dv_windows(
         )
 
 
+@pytest.mark.tier3
+@pytest.mark.polarion("CNV-3638")
+def test_successful_vm_from_cloned_dv_windows_with_vtpm(
+    admin_client,  # only needed for `verify_vtpm_in_windows_vm`, to be deleted once verified
+    unprivileged_client,
+    namespace,
+    cloned_windows_dv_from_registry,
+):
+    """Test cloning Windows 2022 DV and creating VM with vTPM using instance types."""
+
+    with VirtualMachineForTests(
+        name=f"vm-{WIN_2K22}-vtpm",
+        namespace=namespace.name,
+        client=unprivileged_client,
+        os_flavor=OS_FLAVOR_WIN_CONTAINER_DISK,
+        vm_instance_type=VirtualMachineClusterInstancetype(name=U1_LARGE, client=unprivileged_client),
+        vm_preference=VirtualMachineClusterPreference(name=WINDOWS_2K22_PREFERENCE, client=unprivileged_client),
+        data_volume_template={
+            "metadata": cloned_windows_dv_from_registry.res["metadata"],
+            "spec": cloned_windows_dv_from_registry.res["spec"],
+        },
+    ) as vm:
+        vm.start()
+        wait_for_windows_vm(vm=vm, version="2022", timeout=TIMEOUT_40MIN)
+        validate_os_info_vmi_vs_windows_os(vm=vm)
+        verify_vtpm_in_windows_vm(vm=vm, admin_client=admin_client)
+
+
 @pytest.mark.parametrize(
     "data_volume_snapshot_capable_storage_scope_function",
     [

tests/storage/utils.py

@@ -536,3 +536,48 @@ def check_file_in_vm(
         vm_console.expect(pattern=file_name, timeout=TIMEOUT_20SEC)
         vm_console.sendline(f"cat {file_name}")
         vm_console.expect(pattern=file_content, timeout=TIMEOUT_20SEC)
+
+
+def verify_vtpm_in_windows_vm(vm, admin_client):  # TODO: Remove - debug only
+    """
+    Verify vTPM is properly configured and detected in a Windows VM.
+
+    **DEBUG ONLY - MARKED FOR DELETION**
+
+    Performs two-layer validation:
+    1. KubeVirt configuration validation via VM XML
+    2. Windows OS TPM detection via WMI
+
+    Args:
+        vm (VirtualMachineForTests): Running Windows VM instance
+        admin_client (DynamicClient): Admin client for privileged VM XML access
+
+    Raises:
+        AssertionError: If vTPM is not properly configured or detected
+    """
+    LOGGER.info(f"Validating vTPM configuration and detection for VM: {vm.name}")
+
+    # Layer 1: Validate TPM in KubeVirt VM XML
+    LOGGER.info("Validating vTPM configuration in KubeVirt VM spec")
+    xml_dict_tpm = vm.vmi.get_xml_dict(privileged_client=admin_client)["domain"]["devices"]["tpm"]
+
+    tpm_model = xml_dict_tpm.get("@model")
+    assert tpm_model == "tpm-crb", f"Expected TPM model 'tpm-crb', got '{tpm_model}'"
+
+    persistent_state = xml_dict_tpm.get("backend", {}).get("@persistent_state")
+    assert persistent_state == "yes", f"Expected TPM persistent state 'yes', got '{persistent_state}'"
+    LOGGER.info("vTPM configuration validated in VM spec")
+
+    # Layer 2: Validate TPM detection in Windows OS
+    LOGGER.info("Validating vTPM detection in Windows OS")
+    vtpm_enabled = run_ssh_commands(
+        host=vm.ssh_exec,
+        commands=shlex.split(
+            r"wmic /namespace:\\root\cimv2\security\microsofttpm path Win32_Tpm get IsEnabled_InitialValue",
+            posix=False,
+        ),
+    )[0]
+    assert "TRUE" in vtpm_enabled, f"TPM not enabled in Windows. Output: {vtpm_enabled}"
+    LOGGER.info("vTPM detection validated in Windows OS")
+
+    LOGGER.info(f"vTPM validation completed for VM: {vm.name}")

utilities/constants.py

@@ -812,6 +812,7 @@ class NamespacesNames:
 RHEL8_PREFERENCE = "rhel.8"
 RHEL9_PREFERENCE = "rhel.9"
 RHEL10_PREFERENCE = "rhel.10"
+WINDOWS_2K22_PREFERENCE = "windows.2k22"
 U1_SMALL = "u1.small"
 U1_LARGE = "u1.large"
 PROMETHEUS_K8S = "prometheus-k8s"