[IUO] Migrate install and EUS upgrade paths to Konflux IDMS

CNV operator builds migrated from Brew to Konflux, so the install and
EUS upgrade test paths need 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 both install and EUS upgrade
conftest files, and replaces them with new Konflux-based fixtures:

- Install path: installed_konflux_idms (product_install/conftest.py)
- EUS upgrade path: eus_updated_konflux_idms (product_upgrade/conftest.py)

Follows up on PR #4061 which fixed the upgrade path only.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: rlobillo

tests/install_upgrade_operators/product_install/conftest.py

@@ -5,23 +5,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.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.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,
@@ -39,17 +43,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,
@@ -65,54 +64,32 @@
 
 
 @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(
     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)
+    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")
@@ -206,7 +183,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,

tests/install_upgrade_operators/product_upgrade/conftest.py

@@ -12,9 +12,6 @@
 
 from tests.install_upgrade_operators.constants import WORKLOAD_UPDATE_STRATEGY_KEY_NAME, WORKLOADUPDATEMETHODS
 from tests.install_upgrade_operators.product_upgrade.utils import (
-    KONFLUX_IDMS_NAME,
-    KONFLUX_MIRROR_BASE_URL,
-    apply_konflux_idms,
     approve_cnv_upgrade_install_plan,
     extract_ocp_version_from_ocp_image,
     get_alerts_fired_during_upgrade,
@@ -23,7 +20,6 @@
     get_nodes_labels,
     get_nodes_taints,
     get_shortest_upgrade_path,
-    idms_has_all_mirrors,
     perform_cnv_upgrade,
     run_ocp_upgrade_command,
     set_workload_update_methods_hco,
@@ -34,7 +30,13 @@
     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,
@@ -53,8 +55,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,
@@ -360,7 +360,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)
@@ -407,44 +407,31 @@ 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(
     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)
+    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,
     )
 
 

tests/install_upgrade_operators/product_upgrade/utils.py

@@ -13,11 +13,9 @@
 from ocp_resources.cluster_service_version import ClusterServiceVersion
 from ocp_resources.cluster_version import ClusterVersion
 from ocp_resources.hyperconverged import HyperConverged
-from ocp_resources.image_digest_mirror_set import ImageDigestMirrorSet
 from ocp_resources.kubevirt import KubeVirt
 from ocp_resources.machine_config_pool import MachineConfigPool
 from ocp_resources.namespace import Namespace
-from ocp_resources.node import Node
 from ocp_resources.resource import Resource, ResourceEditor
 from packaging.version import Version
 from pyhelper_utils.shell import run_command
@@ -61,9 +59,6 @@
 
 LOGGER = logging.getLogger(__name__)
 TIER_2_PODS_TYPE = "tier-2"
-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"
 
 # list of whitelisted alerts
 WHITELIST_ALERTS_UPGRADE_LIST = ["OutdatedVirtualMachineInstanceWorkloads"]
@@ -718,54 +713,3 @@ def _get_updated_odf_csv(_target_version: str, _admin_client: DynamicClient) ->
         if not sample:
             return
         LOGGER.info(f"Following odf csvs are not updated: {','.join(sample)}")
-
-
-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,
-                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."""
-    source_entry = next(
-        (entry for entry in idms.instance.spec.imageDigestMirrors if entry["source"] == KONFLUX_IDMS_SOURCE),
-        None,
-    )
-    if not source_entry:
-        return False
-    existing_urls = {str(mirror) for mirror in source_entry["mirrors"]}
-    return all(mirror in existing_urls for mirror in required_mirrors)