refactor: manage Artifactory credentials via context manager

.flake8

@@ -103,6 +103,8 @@ fcn_exclude_functions =
     mkdtemp,
     tempfile,
     urllib3,
+    enter_context,
+    callback,
 
 nit_exclude_imports =
     os_params,

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")

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",

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")

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()

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")

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")

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()