Storage: Add windows OADP test

tests/data_protection/oadp/conftest.py

@@ -1,14 +1,30 @@
 import pytest
+from ocp_resources.data_protection_application import DataProtectionApplication
 from ocp_resources.datavolume import DataVolume
 from ocp_resources.namespace import Namespace
-
+from ocp_resources.resource import ResourceEditor
+from pytest_testconfig import config as py_config
+
+from tests.data_protection.oadp.utils import (
+    OADP_DPA_NAME,
+    OADP_VELERO_IMAGE_FQIN_OVERRIDE,
+    create_windows_vm_from_dv_template,
+    write_file_windows_vm_for_oadp,
+)
+from utilities.artifactory import get_test_artifact_server_url
 from utilities.constants import (
+    ADP_NAMESPACE,
     BACKUP_STORAGE_LOCATION,
+    DV_SIZE_STR,
     FILE_NAME_FOR_BACKUP,
+    IMAGE_PATH_STR,
     OS_FLAVOR_RHEL,
+    OS_VERSION_STR,
+    TEMPLATE_LABELS_STR,
     TEXT_TO_TEST,
     TIMEOUT_8MIN,
     TIMEOUT_15MIN,
+    TIMEOUT_60MIN,
     Images,
 )
 from utilities.infra import create_ns
@@ -26,7 +42,20 @@
     virtctl_upload_dv,
     write_file,
 )
-from utilities.virt import running_vm
+from utilities.virt import running_vm, wait_for_windows_vm
+
+
+@pytest.fixture()
+def dpa_velero_image_override(admin_client):
+    """Temporarily override DPA Velero image, restored on teardown."""
+    dpa = DataProtectionApplication(
+        name=OADP_DPA_NAME,
+        namespace=ADP_NAMESPACE,
+        client=admin_client,
+    )
+    patch = {"spec": {"unsupportedOverrides": {"veleroImageFqin": OADP_VELERO_IMAGE_FQIN_OVERRIDE}}}
+    with ResourceEditor(patches={dpa: patch}):
+        yield dpa
 
 
 @pytest.fixture()
@@ -137,6 +166,69 @@ def rhel_vm_with_data_volume_template(
         yield vm
 
 
+@pytest.fixture()
+def windows_vm_with_data_volume_template(
+    admin_client,
+    dpa_velero_image_override,
+    namespace_for_backup,
+    snapshot_storage_class_name_scope_module,
+    modern_cpu_for_migration,
+):
+    """Windows VM in the backup namespace for OADP backup testing."""
+    latest_windows = py_config["latest_windows_os_dict"]
+    with create_windows_vm_from_dv_template(
+        storage_class=snapshot_storage_class_name_scope_module,
+        namespace=namespace_for_backup.name,
+        dv_name="oadp-windows-dv",
+        vm_name="oadp-windows-vm",
+        image_url=f"{get_test_artifact_server_url()}{latest_windows.get(IMAGE_PATH_STR)}",
+        dv_size=latest_windows.get(DV_SIZE_STR),
+        template_labels=latest_windows.get(TEMPLATE_LABELS_STR, {}),
+        client=admin_client,
+        cpu_model=modern_cpu_for_migration,
+        dv_wait_timeout=TIMEOUT_60MIN,
+    ) as vm:
+        wait_for_windows_vm(
+            vm=vm,
+            version=latest_windows.get(OS_VERSION_STR),
+            timeout=TIMEOUT_60MIN,
+        )
+        write_file_windows_vm_for_oadp(vm=vm)
+        yield vm
+
+
+@pytest.fixture()
+def velero_backup_first_namespace_without_datamover(
+    admin_client,
+    namespace_for_backup,
+    windows_vm_with_data_volume_template,
+):
+    with VeleroBackup(
+        client=admin_client,
+        included_namespaces=[
+            namespace_for_backup.name,
+        ],
+        name="backup-windows-dvt-ns",
+    ) as backup:
+        yield backup
+
+
+@pytest.fixture()
+def velero_restore_first_namespace_without_datamover(
+    admin_client,
+    velero_backup_first_namespace_without_datamover,
+):
+    Namespace(name=velero_backup_first_namespace_without_datamover.included_namespaces[0]).delete(wait=True)
+    with VeleroRestore(
+        client=admin_client,
+        included_namespaces=velero_backup_first_namespace_without_datamover.included_namespaces,
+        name="restore-windows-dvt-ns",
+        backup_name=velero_backup_first_namespace_without_datamover.name,
+        timeout=TIMEOUT_8MIN,
+    ) as restore:
+        yield restore
+
+
 @pytest.fixture()
 def velero_backup_first_namespace_using_datamover(admin_client, namespace_for_backup):
     with VeleroBackup(

tests/data_protection/oadp/test_velero.py

@@ -1,14 +1,16 @@
 import pytest
 from ocp_resources.datavolume import DataVolume
 
-from tests.data_protection.oadp.utils import wait_for_restored_dv
+from tests.data_protection.oadp.utils import FILE_PATH_FOR_WINDOWS_BACKUP, wait_for_restored_dv
 from utilities.constants import (
     FILE_NAME_FOR_BACKUP,
     TEXT_TO_TEST,
     TIMEOUT_10SEC,
+    TIMEOUT_15MIN,
     Images,
 )
 from utilities.oadp import check_file_in_running_vm
+from utilities.virt import verify_file_in_windows_vm, wait_for_running_vm
 
 pytestmark = pytest.mark.usefixtures("skip_if_no_storage_class_for_snapshot")
 
@@ -96,6 +98,38 @@ def test_backup_vm_data_volume_template_with_datamover(rhel_vm_with_data_volume_
     )
 
 
+@pytest.mark.tier3
+@pytest.mark.polarion("CNV-8696")
+@pytest.mark.usefixtures("velero_restore_first_namespace_without_datamover")
+def test_backup_and_restore_windows_vm(windows_vm_with_data_volume_template):
+    """
+    Test Windows VM backup and restore without Data Mover using Velero snapshot.
+
+    Preconditions:
+        - Windows VM with a marker file containing test data
+        - Velero backup created without Data Mover
+        - Velero restore completed
+
+    Steps:
+        1. Wait for Windows VM to reach Running state
+        2. Verify marker file exists at expected path
+        3. Verify file content matches pre-backup text
+
+    Expected:
+        - Windows VM is Running
+        - Marker file content equals TEXT_TO_TEST
+    """
+    wait_for_running_vm(
+        vm=windows_vm_with_data_volume_template,
+        wait_until_running_timeout=TIMEOUT_15MIN,
+    )
+    verify_file_in_windows_vm(
+        windows_vm=windows_vm_with_data_volume_template,
+        file_name_with_path=FILE_PATH_FOR_WINDOWS_BACKUP,
+        file_content=TEXT_TO_TEST,
+    )
+
+
 @pytest.mark.s390x
 @pytest.mark.polarion("CNV-10589")
 @pytest.mark.usefixtures("velero_restore_second_namespace_with_datamover")

tests/data_protection/oadp/utils.py

@@ -1,15 +1,116 @@
-import logging
+from __future__ import annotations
 
+from collections.abc import Generator
+from contextlib import contextmanager
+from typing import Any
+
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.datavolume import DataVolume
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
+from ocp_resources.template import Template
+from pyhelper_utils.shell import run_ssh_commands
 
+from utilities.artifactory import (
+    cleanup_artifactory_secret_and_config_map,
+    get_artifactory_config_map,
+    get_artifactory_secret,
+)
 from utilities.constants import (
+    TEXT_TO_TEST,
     TIMEOUT_10SEC,
     TIMEOUT_15SEC,
+    TIMEOUT_60MIN,
 )
+from utilities.virt import VirtualMachineForTests, VirtualMachineForTestsFromTemplate, running_vm
+
+FILE_PATH_FOR_WINDOWS_BACKUP = "C:/oadp_file_before_backup.txt"
 
-LOGGER = logging.getLogger(__name__)
+OADP_DPA_NAME = "dpa"
+OADP_VELERO_IMAGE_FQIN_OVERRIDE = "quay.io/sseago/velero:csi-quick-poll"
 
 
-def wait_for_restored_dv(dv):
+def wait_for_restored_dv(dv: DataVolume) -> None:
     dv.pvc.wait_for_status(status=PersistentVolumeClaim.Status.BOUND, timeout=TIMEOUT_15SEC)
     dv.wait_for_dv_success(timeout=TIMEOUT_10SEC)
+
+
+def write_file_windows_vm_for_oadp(vm: VirtualMachineForTests) -> None:
+    """Write test data to marker file on Windows VM for OADP backup verification."""
+    value = TEXT_TO_TEST.replace("'", "''")
+    cmd = [
+        "powershell",
+        "-NoProfile",
+        "-Command",
+        f"Set-Content -LiteralPath '{FILE_PATH_FOR_WINDOWS_BACKUP}' -Value '{value}' -Encoding ascii",
+    ]
+    run_ssh_commands(host=vm.ssh_exec, commands=cmd)
+
+
+@contextmanager
+def create_windows_vm_from_dv_template(
+    storage_class: str,
+    namespace: str,
+    dv_name: str,
+    vm_name: str,
+    image_url: str,
+    dv_size: str,
+    template_labels: dict[str, Any],
+    client: DynamicClient,
+    cpu_model: str | None = None,
+    wait_running: bool = True,
+    dv_wait_timeout: int = TIMEOUT_60MIN,
+) -> Generator[VirtualMachineForTests, None, None]:
+    """
+    Create Windows VM from template with HTTP DataVolume.
+
+    Args:
+        storage_class: Storage class for the DataVolume.
+        namespace: Target namespace.
+        dv_name: DataVolume name.
+        vm_name: VirtualMachine name.
+        image_url: HTTP URL to Windows image.
+        dv_size: DataVolume size.
+        template_labels: Labels to identify the Windows template.
+        client: Kubernetes dynamic client.
+        cpu_model: CPU model for the VM.
+        wait_running: Wait for VM to reach Running state.
+        dv_wait_timeout: DataVolume import timeout.
+
+    Yields:
+        VirtualMachineForTests instance.
+    """
+    artifactory_secret = None
+    artifactory_config_map = None
+
+    try:
+        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="http",
+            url=image_url,
+            size=dv_size,
+            client=client,
+            api_name="storage",
+            secret=artifactory_secret,
+            cert_configmap=artifactory_config_map.name,
+        )
+        dv.to_dict()
+        with VirtualMachineForTestsFromTemplate(
+            name=vm_name,
+            namespace=namespace,
+            client=client,
+            labels=Template.generate_template_labels(**template_labels),
+            cpu_model=cpu_model,
+            data_volume_template={"metadata": dv.res["metadata"], "spec": dv.res["spec"]},
+        ) as vm:
+            if wait_running:
+                running_vm(vm=vm, dv_wait_timeout=dv_wait_timeout)
+            yield vm
+    finally:
+        cleanup_artifactory_secret_and_config_map(
+            artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map
+        )

tests/storage/storage_migration/test_storage_class_migration.py

@@ -11,13 +11,12 @@
 )
 from tests.storage.storage_migration.utils import (
     verify_file_in_hotplugged_disk,
-    verify_file_in_windows_vm,
     verify_storage_migration_succeeded,
     verify_vm_storage_class_updated,
     verify_vms_boot_time_after_storage_migration,
 )
 from utilities.constants import TIMEOUT_60MIN
-from utilities.virt import migrate_vm_and_verify
+from utilities.virt import migrate_vm_and_verify, verify_file_in_windows_vm
 
 TESTS_CLASS_NAME_A_TO_B = "TestStorageClassMigrationAtoB"
 TESTS_CLASS_NAME_B_TO_A = "TestStorageClassMigrationBtoA"

tests/storage/storage_migration/utils.py

@@ -97,14 +97,6 @@ def verify_file_in_hotplugged_disk(vm: VirtualMachineForTests, file_name: str, f
     assert output.strip() == file_content, f"'{output}' does not equal '{file_content}'"
 
 
-def verify_file_in_windows_vm(windows_vm: VirtualMachineForTests, file_name_with_path: str, file_content: str) -> None:
-    cmd = shlex.split(f'powershell -command "Get-Content {file_name_with_path}"')
-    out = run_ssh_commands(host=windows_vm.ssh_exec, commands=cmd, wait_timeout=TIMEOUT_2MIN, sleep=TIMEOUT_5SEC)[
-        0
-    ].strip()
-    assert out.strip() == file_content, f"'{out}' does not equal '{file_content}'"
-
-
 def wait_for_storage_migration_completed(
     mig_migration: MultiNamespaceVirtualMachineStorageMigration, timeout: int = TIMEOUT_10MIN
 ) -> None:

utilities/virt.py

@@ -1725,6 +1725,28 @@ def get_guest_os_info(vmi):
         raise
 
 
+def verify_file_in_windows_vm(windows_vm: VirtualMachineForTests, file_name_with_path: str, file_content: str) -> None:
+    """
+    Verify that a file on a Windows VM contains the expected content.
+
+    Args:
+        windows_vm: The Windows VM to check.
+        file_name_with_path: Full path to the file on the Windows guest (e.g., "C:/test.txt").
+        file_content: Expected file content.
+
+    Raises:
+        AssertionError: If file content does not match expected content.
+    """
+    cmd = [
+        "powershell",
+        "-NoProfile",
+        "-Command",
+        f"Get-Content -LiteralPath '{file_name_with_path}'",
+    ]
+    out = run_ssh_commands(host=windows_vm.ssh_exec, commands=cmd)[0].strip()
+    assert out == file_content, f"'{out}' does not equal '{file_content}'"
+
+
 def get_windows_os_dict(windows_version: str) -> dict[str, Any]:
     """
     Returns a dictionary of Windows os information from the system_windows_os_matrix in py_config.