Add Windows golden image storage tests (T2)

tests/storage/windows/__init__.py

@@ -0,0 +1 @@
+# Windows golden image tests for self-validation

tests/storage/windows/conftest.py

@@ -0,0 +1,120 @@
+"""
+Fixtures for Windows storage tests using self-validation golden image.
+
+These tests require a Windows 11 golden image DataSource to be created beforehand
+via the self-validation setup script (setup-golden-image.sh).
+"""
+
+import os
+
+import pytest
+from ocp_resources.data_source import DataSource
+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 ocp_resources.virtual_machine_instance import VirtualMachineInstance
+
+from utilities.constants import TIMEOUT_5SEC, TIMEOUT_10MIN, U1_LARGE
+from utilities.storage import add_dv_to_vm, data_volume_template_with_source_ref_dict
+from utilities.virt import VirtualMachineForTests, running_vm
+
+WINDOWS_GOLDEN_IMAGE_NAME = "windows11-golden-image"
+WINDOWS_GOLDEN_IMAGE_NAMESPACE = "openshift-virtualization-os-images"
+BLANK_DATA_DISK_SIZE = "1Gi"
+WINDOWS_11_PREFERENCE = "windows.11"
+
+
+@pytest.fixture(scope="session")
+def skip_if_windows_eula_not_accepted():
+    """Skip Windows tests if ACCEPT_WINDOWS_EULA is not set to true."""
+    if os.environ.get("ACCEPT_WINDOWS_EULA", "").lower() != "true":
+        pytest.skip(
+            "Windows tests require ACCEPT_WINDOWS_EULA=true. "
+            "Set this environment variable to accept Microsoft EULA and enable Windows testing."
+        )
+
+
+@pytest.fixture(scope="module")
+def windows11_golden_image_data_source(unprivileged_client, golden_images_namespace):
+    """Get the Windows 11 golden image DataSource created by self-validation setup."""
+    data_source = DataSource(
+        client=unprivileged_client,
+        name=WINDOWS_GOLDEN_IMAGE_NAME,
+        namespace=golden_images_namespace.name,
+    )
+    if not data_source.exists:
+        pytest.skip(
+            f"Windows golden image DataSource '{WINDOWS_GOLDEN_IMAGE_NAME}' not found in "
+            f"'{golden_images_namespace.name}'. Run self-validation with ACCEPT_WINDOWS_EULA=true to create it."
+        )
+    data_source.wait_for_condition(
+        condition=data_source.Condition.READY,
+        status=data_source.Condition.Status.TRUE,
+        timeout=TIMEOUT_5SEC,
+    )
+    return data_source
+
+
+@pytest.fixture(scope="class")
+def windows_vm_from_golden_image(
+    unprivileged_client,
+    namespace,
+    windows11_golden_image_data_source,
+):
+    """Create a Windows VM from the self-validation golden image DataSource."""
+    with VirtualMachineForTests(
+        client=unprivileged_client,
+        name=f"{windows11_golden_image_data_source.name}-test-vm",
+        namespace=namespace.name,
+        vm_instance_type=VirtualMachineClusterInstancetype(client=unprivileged_client, name=U1_LARGE),
+        vm_preference=VirtualMachineClusterPreference(client=unprivileged_client, name=WINDOWS_11_PREFERENCE),
+        data_volume_template=data_volume_template_with_source_ref_dict(
+            data_source=windows11_golden_image_data_source,
+        ),
+    ) as vm:
+        yield vm
+
+
+@pytest.fixture(scope="class")
+def blank_data_disk_template(namespace, snapshot_storage_class_name_scope_module):
+    """Create a blank DataVolume template dict for use as a second disk."""
+    dv = DataVolume(
+        name="windows-data-disk",
+        namespace=namespace.name,
+        source="blank",
+        size=BLANK_DATA_DISK_SIZE,
+        storage_class=snapshot_storage_class_name_scope_module,
+        api_name="storage",
+    )
+    dv.to_dict()
+    return dv.res
+
+
+@pytest.fixture(scope="class")
+def windows_vm_with_data_disk(
+    unprivileged_client,
+    namespace,
+    windows11_golden_image_data_source,
+    snapshot_storage_class_name_scope_module,
+    blank_data_disk_template,
+):
+    """Create a running Windows VM with boot disk + blank data disk, guest agent connected."""
+    with VirtualMachineForTests(
+        client=unprivileged_client,
+        name="windows-snapshot-test-vm",
+        namespace=namespace.name,
+        vm_instance_type=VirtualMachineClusterInstancetype(client=unprivileged_client, name=U1_LARGE),
+        vm_preference=VirtualMachineClusterPreference(client=unprivileged_client, name=WINDOWS_11_PREFERENCE),
+        data_volume_template=data_volume_template_with_source_ref_dict(
+            data_source=windows11_golden_image_data_source,
+            storage_class=snapshot_storage_class_name_scope_module,
+        ),
+    ) as vm:
+        add_dv_to_vm(vm=vm, template_dv=blank_data_disk_template)
+        running_vm(vm=vm, wait_for_interfaces=True, check_ssh_connectivity=False)
+        vm.vmi.wait_for_condition(
+            condition=VirtualMachineInstance.Condition.Type.AGENT_CONNECTED,
+            status=VirtualMachineInstance.Condition.Status.TRUE,
+            timeout=TIMEOUT_10MIN,
+        )
+        yield vm

tests/storage/windows/test_windows_app_consistent_snapshot.py

@@ -0,0 +1,80 @@
+"""
+Windows app-consistent snapshot test.
+
+Verifies that an online VirtualMachineSnapshot of a multi-disk Windows VM
+(boot + data disk) freezes for less than 10 seconds, proving guest agent
+freeze/thaw works correctly on the storage provider.
+"""
+
+import logging
+
+import pytest
+from dateutil import parser as date_parser
+from ocp_resources.virtual_machine_snapshot import VirtualMachineSnapshot
+from timeout_sampler import TimeoutSampler
+
+from tests.storage.utils import check_snapshot_indication
+from utilities.constants import TIMEOUT_10MIN
+
+LOGGER = logging.getLogger(__name__)
+
+FREEZE_THRESHOLD_SECONDS = 20
+
+
+pytestmark = [
+    pytest.mark.windows,
+    pytest.mark.conformance,
+    pytest.mark.storage,
+    pytest.mark.high_resource_vm,
+    pytest.mark.usefixtures("skip_if_windows_eula_not_accepted"),
+]
+
+
+class TestWindowsAppConsistentSnapshot:
+    """Test app-consistent online snapshot of a multi-disk Windows VM."""
+
+    @pytest.mark.polarion("CNV-16100")
+    def test_windows_multi_disk_snapshot_freeze_within_threshold(
+        self,
+        windows_vm_with_data_disk,
+    ):
+        """
+        Test that the freeze window of an online snapshot of a 2-disk Windows VM
+        completes within 20 seconds.
+
+        Measures the time between snapshot creation and status.creationTime
+        (point-in-time capture), which represents the guest agent freeze duration.
+        The backend may take longer to finalize, but the VM is already unfrozen.
+        """
+        vm = windows_vm_with_data_disk
+
+        LOGGER.info(f"Creating online snapshot of Windows VM {vm.name} with 2 disks...")
+        with VirtualMachineSnapshot(
+            name=f"snapshot-{vm.name}",
+            namespace=vm.namespace,
+            vm_name=vm.name,
+        ) as snapshot:
+            for creation_time in TimeoutSampler(
+                wait_timeout=TIMEOUT_10MIN,
+                sleep=1,
+                func=lambda: snapshot.instance.get("status", {}).get("creationTime"),
+            ):
+                if creation_time:
+                    break
+
+            snapshot_created = date_parser.parse(timestr=snapshot.instance.metadata.creationTimestamp)
+            snapshot_captured = date_parser.parse(timestr=creation_time)
+            freeze_seconds = (snapshot_captured - snapshot_created).total_seconds()
+
+            LOGGER.info(
+                f"Freeze window: {freeze_seconds:.1f}s (created={snapshot_created}, captured={snapshot_captured})"
+            )
+
+            snapshot.wait_snapshot_done(timeout=TIMEOUT_10MIN)
+
+            check_snapshot_indication(snapshot=snapshot, is_online=True)
+            LOGGER.info("Online indication confirmed - app-consistent snapshot verified")
+
+            assert freeze_seconds < FREEZE_THRESHOLD_SECONDS, (
+                f"Freeze took {freeze_seconds:.1f}s, expected < {FREEZE_THRESHOLD_SECONDS}s"
+            )

tests/storage/windows/test_windows_golden_image.py

@@ -0,0 +1,68 @@
+"""
+Windows golden image storage tests.
+
+These tests verify that Windows VMs can be created from the golden image
+DataSource created by the self-validation setup (setup-golden-image.sh).
+
+Requirements:
+- Windows golden image DataSource must exist (created when ACCEPT_WINDOWS_EULA=true)
+- ACCEPT_WINDOWS_EULA=true environment variable must be set
+"""
+
+import logging
+
+import pytest
+from ocp_resources.virtual_machine_instance import VirtualMachineInstance
+
+from utilities.constants import TIMEOUT_10MIN
+from utilities.virt import get_guest_os_info, running_vm
+
+LOGGER = logging.getLogger(__name__)
+
+pytestmark = [
+    pytest.mark.windows,
+    pytest.mark.conformance,
+    pytest.mark.storage,
+    pytest.mark.high_resource_vm,
+    pytest.mark.usefixtures("skip_if_windows_eula_not_accepted"),
+]
+
+
+class TestWindowsGoldenImage:
+    """Test Windows VM creation from self-validation golden image DataSource."""
+
+    @pytest.mark.polarion("CNV-16101")
+    def test_windows_vm_boots_from_golden_image(
+        self,
+        windows_vm_from_golden_image,
+    ):
+        """
+        Test that a Windows VM can boot from the golden image.
+
+        This test verifies:
+        1. VM can be created from the Windows golden image DataSource
+        2. VM boots successfully
+        3. Guest agent connects (indicates Windows is running properly)
+        4. Guest agent reports Windows OS info
+        """
+        vm = windows_vm_from_golden_image
+
+        LOGGER.info(f"Starting Windows VM {vm.name} from golden image...")
+        running_vm(vm=vm, wait_for_interfaces=True, check_ssh_connectivity=False)
+
+        LOGGER.info("Waiting for Windows guest agent to connect...")
+        vm.vmi.wait_for_condition(
+            condition=VirtualMachineInstance.Condition.Type.AGENT_CONNECTED,
+            status=VirtualMachineInstance.Condition.Status.TRUE,
+            timeout=TIMEOUT_10MIN,
+        )
+
+        LOGGER.info("Validating Windows OS info from guest agent...")
+        os_info = get_guest_os_info(vmi=vm.vmi)
+        assert os_info, "VMI doesn't have guest agent data"
+
+        os_name = os_info.get("name", "").lower()
+        LOGGER.info(f"Guest agent reports OS: {os_name}")
+        assert "windows" in os_name, f"Expected Windows OS, but got: {os_name}"
+
+        LOGGER.info("Windows VM booted successfully from golden image!")