diff --git a/.flake8 b/.flake8 index b149f33d4a..f76aea7f5d 100644 --- a/.flake8 +++ b/.flake8 @@ -103,6 +103,8 @@ fcn_exclude_functions = mkdtemp, tempfile, urllib3, + enter_context, + callback, nit_exclude_imports = os_params, diff --git a/tests/chaos/conftest.py b/tests/chaos/conftest.py index 9470c96db4..c3da2b8d3d 100644 --- a/tests/chaos/conftest.py +++ b/tests/chaos/conftest.py @@ -17,7 +17,7 @@ pod_deleting_process_recover, terminate_process, ) -from utilities.artifactory import get_artifactory_config_map, get_artifactory_secret +from utilities.artifactory import artifactory_credentials from utilities.constants import Images from utilities.constants.components import KUBEMACPOOL_MAC_CONTROLLER_MANAGER from utilities.constants.images import OS_FLAVOR_RHEL @@ -90,15 +90,14 @@ def chaos_dv_rhel9( admin_client, chaos_namespace, rhel9_http_image_url, - artifactory_secret_chaos_namespace_scope_module, - artifactory_config_map_chaos_namespace_scope_module, + artifactory_credentials_chaos_namespace_scope_module, ): yield DataVolume( source_dict=construct_datavolume_source_dict( source="http", url=rhel9_http_image_url, - secret_name=artifactory_secret_chaos_namespace_scope_module.name, - cert_configmap_name=artifactory_config_map_chaos_namespace_scope_module.name, + secret_name=artifactory_credentials_chaos_namespace_scope_module.secret_name, + cert_configmap_name=artifactory_credentials_chaos_namespace_scope_module.cert_configmap_name, ), name="chaos-dv", api_name="storage", @@ -355,19 +354,9 @@ def vm_node_with_chaos_label(vm_with_nginx_service): @pytest.fixture(scope="module") -def artifactory_secret_chaos_namespace_scope_module(chaos_namespace): - artifactory_secret = get_artifactory_secret(namespace=chaos_namespace.name) - yield artifactory_secret - if artifactory_secret.exists: - artifactory_secret.clean_up() - - -@pytest.fixture(scope="module") -def artifactory_config_map_chaos_namespace_scope_module(chaos_namespace): - artifactory_config_map = get_artifactory_config_map(namespace=chaos_namespace.name) - yield artifactory_config_map - if artifactory_config_map.exists: - artifactory_config_map.clean_up() +def artifactory_credentials_chaos_namespace_scope_module(chaos_namespace): + with artifactory_credentials(namespace=chaos_namespace.name, client=chaos_namespace.client) as credentials: + yield credentials @pytest.fixture(scope="class") diff --git a/tests/chaos/snapshot/conftest.py b/tests/chaos/snapshot/conftest.py index 393223f6b6..bab6a81f02 100644 --- a/tests/chaos/snapshot/conftest.py +++ b/tests/chaos/snapshot/conftest.py @@ -18,15 +18,14 @@ def chaos_dv_rhel9_for_snapshot( chaos_namespace, storage_class_matrix_snapshot_matrix__function__, rhel9_http_image_url, - artifactory_secret_chaos_namespace_scope_module, - artifactory_config_map_chaos_namespace_scope_module, + artifactory_credentials_chaos_namespace_scope_module, ): yield DataVolume( source_dict=construct_datavolume_source_dict( source="http", url=rhel9_http_image_url, - secret_name=artifactory_secret_chaos_namespace_scope_module.name, - cert_configmap_name=artifactory_config_map_chaos_namespace_scope_module.name, + secret_name=artifactory_credentials_chaos_namespace_scope_module.secret_name, + cert_configmap_name=artifactory_credentials_chaos_namespace_scope_module.cert_configmap_name, ), name="chaos-dv", api_name="storage", diff --git a/tests/conftest.py b/tests/conftest.py index cc4728e2f8..13e46a44a6 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,6 +14,7 @@ import tempfile from bisect import bisect_left from collections import defaultdict +from contextlib import ExitStack from datetime import UTC, datetime from signal import SIGINT, SIGTERM, getsignal, signal @@ -73,7 +74,7 @@ from libs.net.ip import filter_link_local_addresses, random_cidr_addresses_by_family from libs.net.vmspec import lookup_iface_status from tests.utils import download_and_extract_tar -from utilities.artifactory import get_artifactory_header, get_test_artifact_server_url +from utilities.artifactory import artifactory_credentials, get_artifactory_header, get_test_artifact_server_url from utilities.cluster import cache_admin_client, get_oc_whoami_username from utilities.constants import Images from utilities.constants.aaq import ( @@ -2485,42 +2486,37 @@ def dvs_for_upgrade( worker_node1, rhel_latest_os_params, updated_default_storage_class_ocs_virt, + golden_images_namespace, ): - golden_images_namespace_name = py_config["golden_images_namespace"] - dvs_list = [] - artifactory_secret = utilities.artifactory.get_artifactory_secret(namespace=golden_images_namespace_name) - artifactory_config_map = utilities.artifactory.get_artifactory_config_map(namespace=golden_images_namespace_name) - - for sc in py_config["storage_class_matrix"]: - storage_class = [*sc][0] - dv = DataVolume( - client=admin_client, - name=f"dv-for-product-upgrade-{storage_class}", - namespace=golden_images_namespace_name, - source_dict=construct_datavolume_source_dict( - source="http", - url=rhel_latest_os_params["rhel_image_path"], - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ), - storage_class=storage_class, - size=rhel_latest_os_params["rhel_dv_size"], - annotations=BIND_IMMEDIATE_ANNOTATION, - api_name="storage", + with ExitStack() as stack: + artifactory = stack.enter_context( + artifactory_credentials(namespace=golden_images_namespace.name, client=golden_images_namespace.client) ) - dv.create() - dvs_list.append(dv) - for dv in dvs_list: - dv.wait_for_dv_success() - - yield dvs_list - - for dv in dvs_list: - dv.clean_up() - utilities.artifactory.cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, - artifactory_config_map=artifactory_config_map, - ) + dvs_list = [] + for sc in py_config["storage_class_matrix"]: + storage_class = [*sc][0] + dv = DataVolume( + client=admin_client, + name=f"dv-for-product-upgrade-{storage_class}", + namespace=golden_images_namespace.name, + source_dict=construct_datavolume_source_dict( + source="http", + url=rhel_latest_os_params["rhel_image_path"], + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ), + storage_class=storage_class, + size=rhel_latest_os_params["rhel_dv_size"], + annotations=BIND_IMMEDIATE_ANNOTATION, + api_name="storage", + ) + dv.create() + stack.callback(dv.clean_up) + dvs_list.append(dv) + for dv in dvs_list: + dv.wait_for_dv_success() + + yield dvs_list @pytest.fixture(scope="class") diff --git a/tests/data_protection/oadp/conftest.py b/tests/data_protection/oadp/conftest.py index 4e19b04de6..a97ae3483a 100644 --- a/tests/data_protection/oadp/conftest.py +++ b/tests/data_protection/oadp/conftest.py @@ -9,9 +9,7 @@ FILE_PATH_FOR_WINDOWS_BACKUP, ) from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, + artifactory_credentials, get_test_artifact_server_url, ) from utilities.constants import Images @@ -163,13 +161,9 @@ def windows_vm_with_data_volume_template( snapshot_storage_class_name_scope_module, ): """Windows 2022 VM with InstanceType and Preference in the backup namespace for OADP backup testing.""" - artifactory_secret = None - artifactory_config_map = None - - try: - artifactory_secret = get_artifactory_secret(namespace=namespace_for_backup.name) - artifactory_config_map = get_artifactory_config_map(namespace=namespace_for_backup.name) - + with artifactory_credentials( + namespace=namespace_for_backup.name, client=namespace_for_backup.client + ) as artifactory: dv = DataVolume( name="oadp-windows-dv", namespace=namespace_for_backup.name, @@ -180,8 +174,8 @@ def windows_vm_with_data_volume_template( f"{get_test_artifact_server_url(schema='registry')}/" f"{py_config['latest_windows_os_dict'][CONTAINER_DISK_IMAGE_PATH_STR]}" ), - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, ), size=Images.Windows.CONTAINER_DISK_DV_SIZE, client=admin_client, @@ -201,10 +195,6 @@ def windows_vm_with_data_volume_template( running_vm(vm=vm) write_file_windows_vm(vm=vm, file_path=FILE_PATH_FOR_WINDOWS_BACKUP, content=TEXT_TO_TEST) yield vm - finally: - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map - ) @pytest.fixture() diff --git a/tests/fixtures/images/validation_os_images.py b/tests/fixtures/images/validation_os_images.py index f2f93d719f..3bca87a8d2 100644 --- a/tests/fixtures/images/validation_os_images.py +++ b/tests/fixtures/images/validation_os_images.py @@ -8,9 +8,7 @@ from pytest_testconfig import config as py_config from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, + artifactory_credentials, get_test_artifact_server_url, ) from utilities.constants import Images @@ -100,31 +98,27 @@ def windows_validation_os_images_data_volume_scope_session( " Self-validation requires the Windows image to be pre-created." ) - artifactory_secret = get_artifactory_secret( - namespace=validation_os_images_role_binding.namespace, client=validation_os_images_role_binding.client - ) - artifactory_config_map = get_artifactory_config_map( - namespace=validation_os_images_role_binding.namespace, client=validation_os_images_role_binding.client - ) + with artifactory_credentials( + namespace=validation_os_images_role_binding.namespace, + client=validation_os_images_role_binding.client, + ) as artifactory: + win_dv.storage_class = py_config["default_storage_class"] + win_dv.source_dict = construct_datavolume_source_dict( + source=REGISTRY_STR, + url=( + f"{get_test_artifact_server_url(schema=REGISTRY_STR)}/" + f"{get_windows_container_disk_path(os_value=WIN_2K22)}" + ), + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ) + win_dv.size = Images.Windows.CONTAINER_DISK_DV_SIZE + win_dv.api_name = "storage" + win_dv.annotations = BIND_IMMEDIATE_ANNOTATION - win_dv.storage_class = py_config["default_storage_class"] - win_dv.source_dict = construct_datavolume_source_dict( - source=REGISTRY_STR, - url=f"{get_test_artifact_server_url(schema=REGISTRY_STR)}/{get_windows_container_disk_path(os_value=WIN_2K22)}", - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ) - win_dv.size = Images.Windows.CONTAINER_DISK_DV_SIZE - win_dv.api_name = "storage" - win_dv.annotations = BIND_IMMEDIATE_ANNOTATION - - with win_dv as wdv: - wdv.wait_for_dv_success(timeout=TIMEOUT_50MIN) - yield wdv - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, - artifactory_config_map=artifactory_config_map, - ) + with win_dv as wdv: + wdv.wait_for_dv_success(timeout=TIMEOUT_50MIN) + yield wdv @pytest.fixture(scope="session") diff --git a/tests/infrastructure/instance_types/supported_os/conftest.py b/tests/infrastructure/instance_types/supported_os/conftest.py index 7664973c3d..e31c9e5ca5 100644 --- a/tests/infrastructure/instance_types/supported_os/conftest.py +++ b/tests/infrastructure/instance_types/supported_os/conftest.py @@ -5,11 +5,7 @@ from pytest_testconfig import config as py_config from tests.infrastructure.instance_types.supported_os.utils import golden_image_vm_with_instance_type -from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, -) +from utilities.artifactory import artifactory_credentials from utilities.constants import Images from utilities.constants.hco import DATA_SOURCE_NAME from utilities.constants.images import OS_FLAVOR_WIN_CONTAINER_DISK @@ -88,16 +84,21 @@ def golden_image_fedora_vm_with_instance_type( ) +@pytest.fixture(scope="module") +def windows_instance_type_artifactory_credentials(namespace): + with artifactory_credentials(namespace=namespace.name, client=namespace.client) as credentials: + yield credentials + + @pytest.fixture(scope="module") def windows_data_volume_template( unprivileged_client, namespace, windows_os_matrix__module__, + windows_instance_type_artifactory_credentials, ): os_matrix_key = [*windows_os_matrix__module__][0] os_params = windows_os_matrix__module__[os_matrix_key] - secret = get_artifactory_secret(namespace=namespace.name) - cert = get_artifactory_config_map(namespace=namespace.name) win_dv = DataVolume( client=unprivileged_client, name=f"{os_matrix_key}-dv", @@ -106,15 +107,14 @@ def windows_data_volume_template( source_dict=construct_datavolume_source_dict( source="registry", url=f"{get_test_artifact_server_url(schema='registry')}/{os_params[CONTAINER_DISK_IMAGE_PATH_STR]}", - secret_name=secret.name, - cert_configmap_name=cert.name, + secret_name=windows_instance_type_artifactory_credentials.secret_name, + cert_configmap_name=windows_instance_type_artifactory_credentials.cert_configmap_name, ), size=Images.Windows.CONTAINER_DISK_DV_SIZE, storage_class=py_config["default_storage_class"], ) win_dv.to_dict() yield win_dv - cleanup_artifactory_secret_and_config_map(artifactory_secret=secret, artifactory_config_map=cert) @pytest.fixture(scope="class") diff --git a/tests/infrastructure/tekton/conftest.py b/tests/infrastructure/tekton/conftest.py index bdc38361f1..0dbe06508c 100644 --- a/tests/infrastructure/tekton/conftest.py +++ b/tests/infrastructure/tekton/conftest.py @@ -23,7 +23,7 @@ win_iso_download_url_for_pipelineref, yaml_files_in_dir, ) -from utilities.artifactory import get_artifactory_config_map, get_artifactory_secret +from utilities.artifactory import artifactory_credentials from utilities.constants.images import OS_FLAVOR_FEDORA from utilities.constants.tekton import ( TEKTON_AVAILABLE_PIPELINEREF, @@ -216,8 +216,7 @@ def processed_yaml_files( def resource_editor_efi_pipelines( admin_client, custom_pipeline_namespace, - artifactory_secret_custom_pipeline_namespace, - artifactory_config_map_custom_pipeline_namespace, + artifactory_credentials_custom_pipeline_namespace, ): pipeline = Pipeline(client=admin_client, name=WINDOWS_EFI_INSTALLER_STR, namespace=custom_pipeline_namespace.name) pipeline_dict = pipeline.instance.to_dict() @@ -229,10 +228,10 @@ def resource_editor_efi_pipelines( manifest = yaml.safe_load(param["value"]) if manifest["spec"]["source"]["http"]["url"] == "$(params.winImageDownloadURL)": manifest["spec"]["source"]["http"]["secretRef"] = ( - artifactory_secret_custom_pipeline_namespace.name + artifactory_credentials_custom_pipeline_namespace.secret_name ) manifest["spec"]["source"]["http"]["certConfigMap"] = ( - artifactory_config_map_custom_pipeline_namespace.name + artifactory_credentials_custom_pipeline_namespace.cert_configmap_name ) param["value"] = yaml.dump(manifest) @@ -248,19 +247,12 @@ def custom_pipeline_namespace(unprivileged_client, admin_client): @pytest.fixture(scope="module") -def artifactory_secret_custom_pipeline_namespace(custom_pipeline_namespace): - artifactory_secret = get_artifactory_secret(namespace=custom_pipeline_namespace.name) - yield artifactory_secret - if artifactory_secret.exists: - artifactory_secret.clean_up() - - -@pytest.fixture(scope="module") -def artifactory_config_map_custom_pipeline_namespace(custom_pipeline_namespace): - artifactory_config_map = get_artifactory_config_map(namespace=custom_pipeline_namespace.name) - yield artifactory_config_map - if artifactory_config_map.exists: - artifactory_config_map.clean_up() +def artifactory_credentials_custom_pipeline_namespace(custom_pipeline_namespace): + with artifactory_credentials( + namespace=custom_pipeline_namespace.name, + client=custom_pipeline_namespace.client, + ) as credentials: + yield credentials @pytest.fixture() diff --git a/tests/observability/metrics/utils.py b/tests/observability/metrics/utils.py index 7171e347c5..7b849e83e9 100644 --- a/tests/observability/metrics/utils.py +++ b/tests/observability/metrics/utils.py @@ -29,9 +29,7 @@ KUBEVIRT_VMI_FILESYSTEM_BYTES_WITH_MOUNT_POINT, ) from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, + artifactory_credentials, get_test_artifact_server_url, ) from utilities.constants import Images @@ -657,53 +655,47 @@ def create_windows11_wsl2_vm( storage_class (str): The storage class to use for the DataVolume """ windows_preference_name = "windows.11" - artifactory_secret = get_artifactory_secret(namespace=namespace) - artifactory_config_map = get_artifactory_config_map(namespace=namespace) - dv = DataVolume( - client=client, - name=dv_name, - namespace=namespace, - api_name="storage", - source_dict=construct_datavolume_source_dict( - source=REGISTRY_STR, - url=f"{get_test_artifact_server_url(schema=REGISTRY_STR)}/docker-local/windows-qe/win_11:virtio", - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ), - size=Images.Windows.CONTAINER_DISK_DV_SIZE, - storage_class=storage_class, - ) - dv.to_dict() - base_preference = VirtualMachineClusterPreference(client=client, name=windows_preference_name) - base_spec = base_preference.instance.to_dict()["spec"] - - with VirtualMachinePreference( - client=client, - namespace=namespace, - name=f"{vm_name}-{windows_preference_name}-preference", - cpu={"preferredCPUTopology": "cores"}, - clock=base_spec.get("clock"), - devices=base_spec.get("devices"), - features=base_spec.get("features"), - firmware=base_spec.get("firmware"), - requirements=base_spec.get("requirements"), - ) as preference: - with VirtualMachineForTests( - os_flavor=OS_FLAVOR_WINDOWS, - name=vm_name, + with artifactory_credentials(namespace=namespace, client=client) as artifactory: + dv = DataVolume( + client=client, + name=dv_name, namespace=namespace, + api_name="storage", + source_dict=construct_datavolume_source_dict( + source=REGISTRY_STR, + url=f"{get_test_artifact_server_url(schema=REGISTRY_STR)}/docker-local/windows-qe/win_11:virtio", + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ), + size=Images.Windows.CONTAINER_DISK_DV_SIZE, + storage_class=storage_class, + ) + dv.to_dict() + base_preference = VirtualMachineClusterPreference(client=client, name=windows_preference_name) + base_spec = base_preference.instance.to_dict()["spec"] + + with VirtualMachinePreference( client=client, - vm_instance_type=VirtualMachineClusterInstancetype(client=client, name="u1.large"), - vm_preference=preference, - data_volume_template={"metadata": dv.res["metadata"], "spec": dv.res["spec"]}, - ) as vm: - try: + namespace=namespace, + name=f"{vm_name}-{windows_preference_name}-preference", + cpu={"preferredCPUTopology": "cores"}, + clock=base_spec.get("clock"), + devices=base_spec.get("devices"), + features=base_spec.get("features"), + firmware=base_spec.get("firmware"), + requirements=base_spec.get("requirements"), + ) as preference: + with VirtualMachineForTests( + os_flavor=OS_FLAVOR_WINDOWS, + name=vm_name, + namespace=namespace, + client=client, + vm_instance_type=VirtualMachineClusterInstancetype(client=client, name="u1.large"), + vm_preference=preference, + data_volume_template={"metadata": dv.res["metadata"], "spec": dv.res["spec"]}, + ) as vm: running_vm(vm=vm, dv_wait_timeout=TIMEOUT_40MIN) yield vm - finally: - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map - ) def get_vm_comparison_info_dict(vm: VirtualMachineForTests) -> dict[str, str]: diff --git a/tests/scale/test_scale_benchmark.py b/tests/scale/test_scale_benchmark.py index 615e28e336..469a87184a 100644 --- a/tests/scale/test_scale_benchmark.py +++ b/tests/scale/test_scale_benchmark.py @@ -5,6 +5,7 @@ import shlex import time from collections import Counter +from contextlib import ExitStack import pytest import yaml @@ -26,11 +27,7 @@ WINDOWS_LATEST, WINDOWS_LATEST_LABELS, ) -from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, -) +from utilities.artifactory import artifactory_credentials from utilities.constants.cluster import NODE_STR from utilities.constants.images import ( OS_FLAVOR_FEDORA, @@ -254,44 +251,42 @@ def vms_info(scale_test_param): @pytest.fixture(scope="class") -def golden_images_scale_dvs(request, keep_resources, admin_client, golden_images_namespace, dvs_info): +def golden_images_scale_dvs(keep_resources, admin_client, golden_images_namespace, dvs_info): dvs_list = [] - def _delete_resources(): - delete_resources(resources=dvs_list) - - if not keep_resources: - request.addfinalizer(_delete_resources) - - artifactory_secret = get_artifactory_secret(namespace=golden_images_namespace.name) - artifactory_config_map = get_artifactory_config_map(namespace=golden_images_namespace.name) - - for os_name, dv_info in dvs_info.items(): - storage_types_used = [storage_type_key for storage_type_key in SCALE_STORAGE_TYPES if dv_info[storage_type_key]] - for storage_type in storage_types_used: - golden_images_scale_dv = DataVolume( - name=f"{os_name}-{storage_type}-dv", - namespace=golden_images_namespace.name, - storage_class=SCALE_STORAGE_TYPES[storage_type], - api_name="storage", - source_dict=construct_datavolume_source_dict( - source="http", - url=f"{get_test_artifact_server_url()}{dv_info['url']}", - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ), - size=dv_info["size"], - client=admin_client, - ) - golden_images_scale_dv.deploy() - dvs_list.append(golden_images_scale_dv) - for dv in dvs_list: - dv.wait_for_status(status=DataVolume.Status.SUCCEEDED, timeout=TIMEOUT_30MIN) - yield dvs_list - - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map - ) + with ExitStack() as stack: + artifactory = stack.enter_context( + artifactory_credentials(namespace=golden_images_namespace.name, client=golden_images_namespace.client) + ) + if not keep_resources: + stack.callback(delete_resources, resources=dvs_list) + + for os_name, dv_info in dvs_info.items(): + storage_types_used = [ + storage_type_key for storage_type_key in SCALE_STORAGE_TYPES if dv_info[storage_type_key] + ] + for storage_type in storage_types_used: + golden_images_scale_dv = DataVolume( + name=f"{os_name}-{storage_type}-dv", + namespace=golden_images_namespace.name, + storage_class=SCALE_STORAGE_TYPES[storage_type], + api_name="storage", + source_dict=construct_datavolume_source_dict( + source="http", + url=f"{get_test_artifact_server_url()}{dv_info['url']}", + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ), + size=dv_info["size"], + client=admin_client, + ) + dvs_list.append(golden_images_scale_dv) + golden_images_scale_dv.deploy() + for dv in dvs_list: + dv.wait_for_status(status=DataVolume.Status.SUCCEEDED, timeout=TIMEOUT_30MIN) + if keep_resources: + stack.pop_all() + yield dvs_list @pytest.fixture(scope="class") diff --git a/tests/storage/conftest.py b/tests/storage/conftest.py index 657d65bb11..4bd7bd3bff 100644 --- a/tests/storage/conftest.py +++ b/tests/storage/conftest.py @@ -44,7 +44,7 @@ ) from tests.utils import create_cirros_vm from utilities.architecture import get_multiarch_cpu_arch -from utilities.artifactory import get_artifactory_config_map, get_artifactory_secret +from utilities.artifactory import artifactory_credentials from utilities.constants import Images from utilities.constants.cluster import CNV_TEST_SERVICE_ACCOUNT, KUBERNETES_ARCH_LABEL from utilities.constants.components import CDI_OPERATOR, CDI_UPLOADPROXY @@ -370,19 +370,9 @@ def rhel_vm_name(request): @pytest.fixture(scope="module") -def artifactory_secret_scope_module(namespace): - artifactory_secret = get_artifactory_secret(namespace=namespace.name) - yield artifactory_secret - if artifactory_secret: - artifactory_secret.clean_up() - - -@pytest.fixture(scope="module") -def artifactory_config_map_scope_module(namespace): - artifactory_config_map = get_artifactory_config_map(namespace=namespace.name) - yield artifactory_config_map - if artifactory_config_map: - artifactory_config_map.clean_up() +def artifactory_credentials_scope_module(namespace): + with artifactory_credentials(namespace=namespace.name, client=namespace.client) as credentials: + yield credentials @pytest.fixture() diff --git a/tests/storage/cross_cluster_live_migration/conftest.py b/tests/storage/cross_cluster_live_migration/conftest.py index 1b5a344718..262ca07109 100644 --- a/tests/storage/cross_cluster_live_migration/conftest.py +++ b/tests/storage/cross_cluster_live_migration/conftest.py @@ -37,8 +37,7 @@ ) from tests.storage.utils import get_storage_class_for_storage_migration from utilities.artifactory import ( - get_artifactory_config_map, - get_artifactory_secret, + artifactory_credentials, get_test_artifact_server_url, ) from utilities.constants import Images @@ -501,23 +500,12 @@ def vm_for_cclm_with_instance_type( @pytest.fixture(scope="class") -def remote_cluster_artifactory_secret_scope_class(remote_admin_client, remote_cluster_source_test_namespace): - artifactory_secret = get_artifactory_secret( - namespace=remote_cluster_source_test_namespace.name, client=remote_admin_client - ) - yield artifactory_secret - if artifactory_secret: - artifactory_secret.clean_up() - - -@pytest.fixture(scope="class") -def remote_cluster_artifactory_config_map_scope_class(remote_admin_client, remote_cluster_source_test_namespace): - artifactory_config_map = get_artifactory_config_map( - namespace=remote_cluster_source_test_namespace.name, client=remote_admin_client - ) - yield artifactory_config_map - if artifactory_config_map: - artifactory_config_map.clean_up() +def remote_cluster_artifactory_credentials_scope_class(remote_cluster_source_test_namespace): + with artifactory_credentials( + namespace=remote_cluster_source_test_namespace.name, + client=remote_cluster_source_test_namespace.client, + ) as credentials: + yield credentials @pytest.fixture(scope="class") @@ -525,8 +513,7 @@ def vm_for_cclm_windows_with_instance_type( remote_admin_client, remote_cluster_source_test_namespace, remote_cluster_source_storage_class, - remote_cluster_artifactory_secret_scope_class, - remote_cluster_artifactory_config_map_scope_class, + remote_cluster_artifactory_credentials_scope_class, ): dv = DataVolume( client=remote_admin_client, @@ -536,8 +523,8 @@ def vm_for_cclm_windows_with_instance_type( source_dict=construct_datavolume_source_dict( source="registry", url=f"{get_test_artifact_server_url(schema='registry')}/{WINDOWS_2022[CONTAINER_DISK_IMAGE_PATH_STR]}", - secret_name=remote_cluster_artifactory_secret_scope_class.name, - cert_configmap_name=remote_cluster_artifactory_config_map_scope_class.name, + secret_name=remote_cluster_artifactory_credentials_scope_class.secret_name, + cert_configmap_name=remote_cluster_artifactory_credentials_scope_class.cert_configmap_name, ), size=Images.Windows.CONTAINER_DISK_DV_SIZE, storage_class=remote_cluster_source_storage_class, diff --git a/tests/storage/storage_migration/conftest.py b/tests/storage/storage_migration/conftest.py index 3e6e9170dd..a10497a291 100644 --- a/tests/storage/storage_migration/conftest.py +++ b/tests/storage/storage_migration/conftest.py @@ -165,8 +165,7 @@ def vm_for_storage_class_migration_from_template_with_dv( source_storage_class, cpu_for_migration, rhel_latest_os_params, - artifactory_secret_scope_module, - artifactory_config_map_scope_module, + artifactory_credentials_scope_module, ): dv = DataVolume( name="dv-rhel-imported", @@ -174,8 +173,8 @@ def vm_for_storage_class_migration_from_template_with_dv( source_dict=construct_datavolume_source_dict( source="http", url=rhel_latest_os_params["rhel_image_path"], - secret_name=artifactory_secret_scope_module.name, - cert_configmap_name=artifactory_config_map_scope_module.name, + secret_name=artifactory_credentials_scope_module.secret_name, + cert_configmap_name=artifactory_credentials_scope_module.cert_configmap_name, ), size=Images.Rhel.DEFAULT_DV_SIZE, storage_class=source_storage_class, diff --git a/tests/utils.py b/tests/utils.py index 061270a9c4..aa8fc9b786 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -27,9 +27,8 @@ from timeout_sampler import TimeoutExpiredError, TimeoutSampler, retry from utilities.artifactory import ( - get_artifactory_config_map, + artifactory_credentials, get_artifactory_header, - get_artifactory_secret, get_http_image_url, ) from utilities.constants import Images @@ -558,41 +557,39 @@ def create_cirros_vm( cpu_model: str | None = None, annotations: dict[str, str] | None = None, ) -> Generator[VirtualMachineForTests]: - artifactory_secret = get_artifactory_secret(namespace=namespace) - artifactory_config_map = get_artifactory_config_map(namespace=namespace) - - dv = DataVolume( - client=client, - name=dv_name, - namespace=namespace, - source_dict=construct_datavolume_source_dict( - source="http", - url=get_http_image_url(image_directory=Images.Cirros.DIR, image_name=Images.Cirros.QCOW2_IMG), - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ), - storage_class=storage_class, - size=Images.Cirros.DEFAULT_DV_SIZE, - api_name="storage", - volume_mode=volume_mode, - ) - dv.to_dict() - dv_metadata = dv.res["metadata"] - with VirtualMachineForTests( - client=client, - name=vm_name, - namespace=dv_metadata["namespace"], - os_flavor=Images.Cirros.OS_FLAVOR, - memory_guest=Images.Cirros.DEFAULT_MEMORY_SIZE, - data_volume_template={"metadata": dv_metadata, "spec": dv.res["spec"]}, - node_selector=node, - run_strategy=VirtualMachine.RunStrategy.ALWAYS, - cpu_model=cpu_model, - annotations=annotations, - ) as vm: - if wait_running: - running_vm(vm=vm, wait_for_interfaces=False) - yield vm + with artifactory_credentials(namespace=namespace, client=client) as artifactory: + dv = DataVolume( + client=client, + name=dv_name, + namespace=namespace, + source_dict=construct_datavolume_source_dict( + source="http", + url=get_http_image_url(image_directory=Images.Cirros.DIR, image_name=Images.Cirros.QCOW2_IMG), + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ), + storage_class=storage_class, + size=Images.Cirros.DEFAULT_DV_SIZE, + api_name="storage", + volume_mode=volume_mode, + ) + dv.to_dict() + dv_metadata = dv.res["metadata"] + with VirtualMachineForTests( + client=client, + name=vm_name, + namespace=dv_metadata["namespace"], + os_flavor=Images.Cirros.OS_FLAVOR, + memory_guest=Images.Cirros.DEFAULT_MEMORY_SIZE, + data_volume_template={"metadata": dv_metadata, "spec": dv.res["spec"]}, + node_selector=node, + run_strategy=VirtualMachine.RunStrategy.ALWAYS, + cpu_model=cpu_model, + annotations=annotations, + ) as vm: + if wait_running: + running_vm(vm=vm, wait_for_interfaces=False) + yield vm def start_stress_on_vm(vm: VirtualMachineForTests, stress_command: str) -> None: diff --git a/tests/virt/cluster/longevity_tests/utils.py b/tests/virt/cluster/longevity_tests/utils.py index 7786b532c8..9817c8a8fb 100644 --- a/tests/virt/cluster/longevity_tests/utils.py +++ b/tests/virt/cluster/longevity_tests/utils.py @@ -19,11 +19,7 @@ WINDOWS_OS_PREFIX, ) from tests.virt.utils import migrate_and_verify_multi_vms -from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, -) +from utilities.artifactory import artifactory_credentials from utilities.constants.timeouts import ( TCP_TIMEOUT_30SEC, TIMEOUT_5MIN, @@ -356,30 +352,26 @@ def create_dv_vms( def create_multi_dvs(namespace, client, dv_params): namespace_name = namespace.name - artifactory_secret = get_artifactory_secret(namespace=namespace_name) - artifactory_config_map = get_artifactory_config_map(namespace=namespace_name) - dvs = {} - for dv in dv_params: - dv_name = [*dv][0] - dvs[dv_name] = DataVolume( - name=dv_name, - client=client, - namespace=namespace_name, - source_dict=construct_datavolume_source_dict( - source="http", - url=f"{get_test_artifact_server_url()}{dv[dv_name].get('image_path')}", - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, - ), - size=dv[dv_name].get("dv_size"), - storage_class=dv[dv_name].get("storage_class"), - api_name="storage", - ) + with artifactory_credentials(namespace=namespace_name, client=client) as artifactory: + dvs = {} + for dv in dv_params: + dv_name = [*dv][0] + dvs[dv_name] = DataVolume( + name=dv_name, + client=client, + namespace=namespace_name, + source_dict=construct_datavolume_source_dict( + source="http", + url=f"{get_test_artifact_server_url()}{dv[dv_name].get('image_path')}", + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, + ), + size=dv[dv_name].get("dv_size"), + storage_class=dv[dv_name].get("storage_class"), + api_name="storage", + ) - yield from deploy_and_wait_for_dvs(dv_dict=dvs) - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map - ) + yield from deploy_and_wait_for_dvs(dv_dict=dvs) def create_multi_datasources(client, dvs): diff --git a/utilities/artifactory.py b/utilities/artifactory.py index 240037a394..5ea772b58b 100644 --- a/utilities/artifactory.py +++ b/utilities/artifactory.py @@ -1,6 +1,9 @@ import logging import os import ssl +from collections.abc import Generator +from contextlib import contextmanager +from dataclasses import dataclass import requests from kubernetes.dynamic import DynamicClient @@ -18,9 +21,51 @@ LOGGER = logging.getLogger(__name__) ARTIFACTORY_SECRET_NAME = "cnv-tests-artifactory-secret" +ARTIFACTORY_CONFIG_MAP_NAME = "artifactory-configmap" BASE_ARTIFACTORY_LOCATION = "artifactory/cnv-qe-server-local" +@dataclass(frozen=True) +class ArtifactoryCredentials: + """Namespace-scoped Artifactory Secret and/or TLS ConfigMap. + + Attributes: + secret: Artifactory Secret when created; None if ``create_secret=False``. + config_map: Artifactory ConfigMap when created; None if ``create_config_map=False``. + """ + + secret: Secret | None = None + config_map: ConfigMap | None = None + + @property + def secret_name(self) -> str: + """Return the Artifactory Secret name. + + Returns: + str: Name of the created Secret. + + Raises: + ValueError: If the Secret was not created (``create_secret=False``). + """ + if self.secret is None: + raise ValueError("Artifactory secret was not created") + return self.secret.name + + @property + def cert_configmap_name(self) -> str: + """Return the Artifactory ConfigMap name. + + Returns: + str: Name of the created ConfigMap. + + Raises: + ValueError: If the ConfigMap was not created (``create_config_map=False``). + """ + if self.config_map is None: + raise ValueError("Artifactory ConfigMap was not created") + return self.config_map.name + + def get_test_artifact_server_url(schema: str = "https") -> str: # type: ignore[return] """ Verify https server server connectivity (regardless of schema). @@ -88,8 +133,8 @@ def get_artifactory_header() -> dict[str, str]: def get_artifactory_secret( namespace: str, - client: DynamicClient | None = None, -) -> Secret: + client: DynamicClient, +) -> tuple[Secret, bool]: """ Create or retrieve an Artifactory authentication secret in the specified namespace. @@ -99,14 +144,21 @@ def get_artifactory_secret( Args: namespace (str): The Kubernetes namespace where the secret should be created or retrieved. - client (DynamicClient | None): Optional Kubernetes client. If None, uses the default client. + client (DynamicClient): Kubernetes dynamic client used to create or look up the secret. Returns: - Secret: The Artifactory Secret resource object. + tuple[Secret, bool]: The Artifactory Secret and whether this call deployed it. Raises: KeyError: If ARTIFACTORY_USER or ARTIFACTORY_TOKEN environment variables are not set. """ + existing_secret = Secret( + name=ARTIFACTORY_SECRET_NAME, + namespace=namespace, + client=client, + ) + if existing_secret.exists: + return existing_secret, False artifactory_secret = Secret( name=ARTIFACTORY_SECRET_NAME, namespace=namespace, @@ -114,15 +166,14 @@ def get_artifactory_secret( secretkey=base64_encode_str(os.environ["ARTIFACTORY_TOKEN"]), client=client, ) - if not artifactory_secret.exists: - artifactory_secret.deploy() - return artifactory_secret + artifactory_secret.deploy() + return artifactory_secret, True def get_artifactory_config_map( namespace: str, - client: DynamicClient | None = None, -) -> ConfigMap: + client: DynamicClient, +) -> tuple[ConfigMap, bool]: """ Create or retrieve an Artifactory TLS certificate ConfigMap in the specified namespace. @@ -133,24 +184,30 @@ def get_artifactory_config_map( Args: namespace (str): The Kubernetes namespace where the ConfigMap should be created or retrieved. - client (DynamicClient | None): Optional Kubernetes client. If None, uses the default client. + client (DynamicClient): Kubernetes dynamic client used to create or look up the ConfigMap. Returns: - ConfigMap: The Artifactory ConfigMap resource object containing the TLS certificate. + tuple[ConfigMap, bool]: The Artifactory ConfigMap and whether this call deployed it. Raises: KeyError: If server_url is not found in py_config. OSError: If SSL connection to the server fails. """ + existing_cm = ConfigMap( + name=ARTIFACTORY_CONFIG_MAP_NAME, + namespace=namespace, + client=client, + ) + if existing_cm.exists: + return existing_cm, False artifactory_cm = ConfigMap( - name="artifactory-configmap", + name=ARTIFACTORY_CONFIG_MAP_NAME, namespace=namespace, data={"tlsregistry.crt": ssl.get_server_certificate(addr=(py_config["server_url"], 443))}, client=client, ) - if not artifactory_cm.exists: - artifactory_cm.deploy() - return artifactory_cm + artifactory_cm.deploy() + return artifactory_cm, True def cleanup_artifactory_secret_and_config_map( @@ -161,19 +218,61 @@ def cleanup_artifactory_secret_and_config_map( Clean up Artifactory Secret and ConfigMap resources from the cluster. Deletes the provided Artifactory Secret and/or ConfigMap resources if they exist. - This is typically used in test cleanup to remove temporary Artifactory credentials - and certificates from the cluster. + Prefer ``artifactory_credentials`` for new code; this remains for low-level cleanup. Args: - artifactory_secret (Secret | None): The Artifactory Secret resource to delete. + artifactory_secret (Secret | None): The Artifactory Secret resource to delete. If None, no secret cleanup is performed. artifactory_config_map (ConfigMap | None): The Artifactory ConfigMap resource to delete. If None, no ConfigMap cleanup is performed. - - Returns: - None """ if artifactory_secret: artifactory_secret.clean_up() if artifactory_config_map: artifactory_config_map.clean_up() + + +@contextmanager +def artifactory_credentials( + namespace: str, + client: DynamicClient, + *, + create_secret: bool = True, + create_config_map: bool = True, +) -> Generator[ArtifactoryCredentials]: + """ + Create Artifactory Secret and/or ConfigMap and clean up owned resources on exit. + + Only resources deployed by this context are deleted. Pre-existing Secret/ConfigMap + resources in the namespace are left in place for longer-lived owners. + + Args: + namespace: Kubernetes namespace for the credentials resources. + client: Kubernetes dynamic client used to create and delete the resources. + create_secret: Whether to create/retrieve the Artifactory Secret. Defaults to True. + create_config_map: Whether to create/retrieve the Artifactory ConfigMap. Defaults to True. + + Yields: + ArtifactoryCredentials: Created or retrieved Secret and/or ConfigMap for DV/source auth. + + Raises: + ValueError: If both ``create_secret`` and ``create_config_map`` are False. + """ + if not create_secret and not create_config_map: + raise ValueError("At least one of create_secret or create_config_map must be True") + + secret: Secret | None = None + config_map: ConfigMap | None = None + owned_secret = False + owned_config_map = False + try: + if create_secret: + secret, owned_secret = get_artifactory_secret(namespace=namespace, client=client) + if create_config_map: + config_map, owned_config_map = get_artifactory_config_map(namespace=namespace, client=client) + yield ArtifactoryCredentials(secret=secret, config_map=config_map) + finally: + cleanup_artifactory_secret_and_config_map( + artifactory_secret=secret if owned_secret else None, + artifactory_config_map=config_map if owned_config_map else None, + ) diff --git a/utilities/oadp.py b/utilities/oadp.py index bd3d15d654..3798d9635d 100644 --- a/utilities/oadp.py +++ b/utilities/oadp.py @@ -14,12 +14,7 @@ from ocp_resources.storage_profile import StorageProfile from ocp_resources.virtual_machine import VirtualMachine -from utilities.artifactory import ( - cleanup_artifactory_secret_and_config_map, - get_artifactory_config_map, - get_artifactory_secret, - get_http_image_url, -) +from utilities.artifactory import artifactory_credentials, get_http_image_url from utilities.console import Console from utilities.constants import Images from utilities.constants.cluster import LS_COMMAND @@ -150,13 +145,7 @@ def create_rhel_vm( wait_running: bool = True, volume_mode: str | None = None, ) -> Generator[VirtualMachineForTests]: - artifactory_secret = None - artifactory_config_map = None - - try: - artifactory_secret = get_artifactory_secret(namespace=namespace) - artifactory_config_map = get_artifactory_config_map(namespace=namespace) - + with artifactory_credentials(namespace=namespace, client=client) as artifactory: dv = DataVolume( name=dv_name, namespace=namespace, @@ -166,8 +155,8 @@ def create_rhel_vm( image_directory=Images.Rhel.DIR, image_name=rhel_image, ), - secret_name=artifactory_secret.name, - cert_configmap_name=artifactory_config_map.name, + secret_name=artifactory.secret_name, + cert_configmap_name=artifactory.cert_configmap_name, ), storage_class=storage_class, size=Images.Rhel.DEFAULT_DV_SIZE, @@ -188,10 +177,6 @@ def create_rhel_vm( if wait_running: running_vm(vm=vm) yield vm - finally: - cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map - ) class VeleroRestore(Restore): diff --git a/utilities/storage.py b/utilities/storage.py index 64f381ff03..252809f900 100644 --- a/utilities/storage.py +++ b/utilities/storage.py @@ -3,7 +3,7 @@ import os import shlex from collections.abc import Collection, Generator -from contextlib import contextmanager +from contextlib import ExitStack, contextmanager from typing import Any import cachetools.func @@ -35,7 +35,7 @@ import utilities.virt as virt_util from utilities import console from utilities.architecture import get_multiarch_cpu_arch -from utilities.artifactory import get_test_artifact_server_url +from utilities.artifactory import artifactory_credentials, get_test_artifact_server_url from utilities.constants import Images from utilities.constants.components import HPP_POOL from utilities.constants.images import OS_FLAVOR_WINDOWS @@ -193,8 +193,10 @@ def create_dv( Context manager that constructs a DataVolume from either a pre-built ``source_dict``/``source_ref`` or by building one via ``construct_datavolume_source_dict`` from the ``source`` parameter. - When ``use_artifactory`` is True for http/registry sources, creates namespace-scoped - Artifactory Secret and ConfigMap resources that are cleaned up on exit. + When ``use_artifactory`` is True for http/registry sources, creates only the + namespace-scoped Artifactory Secret and/or ConfigMap that the caller did not supply + via ``secret_name``/``cert_configmap_name``. Resources deployed by this call are + deleted on exit; pre-existing and caller-supplied resources are left in place. Args: dv_name: Name for the DataVolume resource. @@ -227,10 +229,7 @@ def create_dv( Raises: ValueError: If ``source`` is not provided when ``source_dict`` and ``source_ref`` are both None. """ - artifactory_secret = None - artifactory_config_map = None - - try: + with ExitStack() as stack: if source_dict is None and source_ref is None: if not source: raise ValueError("'source' is required when 'source_dict' and 'source_ref' are not provided") @@ -241,16 +240,21 @@ def create_dv( LOGGER.info(f"Creating artifactory resources for DV '{dv_name}' in namespace '{namespace}'") LOGGER.info(f"DV source is '{source}' with url: {url}") - if not secret_name: - artifactory_secret = utilities.artifactory.get_artifactory_secret( - namespace=namespace, client=client - ) - secret_name = artifactory_secret.name - if not cert_configmap_name: - artifactory_config_map = utilities.artifactory.get_artifactory_config_map( - namespace=namespace, client=client + if not secret_name or not cert_configmap_name: + create_secret = not secret_name + create_config_map = not cert_configmap_name + artifactory = stack.enter_context( + artifactory_credentials( + namespace=namespace, + client=client, + create_secret=create_secret, + create_config_map=create_config_map, + ) ) - cert_configmap_name = artifactory_config_map.name + if create_secret: + secret_name = artifactory.secret_name + if create_config_map: + cert_configmap_name = artifactory.cert_configmap_name source_dict = construct_datavolume_source_dict( source=source, @@ -261,30 +265,27 @@ def create_dv( source_pvc_namespace=source_pvc_namespace, ) - with DataVolume( - name=dv_name, - namespace=namespace, - client=client, - content_type=content_type, - size=size, - storage_class=storage_class, - access_modes=access_modes, - volume_mode=volume_mode, - annotations=annotations, - teardown=teardown, - preallocation=preallocation, - api_name=api_name, - source_ref=source_ref, - source_dict=source_dict, - ) as dv: - if storage_class and sc_volume_binding_mode_is_wffc(sc=storage_class, client=client) and consume_wffc: - create_dummy_first_consumer_pod(dv=dv) - yield dv - - finally: - utilities.artifactory.cleanup_artifactory_secret_and_config_map( - artifactory_secret=artifactory_secret, artifactory_config_map=artifactory_config_map + dv = stack.enter_context( + DataVolume( + name=dv_name, + namespace=namespace, + client=client, + content_type=content_type, + size=size, + storage_class=storage_class, + access_modes=access_modes, + volume_mode=volume_mode, + annotations=annotations, + teardown=teardown, + preallocation=preallocation, + api_name=api_name, + source_ref=source_ref, + source_dict=source_dict, + ) ) + if storage_class and sc_volume_binding_mode_is_wffc(sc=storage_class, client=client) and consume_wffc: + create_dummy_first_consumer_pod(dv=dv) + yield dv def data_volume( diff --git a/utilities/unittests/test_artifactory.py b/utilities/unittests/test_artifactory.py index 548cfe0719..7400a1616a 100644 --- a/utilities/unittests/test_artifactory.py +++ b/utilities/unittests/test_artifactory.py @@ -15,8 +15,11 @@ sys.path.insert(0, str(Path(__file__).parent.parent)) from utilities.artifactory import ( + ARTIFACTORY_CONFIG_MAP_NAME, ARTIFACTORY_SECRET_NAME, BASE_ARTIFACTORY_LOCATION, + ArtifactoryCredentials, + artifactory_credentials, cleanup_artifactory_secret_and_config_map, get_artifactory_config_map, get_artifactory_header, @@ -283,16 +286,23 @@ def test_get_artifactory_secret_creates_secret_with_correct_parameters(self, moc mock_secret_instance = MagicMock() mock_secret_instance.exists = False mock_secret_class.return_value = mock_secret_instance + mock_client = MagicMock() - result = get_artifactory_secret(namespace="test-namespace") + secret, created = get_artifactory_secret(namespace="test-namespace", client=mock_client) - # Verify Secret was created with correct parameters - mock_secret_class.assert_called_once_with( + # Verify Secret was checked and created with correct parameters + assert mock_secret_class.call_count == 2 + mock_secret_class.assert_any_call( + name=ARTIFACTORY_SECRET_NAME, + namespace="test-namespace", + client=mock_client, + ) + mock_secret_class.assert_any_call( name=ARTIFACTORY_SECRET_NAME, namespace="test-namespace", accesskeyid="base64_test-user", secretkey="base64_test-token", - client=None, + client=mock_client, ) # Verify base64 encoding was called @@ -300,7 +310,8 @@ def test_get_artifactory_secret_creates_secret_with_correct_parameters(self, moc mock_base64_encode.assert_any_call("test-user") mock_base64_encode.assert_any_call("test-token") - assert result == mock_secret_instance + assert secret == mock_secret_instance + assert created is True @patch("utilities.artifactory.Secret") @patch("utilities.artifactory.base64_encode_str") @@ -314,11 +325,12 @@ def test_get_artifactory_secret_deploys_if_not_exists(self, mock_base64_encode, mock_secret_instance.exists = False mock_secret_class.return_value = mock_secret_instance - result = get_artifactory_secret(namespace="test-namespace") + secret, created = get_artifactory_secret(namespace="test-namespace", client=MagicMock()) # Verify deploy was called mock_secret_instance.deploy.assert_called_once() - assert result == mock_secret_instance + assert secret == mock_secret_instance + assert created is True @patch("utilities.artifactory.Secret") @patch("utilities.artifactory.base64_encode_str") @@ -332,25 +344,34 @@ def test_get_artifactory_secret_returns_existing_if_exists(self, mock_base64_enc mock_secret_instance.exists = True mock_secret_class.return_value = mock_secret_instance - result = get_artifactory_secret(namespace="test-namespace") + secret, created = get_artifactory_secret(namespace="test-namespace", client=MagicMock()) # Verify deploy was NOT called mock_secret_instance.deploy.assert_not_called() - assert result == mock_secret_instance + assert secret == mock_secret_instance + assert created is False @patch("utilities.artifactory.Secret") def test_get_artifactory_secret_raises_key_error_if_user_not_set(self, mock_secret_class): """Test raises KeyError if ARTIFACTORY_USER not set""" + mock_secret_instance = MagicMock() + mock_secret_instance.exists = False + mock_secret_class.return_value = mock_secret_instance + with patch.dict(os.environ, {"ARTIFACTORY_TOKEN": "test-token"}, clear=True): with pytest.raises(KeyError): - get_artifactory_secret(namespace="test-namespace") + get_artifactory_secret(namespace="test-namespace", client=MagicMock()) @patch("utilities.artifactory.Secret") def test_get_artifactory_secret_raises_key_error_if_token_not_set(self, mock_secret_class): """Test raises KeyError if ARTIFACTORY_TOKEN not set""" + mock_secret_instance = MagicMock() + mock_secret_instance.exists = False + mock_secret_class.return_value = mock_secret_instance + with patch.dict(os.environ, {"ARTIFACTORY_USER": "test-user"}, clear=True): with pytest.raises(KeyError): - get_artifactory_secret(namespace="test-namespace") + get_artifactory_secret(namespace="test-namespace", client=MagicMock()) @patch("utilities.artifactory.Secret") @patch("utilities.artifactory.base64_encode_str") @@ -363,7 +384,7 @@ def test_get_artifactory_secret_uses_correct_secret_name(self, mock_base64_encod mock_secret_instance.exists = False mock_secret_class.return_value = mock_secret_instance - get_artifactory_secret(namespace="test-namespace") + get_artifactory_secret(namespace="test-namespace", client=MagicMock()) # Verify the name parameter matches the constant call_kwargs = mock_secret_class.call_args[1] @@ -387,21 +408,29 @@ def test_get_artifactory_config_map_creates_with_correct_parameters(self, mock_g mock_cm_instance = MagicMock() mock_cm_instance.exists = False mock_cm_class.return_value = mock_cm_instance + mock_client = MagicMock() - result = get_artifactory_config_map(namespace="test-namespace") + config_map, created = get_artifactory_config_map(namespace="test-namespace", client=mock_client) - # Verify ConfigMap was created with correct parameters - mock_cm_class.assert_called_once_with( + # Verify ConfigMap was checked and created with correct parameters + assert mock_cm_class.call_count == 2 + mock_cm_class.assert_any_call( + name="artifactory-configmap", + namespace="test-namespace", + client=mock_client, + ) + mock_cm_class.assert_any_call( name="artifactory-configmap", namespace="test-namespace", data={"tlsregistry.crt": mock_cert}, - client=None, + client=mock_client, ) # Verify SSL certificate was retrieved mock_get_cert.assert_called_once_with(addr=("test.artifactory.com", 443)) - assert result == mock_cm_instance + assert config_map == mock_cm_instance + assert created is True @patch("utilities.artifactory.ConfigMap") @patch("utilities.artifactory.ssl.get_server_certificate") @@ -416,11 +445,12 @@ def test_get_artifactory_config_map_deploys_if_not_exists(self, mock_get_cert, m mock_cm_instance.exists = False mock_cm_class.return_value = mock_cm_instance - result = get_artifactory_config_map(namespace="test-namespace") + config_map, created = get_artifactory_config_map(namespace="test-namespace", client=MagicMock()) # Verify deploy was called mock_cm_instance.deploy.assert_called_once() - assert result == mock_cm_instance + assert config_map == mock_cm_instance + assert created is True @patch("utilities.artifactory.ConfigMap") @patch("utilities.artifactory.ssl.get_server_certificate") @@ -435,30 +465,38 @@ def test_get_artifactory_config_map_returns_existing_if_exists(self, mock_get_ce mock_cm_instance.exists = True mock_cm_class.return_value = mock_cm_instance - result = get_artifactory_config_map(namespace="test-namespace") + config_map, created = get_artifactory_config_map(namespace="test-namespace", client=MagicMock()) # Verify deploy was NOT called mock_cm_instance.deploy.assert_not_called() - assert result == mock_cm_instance + assert config_map == mock_cm_instance + assert created is False @patch("utilities.artifactory.ConfigMap") - @patch("utilities.artifactory.ssl.get_server_certificate") @patch("utilities.artifactory.py_config", {}) - def test_get_artifactory_config_map_raises_key_error_if_server_url_missing(self, mock_get_cert, mock_cm_class): + def test_get_artifactory_config_map_raises_key_error_if_server_url_missing(self, mock_cm_class): """Test raises KeyError if server_url not in py_config""" + mock_cm_instance = MagicMock() + mock_cm_instance.exists = False + mock_cm_class.return_value = mock_cm_instance + with pytest.raises(KeyError): - get_artifactory_config_map(namespace="test-namespace") + get_artifactory_config_map(namespace="test-namespace", client=MagicMock()) @patch("utilities.artifactory.ConfigMap") @patch("utilities.artifactory.ssl.get_server_certificate") @patch("utilities.artifactory.py_config", {"server_url": "test.artifactory.com"}) def test_get_artifactory_config_map_ssl_connection_failure(self, mock_get_cert, mock_cm_class): """Test OSError is raised on SSL connection failure""" + mock_cm_instance = MagicMock() + mock_cm_instance.exists = False + mock_cm_class.return_value = mock_cm_instance + # Mock SSL connection failure mock_get_cert.side_effect = OSError("Connection refused") with pytest.raises(OSError): - get_artifactory_config_map(namespace="test-namespace") + get_artifactory_config_map(namespace="test-namespace", client=MagicMock()) mock_get_cert.assert_called_once_with(addr=("test.artifactory.com", 443)) @@ -474,7 +512,7 @@ def test_get_artifactory_config_map_uses_custom_server_url(self, mock_get_cert, mock_cm_instance.exists = False mock_cm_class.return_value = mock_cm_instance - get_artifactory_config_map(namespace="test-namespace") + get_artifactory_config_map(namespace="test-namespace", client=MagicMock()) # Verify SSL certificate was retrieved from custom server mock_get_cert.assert_called_once_with(addr=("custom.server.com", 443)) @@ -550,6 +588,267 @@ def test_cleanup_artifactory_secret_and_config_map_returns_none(self): ) +class TestArtifactoryCredentials: + """Test cases for ArtifactoryCredentials and artifactory_credentials""" + + def test_artifactory_credentials_name_properties(self): + """Test secret_name and cert_configmap_name properties""" + mock_secret = MagicMock() + mock_secret.name = "cnv-tests-artifactory-secret" + mock_config_map = MagicMock() + mock_config_map.name = "artifactory-configmap" + + credentials = ArtifactoryCredentials(secret=mock_secret, config_map=mock_config_map) + + assert credentials.secret_name == "cnv-tests-artifactory-secret" + assert credentials.cert_configmap_name == "artifactory-configmap" + + def test_artifactory_credentials_secret_name_raises_when_missing(self): + """Test secret_name raises when secret was not created""" + credentials = ArtifactoryCredentials(config_map=MagicMock()) + + with pytest.raises(ValueError, match="Artifactory secret was not created"): + _ = credentials.secret_name + + def test_artifactory_credentials_cert_configmap_name_raises_when_missing(self): + """Test cert_configmap_name raises when ConfigMap was not created""" + credentials = ArtifactoryCredentials(secret=MagicMock()) + + with pytest.raises(ValueError, match="Artifactory ConfigMap was not created"): + _ = credentials.cert_configmap_name + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_context_manager_yields_pair( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test context manager yields credentials and cleans up owned resources on exit""" + mock_secret = MagicMock() + mock_secret.name = "cnv-tests-artifactory-secret" + mock_config_map = MagicMock() + mock_config_map.name = "artifactory-configmap" + mock_get_secret.return_value = (mock_secret, True) + mock_get_config_map.return_value = (mock_config_map, True) + mock_client = MagicMock() + + with artifactory_credentials(namespace="test-namespace", client=mock_client) as credentials: + assert credentials.secret is mock_secret + assert credentials.config_map is mock_config_map + assert credentials.secret_name == mock_secret.name + assert credentials.cert_configmap_name == mock_config_map.name + + mock_get_secret.assert_called_once_with(namespace="test-namespace", client=mock_client) + mock_get_config_map.assert_called_once_with(namespace="test-namespace", client=mock_client) + mock_cleanup.assert_called_once_with( + artifactory_secret=mock_secret, + artifactory_config_map=mock_config_map, + ) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_create_secret_only( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test context manager can create only the Secret""" + mock_secret = MagicMock() + mock_secret.name = "cnv-tests-artifactory-secret" + mock_get_secret.return_value = (mock_secret, True) + mock_client = MagicMock() + + with artifactory_credentials( + namespace="test-namespace", + client=mock_client, + create_secret=True, + create_config_map=False, + ) as credentials: + assert credentials.secret is mock_secret + assert credentials.config_map is None + assert credentials.secret_name == mock_secret.name + + mock_get_secret.assert_called_once_with(namespace="test-namespace", client=mock_client) + mock_get_config_map.assert_not_called() + mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=None) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_create_config_map_only( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test context manager can create only the ConfigMap""" + mock_config_map = MagicMock() + mock_config_map.name = "artifactory-configmap" + mock_get_config_map.return_value = (mock_config_map, True) + mock_client = MagicMock() + + with artifactory_credentials( + namespace="test-namespace", + client=mock_client, + create_secret=False, + create_config_map=True, + ) as credentials: + assert credentials.secret is None + assert credentials.config_map is mock_config_map + assert credentials.cert_configmap_name == mock_config_map.name + + mock_get_secret.assert_not_called() + mock_get_config_map.assert_called_once_with(namespace="test-namespace", client=mock_client) + mock_cleanup.assert_called_once_with(artifactory_secret=None, artifactory_config_map=mock_config_map) + + def test_artifactory_credentials_raises_when_both_disabled(self): + """Test context manager rejects creating neither resource""" + with pytest.raises(ValueError, match="At least one of create_secret or create_config_map must be True"): + with artifactory_credentials( + namespace="test-namespace", + client=MagicMock(), + create_secret=False, + create_config_map=False, + ): + pass + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_cleans_up_secret_if_config_map_create_fails( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test owned Secret is cleaned up when ConfigMap creation fails after Secret creation""" + mock_secret = MagicMock() + mock_get_secret.return_value = (mock_secret, True) + mock_get_config_map.side_effect = OSError("SSL connection failed") + + with pytest.raises(OSError, match="SSL connection failed"): + with artifactory_credentials(namespace="test-namespace", client=MagicMock()): + pass + + mock_cleanup.assert_called_once_with( + artifactory_secret=mock_secret, + artifactory_config_map=None, + ) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_skips_preexisting_secret_if_config_map_create_fails( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test pre-existing Secret is not deleted when ConfigMap creation fails""" + mock_secret = MagicMock() + mock_get_secret.return_value = (mock_secret, False) + mock_get_config_map.side_effect = OSError("SSL connection failed") + + with pytest.raises(OSError, match="SSL connection failed"): + with artifactory_credentials(namespace="test-namespace", client=MagicMock()): + pass + + mock_cleanup.assert_called_once_with( + artifactory_secret=None, + artifactory_config_map=None, + ) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_context_manager_cleanup_on_exception( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test context manager cleans up owned resources when body raises""" + mock_secret = MagicMock() + mock_config_map = MagicMock() + mock_get_secret.return_value = (mock_secret, True) + mock_get_config_map.return_value = (mock_config_map, True) + + with pytest.raises(RuntimeError, match="boom"): + with artifactory_credentials(namespace="test-namespace", client=MagicMock()): + raise RuntimeError("boom") + + mock_cleanup.assert_called_once_with( + artifactory_secret=mock_secret, + artifactory_config_map=mock_config_map, + ) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_skips_cleanup_for_preexisting_resources( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test pre-existing Secret/ConfigMap are not deleted on exit""" + mock_secret = MagicMock() + mock_config_map = MagicMock() + mock_get_secret.return_value = (mock_secret, False) + mock_get_config_map.return_value = (mock_config_map, False) + + with artifactory_credentials(namespace="test-namespace", client=MagicMock()) as credentials: + assert credentials.secret is mock_secret + assert credentials.config_map is mock_config_map + + mock_cleanup.assert_called_once_with(artifactory_secret=None, artifactory_config_map=None) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_cleans_up_only_owned_secret( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test only the newly deployed Secret is cleaned up when ConfigMap already existed""" + mock_secret = MagicMock() + mock_config_map = MagicMock() + mock_get_secret.return_value = (mock_secret, True) + mock_get_config_map.return_value = (mock_config_map, False) + + with artifactory_credentials(namespace="test-namespace", client=MagicMock()): + pass + + mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=None) + + @patch("utilities.artifactory.cleanup_artifactory_secret_and_config_map") + @patch("utilities.artifactory.get_artifactory_config_map") + @patch("utilities.artifactory.get_artifactory_secret") + def test_artifactory_credentials_cleans_up_only_owned_config_map( + self, + mock_get_secret, + mock_get_config_map, + mock_cleanup, + ): + """Test only the newly deployed ConfigMap is cleaned up when Secret already existed""" + mock_secret = MagicMock() + mock_config_map = MagicMock() + mock_get_secret.return_value = (mock_secret, False) + mock_get_config_map.return_value = (mock_config_map, True) + + with artifactory_credentials(namespace="test-namespace", client=MagicMock()): + pass + + mock_cleanup.assert_called_once_with(artifactory_secret=None, artifactory_config_map=mock_config_map) + + class TestArtifactoryConstants: """Test cases for artifactory module constants""" @@ -557,6 +856,10 @@ def test_artifactory_secret_name_constant(self): """Test ARTIFACTORY_SECRET_NAME constant is defined""" assert ARTIFACTORY_SECRET_NAME == "cnv-tests-artifactory-secret" + def test_artifactory_config_map_name_constant(self): + """Test ARTIFACTORY_CONFIG_MAP_NAME constant is defined""" + assert ARTIFACTORY_CONFIG_MAP_NAME == "artifactory-configmap" + def test_base_artifactory_location_constant(self): """Test BASE_ARTIFACTORY_LOCATION constant is defined""" assert BASE_ARTIFACTORY_LOCATION == "artifactory/cnv-qe-server-local" diff --git a/utilities/unittests/test_oadp.py b/utilities/unittests/test_oadp.py index 7f92ac54b5..5b82e9e5d3 100644 --- a/utilities/unittests/test_oadp.py +++ b/utilities/unittests/test_oadp.py @@ -280,43 +280,46 @@ def test_velero_backup_exit_calls_teardown_and_super(self, mock_backup_init, moc class TestCreateRhelVm: """Test cases for create_rhel_vm context manager""" - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") + def _mock_artifactory_credentials(self, mock_artifactory_credentials): + mock_credentials = MagicMock() + mock_credentials.secret_name = "cnv-tests-artifactory-secret" + mock_credentials.cert_configmap_name = "artifactory-configmap" + mock_artifactory_credentials.return_value.__enter__.return_value = mock_credentials + mock_artifactory_credentials.return_value.__exit__.return_value = None + return mock_credentials + + def _mock_dv_and_vm(self, mock_dv_class, mock_vm_class): + mock_dv = MagicMock() + mock_dv.res = { + "metadata": {"name": "test-dv", "namespace": "test-namespace"}, + "spec": {"source": "http"}, + } + mock_dv_class.return_value = mock_dv + + mock_vm = MagicMock() + mock_vm.__enter__ = MagicMock(return_value=mock_vm) + mock_vm.__exit__ = MagicMock(return_value=None) + mock_vm_class.return_value = mock_vm + return mock_dv, mock_vm + @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_success_with_wait( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm creates VM and waits for running""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" - - mock_dv = MagicMock() - mock_dv.res = { - "metadata": {"name": "test-dv", "namespace": "test-namespace"}, - "spec": {"source": "http"}, - } - mock_dv_class.return_value = mock_dv - - mock_vm = MagicMock() - mock_vm.__enter__ = MagicMock(return_value=mock_vm) - mock_vm.__exit__ = MagicMock(return_value=None) - mock_vm_class.return_value = mock_vm + mock_dv, mock_vm = self._mock_dv_and_vm(mock_dv_class, mock_vm_class) with create_rhel_vm( storage_class="ocs-storagecluster-ceph-rbd", @@ -329,50 +332,30 @@ def test_create_rhel_vm_success_with_wait( ) as vm: assert vm == mock_vm - mock_get_secret.assert_called_once_with(namespace="test-namespace") - mock_get_config_map.assert_called_once_with(namespace="test-namespace") + mock_artifactory_credentials.assert_called_once_with(namespace="test-namespace", client=mock_client) mock_get_url.assert_called_once() mock_dv.to_dict.assert_called_once() mock_running_vm.assert_called_once_with(vm=mock_vm) - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_success_without_wait( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm creates VM without waiting for running""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" - - mock_dv = MagicMock() - mock_dv.res = { - "metadata": {"name": "test-dv", "namespace": "test-namespace"}, - "spec": {"source": "http"}, - } - mock_dv_class.return_value = mock_dv - - mock_vm = MagicMock() - mock_vm.__enter__ = MagicMock(return_value=mock_vm) - mock_vm.__exit__ = MagicMock(return_value=None) - mock_vm_class.return_value = mock_vm + _mock_dv, mock_vm = self._mock_dv_and_vm(mock_dv_class, mock_vm_class) with create_rhel_vm( storage_class="ocs-storagecluster-ceph-rbd", @@ -386,45 +369,26 @@ def test_create_rhel_vm_success_without_wait( assert vm == mock_vm mock_running_vm.assert_not_called() - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_with_volume_mode( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm with volume_mode parameter""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" - - mock_dv = MagicMock() - mock_dv.res = { - "metadata": {"name": "test-dv", "namespace": "test-namespace"}, - "spec": {"source": "http"}, - } - mock_dv_class.return_value = mock_dv - - mock_vm = MagicMock() - mock_vm.__enter__ = MagicMock(return_value=mock_vm) - mock_vm.__exit__ = MagicMock(return_value=None) - mock_vm_class.return_value = mock_vm + _mock_dv, mock_vm = self._mock_dv_and_vm(mock_dv_class, mock_vm_class) with create_rhel_vm( storage_class="ocs-storagecluster-ceph-rbd", @@ -438,34 +402,25 @@ def test_create_rhel_vm_with_volume_mode( ) as vm: assert vm == mock_vm - # Verify DataVolume was created with volume_mode assert mock_dv_class.call_args.kwargs["volume_mode"] == "Block" - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_cleanup_on_exception( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm cleanup happens on exception""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" mock_dv = MagicMock() @@ -474,8 +429,6 @@ def test_create_rhel_vm_cleanup_on_exception( "spec": {"source": "http"}, } mock_dv_class.return_value = mock_dv - - # Make VirtualMachineForTests raise exception on enter mock_vm_class.return_value.__enter__.side_effect = Exception("VM creation failed") with pytest.raises(Exception, match="VM creation failed"): @@ -490,46 +443,26 @@ def test_create_rhel_vm_cleanup_on_exception( ): pass - # Cleanup should still be called - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_cleanup_on_success( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm cleanup happens on successful completion""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" - - mock_dv = MagicMock() - mock_dv.res = { - "metadata": {"name": "test-dv", "namespace": "test-namespace"}, - "spec": {"source": "http"}, - } - mock_dv_class.return_value = mock_dv - - mock_vm = MagicMock() - mock_vm.__enter__ = MagicMock(return_value=mock_vm) - mock_vm.__exit__ = MagicMock(return_value=None) - mock_vm_class.return_value = mock_vm + self._mock_dv_and_vm(mock_dv_class, mock_vm_class) with create_rhel_vm( storage_class="ocs-storagecluster-ceph-rbd", @@ -542,47 +475,26 @@ def test_create_rhel_vm_cleanup_on_success( ): pass - # Cleanup should be called - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() - @patch("utilities.oadp.cleanup_artifactory_secret_and_config_map") @patch("utilities.oadp.running_vm") @patch("utilities.oadp.VirtualMachineForTests") @patch("utilities.oadp.DataVolume") @patch("utilities.oadp.get_http_image_url") - @patch("utilities.oadp.get_artifactory_config_map") - @patch("utilities.oadp.get_artifactory_secret") + @patch("utilities.oadp.artifactory_credentials") def test_create_rhel_vm_running_vm_exception( self, - mock_get_secret, - mock_get_config_map, + mock_artifactory_credentials, mock_get_url, mock_dv_class, mock_vm_class, mock_running_vm, - mock_cleanup, ): """Test create_rhel_vm handles running_vm exception and still cleans up""" mock_client = MagicMock() - mock_secret = MagicMock() - mock_config_map = MagicMock() - mock_config_map.name = "artifactory-cert" - mock_get_secret.return_value = mock_secret - mock_get_config_map.return_value = mock_config_map + self._mock_artifactory_credentials(mock_artifactory_credentials) mock_get_url.return_value = "http://example.com/rhel-9.6.qcow2" - - mock_dv = MagicMock() - mock_dv.res = { - "metadata": {"name": "test-dv", "namespace": "test-namespace"}, - "spec": {"source": "http"}, - } - mock_dv_class.return_value = mock_dv - - mock_vm = MagicMock() - mock_vm.__enter__ = MagicMock(return_value=mock_vm) - mock_vm.__exit__ = MagicMock(return_value=None) - mock_vm_class.return_value = mock_vm - + self._mock_dv_and_vm(mock_dv_class, mock_vm_class) mock_running_vm.side_effect = Exception("VM failed to start") with pytest.raises(Exception, match="VM failed to start"): @@ -597,8 +509,7 @@ def test_create_rhel_vm_running_vm_exception( ): pass - # Cleanup should still be called - mock_cleanup.assert_called_once_with(artifactory_secret=mock_secret, artifactory_config_map=mock_config_map) + mock_artifactory_credentials.return_value.__exit__.assert_called_once() class TestVeleroRestore: diff --git a/utilities/unittests/test_storage.py b/utilities/unittests/test_storage.py index 7967454b70..983a8fc975 100644 --- a/utilities/unittests/test_storage.py +++ b/utilities/unittests/test_storage.py @@ -1,8 +1,9 @@ -"""Unit tests for construct_datavolume_source_dict in utilities/storage.py""" +"""Unit tests for construct_datavolume_source_dict and create_dv in utilities/storage.py""" import importlib import sys -from unittest.mock import patch +from contextlib import contextmanager +from unittest.mock import MagicMock, patch import pytest @@ -15,7 +16,7 @@ importlib.reload(utilities.storage) -from utilities.storage import construct_datavolume_source_dict +from utilities.storage import construct_datavolume_source_dict, create_dv class TestConstructDatavolumeSourceDictHttp: @@ -163,3 +164,213 @@ class TestConstructDatavolumeSourceDictUnsupported: def test_unsupported_source_raises_value_error(self): with pytest.raises(ValueError, match="Unsupported source type: ftp"): construct_datavolume_source_dict(source="ftp") + + +class TestCreateDvArtifactory: + """Unit tests for create_dv Artifactory credential wiring via ExitStack.""" + + @staticmethod + def _credentials(*, secret_name: str = "artifactory-secret", config_map_name: str = "artifactory-configmap"): + credentials = MagicMock() + credentials.secret_name = secret_name + credentials.cert_configmap_name = config_map_name + return credentials + + @staticmethod + @contextmanager + def _artifactory_cm(credentials): + yield credentials + + def _mock_data_volume(self, mock_data_volume_class): + mock_dv = MagicMock() + mock_data_volume_class.return_value = mock_dv + mock_data_volume_class.return_value.__enter__ = MagicMock(return_value=mock_dv) + mock_data_volume_class.return_value.__exit__ = MagicMock(return_value=None) + return mock_dv + + @patch("utilities.storage.sc_volume_binding_mode_is_wffc", return_value=False) + @patch("utilities.storage.DataVolume") + @patch("utilities.storage.validate_file_exists_in_url") + @patch("utilities.infra.url_excluded_from_validation", return_value=True) + @patch("utilities.storage.artifactory_credentials") + def test_create_dv_creates_both_artifactory_resources( + self, + mock_artifactory_credentials, + _mock_excluded, + _mock_validate, + mock_data_volume_class, + _mock_wffc, + ): + credentials = self._credentials() + mock_client = MagicMock() + mock_artifactory_credentials.side_effect = lambda **_kwargs: self._artifactory_cm(credentials) + mock_dv = self._mock_data_volume(mock_data_volume_class) + + with create_dv( + dv_name="test-dv", + namespace="test-ns", + client=mock_client, + source="http", + url="https://example.com/image.qcow2", + use_artifactory=True, + ) as dv: + assert dv is mock_dv + + mock_artifactory_credentials.assert_called_once_with( + namespace="test-ns", + client=mock_client, + create_secret=True, + create_config_map=True, + ) + source_dict = mock_data_volume_class.call_args.kwargs["source_dict"] + assert source_dict["http"]["secretRef"] == "artifactory-secret" + assert source_dict["http"]["certConfigMap"] == "artifactory-configmap" + + @patch("utilities.storage.sc_volume_binding_mode_is_wffc", return_value=False) + @patch("utilities.storage.DataVolume") + @patch("utilities.storage.validate_file_exists_in_url") + @patch("utilities.infra.url_excluded_from_validation", return_value=True) + @patch("utilities.storage.artifactory_credentials") + def test_create_dv_creates_only_missing_secret( + self, + mock_artifactory_credentials, + _mock_excluded, + _mock_validate, + mock_data_volume_class, + _mock_wffc, + ): + credentials = self._credentials(secret_name="created-secret") + mock_client = MagicMock() + mock_artifactory_credentials.side_effect = lambda **_kwargs: self._artifactory_cm(credentials) + self._mock_data_volume(mock_data_volume_class) + + with create_dv( + dv_name="test-dv", + namespace="test-ns", + client=mock_client, + source="http", + url="https://example.com/image.qcow2", + use_artifactory=True, + cert_configmap_name="existing-cm", + ): + pass + + mock_artifactory_credentials.assert_called_once_with( + namespace="test-ns", + client=mock_client, + create_secret=True, + create_config_map=False, + ) + source_dict = mock_data_volume_class.call_args.kwargs["source_dict"] + assert source_dict["http"]["secretRef"] == "created-secret" + assert source_dict["http"]["certConfigMap"] == "existing-cm" + + @patch("utilities.storage.sc_volume_binding_mode_is_wffc", return_value=False) + @patch("utilities.storage.DataVolume") + @patch("utilities.storage.validate_file_exists_in_url") + @patch("utilities.infra.url_excluded_from_validation", return_value=True) + @patch("utilities.storage.artifactory_credentials") + def test_create_dv_creates_only_missing_config_map( + self, + mock_artifactory_credentials, + _mock_excluded, + _mock_validate, + mock_data_volume_class, + _mock_wffc, + ): + credentials = self._credentials(config_map_name="created-cm") + mock_client = MagicMock() + mock_artifactory_credentials.side_effect = lambda **_kwargs: self._artifactory_cm(credentials) + self._mock_data_volume(mock_data_volume_class) + + with create_dv( + dv_name="test-dv", + namespace="test-ns", + client=mock_client, + source="http", + url="https://example.com/image.qcow2", + use_artifactory=True, + secret_name="existing-secret", + ): + pass + + mock_artifactory_credentials.assert_called_once_with( + namespace="test-ns", + client=mock_client, + create_secret=False, + create_config_map=True, + ) + source_dict = mock_data_volume_class.call_args.kwargs["source_dict"] + assert source_dict["http"]["secretRef"] == "existing-secret" + assert source_dict["http"]["certConfigMap"] == "created-cm" + + @patch("utilities.storage.sc_volume_binding_mode_is_wffc", return_value=False) + @patch("utilities.storage.DataVolume") + @patch("utilities.storage.validate_file_exists_in_url") + @patch("utilities.infra.url_excluded_from_validation", return_value=True) + @patch("utilities.storage.artifactory_credentials") + def test_create_dv_skips_artifactory_when_both_names_provided( + self, + mock_artifactory_credentials, + _mock_excluded, + _mock_validate, + mock_data_volume_class, + _mock_wffc, + ): + self._mock_data_volume(mock_data_volume_class) + + with create_dv( + dv_name="test-dv", + namespace="test-ns", + client=MagicMock(), + source="http", + url="https://example.com/image.qcow2", + use_artifactory=True, + secret_name="existing-secret", + cert_configmap_name="existing-cm", + ): + pass + + mock_artifactory_credentials.assert_not_called() + source_dict = mock_data_volume_class.call_args.kwargs["source_dict"] + assert source_dict["http"]["secretRef"] == "existing-secret" + assert source_dict["http"]["certConfigMap"] == "existing-cm" + + @patch("utilities.storage.sc_volume_binding_mode_is_wffc", return_value=False) + @patch("utilities.storage.DataVolume") + @patch("utilities.storage.validate_file_exists_in_url") + @patch("utilities.infra.url_excluded_from_validation", return_value=True) + @patch("utilities.storage.artifactory_credentials") + def test_create_dv_unwinds_artifactory_on_data_volume_failure( + self, + mock_artifactory_credentials, + _mock_excluded, + _mock_validate, + mock_data_volume_class, + _mock_wffc, + ): + credentials = self._credentials() + exit_mock = MagicMock(return_value=None) + + @contextmanager + def artifactory_cm(**_kwargs): + try: + yield credentials + finally: + exit_mock() + + mock_artifactory_credentials.side_effect = artifactory_cm + mock_data_volume_class.side_effect = RuntimeError("DV create failed") + + with pytest.raises(RuntimeError, match="DV create failed"): + with create_dv( + dv_name="test-dv", + namespace="test-ns", + client=MagicMock(), + source="http", + url="https://example.com/image.qcow2", + use_artifactory=True, + ): + pass + + exit_mock.assert_called_once()