diff --git a/tests/data_protection/oadp/conftest.py b/tests/data_protection/oadp/conftest.py index f583c00881..c656c4daa5 100644 --- a/tests/data_protection/oadp/conftest.py +++ b/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, ): diff --git a/tests/data_protection/oadp/test_velero.py b/tests/data_protection/oadp/test_velero.py index 00ab10397b..5239833919 100644 --- a/tests/data_protection/oadp/test_velero.py +++ b/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 diff --git a/tests/data_protection/oadp/utils.py b/tests/data_protection/oadp/utils.py index d688989573..11e09c2227 100644 --- a/tests/data_protection/oadp/utils.py +++ b/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) diff --git a/utilities/oadp.py b/utilities/oadp.py index 7846bcc1aa..896f175821 100644 --- a/utilities/oadp.py +++ b/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) diff --git a/utilities/unittests/test_oadp.py b/utilities/unittests/test_oadp.py index 8432a8597f..48525fba6d 100644 --- a/utilities/unittests/test_oadp.py +++ b/utilities/unittests/test_oadp.py @@ -2,14 +2,19 @@ """Unit tests for oadp module""" +# flake8: noqa: E402 import sys +from re import escape +from shlex import quote from unittest.mock import MagicMock, patch +import pexpect import pytest # Need to mock circular imports for oadp import utilities +# mock must be before importing oadp to prevent circular import mock_virt = MagicMock() mock_infra = MagicMock() sys.modules["utilities.virt"] = mock_virt @@ -18,75 +23,146 @@ utilities.infra = mock_infra # Import after setting up mocks to avoid circular dependency -from utilities.oadp import ( # noqa: E402 +from kubernetes.dynamic.exceptions import ResourceNotFoundError +from ocp_resources.exceptions import ResourceTeardownError + +from utilities.constants import ( + ADP_NAMESPACE, + LS_COMMAND, + TIMEOUT_20SEC, +) +from utilities.oadp import ( VeleroBackup, + VeleroRestore, + _velero_teardown, + check_file_in_running_vm, create_rhel_vm, delete_velero_resource, + is_storage_class_support_volume_mode, ) class TestDeleteVeleroResource: """Test cases for delete_velero_resource function""" + @pytest.mark.parametrize( + ("kind", "name"), + [ + ("Backup", "test-backup"), + ("Restore", "test-restore"), + ], + ) @patch("utilities.oadp.get_pod_by_name_prefix") - def test_delete_velero_resource_success(self, mock_get_pod): - """Test successful deletion of Velero resource""" + @patch("utilities.oadp.LOGGER") + def test_delete_velero_resource_success(self, mock_logger, mock_get_pod, kind, name): mock_client = MagicMock() mock_resource = MagicMock() - mock_resource.kind = "Backup" - mock_resource.name = "test-backup" + mock_resource.kind = kind + mock_resource.name = name mock_pod = MagicMock() - mock_pod.execute = MagicMock() mock_get_pod.return_value = mock_pod delete_velero_resource(resource=mock_resource, client=mock_client) - mock_get_pod.assert_called_once_with(client=mock_client, pod_prefix="velero", namespace="openshift-adp") - mock_pod.execute.assert_called_once_with(command=["./velero", "delete", "backup", "test-backup", "--confirm"]) + mock_get_pod.assert_called_once_with( + client=mock_client, + pod_prefix="velero", + namespace=ADP_NAMESPACE, + ) + mock_pod.execute.assert_called_once_with(command=["./velero", "delete", kind.lower(), name, "--confirm"]) + expected_message = ( + f"Deleting Velero resource: kind={kind} name={name} command=./velero delete {kind.lower()} {name} --confirm" + ) + mock_logger.info.assert_called_once_with(expected_message) + + @pytest.mark.parametrize( + ("kind", "name"), + [ + ("Backup", "test-backup"), + ("Restore", "test-restore"), + ], + ) @patch("utilities.oadp.get_pod_by_name_prefix") - def test_delete_velero_resource_restore(self, mock_get_pod): - """Test successful deletion of Velero restore resource""" + @patch("utilities.oadp.LOGGER") + def test_delete_velero_resource_not_found_exception(self, mock_logger, mock_get_pod, kind, name): + """Test delete_velero_resource when get_pod_by_name_prefix raises exception""" mock_client = MagicMock() mock_resource = MagicMock() - mock_resource.kind = "Restore" - mock_resource.name = "test-restore" + mock_resource.kind = kind + mock_resource.name = name - mock_pod = MagicMock() - mock_pod.execute = MagicMock() - mock_get_pod.return_value = mock_pod + # simulate get_pod_by_name_prefix raising an exception + mock_get_pod.side_effect = ResourceNotFoundError("Velero resource not found") - delete_velero_resource(resource=mock_resource, client=mock_client) + with pytest.raises(ResourceNotFoundError, match="Velero resource not found"): + delete_velero_resource(resource=mock_resource, client=mock_client) - mock_get_pod.assert_called_once_with(client=mock_client, pod_prefix="velero", namespace="openshift-adp") - mock_pod.execute.assert_called_once_with(command=["./velero", "delete", "restore", "test-restore", "--confirm"]) + # ensure exception logging is called + mock_logger.error.assert_called_once() - @patch("utilities.oadp.get_pod_by_name_prefix") - def test_delete_velero_resource_pod_not_found(self, mock_get_pod): - """Test delete_velero_resource when velero pod is not found""" - mock_client = MagicMock() + log_call = mock_logger.error.call_args + log_message = log_call.args[0] + expected_message = ( + f"Failed to delete Velero resource: kind={kind} " + f"name={name} command=./velero delete {kind.lower()} {name} --confirm" + ) + assert log_message == expected_message, "Unexpected delete_velero_resource error log message" + + +class TestVeleroTeardown: + """Test cases for _velero_teardown function""" + + @patch("utilities.oadp.delete_velero_resource") + def test_velero_teardown_with_teardown_true_calls_delete(self, mock_delete): mock_resource = MagicMock() - mock_resource.kind = "Backup" - mock_resource.name = "test-backup" + mock_resource.teardown = True + mock_resource.client = MagicMock() - mock_get_pod.return_value = None + _velero_teardown(resource=mock_resource, exception_type=None, exception_value=None, traceback=None) - with pytest.raises(AttributeError): - delete_velero_resource(resource=mock_resource, client=mock_client) + mock_delete.assert_called_once_with(resource=mock_resource, client=mock_resource.client) - @patch("utilities.oadp.get_pod_by_name_prefix") - def test_delete_velero_resource_pod_exception(self, mock_get_pod): - """Test delete_velero_resource when getting pod raises exception""" - mock_client = MagicMock() + @pytest.mark.parametrize( + ("kind", "name"), + [ + ("Backup", "test-backup"), + ("Restore", "test-restore"), + ], + ) + @patch("utilities.oadp.LOGGER") + @patch("utilities.oadp.delete_velero_resource") + def test_velero_teardown_with_teardown_false_logs_and_no_delete(self, mock_delete, mock_logger, kind, name): mock_resource = MagicMock() - mock_resource.kind = "Backup" - mock_resource.name = "test-backup" + mock_resource.teardown = False + mock_resource.kind = kind + mock_resource.name = name - mock_get_pod.side_effect = Exception("Pod not found") + _velero_teardown(resource=mock_resource, exception_type=None, exception_value=None, traceback=None) - with pytest.raises(Exception, match="Pod not found"): - delete_velero_resource(resource=mock_resource, client=mock_client) + mock_delete.assert_not_called() + + mock_logger.info.assert_called_once() + log_call = mock_logger.info.call_args + log_message = log_call.args[0] + + expected_message = f"Skipping Velero delete: kind={kind} name={name} teardown=False" + + assert log_message == expected_message, "Unexpected _velero_teardown skip log message" + + @patch("utilities.oadp.delete_velero_resource") + def test_velero_teardown_raises_resource_teardown_error_when_delete_fails(self, mock_delete): + mock_resource = MagicMock() + mock_resource.teardown = True + mock_resource.client = MagicMock() + + mock_delete.side_effect = Exception("delete failed") + + with pytest.raises(ResourceTeardownError): + _velero_teardown(resource=mock_resource, exception_type=None, exception_value=None, traceback=None) + + mock_delete.assert_called_once_with(resource=mock_resource, client=mock_resource.client) class TestVeleroBackup: @@ -125,8 +201,8 @@ def test_velero_backup_init(self, mock_backup_init, mock_unique_name): storage_location="default", snapshot_move_data=True, ) - assert backup.wait_complete is True - assert backup.timeout == 600 + assert backup.wait_complete, "wait_complete should default to True" + assert backup.timeout == 600, "Expected timeout to be 600 seconds when provided" @patch("utilities.oadp.unique_name") @patch("utilities.oadp.Backup.__init__") @@ -139,8 +215,8 @@ def test_velero_backup_init_defaults(self, mock_backup_init, mock_unique_name): backup = VeleroBackup(name="test-backup", client=mock_client) mock_unique_name.assert_called_once_with(name="test-backup") - assert backup.wait_complete is True - assert backup.timeout == 300 # TIMEOUT_5MIN + assert backup.wait_complete, "wait_complete should default to True" + assert backup.timeout == 300, "Expected timeout to be 300 seconds when provided" # TIMEOUT_5MIN @patch("utilities.oadp.unique_name") @patch("utilities.oadp.Backup.__init__") @@ -161,7 +237,7 @@ def test_velero_backup_enter_with_wait_complete(self, mock_backup_enter, mock_ba mock_backup_enter.assert_called_once() backup.wait_for_status.assert_called_once_with(status="Completed", timeout=300) - assert result == backup + assert result == backup, "Expected __enter__ to return the backup instance" @patch("utilities.oadp.unique_name") @patch("utilities.oadp.Backup.__init__") @@ -180,90 +256,27 @@ def test_velero_backup_enter_without_wait_complete(self, mock_backup_enter, mock mock_backup_enter.assert_called_once() backup.wait_for_status.assert_not_called() - assert result == backup + assert result == backup, "Expected __enter__ to return the backup instance" - @patch("utilities.oadp.unique_name") - @patch("utilities.oadp.Backup.__init__") + @patch("utilities.oadp._velero_teardown") @patch("utilities.oadp.Backup.__exit__") - @patch("utilities.oadp.delete_velero_resource") - def test_velero_backup_exit_with_teardown( - self, mock_delete_resource, mock_backup_exit, mock_backup_init, mock_unique_name - ): - """Test VeleroBackup __exit__ calls delete_velero_resource when teardown=True""" - # Mock Backup.__init__ to not raise error and allow attribute setting - mock_backup_init.return_value = None - mock_unique_name.return_value = "test-backup-unique" - mock_client = MagicMock() - - backup = VeleroBackup(name="test-backup", client=mock_client, teardown=True) - # Manually set teardown since the mock doesn't do it - backup.teardown = True - backup.client = mock_client - backup.kind = "Backup" - backup.name = "test-backup-unique" - - backup.__exit__(None, None, None) - - mock_delete_resource.assert_called_once_with(resource=backup, client=mock_client) - mock_backup_exit.assert_called_once_with(None, None, None) - - @patch("utilities.oadp.unique_name") @patch("utilities.oadp.Backup.__init__") - @patch("utilities.oadp.Backup.__exit__") - @patch("utilities.oadp.delete_velero_resource") - @patch("utilities.oadp.LOGGER") - def test_velero_backup_exit_without_teardown( - self, mock_logger, mock_delete_resource, mock_backup_exit, mock_backup_init, mock_unique_name - ): - """Test VeleroBackup __exit__ skips delete when teardown=False""" - # Mock Backup.__init__ to not raise error and allow attribute setting + def test_velero_backup_exit_calls_teardown_and_super(self, mock_backup_init, mock_backup_exit, mock_teardown): mock_backup_init.return_value = None - mock_unique_name.return_value = "test-backup-unique" - mock_client = MagicMock() - backup = VeleroBackup(name="test-backup", client=mock_client, teardown=False) - # Manually set teardown since the mock doesn't do it - backup.teardown = False - backup.kind = "Backup" - backup.name = "test-backup-unique" + backup = VeleroBackup(name="test-backup", client=MagicMock(), teardown=True) backup.__exit__(None, None, None) - mock_delete_resource.assert_not_called() - mock_logger.info.assert_called_once_with( - "Skipping Velero delete for Backup test-backup-unique (teardown=False)" + # 1. Ensure _velero_teardown was called + mock_teardown.assert_called_once_with( + resource=backup, + exception_type=None, + exception_value=None, + traceback=None, ) - mock_backup_exit.assert_called_once_with(None, None, None) - @patch("utilities.oadp.unique_name") - @patch("utilities.oadp.Backup.__init__") - @patch("utilities.oadp.Backup.__exit__") - @patch("utilities.oadp.delete_velero_resource") - @patch("utilities.oadp.LOGGER") - def test_velero_backup_exit_delete_exception( - self, mock_logger, mock_delete_resource, mock_backup_exit, mock_backup_init, mock_unique_name - ): - """Test VeleroBackup __exit__ handles delete exception gracefully""" - # Mock Backup.__init__ to not raise error and allow attribute setting - mock_backup_init.return_value = None - mock_unique_name.return_value = "test-backup-unique" - mock_client = MagicMock() - - backup = VeleroBackup(name="test-backup", client=mock_client, teardown=True) - # Manually set teardown since the mock doesn't do it - backup.teardown = True - backup.client = mock_client - backup.kind = "Backup" - backup.name = "test-backup-unique" - - mock_delete_resource.side_effect = Exception("Delete failed") - - # Should not raise exception - backup.__exit__(None, None, None) - - mock_delete_resource.assert_called_once_with(resource=backup, client=mock_client) - mock_logger.exception.assert_called_once_with("Failed to delete Velero Backup test-backup-unique") - # Parent __exit__ should still be called + # 2. Ensure super().__exit__ was called mock_backup_exit.assert_called_once_with(None, None, None) @@ -589,3 +602,199 @@ def test_create_rhel_vm_running_vm_exception( # Cleanup should still be called mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + + +class TestVeleroRestore: + """Test cases for VeleroRestore class""" + + @patch("utilities.oadp.unique_name") + @patch("utilities.oadp.Restore.__init__") + def test_velero_restore_init(self, mock_restore_init, mock_unique_name): + mock_restore_init.return_value = None + mock_unique_name.return_value = "test-restore-unique" + mock_client = MagicMock() + + restore = VeleroRestore( + name="test-restore", + namespace="test-namespace", + included_namespaces=["ns1"], + backup_name="backup-1", + client=mock_client, + teardown=True, + wait_complete=True, + timeout=600, + ) + + mock_unique_name.assert_called_once_with(name="test-restore") + mock_restore_init.assert_called_once_with( + name="test-restore-unique", + namespace="test-namespace", + included_namespaces=["ns1"], + backup_name="backup-1", + client=mock_client, + teardown=True, + ) + + assert restore.wait_complete, "wait_complete should default to True" + assert restore.timeout == 600, "Expected timeout to be 600 seconds when provided" + + @patch("utilities.oadp.Restore.__init__") + @patch("utilities.oadp.Restore.__enter__") + def test_velero_restore_enter_with_wait_complete(self, mock_restore_enter, mock_restore_init): + mock_restore_init.return_value = None + + restore = VeleroRestore(name="test-restore", client=MagicMock(), wait_complete=True) + restore.wait_for_status = MagicMock() + restore.Status = MagicMock() + restore.Status.COMPLETED = "Completed" + + mock_restore_enter.return_value = restore + + result = restore.__enter__() + + mock_restore_enter.assert_called_once() + restore.wait_for_status.assert_called_once_with(status="Completed", timeout=300) + assert result == restore, "Expected __enter__ to return the restore instance" + + @patch("utilities.oadp.Restore.__init__") + @patch("utilities.oadp.Restore.__enter__") + def test_velero_restore_enter_without_wait_complete(self, mock_restore_enter, mock_restore_init): + mock_restore_init.return_value = None + + restore = VeleroRestore(name="test-restore", client=MagicMock(), wait_complete=False) + restore.wait_for_status = MagicMock() + + mock_restore_enter.return_value = restore + + restore.__enter__() + + restore.wait_for_status.assert_not_called() + + @patch("utilities.oadp._velero_teardown") + @patch("utilities.oadp.Restore.__exit__") + @patch("utilities.oadp.Restore.__init__") + def test_velero_restore_exit_calls_teardown_and_super(self, mock_restore_init, mock_restore_exit, mock_teardown): + mock_restore_init.return_value = None + + restore = VeleroRestore(name="test-restore", client=MagicMock(), teardown=True) + + restore.__exit__(None, None, None) + + # 1. Ensure _velero_teardown was called + mock_teardown.assert_called_once_with( + resource=restore, + exception_type=None, + exception_value=None, + traceback=None, + ) + + # 2. Ensure super().__exit__ was called + mock_restore_exit.assert_called_once_with(None, None, None) + + +class TestCheckFileInRunningVm: + """Test cases for check_file_in_running_vm function that verifies file existence and content in a running VM.""" + + @patch("utilities.oadp.Console") + def test_check_file_in_running_vm(self, mock_console_cls): + mock_vm = MagicMock() + mock_vm.ready = True + + mock_console = MagicMock() + mock_console_cls.return_value.__enter__.return_value = mock_console + + check_file_in_running_vm( + vm=mock_vm, + file_name="test-file", + file_content="hello world", + ) + + mock_vm.start.assert_not_called() + + mock_console.sendline.assert_any_call(LS_COMMAND) + mock_console.expect.assert_any_call(pattern=escape("test-file"), timeout=TIMEOUT_20SEC) + mock_console.sendline.assert_any_call(f"cat {quote('test-file')}") + mock_console.expect.assert_any_call(pattern=escape("hello world"), timeout=TIMEOUT_20SEC) + + @patch("utilities.oadp.Console") + def test_check_file_in_running_vm_file_not_found(self, mock_console_cls): + mock_vm = MagicMock() + mock_vm.ready = True + + mock_console = MagicMock() + mock_console.expect.side_effect = pexpect.TIMEOUT("file not found") + mock_console_cls.return_value.__enter__.return_value = mock_console + + with pytest.raises(pexpect.TIMEOUT): + check_file_in_running_vm( + vm=mock_vm, + file_name="missing-file", + file_content="hello world", + ) + + @patch("utilities.oadp.Console") + def test_check_file_in_running_vm_content_mismatch(self, mock_console_cls): + mock_vm = MagicMock() + mock_vm.ready = True + + mock_console = MagicMock() + mock_console.sendline.side_effect = [ + None, # ls + None, # cat + ] + mock_console.expect.side_effect = [ + None, # find file + Exception("content mismatch"), + ] + + mock_console_cls.return_value.__enter__.return_value = mock_console + + with pytest.raises(Exception, match="content mismatch"): + check_file_in_running_vm( + vm=mock_vm, + file_name="test-file", + file_content="expected content", + ) + + +class TestIsStorageClassSupportVolumeMode: + """Test cases for is_storage_class_support_volume_mode function that checks StorageProfile volume mode support.""" + + @patch("utilities.oadp.StorageProfile") + def test_volume_mode_supported(self, mock_profile): + admin_client = MagicMock() + mock_profile.return_value.claim_property_sets = [ + MagicMock(volumeMode="Filesystem"), + MagicMock(volumeMode="Block"), + ] + + assert is_storage_class_support_volume_mode( + admin_client=admin_client, storage_class_name="sc-name", requested_volume_mode="Block" + ), "Expected Block volume mode to be supported" + + @patch("utilities.oadp.StorageProfile") + def test_volume_mode_not_supported(self, mock_profile): + admin_client = MagicMock() + mock_profile.return_value.claim_property_sets = [ + MagicMock(volumeMode="Filesystem"), + ] + + assert not is_storage_class_support_volume_mode( + admin_client=admin_client, storage_class_name="sc-name", requested_volume_mode="Block" + ), "Expected Block volume mode to be unsupported" + + @pytest.mark.parametrize( + "claim_property_sets", + [ + None, + [], + ], + ) + @patch("utilities.oadp.StorageProfile") + def test_volume_mode_no_claim_property_sets(self, mock_profile, claim_property_sets): + admin_client = MagicMock() + mock_profile.return_value.claim_property_sets = claim_property_sets + + assert not is_storage_class_support_volume_mode( + admin_client=admin_client, storage_class_name="sc-name", requested_volume_mode="Block" + ), "Expected Block volume mode to be unsupported when claim_property_sets is empty"