Move and type OADP related functions from tests directory to the upper level utilities directory and add unit tests

tests/data_protection/oadp/conftest.py

@@ -2,10 +2,6 @@
 from ocp_resources.datavolume import DataVolume
 from ocp_resources.namespace import Namespace
 
-from tests.data_protection.oadp.utils import (
-    VeleroRestore,
-    is_storage_class_support_volume_mode,
-)
 from utilities.constants import (
     BACKUP_STORAGE_LOCATION,
     FILE_NAME_FOR_BACKUP,
@@ -18,7 +14,9 @@
 from utilities.infra import create_ns
 from utilities.oadp import (
     VeleroBackup,
+    VeleroRestore,
     create_rhel_vm,
+    is_storage_class_support_volume_mode,
 )
 from utilities.storage import (
     check_upload_virtctl_result,
@@ -113,6 +111,7 @@ def rhel_vm_with_data_volume_template(
 ):
     volume_mode = request.param.get("volume_mode")
     if not is_storage_class_support_volume_mode(
+        admin_client=admin_client,
         storage_class_name=snapshot_storage_class_name_scope_module,
         requested_volume_mode=volume_mode,
     ):

tests/data_protection/oadp/test_velero.py

@@ -1,8 +1,14 @@
 import pytest
 from ocp_resources.datavolume import DataVolume
 
-from tests.data_protection.oadp.utils import check_file_in_vm, wait_for_restored_dv
-from utilities.constants import TIMEOUT_10SEC, Images
+from tests.data_protection.oadp.utils import wait_for_restored_dv
+from utilities.constants import (
+    FILE_NAME_FOR_BACKUP,
+    TEXT_TO_TEST,
+    TIMEOUT_10SEC,
+    Images,
+)
+from utilities.oadp import check_file_in_running_vm
 
 pytestmark = pytest.mark.usefixtures("skip_if_no_storage_class_for_snapshot")
 
@@ -53,7 +59,9 @@ def test_restore_multiple_namespaces(
         timeout=TIMEOUT_10SEC,
         stop_status=DataVolume.Status.IMPORT_IN_PROGRESS,
     )
-    check_file_in_vm(vm=rhel_vm_with_data_volume_template)
+    check_file_in_running_vm(
+        vm=rhel_vm_with_data_volume_template, file_name=FILE_NAME_FOR_BACKUP, file_content=TEXT_TO_TEST
+    )
 
 
 @pytest.mark.s390x
@@ -81,16 +89,18 @@ def test_restore_multiple_namespaces(
     ],
     indirect=True,
 )
-def test_backup_vm_data_volume_template_with_datamover(
-    rhel_vm_with_data_volume_template, velero_restore_first_namespace_with_datamover
-):
-    check_file_in_vm(vm=rhel_vm_with_data_volume_template)
+@pytest.mark.usefixtures("velero_restore_first_namespace_with_datamover")
+def test_backup_vm_data_volume_template_with_datamover(rhel_vm_with_data_volume_template):
+    check_file_in_running_vm(
+        vm=rhel_vm_with_data_volume_template, file_name=FILE_NAME_FOR_BACKUP, file_content=TEXT_TO_TEST
+    )
 
 
 @pytest.mark.s390x
 @pytest.mark.polarion("CNV-10589")
-def test_restore_vm_with_existing_dv(rhel_vm_from_existing_dv, velero_restore_second_namespace_with_datamover):
-    check_file_in_vm(vm=rhel_vm_from_existing_dv)
+@pytest.mark.usefixtures("velero_restore_second_namespace_with_datamover")
+def test_restore_vm_with_existing_dv(rhel_vm_from_existing_dv):
+    check_file_in_running_vm(vm=rhel_vm_from_existing_dv, file_name=FILE_NAME_FOR_BACKUP, file_content=TEXT_TO_TEST)
 
 
 @pytest.mark.s390x

tests/data_protection/oadp/utils.py

@@ -1,83 +1,15 @@
 import logging
 
 from ocp_resources.persistent_volume_claim import PersistentVolumeClaim
-from ocp_resources.restore import Restore
-from ocp_resources.storage_profile import StorageProfile
 
-from utilities import console
 from utilities.constants import (
-    ADP_NAMESPACE,
-    FILE_NAME_FOR_BACKUP,
-    LS_COMMAND,
-    TEXT_TO_TEST,
-    TIMEOUT_5MIN,
     TIMEOUT_10SEC,
     TIMEOUT_15SEC,
-    TIMEOUT_20SEC,
 )
-from utilities.infra import (
-    unique_name,
-)
-from utilities.oadp import delete_velero_resource
 
 LOGGER = logging.getLogger(__name__)
 
 
-class VeleroRestore(Restore):
-    def __init__(
-        self,
-        name,
-        namespace=ADP_NAMESPACE,
-        included_namespaces=None,
-        backup_name=None,
-        client=None,
-        teardown=False,
-        yaml_file=None,
-        wait_complete=True,
-        timeout=TIMEOUT_5MIN,
-        **kwargs,
-    ):
-        super().__init__(
-            name=unique_name(name=name),
-            namespace=namespace,
-            included_namespaces=included_namespaces,
-            backup_name=backup_name,
-            client=client,
-            teardown=teardown,
-            yaml_file=yaml_file,
-            **kwargs,
-        )
-        self.wait_complete = wait_complete
-        self.timeout = timeout
-
-    def __enter__(self):
-        super().__enter__()
-        if self.wait_complete:
-            self.wait_for_status(
-                status=self.Status.COMPLETED,
-                timeout=self.timeout,
-            )
-        return self
-
-    def __exit__(self, exception_type, exception_value, traceback):
-        delete_velero_resource(resource=self, client=self.client)
-
-
-def check_file_in_vm(vm):
-    with console.Console(vm=vm) as vm_console:
-        vm_console.sendline(LS_COMMAND)
-        vm_console.expect(FILE_NAME_FOR_BACKUP, timeout=TIMEOUT_20SEC)
-        vm_console.sendline(f"cat {FILE_NAME_FOR_BACKUP}")
-        vm_console.expect(TEXT_TO_TEST, timeout=TIMEOUT_20SEC)
-
-
-def is_storage_class_support_volume_mode(storage_class_name, requested_volume_mode):
-    for claim_property_set in StorageProfile(name=storage_class_name).claim_property_sets:
-        if claim_property_set.volumeMode == requested_volume_mode:
-            return True
-    return False
-
-
 def wait_for_restored_dv(dv):
     dv.pvc.wait_for_status(status=PersistentVolumeClaim.Status.BOUND, timeout=TIMEOUT_15SEC)
     dv.wait_for_dv_success(timeout=TIMEOUT_10SEC)

utilities/oadp.py

@@ -1,10 +1,17 @@
 import logging
+from collections.abc import Generator
 from contextlib import contextmanager
-from typing import Generator
+from re import escape
+from shlex import quote
+from typing import Any, Self
 
 from kubernetes.dynamic import DynamicClient
+from kubernetes.dynamic.exceptions import ResourceNotFoundError
 from ocp_resources.backup import Backup
 from ocp_resources.datavolume import DataVolume
+from ocp_resources.exceptions import ResourceTeardownError
+from ocp_resources.restore import Restore
+from ocp_resources.storage_profile import StorageProfile
 from ocp_resources.virtual_machine import VirtualMachine
 
 from utilities.artifactory import (
@@ -13,10 +20,13 @@
     get_artifactory_secret,
     get_http_image_url,
 )
+from utilities.console import Console
 from utilities.constants import (
     ADP_NAMESPACE,
+    LS_COMMAND,
     OS_FLAVOR_RHEL,
     TIMEOUT_5MIN,
+    TIMEOUT_20SEC,
     Images,
 )
 from utilities.infra import (
@@ -28,19 +38,66 @@
 LOGGER = logging.getLogger(__name__)
 
 
-def delete_velero_resource(resource, client):
-    velero_pod = get_pod_by_name_prefix(client=client, pod_prefix="velero", namespace=ADP_NAMESPACE)
+def delete_velero_resource(resource: Backup | Restore, client: DynamicClient) -> None:
+    """
+    Delete a Velero resource using the Velero CLI inside the Velero pod.
+
+    Args:
+        resource (Backup | Restore):
+            The Velero resource to delete.
+        client (DynamicClient):
+            Kubernetes dynamic client used to locate the Velero pod.
+
+    Raises:
+        ResourceNotFoundError:
+            If the Velero pod or resource cannot be found.
+    """
+
     command = ["./velero", "delete", resource.kind.lower(), resource.name, "--confirm"]
-    velero_pod.execute(command=command)
+
+    try:
+        velero_pod = get_pod_by_name_prefix(client=client, pod_prefix="velero", namespace=ADP_NAMESPACE)
+
+        LOGGER.info(f"Deleting Velero resource: kind={resource.kind} name={resource.name} command={' '.join(command)}")
+
+        velero_pod.execute(command=command)
+
+    except ResourceNotFoundError:
+        LOGGER.error(
+            f"Failed to delete Velero resource: kind={resource.kind} name={resource.name} command={' '.join(command)}",
+            exc_info=True,
+        )
+
+        raise
+
+
+def _velero_teardown(resource, exception_type, exception_value, traceback):
+    teardown_error = None
+
+    if resource.teardown:
+        try:
+            delete_velero_resource(resource=resource, client=resource.client)
+        except Exception as error:
+            LOGGER.error(
+                f"Failed to delete Velero resource during teardown: kind={resource.kind} name={resource.name}",
+                exc_info=True,
+            )
+            teardown_error = error
+
+    else:
+        LOGGER.info(f"Skipping Velero delete: kind={resource.kind} name={resource.name} teardown=False")
+
+    if teardown_error is not None and exception_type is None:
+        raise ResourceTeardownError(resource=resource) from teardown_error
 
 
 class VeleroBackup(Backup):
     def __init__(
         self,
         name: str,
+        client: DynamicClient,
         namespace: str = ADP_NAMESPACE,
         included_namespaces: list[str] | None = None,
-        client: DynamicClient = None,
         teardown: bool = False,
         yaml_file: str | None = None,
         excluded_resources: list[str] | None = None,
@@ -75,15 +132,10 @@ def __enter__(self) -> "VeleroBackup":
         return self
 
     def __exit__(self, exception_type, exception_value, traceback) -> None:
-        try:
-            if self.teardown:
-                delete_velero_resource(resource=self, client=self.client)
-            else:
-                LOGGER.info(f"Skipping Velero delete for {self.kind} {self.name} (teardown=False)")
-        except Exception:
-            LOGGER.exception(f"Failed to delete Velero {self.kind} {self.name}")
-        finally:
-            super().__exit__(exception_type, exception_value, traceback)
+        _velero_teardown(
+            resource=self, exception_type=exception_type, exception_value=exception_value, traceback=traceback
+        )
+        return super().__exit__(exception_type, exception_value, traceback)
 
 
 @contextmanager
@@ -137,3 +189,99 @@ def create_rhel_vm(
         cleanup_artifactory_secret_and_config_map(
             artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map
         )
+
+
+class VeleroRestore(Restore):
+    """
+    Context manager for managing a Velero Restore resource.
+
+    Args:
+        wait_complete (bool):
+            Whether to wait for the Restore to reach COMPLETED status on context entry.
+    """
+
+    def __init__(
+        self,
+        name: str,
+        namespace: str = ADP_NAMESPACE,
+        teardown: bool = True,
+        wait_complete: bool = True,
+        timeout: int = TIMEOUT_5MIN,
+        **kwargs: Any,
+    ) -> None:
+        super().__init__(
+            name=unique_name(name=name),
+            namespace=namespace,
+            teardown=teardown,
+            **kwargs,
+        )
+        self.wait_complete = wait_complete
+        self.timeout = timeout
+
+    def __enter__(self) -> Self:
+        super().__enter__()
+        if self.wait_complete:
+            self.wait_for_status(
+                status=self.Status.COMPLETED,
+                timeout=self.timeout,
+            )
+        return self
+
+    def __exit__(self, exception_type, exception_value, traceback) -> None:
+        _velero_teardown(
+            resource=self, exception_type=exception_type, exception_value=exception_value, traceback=traceback
+        )
+        return super().__exit__(exception_type, exception_value, traceback)
+
+
+def check_file_in_running_vm(vm: VirtualMachineForTests, file_name: str, file_content: str) -> None:
+    """
+    Verify that a file exists in a running VM and contains the expected content.
+    VM must be running before calling this function.
+
+    This function opens a console session to the given virtual machine,
+    verifies that the specified file exists, and checks that its content matches the expected value.
+
+    Args:
+        vm: Virtual machine instance to check.
+        file_name: Name of the file expected to exist in the VM.
+        file_content: Expected content of the file.
+    """
+    LOGGER.info(f"Starting file verification in VM: vm={vm.name}, file={file_name}")
+
+    with Console(vm=vm) as vm_console:
+        LOGGER.info(f"Listing files in VM: vm={vm.name}")
+        vm_console.sendline(LS_COMMAND)
+        vm_console.expect(pattern=escape(file_name), timeout=TIMEOUT_20SEC)
+        LOGGER.info(f"Verifying file content in VM: vm={vm.name}, file={file_name}")
+        vm_console.sendline(f"cat {quote(file_name)}")
+        vm_console.expect(pattern=escape(file_content), timeout=TIMEOUT_20SEC)
+        LOGGER.info(f"File verification succeeded: vm={vm.name}, file={file_name}")
+
+
+def is_storage_class_support_volume_mode(
+    admin_client: DynamicClient, storage_class_name: str, requested_volume_mode: str
+) -> bool:
+    """
+    Check whether a storage class supports a specific volume mode.
+
+    This function inspects the StorageProfile associated with the given
+    storage class and determines whether the requested volume mode
+    (e.g. 'Filesystem' or 'Block') is listed in its claim property sets.
+
+    Args:
+        admin_client: OpenShift DynamicClient with sufficient permissions to access StorageProfile resources.
+        storage_class_name: Name of the StorageClass to be checked.
+        requested_volume_mode: Requested volume mode to validate (e.g. 'Filesystem' or 'Block').
+
+    Returns:
+        True if the storage class supports the requested volume mode;
+        False otherwise.
+    """
+    profile = StorageProfile(client=admin_client, name=storage_class_name)
+
+    claim_property_sets = profile.claim_property_sets
+    if not claim_property_sets:
+        return False
+
+    return any(prop.volumeMode == requested_volume_mode for prop in claim_property_sets)