From 8b383a2ea3bbcb110141b9e53fd320a29c302262 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Ram=C3=B3n=20Lobillo?= Date: Wed, 4 Mar 2026 15:16:38 +0100 Subject: [PATCH] [IUO] Migrate install, upgrade and EUS upgrade paths to Konflux IDMS CNV operator builds migrated from Brew to Konflux, so install, upgrade and EUS upgrade test paths now require Konflux IDMS instead of the legacy ICSP/IDMS generated via `oc adm catalog mirror`. This change consolidates Konflux IDMS helpers (apply_konflux_idms, idms_has_all_mirrors) into tests/install_upgrade_operators/utils.py, removes the legacy ICSP fixtures from install, upgrade and EUS upgrade conftest files, and replaces them with Konflux-based fixtures: - Install path: installed_konflux_idms (product_install/conftest.py) - Upgrade path: updated_konflux_idms (product_upgrade/conftest.py) - EUS upgrade path: eus_updated_konflux_idms (product_upgrade/conftest.py) Co-Authored-By: Claude Opus 4.6 --- tests/conftest.py | 11 -- .../product_install/conftest.py | 78 ++++------ .../product_upgrade/conftest.py | 122 +++++++-------- .../product_upgrade/test_upgrade.py | 2 +- tests/install_upgrade_operators/utils.py | 64 +++++++- utilities/operator.py | 145 ------------------ 6 files changed, 153 insertions(+), 269 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index 35697a4628..4fe5b50849 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -171,7 +171,6 @@ wait_for_ovs_status, ) from utilities.operator import ( - cluster_with_icsp, disable_default_sources_in_operatorhub, get_hco_csv_name_by_version, get_machine_config_pool_by_name, @@ -488,11 +487,6 @@ def utility_daemonset( yield ds -@pytest.fixture(scope="session") -def pull_secret_directory(tmpdir_factory): - yield tmpdir_factory.mktemp("pullsecret-folder") - - @pytest.fixture(scope="session") def generated_pulled_secret( is_production_source, @@ -2383,11 +2377,6 @@ def worker_machine1(worker_node1): raise ResourceNotFoundError(f"Machine object for {worker_node1.name} doesn't exists") -@pytest.fixture(scope="session") -def is_idms_cluster(): - return not cluster_with_icsp() - - @pytest.fixture(scope="session") def available_storage_classes_names(): return [[*sc][0] for sc in py_config["storage_class_matrix"]] diff --git a/tests/install_upgrade_operators/product_install/conftest.py b/tests/install_upgrade_operators/product_install/conftest.py index eb7bce786f..ee38777f29 100644 --- a/tests/install_upgrade_operators/product_install/conftest.py +++ b/tests/install_upgrade_operators/product_install/conftest.py @@ -6,23 +6,27 @@ from ocp_resources.cluster_service_version import ClusterServiceVersion from ocp_resources.hostpath_provisioner import HostPathProvisioner from ocp_resources.hyperconverged import HyperConverged +from ocp_resources.image_digest_mirror_set import ImageDigestMirrorSet from ocp_resources.installplan import InstallPlan from ocp_resources.persistent_volume import PersistentVolume from ocp_resources.resource import get_client +from packaging.version import Version from pytest_testconfig import py_config from tests.install_upgrade_operators.product_install.constants import ( HCO_NOT_INSTALLED_ALERT, - OPENSHIFT_VIRTUALIZATION, ) from tests.install_upgrade_operators.product_install.utils import get_all_resources +from tests.install_upgrade_operators.utils import ( + KONFLUX_IDMS_NAME, + KONFLUX_MIRROR_BASE_URL, + apply_konflux_idms, + idms_has_all_mirrors, +) from utilities.constants import ( - BREW_REGISTERY_SOURCE, CRITICAL_STR, HCO_CATALOG_SOURCE, HCO_SUBSCRIPTION, - ICSP_FILE, - IDMS_FILE, INFO_STR, PENDING_STR, PRODUCTION_CATALOG_SOURCE, @@ -42,17 +46,12 @@ ) from utilities.operator import ( create_catalog_source, - create_icsp_idms_from_file, create_operator, create_operator_group, create_subscription, - generate_icsp_idms_file, get_hco_csv_name_by_version, get_install_plan_from_subscription, - get_mcp_updating_transition_times, wait_for_catalogsource_ready, - wait_for_mcp_update_end, - wait_for_mcp_update_start, ) from utilities.storage import ( HppCsiStorageClass, @@ -81,54 +80,33 @@ def after_installation_all_resources(installation_data_dir): @pytest.fixture(scope="module") -def hyperconverged_directory(tmpdir_factory, is_production_source): - if is_production_source: - yield - else: - yield tmpdir_factory.mktemp(f"{OPENSHIFT_VIRTUALIZATION}-folder") - - -@pytest.fixture(scope="module") -def generated_hyperconverged_icsp_idms( +def installed_konflux_idms( + admin_client, is_production_source, - is_idms_cluster, - hyperconverged_directory, - generated_pulled_secret, - cnv_image_url, + cnv_version_to_install_info, + nodes, + machine_config_pools, + machine_config_pools_conditions_scope_module, ): if is_production_source: - LOGGER.info("This is installation from production source, icsp update is not needed.") + LOGGER.info("Production source install, IDMS update not needed.") return - folder_name = f"{hyperconverged_directory}/{OPENSHIFT_VIRTUALIZATION}-manifest" - LOGGER.info(f"Create CNV ICSP/IDMS file {ICSP_FILE}/{IDMS_FILE} in {hyperconverged_directory}") - mirror_cmd = ( - f"oc adm catalog mirror {cnv_image_url} {BREW_REGISTERY_SOURCE} --manifests-only" - f" --to-manifests {folder_name} --registry-config={generated_pulled_secret}" - ) - - return generate_icsp_idms_file(folder_name=folder_name, command=mirror_cmd, is_idms_file=is_idms_cluster) + version = Version(version=cnv_version_to_install_info["version"]) + required_mirrors = [f"{KONFLUX_MIRROR_BASE_URL}/v{version.major}-{version.minor}"] -@pytest.fixture(scope="module") -def updated_icsp_hyperconverged( - is_production_source, - generated_hyperconverged_icsp_idms, - machine_config_pools, - machine_config_pools_conditions_scope_module, -): - initial_updating_transition_times = get_mcp_updating_transition_times( - mcp_conditions=machine_config_pools_conditions_scope_module - ) - if is_production_source: - LOGGER.info("This is installation from production source, icsp/idms update is not needed.") + idms = ImageDigestMirrorSet(name=KONFLUX_IDMS_NAME, client=admin_client) + if idms.exists and idms_has_all_mirrors(idms=idms, required_mirrors=required_mirrors): + LOGGER.info(f"IDMS {KONFLUX_IDMS_NAME} already contains required mirrors.") return - create_icsp_idms_from_file(file_path=generated_hyperconverged_icsp_idms) - LOGGER.info("Wait for MCP update after ICSP/IDMS modification.") - wait_for_mcp_update_start( - machine_config_pools_list=machine_config_pools, - initial_transition_times=initial_updating_transition_times, + + apply_konflux_idms( + idms=idms, + required_mirrors=required_mirrors, + machine_config_pools=machine_config_pools, + mcp_conditions=machine_config_pools_conditions_scope_module, + nodes=nodes, ) - wait_for_mcp_update_end(machine_config_pools_list=machine_config_pools) @pytest.fixture(scope="module") @@ -218,7 +196,7 @@ def cnv_install_plan_installed( def installed_openshift_virtualization( admin_client, disabled_default_sources_in_operatorhub_scope_module, - updated_icsp_hyperconverged, + installed_konflux_idms, hyperconverged_catalog_source, created_cnv_namespace, created_cnv_operator_group, diff --git a/tests/install_upgrade_operators/product_upgrade/conftest.py b/tests/install_upgrade_operators/product_upgrade/conftest.py index 93286d4d06..0b873b7775 100644 --- a/tests/install_upgrade_operators/product_upgrade/conftest.py +++ b/tests/install_upgrade_operators/product_upgrade/conftest.py @@ -4,6 +4,7 @@ import pytest from ocp_resources.cluster_version import ClusterVersion +from ocp_resources.image_digest_mirror_set import ImageDigestMirrorSet from ocp_resources.resource import ResourceEditor from ocp_utilities.monitoring import Prometheus from packaging.version import Version @@ -29,9 +30,21 @@ wait_for_odf_update, wait_for_pods_replacement_by_type, ) -from tests.install_upgrade_operators.utils import wait_for_operator_condition +from tests.install_upgrade_operators.utils import ( + KONFLUX_IDMS_NAME, + KONFLUX_MIRROR_BASE_URL, + apply_konflux_idms, + idms_has_all_mirrors, + wait_for_operator_condition, +) from tests.upgrade_params import EUS -from utilities.constants import HCO_CATALOG_SOURCE, HOTFIX_STR, TIMEOUT_10MIN, TIMEOUT_180MIN, NamespacesNames +from utilities.constants import ( + HCO_CATALOG_SOURCE, + HOTFIX_STR, + TIMEOUT_10MIN, + TIMEOUT_180MIN, + NamespacesNames, +) from utilities.data_collector import ( get_data_collector_base_directory, ) @@ -43,8 +56,6 @@ get_subscription, ) from utilities.operator import ( - apply_icsp_idms, - get_generated_icsp_idms, get_machine_config_pool_by_name, get_machine_config_pools_conditions, update_image_in_catalog_source, @@ -83,47 +94,46 @@ def nodes_labels_before_upgrade(nodes, cnv_upgrade): return get_nodes_labels(nodes=nodes, cnv_upgrade=cnv_upgrade) +@pytest.fixture(scope="session") +def required_konflux_mirrors(cnv_target_version, cnv_current_version): + target = Version(version=cnv_target_version) + current = Version(version=cnv_current_version) + return [ + f"{KONFLUX_MIRROR_BASE_URL}/v{target.major}-{minor}" for minor in range(target.minor, current.minor - 1, -1) + ] + + @pytest.fixture() -def updated_image_content_source_policy( +def updated_konflux_idms( admin_client, - nodes, - tmpdir_factory, - active_machine_config_pools, - machine_config_pools_conditions, - cnv_image_url, cnv_image_name, + nodes, cnv_source, - cnv_target_version, - cnv_registry_source, - pull_secret_directory, - generated_pulled_secret, + required_konflux_mirrors, is_disconnected_cluster, - is_idms_cluster, + active_machine_config_pools, + machine_config_pools_conditions, ): - """ - Creates a new ImageContentSourcePolicy file with a given CNV image and applies it to the cluster. - """ + """Ensures the Konflux IDMS contains the required mirror entries for the CNV upgrade target version.""" if is_disconnected_cluster: - LOGGER.warning("Skip applying ICSP/IDMS in a disconnected setup.") + LOGGER.warning("Skip applying IDMS in a disconnected setup.") return if cnv_source == HOTFIX_STR: - LOGGER.info("ICSP updates skipped as upgrading using production source/upgrade to hotfix") + LOGGER.info("IDMS updates skipped as upgrading using production source/upgrade to hotfix") return - file_path = get_generated_icsp_idms( - image_url=cnv_image_url, - registry_source=cnv_registry_source["source_map"], - generated_pulled_secret=generated_pulled_secret, - pull_secret_directory=pull_secret_directory, - is_idms_cluster=is_idms_cluster, - ) - apply_icsp_idms( - file_paths=[file_path], + + idms = ImageDigestMirrorSet(name=KONFLUX_IDMS_NAME, client=admin_client) + if idms.exists and idms_has_all_mirrors(idms=idms, required_mirrors=required_konflux_mirrors): + LOGGER.info(f"IDMS {KONFLUX_IDMS_NAME} already contains all required mirrors.") + return + + apply_konflux_idms( + idms=idms, + required_mirrors=required_konflux_mirrors, machine_config_pools=active_machine_config_pools, mcp_conditions=machine_config_pools_conditions, nodes=nodes, - is_idms_file=is_idms_cluster, - delete_file=True, ) @@ -384,6 +394,8 @@ def default_workload_update_strategy(hyperconverged_resource_scope_session): @pytest.fixture() def eus_paused_worker_mcp( worker_machine_config_pools, + worker_machine_config_pools_conditions, + eus_updated_konflux_idms, ): LOGGER.info("Pausing worker MCP updates before starting EUS upgrade.") update_mcp_paused_spec(mcp=worker_machine_config_pools) @@ -431,44 +443,32 @@ def eus_unpaused_workload_update( @pytest.fixture(scope="module") -def created_eus_icsps( - pull_secret_directory, - generated_pulled_secret, - cnv_registry_source, +def eus_updated_konflux_idms( + admin_client, eus_cnv_upgrade_path, - is_idms_cluster, -): - icsp_files = [] - for entry in eus_cnv_upgrade_path: - for version in eus_cnv_upgrade_path[entry]: - icsp_file = get_generated_icsp_idms( - image_url=eus_cnv_upgrade_path[entry][version], - registry_source=cnv_registry_source["source_map"], - generated_pulled_secret=generated_pulled_secret, - pull_secret_directory=pull_secret_directory, - is_idms_cluster=is_idms_cluster, - cnv_version=version, - ) - icsp_files.append(icsp_file) - LOGGER.info(f"EUS ICSP Files created: {icsp_files}") - return icsp_files - - -@pytest.fixture(scope="module") -def eus_applied_all_icsp( nodes, - generated_pulled_secret, machine_config_pools, machine_config_pools_conditions_scope_module, - created_eus_icsps, - is_idms_cluster, ): - apply_icsp_idms( - file_paths=created_eus_icsps, + required_mirrors = [] + for phase in eus_cnv_upgrade_path: + for version in eus_cnv_upgrade_path[phase]: + ver = Version(version=version) + mirror = f"{KONFLUX_MIRROR_BASE_URL}/v{ver.major}-{ver.minor}" + if mirror not in required_mirrors: + required_mirrors.append(mirror) + + idms = ImageDigestMirrorSet(name=KONFLUX_IDMS_NAME, client=admin_client) + if idms.exists and idms_has_all_mirrors(idms=idms, required_mirrors=required_mirrors): + LOGGER.info(f"IDMS {KONFLUX_IDMS_NAME} already has all EUS mirrors.") + return + + apply_konflux_idms( + idms=idms, + required_mirrors=required_mirrors, machine_config_pools=machine_config_pools, mcp_conditions=machine_config_pools_conditions_scope_module, nodes=nodes, - is_idms_file=is_idms_cluster, ) diff --git a/tests/install_upgrade_operators/product_upgrade/test_upgrade.py b/tests/install_upgrade_operators/product_upgrade/test_upgrade.py index d89e26a845..dff141007a 100644 --- a/tests/install_upgrade_operators/product_upgrade/test_upgrade.py +++ b/tests/install_upgrade_operators/product_upgrade/test_upgrade.py @@ -54,7 +54,7 @@ def test_cnv_upgrade_process( cnv_upgrade_stream, fired_alerts_before_upgrade, disabled_default_sources_in_operatorhub, - updated_image_content_source_policy, + updated_konflux_idms, updated_custom_hco_catalog_source_image, updated_cnv_subscription_source, approved_cnv_upgrade_install_plan, diff --git a/tests/install_upgrade_operators/utils.py b/tests/install_upgrade_operators/utils.py index 795657823d..4e8fcc6c7c 100644 --- a/tests/install_upgrade_operators/utils.py +++ b/tests/install_upgrade_operators/utils.py @@ -6,10 +6,13 @@ from benedict import benedict from kubernetes.dynamic import DynamicClient from kubernetes.dynamic.exceptions import ConflictError, ResourceNotFoundError +from ocp_resources.image_digest_mirror_set import ImageDigestMirrorSet from ocp_resources.installplan import InstallPlan +from ocp_resources.machine_config_pool import MachineConfigPool from ocp_resources.network_addons_config import NetworkAddonsConfig +from ocp_resources.node import Node from ocp_resources.operator_condition import OperatorCondition -from ocp_resources.resource import Resource +from ocp_resources.resource import Resource, ResourceEditor from timeout_sampler import TimeoutExpiredError, TimeoutSampler from tests.install_upgrade_operators.constants import KEY_PATH_SEPARATOR @@ -23,8 +26,12 @@ TIMEOUT_40MIN, ) from utilities.infra import get_subscription +from utilities.operator import wait_for_mcp_update_completion LOGGER = logging.getLogger(__name__) +KONFLUX_IDMS_NAME = "zz-cnv-icsp-fallback" +KONFLUX_MIRROR_BASE_URL = "quay.io/openshift-virtualization/konflux-builds" +KONFLUX_IDMS_SOURCE = "registry.redhat.io/container-native-virtualization" def wait_for_operator_condition(dyn_client, hco_namespace, name, upgradable): @@ -277,3 +284,58 @@ def get_resource_key_value(resource, key_name): resource.instance.to_dict()["spec"], keypath_separator=KEY_PATH_SEPARATOR, ).get(key_name) + + +def apply_konflux_idms( + idms: ImageDigestMirrorSet, + required_mirrors: list[str], + machine_config_pools: list[MachineConfigPool], + mcp_conditions: dict[str, list[dict[str, str]]], + nodes: list[Node], +) -> None: + """Creates or patches the Konflux IDMS with the required mirror entries. + + Args: + idms: The Konflux IDMS resource to create or patch. + required_mirrors: Konflux mirror URLs to set on the IDMS. + machine_config_pools: Active machine config pools to pause/wait. + mcp_conditions: Initial MCP conditions for tracking update progress. + nodes: Cluster nodes to verify readiness after MCP update. + """ + image_digest_mirrors = [{"source": KONFLUX_IDMS_SOURCE, "mirrors": required_mirrors}] + LOGGER.info("Pausing MCP updates while modifying IDMS.") + with ResourceEditor(patches={mcp: {"spec": {"paused": True}} for mcp in machine_config_pools}): + if idms.exists: + LOGGER.info(f"Patching IDMS {KONFLUX_IDMS_NAME} with mirrors: {required_mirrors}") + ResourceEditor(patches={idms: {"spec": {"imageDigestMirrors": image_digest_mirrors}}}).update() + else: + LOGGER.info(f"Creating IDMS {KONFLUX_IDMS_NAME} with mirrors: {required_mirrors}") + ImageDigestMirrorSet( + name=KONFLUX_IDMS_NAME, + client=idms.client, + image_digest_mirrors=image_digest_mirrors, + teardown=False, + ).deploy(wait=True) + LOGGER.info("Wait for MCP update after IDMS modification.") + wait_for_mcp_update_completion( + machine_config_pools_list=machine_config_pools, + initial_mcp_conditions=mcp_conditions, + nodes=nodes, + ) + + +def idms_has_all_mirrors(idms: ImageDigestMirrorSet, required_mirrors: list[str]) -> bool: + """Returns True if the IDMS already contains all required Konflux mirror entries.""" + existing_mirrors = idms.instance.spec.imageDigestMirrors + source_entry = next( + (entry for entry in existing_mirrors if entry["source"] == KONFLUX_IDMS_SOURCE), + None, + ) + if not source_entry: + LOGGER.info( + f"IDMS {idms.name} has no entry for source {KONFLUX_IDMS_SOURCE}, mirrors need to be added." + f" Current IDMS mirrors: {existing_mirrors}" + ) + return False + existing_urls = {str(mirror) for mirror in source_entry["mirrors"]} + return all(mirror in existing_urls for mirror in required_mirrors) diff --git a/utilities/operator.py b/utilities/operator.py index 3059c7a137..3121819d3b 100644 --- a/utilities/operator.py +++ b/utilities/operator.py @@ -1,38 +1,28 @@ # TODO: Remove ### unused_code: ignore ### from function docstring once it's used. import logging -import os -import shlex from contextlib import contextmanager from datetime import datetime from pprint import pformat -import yaml from kubernetes.dynamic.exceptions import ResourceNotFoundError from ocp_resources.catalog_source import CatalogSource from ocp_resources.cluster_operator import ClusterOperator from ocp_resources.cluster_service_version import ClusterServiceVersion -from ocp_resources.image_content_source_policy import ImageContentSourcePolicy -from ocp_resources.image_digest_mirror_set import ImageDigestMirrorSet from ocp_resources.machine_config_pool import MachineConfigPool from ocp_resources.namespace import Namespace -from ocp_resources.node import Node from ocp_resources.operator_group import OperatorGroup from ocp_resources.operator_hub import OperatorHub from ocp_resources.pod import Pod from ocp_resources.resource import Resource, ResourceEditor from ocp_resources.subscription import Subscription -from pyhelper_utils.shell import run_command from pytest_testconfig import config as py_config from timeout_sampler import TimeoutExpiredError, TimeoutSampler import utilities.infra from utilities.constants import ( BASE_EXCEPTIONS_DICT, - BREW_REGISTERY_SOURCE, DEFAULT_RESOURCE_CONDITIONS, - ICSP_FILE, - IDMS_FILE, TIMEOUT_5MIN, TIMEOUT_5SEC, TIMEOUT_10MIN, @@ -46,80 +36,6 @@ LOGGER = logging.getLogger(__name__) -def create_icsp_idms_command(image, source_url, folder_name, pull_secret=None, filter_options=""): - """ - Create ImageContentSourcePolicy command. - - Args: - image (str): name of image to be mirrored. - source_url (str): source url of image registry to which contents mirror. - folder_name (str): local path to store manifests. - pull_secret (str): Path to your registry credentials, default set to None(until passed) - filter_options (str): when filter passed it will choose image from multiple variants. - - Returns: - str: base command to create icsp in the cluster. - """ - base_command = ( - f"oc adm catalog mirror {image} {source_url} --manifests-only --to-manifests {folder_name} {filter_options}" - ) - if pull_secret: - base_command = f"{base_command} --registry-config={pull_secret}" - - return base_command - - -def generate_icsp_idms_file(folder_name, command, is_idms_file, cnv_version=None): - rc, _, _ = run_command( - command=shlex.split(command), - verify_stderr=False, - check=False, - ) - assert rc - file_name = IDMS_FILE if is_idms_file else ICSP_FILE - - absolute_file_name = os.path.join(folder_name, file_name) - assert os.path.isfile(absolute_file_name), f"file does not exist in path {absolute_file_name}" - if cnv_version: - absolute_file_name = generate_unique_icsp_idms_file( - file_name=absolute_file_name, - version_string=cnv_version.lstrip("v").replace(".", ""), - ) - return absolute_file_name - - -def generate_unique_icsp_idms_file(file_name, version_string): - # update the metadata.name value to generate unique ICSP/IDMS - with open(file_name, "r") as fd: - file_yaml = yaml.safe_load(fd.read()) - file_yaml["metadata"]["name"] = f"iib-{version_string}" - with open(file_name, "w") as current_mirror_file: - yaml.dump(file_yaml, current_mirror_file) - new_file_name = file_name.replace(file_name, f"{file_name.replace('.yaml', '')}{version_string}.yaml") - os.rename(file_name, new_file_name) - return new_file_name - - -def create_icsp_idms_from_file(file_path): - LOGGER.info(f"Creating icsp/idms using file: {file_path}") - rc, _, _ = run_command( - command=shlex.split(f"oc create -f {file_path}"), - verify_stderr=False, - check=False, - ) - assert rc - - -def delete_existing_icsp_idms(name, is_idms_file): - resource_class = ImageDigestMirrorSet if is_idms_file else ImageContentSourcePolicy - LOGGER.info(f"Deleting {resource_class}.") - for resource_obj in resource_class.get(): - object_name = resource_obj.name - if object_name.startswith(name): - LOGGER.info(f"Deleting {resource_class} {object_name}.") - resource_obj.delete(wait=True) - - def get_mcps_with_different_transition_times(condition_type, machine_config_pools_list, initial_transition_times): """ Return a set of machine config pool (MCP) names with different transition times. @@ -629,11 +545,6 @@ def update_subscription_source( }).update() -def cluster_with_icsp(): - icsp_list = list(ImageContentSourcePolicy.get()) - return len(icsp_list) > 0 - - def get_cluster_operator_status_conditions(admin_client, operator_conditions=None): operator_conditions = operator_conditions or DEFAULT_RESOURCE_CONDITIONS cluster_operator_status = {} @@ -697,59 +608,3 @@ def wait_for_cluster_operator_stabilize(admin_client, wait_timeout=TIMEOUT_20MIN def get_hco_csv_name_by_version(cnv_target_version: str) -> str: return f"kubevirt-hyperconverged-operator.v{cnv_target_version}" - - -def get_generated_icsp_idms( - image_url: str, - registry_source: str, - generated_pulled_secret: str, - pull_secret_directory: str, - is_idms_cluster: bool, - cnv_version: str | None = None, - filter_options: str = "", -) -> str: - pull_secret = None - if image_url.startswith(tuple([BREW_REGISTERY_SOURCE, "quay.io"])): - registry_source = BREW_REGISTERY_SOURCE - pull_secret = generated_pulled_secret - cnv_mirror_cmd = create_icsp_idms_command( - image=image_url, - source_url=registry_source, - folder_name=pull_secret_directory, - pull_secret=pull_secret, - filter_options=filter_options, - ) - icsp_file_path = generate_icsp_idms_file( - folder_name=pull_secret_directory, - command=cnv_mirror_cmd, - is_idms_file=is_idms_cluster, - cnv_version=cnv_version, - ) - - return icsp_file_path - - -def apply_icsp_idms( - file_paths: list[str], - machine_config_pools: list[MachineConfigPool], - mcp_conditions: dict[str, list[dict[str, str]]], - nodes: list[Node], - is_idms_file: bool, - delete_file: bool = False, -) -> None: - LOGGER.info("pausing MCP updates while modifying ICSP/IDMS") - with ResourceEditor(patches={mcp: {"spec": {"paused": True}} for mcp in machine_config_pools}): - if delete_file: - # Due to the amount of annotations in ICSP/IDMS yaml, `oc apply` may fail. Existing ICSP/IDMS is deleted. - LOGGER.info("Deleting existing ICSP/IDMS.") - delete_existing_icsp_idms(name="iib", is_idms_file=is_idms_file) - LOGGER.info("Creating new ICSP/IDMS") - for file_path in file_paths: - create_icsp_idms_from_file(file_path=file_path) - - LOGGER.info("Wait for MCP update after ICSP/IDMS modification.") - wait_for_mcp_update_completion( - machine_config_pools_list=machine_config_pools, - initial_mcp_conditions=mcp_conditions, - nodes=nodes, - )