Manual cherry-pick: [4.21] [Storage] refactor os matrix generator(#3883)

conftest.py

@@ -27,7 +27,9 @@
 from pytest_testconfig import config as py_config
 
 import utilities.cluster
-import utilities.infra
+
+# TODO: Remove this import when utilities modules are refactored...
+import utilities.infra  # noqa
 from libs.storage.config import StorageClassConfig
 from utilities.bitwarden import get_cnv_tests_secret_by_name
 from utilities.constants import (
@@ -52,6 +54,7 @@
     config_default_storage_class,
     deploy_run_in_progress_config_map,
     deploy_run_in_progress_namespace,
+    generate_os_matrix_dicts,
     get_artifactory_server_url,
     get_base_matrix_name,
     get_cnv_version_explorer_url,
@@ -62,6 +65,7 @@
     separator,
     skip_if_pytest_flags_exists,
     stop_if_run_in_progress,
+    update_latest_os_config,
 )
 
 LOGGER = logging.getLogger(__name__)
@@ -755,34 +759,6 @@ def pytest_generate_tests(metafunc):
 
 
 def pytest_sessionstart(session):
-    # TODO: Reduce cognitive complexity
-    def _update_os_related_config():
-        # Save the default windows_os_matrix before it is updated
-        # with runtime windows_os_matrix value(s).
-        # Some tests extract a single OS from the matrix and may fail if running with
-        # passed values from cli
-        if windows_os_matrix := py_config.get("windows_os_matrix"):
-            py_config["system_windows_os_matrix"] = windows_os_matrix
-
-        if rhel_os_matrix := py_config.get("rhel_os_matrix"):
-            py_config["system_rhel_os_matrix"] = rhel_os_matrix
-
-        # Update OS matrix list with the latest OS if running with os_group
-        if session.config.getoption("latest_rhel") and rhel_os_matrix:
-            py_config["rhel_os_matrix"] = [utilities.infra.generate_latest_os_dict(os_list=rhel_os_matrix)]
-            py_config["instance_type_rhel_os_matrix"] = [
-                utilities.infra.generate_latest_os_dict(os_list=py_config["instance_type_rhel_os_matrix"])
-            ]
-
-        if session.config.getoption("latest_windows") and windows_os_matrix:
-            py_config["windows_os_matrix"] = [utilities.infra.generate_latest_os_dict(os_list=windows_os_matrix)]
-
-        if session.config.getoption("latest_centos") and (centos_os_matrix := py_config.get("centos_os_matrix")):
-            py_config["centos_os_matrix"] = [utilities.infra.generate_latest_os_dict(os_list=centos_os_matrix)]
-
-        if session.config.getoption("latest_fedora") and (fedora_os_matrix := py_config.get("fedora_os_matrix")):
-            py_config["fedora_os_matrix"] = [utilities.infra.generate_latest_os_dict(os_list=fedora_os_matrix)]
-
     data_collector_dict = set_data_collector_values(base_dir=session.config.getoption("data_collector_output_dir"))
     shutil.rmtree(
         data_collector_dict["data_collector_base_directory"],
@@ -802,7 +778,8 @@ def _update_os_related_config():
     # with runtime storage_class_matrix value(s)
     py_config["system_storage_class_matrix"] = py_config.get("storage_class_matrix", [])
 
-    _update_os_related_config()
+    generate_os_matrix_dicts(os_dict=py_config)
+    update_latest_os_config(session_config=session.config)
 
     matrix_addoptions = [matrix for matrix in session.config.invocation_params.args if "-matrix=" in matrix]
     for matrix_addoption in matrix_addoptions:

tests/global_config_arm64.py

@@ -12,11 +12,6 @@
     Images,
     StorageClassNames,
 )
-from utilities.infra import get_latest_os_dict_list
-from utilities.os_utils import (
-    generate_linux_instance_type_os_matrix,
-    generate_os_matrix_dict,
-)
 from utilities.storage import HppCsiStorageClass
 
 global config
@@ -51,21 +46,12 @@
 storage_class_a = StorageClassNames.IO2_CSI
 storage_class_b = StorageClassNames.IO2_CSI
 
-rhel_os_matrix = generate_os_matrix_dict(os_name="rhel", supported_operating_systems=["rhel-9-5", "rhel-9-6"])
-fedora_os_matrix = generate_os_matrix_dict(os_name="fedora", supported_operating_systems=["fedora-42"])
-centos_os_matrix = generate_os_matrix_dict(os_name="centos", supported_operating_systems=["centos-stream-9"])
-
-latest_rhel_os_dict, latest_fedora_os_dict, latest_centos_os_dict = get_latest_os_dict_list(
-    os_list=[rhel_os_matrix, fedora_os_matrix, centos_os_matrix]
-)
+rhel_os_list = ["rhel-9-5", "rhel-9-6"]
+fedora_os_list = ["fedora-42"]
+centos_os_list = ["centos-stream-9"]
 
-# Modify instance_type_rhel_os_matrix for arm64
-instance_type_rhel_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name="rhel", preferences=[RHEL10_PREFERENCE], arch_suffix=ARM_64
-)
-instance_type_fedora_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name=OS_FLAVOR_FEDORA, preferences=[OS_FLAVOR_FEDORA], arch_suffix=ARM_64
-)
+instance_type_rhel_os_list = [RHEL10_PREFERENCE]
+instance_type_fedora_os_list = [OS_FLAVOR_FEDORA]
 
 for _dir in dir():
     if not config:  # noqa: F821

tests/global_config_interop.py

@@ -2,14 +2,9 @@
 
 import pytest_testconfig
 
-from utilities.infra import generate_latest_os_dict
-
 global config
 global_config = pytest_testconfig.load_python(py_file="tests/global_config.py", encoding="utf-8")
 
-fedora_os_matrix = [
-    generate_latest_os_dict(os_list=config["fedora_os_matrix"])  # noqa: F821
-]
 
 for _dir in dir():
     if not config:  # noqa: F821

tests/global_config_s390x.py

@@ -1,36 +1,28 @@
 from typing import Any
 
-import utilities.constants
 from utilities.constants import (
+    CENTOS_STREAM9_PREFERENCE,
     EXPECTED_CLUSTER_INSTANCE_TYPE_LABELS,
+    OS_FLAVOR_FEDORA,
     PREFERENCE_STR,
+    RHEL9_PREFERENCE,
     S390X,
 )
-from utilities.infra import get_latest_os_dict_list
-from utilities.os_utils import generate_linux_instance_type_os_matrix, generate_os_matrix_dict
 
 global config
 
 
 EXPECTED_CLUSTER_INSTANCE_TYPE_LABELS[PREFERENCE_STR] = f"rhel.9.{S390X}"
 
-rhel_os_matrix = generate_os_matrix_dict(
-    os_name="rhel",
-    supported_operating_systems=[
-        "rhel-8-10",
-        "rhel-9-6",
-    ],
-)
-fedora_os_matrix = generate_os_matrix_dict(os_name="fedora", supported_operating_systems=["fedora-42"])
-centos_os_matrix = generate_os_matrix_dict(os_name="centos", supported_operating_systems=["centos-stream-9"])
 
-instance_type_rhel_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name="rhel", preferences=[utilities.constants.RHEL9_PREFERENCE], arch_suffix=S390X
-)
+rhel_os_list = ["rhel-8-10", "rhel-9-6"]
+fedora_os_list = ["fedora-42"]
+centos_os_list = ["centos-stream-9"]
+
+instance_type_rhel_os_list = [RHEL9_PREFERENCE]
+instance_type_fedora_os_list = [OS_FLAVOR_FEDORA]
+instance_type_centos_os_list = [CENTOS_STREAM9_PREFERENCE]
 
-latest_rhel_os_dict, latest_fedora_os_dict, latest_centos_os_dict = get_latest_os_dict_list(
-    os_list=[rhel_os_matrix, fedora_os_matrix, centos_os_matrix]
-)
 
 for _dir in dir():
     if not config:  # noqa: F821

tests/global_config_x86_64.py

@@ -8,52 +8,17 @@
     RHEL9_PREFERENCE,
     RHEL10_PREFERENCE,
 )
-from utilities.infra import get_latest_os_dict_list
-from utilities.os_utils import (
-    generate_linux_instance_type_os_matrix,
-    generate_os_matrix_dict,
-)
 
 global config
 
-rhel_os_matrix = generate_os_matrix_dict(
-    os_name="rhel",
-    supported_operating_systems=[
-        "rhel-8-10",
-        "rhel-9-6",
-    ],
-)
-
-windows_os_matrix = generate_os_matrix_dict(
-    os_name="windows",
-    supported_operating_systems=[
-        "win-10",
-        "win-2016",
-        "win-2019",
-        "win-11",
-        "win-2022",
-        "win-2025",
-    ],
-)
-
-fedora_os_matrix = generate_os_matrix_dict(os_name="fedora", supported_operating_systems=["fedora-43"])
-centos_os_matrix = generate_os_matrix_dict(os_name="centos", supported_operating_systems=["centos-stream-9"])
-instance_type_rhel_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name="rhel", preferences=[RHEL8_PREFERENCE, RHEL9_PREFERENCE, RHEL10_PREFERENCE]
-)
-instance_type_centos_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name="centos.stream", preferences=[CENTOS_STREAM9_PREFERENCE, CENTOS_STREAM10_PREFERENCE]
-)
-instance_type_fedora_os_matrix = generate_linux_instance_type_os_matrix(
-    os_name=OS_FLAVOR_FEDORA, preferences=[OS_FLAVOR_FEDORA]
-)
+rhel_os_list = ["rhel-8-10", "rhel-9-6"]
+windows_os_list = ["win-10", "win-2016", "win-2019", "win-11", "win-2022", "win-2025"]
+fedora_os_list = ["fedora-43"]
+centos_os_list = ["centos-stream-9"]
 
-(
-    latest_rhel_os_dict,
-    latest_windows_os_dict,
-    latest_fedora_os_dict,
-    latest_centos_os_dict,
-) = get_latest_os_dict_list(os_list=[rhel_os_matrix, windows_os_matrix, fedora_os_matrix, centos_os_matrix])
+instance_type_rhel_os_list = [RHEL8_PREFERENCE, RHEL9_PREFERENCE, RHEL10_PREFERENCE]
+instance_type_centos_os_list = [CENTOS_STREAM9_PREFERENCE, CENTOS_STREAM10_PREFERENCE]
+instance_type_fedora_os_list = [OS_FLAVOR_FEDORA]
 
 
 for _dir in dir():

utilities/infra.py

@@ -70,7 +70,6 @@
     NamespacesNames,
 )
 from utilities.exceptions import (
-    OsDictNotFoundError,
     UrlNotFoundError,
     UtilityPodNotFoundError,
 )
@@ -165,44 +164,6 @@ def generate_namespace_name(file_path):
     return (file_path.strip(".py").replace("/", "-").replace("_", "-"))[-63:].split("-", 1)[-1]
 
 
-def generate_latest_os_dict(os_list):
-    """
-    Get latest os dict.
-
-    Args:
-        os_list (list): [<os-name>]_os_matrix - a list of dicts.
-
-    Returns:
-        dict: {Latest OS name: latest supported OS dict} else raises an exception.
-
-    Raises:
-        OsDictNotFoundError: If no os matched.
-    """
-    for os_dict in os_list:
-        for os_version, os_values in os_dict.items():
-            if os_values.get("latest_released"):
-                return {os_version: os_values}
-
-    raise OsDictNotFoundError(f"No OS is marked as 'latest_released': {os_list}")
-
-
-def get_latest_os_dict_list(os_list):
-    """
-    Get latest os dict generated by 'generate_latest_os_dict()'
-    This will extract the dict from `generate_latest_os_dict()` without the name key.
-
-    Args:
-        os_list (list): [rhel|windows|fedora]_os_matrix - a list of dicts
-
-    Returns:
-        list: List of oses dict [{latest supported OS dict}]
-    """
-    res = []
-    for _os in os_list:
-        res.append(list(generate_latest_os_dict(os_list=_os).values())[0])
-    return res
-
-
 def get_pods(client: DynamicClient, namespace: Namespace, label: str = "") -> list[Pod]:
     return list(
         Pod.get(

utilities/os_utils.py

@@ -27,6 +27,7 @@
     WORKLOAD_STR,
     Images,
 )
+from utilities.exceptions import OsDictNotFoundError
 
 LOGGER = logging.getLogger(__name__)
 
@@ -294,3 +295,24 @@ def _format_data_source_name(preference_name: str) -> str:
 
         instance_types.append({arch_preference: preference_config})
     return instance_types
+
+
+def generate_latest_os_dict(os_matrix: list[dict[str, Any]]) -> dict[str, Any]:
+    """
+    Get latest os dict.
+
+    Args:
+        os_matrix (list): [<os-name>]_os_matrix - a list of dicts.
+
+    Returns:
+        dict: Latest supported OS dict (the os_values payload) or raises an exception.
+
+    Raises:
+        OsDictNotFoundError: If no os matched.
+    """
+    for matrix in os_matrix:
+        for os_values in matrix.values():
+            if os_values.get(LATEST_RELEASE_STR):
+                return os_values
+
+    raise OsDictNotFoundError(f"No OS is marked as '{LATEST_RELEASE_STR}': {os_matrix}")