From 410f19417c0b9eff2835203d940d99f361872b54 Mon Sep 17 00:00:00 2001 From: Ramon Lobillo Mateos <62110535+rlobillo@users.noreply.github.com> Date: Wed, 4 Mar 2026 17:21:38 +0100 Subject: [PATCH] [CNV4.20][IUO] Migrate install, upgrade and EUS upgrade paths to Konflux IDMS Manual Cherry pick of https://github.com/RedHatQE/openshift-virtualization-tests/pull/4070 and https://github.com/RedHatQE/openshift-virtualization-tests/pull/4061 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) N/A N/A N/A N/A N/A Co-authored-by: Claude Opus 4.6 --- tests/conftest.py | 11 -- .../product_install/conftest.py | 78 ++++------ .../product_upgrade/conftest.py | 121 +++++++-------- .../product_upgrade/test_upgrade.py | 2 +- tests/install_upgrade_operators/utils.py | 64 +++++++- utilities/operator.py | 145 ------------------ 6 files changed, 151 insertions(+), 270 deletions(-) diff --git a/tests/conftest.py b/tests/conftest.py index d9bc9869e0..e10027b207 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -179,7 +179,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, @@ -520,11 +519,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, @@ -2364,11 +2358,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 676a640a80..4835130be6 100644 --- a/tests/install_upgrade_operators/product_install/conftest.py +++ b/tests/install_upgrade_operators/product_install/conftest.py @@ -6,25 +6,29 @@ 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 ocp_resources.storage_class import StorageClass +from packaging.version import Version from pytest_testconfig import py_config from timeout_sampler import TimeoutSampler 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, @@ -45,17 +49,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, @@ -86,54 +85,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") @@ -223,7 +201,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 f401ef19fe..e1b7a6407e 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,20 @@ 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, NamespacesNames +from utilities.constants import ( + HCO_CATALOG_SOURCE, + HOTFIX_STR, + TIMEOUT_10MIN, + NamespacesNames, +) from utilities.data_collector import ( get_data_collector_base_directory, ) @@ -44,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, @@ -85,47 +95,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, ) @@ -349,7 +358,7 @@ def eus_paused_worker_mcp( workers, worker_machine_config_pools, worker_machine_config_pools_conditions, - eus_applied_all_icsp, + eus_updated_konflux_idms, ): LOGGER.info("Pausing worker MCP updates before starting EUS upgrade.") update_mcp_paused_spec(mcp=worker_machine_config_pools) @@ -396,44 +405,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 7399daa688..ab359a3fc2 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 16ee1c20c7..1629dc958c 100644 --- a/tests/install_upgrade_operators/utils.py +++ b/tests/install_upgrade_operators/utils.py @@ -7,10 +7,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 @@ -24,8 +27,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): @@ -278,3 +285,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 273ce40ee8..eec02fc67d 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) 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. @@ -628,11 +544,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 = {} @@ -696,59 +607,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, - )