CherryPicked: [4.18] eus upgrade alignment

conftest.py

@@ -22,6 +22,7 @@
 from _pytest.runner import CallInfo
 from kubernetes.dynamic.exceptions import ConflictError
 from ocp_resources.resource import get_client
+from packaging.version import Version
 from pyhelper_utils.shell import run_command
 from pytest import Item
 from pytest_testconfig import config as py_config
@@ -150,7 +151,6 @@ def pytest_addoption(parser):
         "--eus-ocp-images",
         help="Comma-separated OCP images to use for EUS-to-EUS upgrade.",
     )
-    install_upgrade_group.addoption("--eus-cnv-target-version", help="target CNV version for eus upgrade")
     install_upgrade_group.addoption(
         "--upgrade-skip-default-sc-setup",
         help="Skip the fixture that changes the default sc in upgrade lane",
@@ -302,18 +302,21 @@ def pytest_cmdline_main(config):
     if upgrade_option == "ocp" and not config.getoption("ocp_image"):
         raise ValueError("Running with --upgrade ocp: Missing --ocp-image")
 
-    if upgrade_option == "cnv":
+    if upgrade_option in ("cnv", "eus"):
         if not config.getoption("cnv_version"):
             raise ValueError("Missing --cnv-version")
         if not config.getoption("cnv_image"):
-            if config.getoption("cnv_source") != "production":
+            if upgrade_option == "eus" or config.getoption("cnv_source") != "production":
                 raise ValueError("Missing --cnv-image")
 
-    if upgrade_option == "eus":
+    if upgrade_option == "eus" and not config.option.collectonly:
+        cnv_version = config.getoption("cnv_version")
+        if Version(version=cnv_version).minor % 2:
+            raise ValueError(f"EUS target version {cnv_version} must have an even minor version")
         eus_ocp_images = config.getoption("eus_ocp_images")
         if not (eus_ocp_images and len(eus_ocp_images.split(",")) == 2):
             raise ValueError(
-                f"Two OCP images are needed to perform EUS-to-EUS upgrade with --eus-ocp-images."
+                f"Two OCP images are needed for EUS-to-EUS upgrade with --eus-ocp-images."
                 f" Provided images: {eus_ocp_images}"
             )
 
@@ -793,7 +796,7 @@ def is_skip_must_gather(node: Node) -> bool:
 
 def get_inspect_command_namespace_string(node: Node, test_name: str) -> str:
     namespace_str = ""
-    components = [key for key in NAMESPACE_COLLECTION.keys() if f"tests/{key}/" in test_name]
+    components = [key for key in NAMESPACE_COLLECTION if f"tests/{key}/" in test_name]
     if not components:
         LOGGER.warning(f"{test_name} does not require special data collection on failure")
     else:

tests/conftest.py

@@ -63,7 +63,7 @@
 from ocp_resources.virtual_machine_instancetype import VirtualMachineInstancetype
 from ocp_resources.virtual_machine_preference import VirtualMachinePreference
 from ocp_utilities.monitoring import Prometheus
-from packaging.version import Version, parse
+from packaging.version import parse
 from pytest_testconfig import config as py_config
 from timeout_sampler import TimeoutSampler
 
@@ -1898,27 +1898,14 @@ def hco_target_csv_name(cnv_target_version):
     return get_hco_csv_name_by_version(cnv_target_version=cnv_target_version) if cnv_target_version else None
 
 
-@pytest.fixture(scope="session")
-def eus_hco_target_csv_name(eus_target_cnv_version):
-    return get_hco_csv_name_by_version(cnv_target_version=eus_target_cnv_version)
-
-
 @pytest.fixture(scope="session")
 def cnv_target_version(pytestconfig):
     return pytestconfig.option.cnv_version
 
 
 @pytest.fixture(scope="session")
-def eus_target_cnv_version(pytestconfig, cnv_current_version):
-    cnv_current_version = Version(version=cnv_current_version)
-    minor = cnv_current_version.minor
-    # EUS-to-EUS upgrades are only viable between even-numbered minor versions, exit if non-eus version
-    if minor % 2:
-        exit_pytest_execution(
-            message=f"EUS upgrade can not be performed from non-eus version: {cnv_current_version}",
-            return_code=EUS_ERROR_CODE,
-        )
-    return pytestconfig.option.eus_cnv_target_version or f"{cnv_current_version.major}.{minor + 2}.0"
+def cnv_channel(pytestconfig):
+    return pytestconfig.option.cnv_channel
 
 
 @pytest.fixture()

tests/install_upgrade_operators/product_upgrade/conftest.py

@@ -15,13 +15,12 @@
 )
 from tests.install_upgrade_operators.product_upgrade.utils import (
     approve_cnv_upgrade_install_plan,
+    build_eus_upgrade_path_dict,
     extract_ocp_version_from_ocp_image,
     get_alerts_fired_during_upgrade,
     get_all_firing_cnv_alerts,
-    get_iib_images_of_cnv_versions,
     get_nodes_labels,
     get_nodes_taints,
-    get_upgrade_path,
     perform_cnv_upgrade,
     run_ocp_upgrade_command,
     set_workload_update_methods_hco,
@@ -49,8 +48,8 @@
     get_data_collector_base_directory,
 )
 from utilities.infra import (
+    exit_pytest_execution,
     generate_openshift_pull_secret_file,
-    get_csv_by_name,
     get_prometheus_k8s_token,
     get_related_images_name_and_version,
     get_subscription,
@@ -66,6 +65,7 @@
 
 LOGGER = logging.getLogger(__name__)
 POD_STR_NOT_MANAGED_BY_HCO = "hostpath-"
+EUS_ERROR_CODE = 98
 
 
 @pytest.fixture(scope="session")
@@ -323,64 +323,25 @@ def fired_alerts_during_upgrade(
 
 
 @pytest.fixture(scope="session")
-def eus_shortest_upgrade_path_info(eus_target_cnv_version, cnv_current_version):
-    # if cant get from stable - try candidate
-    if upgrade_paths_target_version := get_upgrade_path(target_version=eus_target_cnv_version):
-        target_channel = "stable"
-    else:
-        target_channel = "candidate"
-        upgrade_paths_target_version = get_upgrade_path(target_version=eus_target_cnv_version, channel=target_channel)
-    assert upgrade_paths_target_version, f"Couldn't find upgrade path for {eus_target_cnv_version} version"
-
-    sorted_upgrade_paths = sorted(
-        upgrade_paths_target_version, key=lambda path: Version(version=str(path["startVersion"])), reverse=True
-    )
-    for path in sorted_upgrade_paths:
-        if intermediate_upgrade_paths := get_upgrade_path(target_version=path["startVersion"]):
-            if intermediate_path_dict := next(
-                (item for item in intermediate_upgrade_paths if item["startVersion"] == f"v{cnv_current_version}"), None
-            ):
-                return {
-                    "target_versions": path["versions"],
-                    "intermediate_versions": intermediate_path_dict["versions"],
-                    "target_channel": target_channel,
-                }
-    raise AssertionError(f"Couldn't find upgrade path for {eus_target_cnv_version} version from {cnv_current_version}")
-
-
-@pytest.fixture(scope="session")
-def eus_target_channel(eus_shortest_upgrade_path_info):
-    return eus_shortest_upgrade_path_info["target_channel"]
-
-
-@pytest.fixture(scope="session")
-def eus_cnv_upgrade_path(eus_shortest_upgrade_path_info, eus_target_channel):
-    if not (
-        target_paths := get_iib_images_of_cnv_versions(
-            versions=eus_shortest_upgrade_path_info["target_versions"],
-            target_channel=eus_target_channel,
-        )
-    ):
-        target_paths = get_iib_images_of_cnv_versions(
-            versions=eus_shortest_upgrade_path_info["target_versions"],
-            errata_status="false",
-            target_channel=eus_target_channel,
+def eus_cnv_upgrade_path(
+    cnv_target_version,
+    cnv_current_version,
+    cnv_channel,
+    cnv_image_url,
+):
+    if Version(version=cnv_current_version).minor % 2:
+        exit_pytest_execution(
+            message=f"EUS upgrade can not be performed from non-eus version: {cnv_current_version}",
+            return_code=EUS_ERROR_CODE,
+            filename="eus_upgrade_failure.txt",
         )
-    intermediate_paths = get_iib_images_of_cnv_versions(
-        versions=eus_shortest_upgrade_path_info["intermediate_versions"],
-    )
-    assert intermediate_paths, (
-        f"Couldn't find build info for {eus_shortest_upgrade_path_info['intermediate_versions']} versions"
+    return build_eus_upgrade_path_dict(
+        current_cnv_version=cnv_current_version,
+        target_cnv_version=cnv_target_version,
+        target_channel=cnv_channel,
+        target_cnv_image_url=cnv_image_url,
     )
 
-    # Return a dictionary with the versions and images for the EUS-to-EUS upgrade
-    upgrade_path_dict = {
-        EUS: target_paths,
-        "non-eus": intermediate_paths,
-    }
-    LOGGER.info(f"Upgrade path for EUS-to-EUS upgrade: {upgrade_path_dict}")
-    return upgrade_path_dict
-
 
 @pytest.fixture(scope="session")
 def default_workload_update_strategy(hyperconverged_resource_scope_session):
@@ -564,18 +525,25 @@ def source_eus_to_non_eus_cnv_upgraded(
     admin_client,
     hco_namespace,
     eus_cnv_upgrade_path,
-    eus_target_channel,
+    cnv_subscription_scope_session,
+    cnv_registry_source,
     hyperconverged_resource_scope_function,
-    updated_cnv_subscription_source,
 ):
-    for version, cnv_image in sorted(eus_cnv_upgrade_path["non-eus"].items()):
+    for version, build_info in sorted(
+        eus_cnv_upgrade_path["non-eus"].items(),
+        key=lambda item: Version(version=item[0]),
+    ):
+        cnv_image = build_info["cnv_image_url"]
         LOGGER.info(f"Cnv upgrade to version {version} using image: {cnv_image}")
         perform_cnv_upgrade(
             admin_client=admin_client,
             cnv_image_url=cnv_image,
             cr_name=hyperconverged_resource_scope_function.name,
             hco_namespace=hco_namespace,
-            cnv_target_version=version.lstrip("v"),
+            cnv_target_version=version,
+            subscription=cnv_subscription_scope_session,
+            subscription_source=cnv_registry_source["cnv_subscription_source"],
+            subscription_channel=build_info["channel"],
         )
     LOGGER.info("Successfully performed cnv upgrades from source EUS to non-EUS version.")
 
@@ -585,32 +553,27 @@ def non_eus_to_target_eus_cnv_upgraded(
     admin_client,
     hco_namespace,
     eus_cnv_upgrade_path,
-    eus_target_channel,
     cnv_subscription_scope_session,
     cnv_registry_source,
     hyperconverged_resource_scope_function,
 ):
-    version, cnv_image = next(iter(eus_cnv_upgrade_path[EUS].items()))
-    LOGGER.info(f"Cnv upgrade to version {version} using image: {cnv_image}")
-    perform_cnv_upgrade(
-        admin_client=admin_client,
-        cnv_image_url=cnv_image,
-        cr_name=hyperconverged_resource_scope_function.name,
-        hco_namespace=hco_namespace,
-        cnv_target_version=version.lstrip("v"),
-        subscription=cnv_subscription_scope_session,
-        subscription_source=cnv_registry_source["cnv_subscription_source"],
-        subscription_channel=eus_target_channel,
-    )
-
-
-@pytest.fixture()
-def eus_created_target_hco_csv(admin_client, hco_namespace, eus_hco_target_csv_name):
-    return get_csv_by_name(
-        csv_name=eus_hco_target_csv_name,
-        admin_client=admin_client,
-        namespace=hco_namespace.name,
-    )
+    for version, build_info in sorted(
+        eus_cnv_upgrade_path[EUS].items(),
+        key=lambda item: Version(version=item[0]),
+    ):
+        cnv_image = build_info["cnv_image_url"]
+        LOGGER.info(f"Cnv upgrade to version {version} using image: {cnv_image}")
+        perform_cnv_upgrade(
+            admin_client=admin_client,
+            cnv_image_url=cnv_image,
+            cr_name=hyperconverged_resource_scope_function.name,
+            hco_namespace=hco_namespace,
+            cnv_target_version=version,
+            subscription=cnv_subscription_scope_session,
+            subscription_source=cnv_registry_source["cnv_subscription_source"],
+            subscription_channel=build_info["channel"],
+        )
+    LOGGER.info("Successfully performed cnv upgrades from non-EUS to target EUS version.")
 
 
 @pytest.fixture()

tests/install_upgrade_operators/product_upgrade/test_eus_upgrade.py

@@ -8,6 +8,11 @@
 
 LOGGER = logging.getLogger(__name__)
 
+pytestmark = pytest.mark.usefixtures(
+    "nodes_taints_before_upgrade",
+    "nodes_labels_before_upgrade",
+)
+
 
 @pytest.mark.product_upgrade_test
 @pytest.mark.upgrade
@@ -20,25 +25,22 @@ def test_eus_upgrade_process(
         self,
         admin_client,
         hco_namespace,
-        eus_target_cnv_version,
         eus_cnv_upgrade_path,
-        worker_machine_config_pools_conditions,
-        eus_applied_all_icsp,
         eus_paused_worker_mcp,
         eus_paused_workload_update,
         source_eus_to_non_eus_ocp_upgraded,
         source_eus_to_non_eus_cnv_upgraded,
         upgraded_odf,
         non_eus_to_target_eus_ocp_upgraded,
         non_eus_to_target_eus_cnv_upgraded,
-        eus_created_target_hco_csv,
+        created_target_hco_csv,
         eus_unpaused_workload_update,
         eus_unpaused_worker_mcp,
     ):
         LOGGER.info("Validate EUS to EUS upgrade process")
         verify_upgrade_cnv(
             dyn_client=admin_client,
             hco_namespace=hco_namespace,
-            expected_images=get_related_images_name_and_version(csv=eus_created_target_hco_csv).values(),
+            expected_images=get_related_images_name_and_version(csv=created_target_hco_csv).values(),
         )
         LOGGER.info("EUS post upgrade validation completed.")