Manual cherry-pick 4.21: Drop cluster net stack fixtures (#4365)

Both ipv4_supported_cluster and ipv6_supported_cluster fixtures are used
frequently as parameters.
For cleaner data usage, these fixtures can be removed and replaced with
cached helper functions that retrieve the cluster network stack once.

Signed-off-by: Asia Khromov <azhivovk@redhat.com>

Author: azhivovk

conftest.py

@@ -21,6 +21,7 @@
 from _pytest.reports import CollectReport, TestReport
 from _pytest.runner import CallInfo
 from kubernetes.dynamic.exceptions import ConflictError
+from ocp_resources.network_config_openshift_io import Network
 from pyhelper_utils.shell import run_command
 from pytest import Item
 from pytest_testconfig import config as py_config
@@ -830,6 +831,9 @@ def _update_os_related_config():
     if not skip_if_pytest_flags_exists(pytest_config=session.config):
         admin_client = utilities.cluster.cache_admin_client()
         py_config["version_explorer_url"] = get_cnv_version_explorer_url(pytest_config=session.config)
+        py_config["cluster_service_network"] = Network(
+            client=admin_client, name="cluster"
+        ).instance.status.serviceNetwork
         if not session.config.getoption("--skip-artifactory-check"):
             py_config["server_url"] = py_config["server_url"] or get_artifactory_server_url(
                 cluster_host_url=admin_client.configuration.host, session=session

libs/net/cluster.py

@@ -0,0 +1,31 @@
+import ipaddress
+import logging
+from functools import cache
+
+from pytest_testconfig import py_config
+
+LOGGER = logging.getLogger(__name__)
+
+
+@cache
+def is_ipv6_single_stack_cluster() -> bool:
+    ipv4_supported = ipv4_supported_cluster()
+    ipv6_supported = ipv6_supported_cluster()
+
+    is_ipv6_only = ipv6_supported and not ipv4_supported
+    LOGGER.info(f"Cluster network detection: IPv4={ipv4_supported}, IPv6={ipv6_supported}, IPv6-only={is_ipv6_only}")
+    return is_ipv6_only
+
+
+@cache
+def ipv4_supported_cluster() -> bool:
+    return _cluster_ip_family_supported(ip_family=4)
+
+
+@cache
+def ipv6_supported_cluster() -> bool:
+    return _cluster_ip_family_supported(ip_family=6)
+
+
+def _cluster_ip_family_supported(ip_family: int) -> bool:
+    return any(ipaddress.ip_network(ip).version == ip_family for ip in py_config.get("cluster_service_network"))

tests/conftest.py

@@ -3,7 +3,6 @@
 """
 
 import copy
-import ipaddress
 import logging
 import os
 import os.path
@@ -42,7 +41,6 @@
 from ocp_resources.mutating_webhook_config import MutatingWebhookConfiguration
 from ocp_resources.namespace import Namespace
 from ocp_resources.network_addons_config import NetworkAddonsConfig
-from ocp_resources.network_config_openshift_io import Network
 from ocp_resources.node import Node
 from ocp_resources.node_network_state import NodeNetworkState
 from ocp_resources.oauth import OAuth
@@ -71,6 +69,7 @@
 from timeout_sampler import TimeoutSampler
 
 import utilities.hco
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from tests.utils import download_and_extract_tar
 from utilities.artifactory import get_artifactory_header, get_http_image_url, get_test_artifact_server_url
 from utilities.bitwarden import get_cnv_tests_secret_by_name
@@ -1437,8 +1436,6 @@ def cluster_info(
     hco_image,
     ocs_current_version,
     kubevirt_resource_scope_session,
-    ipv6_supported_cluster,
-    ipv4_supported_cluster,
     workers_type,
     nodes_cpu_architecture,
 ):
@@ -1458,8 +1455,8 @@ def cluster_info(
         f"\tCNI type: {get_cluster_cni_type(admin_client=admin_client)}\n"
         f"\tWorkers type: {workers_type}\n"
         f"\tCluster CPU Architecture: {nodes_cpu_architecture}\n"
-        f"\tIPv4 cluster: {ipv4_supported_cluster}\n"
-        f"\tIPv6 cluster: {ipv6_supported_cluster}\n"
+        f"\tIPv4 cluster: {ipv4_supported_cluster()}\n"
+        f"\tIPv6 cluster: {ipv6_supported_cluster()}\n"
         f"\tVirtctl version: \n\t{virtctl_client_version}\n\t{virtctl_server_version}\n"
     )
 
@@ -1794,23 +1791,6 @@ def ssp_resource_scope_function(admin_client, hco_namespace):
     return get_ssp_resource(admin_client=admin_client, namespace=hco_namespace)
 
 
-@pytest.fixture(scope="session")
-def cluster_service_network(admin_client):
-    return Network(client=admin_client, name="cluster").instance.status.serviceNetwork
-
-
-@pytest.fixture(scope="session")
-def ipv4_supported_cluster(cluster_service_network):
-    if cluster_service_network:
-        return any([ipaddress.ip_network(ip).version == 4 for ip in cluster_service_network])
-
-
-@pytest.fixture(scope="session")
-def ipv6_supported_cluster(cluster_service_network):
-    if cluster_service_network:
-        return any([ipaddress.ip_network(ip).version == 6 for ip in cluster_service_network])
-
-
 @pytest.fixture()
 def disabled_common_boot_image_import_hco_spec_scope_function(
     admin_client,
@@ -2574,11 +2554,6 @@ def nmstate_namespace(admin_client):
         return None
 
 
-@pytest.fixture()
-def ipv6_single_stack_cluster(ipv4_supported_cluster, ipv6_supported_cluster):
-    return ipv6_supported_cluster and not ipv4_supported_cluster
-
-
 @pytest.fixture(scope="class")
 def ping_process_in_rhel_os():
     def _start_ping(vm):

tests/network/conftest.py

@@ -16,6 +16,7 @@
 from pytest_testconfig import config as py_config
 from timeout_sampler import TimeoutExpiredError
 
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from tests.network.utils import get_vlan_index_number
 from utilities.constants import (
     CLUSTER,
@@ -59,22 +60,15 @@ def virt_handler_pod(admin_client):
     raise ResourceNotFoundError(f"No {VIRT_HANDLER} Pod found.")
 
 
-@pytest.fixture(scope="session")
-def dual_stack_cluster(ipv4_supported_cluster, ipv6_supported_cluster):
-    return ipv4_supported_cluster and ipv6_supported_cluster
-
-
 @pytest.fixture(scope="module")
-def ipv6_primary_interface_cloud_init_data(
-    ipv4_supported_cluster: bool, ipv6_supported_cluster: bool
-) -> dict[str, dict] | None:
-    if ipv6_supported_cluster:
+def ipv6_primary_interface_cloud_init_data() -> dict[str, dict] | None:
+    if ipv6_supported_cluster():
         return {
             "ethernets": {
                 "eth0": {
                     "addresses": ["fd10:0:2::2/120"],
                     "gateway6": "fd10:0:2::1",
-                    "dhcp4": ipv4_supported_cluster,
+                    "dhcp4": ipv4_supported_cluster(),
                     "dhcp6": False,
                 },
             },
@@ -174,8 +168,6 @@ def network_sanity(
     cluster_network_mtu,
     network_overhead,
     sriov_workers,
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
     conformance_tests,
     nmstate_namespace,
     mtv_namespace_scope_session,
@@ -313,8 +305,8 @@ def _verify_mtv_installed():
     _verify_service_mesh()
     _verify_jumbo_frame()
     _verify_sriov()
-    _verify_ip_family(family="ipv4", is_supported_in_cluster=ipv4_supported_cluster)
-    _verify_ip_family(family="ipv6", is_supported_in_cluster=ipv6_supported_cluster)
+    _verify_ip_family(family="ipv4", is_supported_in_cluster=ipv4_supported_cluster())
+    _verify_ip_family(family="ipv6", is_supported_in_cluster=ipv6_supported_cluster())
     _verify_nmstate_running_pods(_admin_client=admin_client, namespace=nmstate_namespace)
     _verify_mtv_installed()
 

tests/network/connectivity/conftest.py

@@ -1,5 +1,6 @@
 import pytest
 
+from libs.net.cluster import ipv6_supported_cluster
 from tests.network.connectivity.utils import create_running_vm
 from utilities.constants import LINUX_BRIDGE, OVS_BRIDGE
 from utilities.data_utils import name_prefix
@@ -28,8 +29,8 @@ def vlan_id_3(vlan_index_number):
 
 
 @pytest.fixture()
-def fail_if_not_ipv6_supported_cluster(ipv6_supported_cluster):
-    if not ipv6_supported_cluster:
+def fail_if_not_ipv6_supported_cluster():
+    if not ipv6_supported_cluster():
         pytest.fail(reason="IPv6 is not supported in this cluster")
 
 

tests/network/general/test_ip_family_services.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from tests.network.utils import basic_expose_command, get_service
 from utilities.constants import SSH_PORT_22
 from utilities.infra import get_node_selector_dict, run_virtctl_command
@@ -75,10 +76,11 @@ def virtctl_expose_service(
     request,
     unprivileged_client,
     running_vm_for_exposure,
-    dual_stack_cluster,
 ):
     ip_family_policy = request.param
-    if ip_family_policy == SERVICE_IP_FAMILY_POLICY_REQUIRE_DUAL_STACK and not dual_stack_cluster:
+    if ip_family_policy == SERVICE_IP_FAMILY_POLICY_REQUIRE_DUAL_STACK and not (
+        ipv4_supported_cluster() and ipv6_supported_cluster()
+    ):
         pytest.skip(
             f"{SERVICE_IP_FAMILY_POLICY_REQUIRE_DUAL_STACK} service cannot be created in a non-dual-stack cluster."
         )
@@ -98,9 +100,11 @@ def virtctl_expose_service(
 
 
 @pytest.fixture()
-def expected_num_families_in_service(request, dual_stack_cluster):
+def expected_num_families_in_service(request):
     ip_family_policy = request.param
-    if ip_family_policy != SERVICE_IP_FAMILY_POLICY_SINGLE_STACK and dual_stack_cluster:
+    if ip_family_policy != SERVICE_IP_FAMILY_POLICY_SINGLE_STACK and (
+        ipv4_supported_cluster() and ipv6_supported_cluster()
+    ):
         return 2
     return 1
 
@@ -165,7 +169,6 @@ def test_vitrctl_expose_services(
         expected_num_families_in_service,
         running_vm_for_exposure,
         virtctl_expose_service,
-        dual_stack_cluster,
         ip_family_policy,
     ):
         assert_svc_ip_params(

tests/network/l2_bridge/vmi_interfaces_stability/conftest.py

@@ -16,8 +16,6 @@
 
 @pytest.fixture(scope="class")
 def running_linux_bridge_vm(
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
     unprivileged_client: DynamicClient,
     namespace: Namespace,
     bridge_nad: NetworkAttachmentDefinition,
@@ -27,8 +25,6 @@ def running_linux_bridge_vm(
         name="vm-iface-stability",
         client=unprivileged_client,
         bridge_network_name=bridge_nad.name,
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
     ) as vm:
         vm.start(wait=True)
         vm.wait_for_agent_connected()

tests/network/l2_bridge/vmi_interfaces_stability/lib_helpers.py

@@ -7,6 +7,7 @@
 from kubernetes.dynamic.resource import ResourceField
 from ocp_resources.virtual_machine_instance import VirtualMachineInstance
 
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from libs.net.ip import random_ipv4_address, random_ipv6_address
 from libs.net.vmspec import lookup_iface_status, lookup_primary_network
 from libs.vm.factory import base_vmspec, fedora_vm
@@ -26,8 +27,6 @@ def secondary_network_vm(
     name: str,
     client: DynamicClient,
     bridge_network_name: str,
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
 ) -> BaseVirtualMachine:
     spec = base_vmspec()
     spec.template.spec.domain.devices.interfaces = [  # type: ignore
@@ -42,24 +41,13 @@ def secondary_network_vm(
     ]
 
     ethernets = {}
-    primary = primary_iface_cloud_init(
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
-    )
+    primary = primary_iface_cloud_init()
     if primary:
         ethernets["eth1"] = primary
 
-    ethernets["eth0"] = secondary_iface_cloud_init(
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
-        host_address=1,
-    )
+    ethernets["eth0"] = secondary_iface_cloud_init(host_address=1)
 
-    ethernets["eth2"] = secondary_iface_cloud_init(
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
-        host_address=2,
-    )
+    ethernets["eth2"] = secondary_iface_cloud_init(host_address=2)
 
     userdata = cloudinit.UserData(users=[])
     disk, volume = cloudinitdisk_storage(
@@ -73,50 +61,33 @@ def secondary_network_vm(
     return fedora_vm(namespace=namespace, name=name, client=client, spec=spec)
 
 
-def primary_iface_cloud_init(
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
-) -> cloudinit.EthernetDevice | None:
-    if not ipv6_supported_cluster:
+def primary_iface_cloud_init() -> cloudinit.EthernetDevice | None:
+    if not ipv6_supported_cluster():
         return None
     return cloudinit.EthernetDevice(
         addresses=["fd10:0:2::2/120"],
         gateway6="fd10:0:2::1",
-        dhcp4=ipv4_supported_cluster,
+        dhcp4=ipv4_supported_cluster(),
         dhcp6=False,
     )
 
 
-def secondary_iface_cloud_init(
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
-    host_address: int,
-) -> cloudinit.EthernetDevice:
-    ips = secondary_iface_ips(
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
-        host_address=host_address,
-    )
+def secondary_iface_cloud_init(host_address: int) -> cloudinit.EthernetDevice:
+    ips = secondary_iface_ips(host_address=host_address)
     addresses = [f"{ip}/64" if ipaddress.ip_address(ip).version == 6 else f"{ip}/24" for ip in ips]
     return cloudinit.EthernetDevice(addresses=addresses)
 
 
-def secondary_iface_ips(
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
-    host_address: int,
-) -> list[str]:
+def secondary_iface_ips(host_address: int) -> list[str]:
     ips = []
-    if ipv4_supported_cluster:
+    if ipv4_supported_cluster():
         ips.append(random_ipv4_address(net_seed=0, host_address=host_address))
-    if ipv6_supported_cluster:
+    if ipv6_supported_cluster():
         ips.append(random_ipv6_address(net_seed=0, host_address=host_address))
     return ips
 
 
-def wait_for_stable_ifaces(
-    vm: BaseVirtualMachine,
-) -> None:
+def wait_for_stable_ifaces(vm: BaseVirtualMachine) -> None:
     primary_network = lookup_primary_network(vm=vm)
 
     secondary_iface_to_ips = {

tests/observability/metrics/test_general_metrics.py

@@ -4,6 +4,7 @@
 from ocp_resources.resource import Resource
 from ocp_resources.virtual_machine import VirtualMachine
 
+from libs.net.cluster import is_ipv6_single_stack_cluster
 from tests.observability.metrics.constants import KUBEVIRT_VMI_NODE_CPU_AFFINITY
 from tests.observability.metrics.utils import validate_vmi_node_cpu_affinity_with_prometheus
 from tests.observability.utils import validate_metrics_value
@@ -90,9 +91,9 @@ class TestVirtHCOSingleStackIpv6:
     @pytest.mark.ipv6
     @pytest.mark.polarion("CNV-11740")
     @pytest.mark.s390x
-    def test_metric_kubevirt_hco_single_stack_ipv6(self, prometheus, ipv6_single_stack_cluster):
+    def test_metric_kubevirt_hco_single_stack_ipv6(self, prometheus):
         validate_metrics_value(
             prometheus=prometheus,
             metric_name="kubevirt_hco_single_stack_ipv6",
-            expected_value="1" if ipv6_single_stack_cluster else "0",
+            expected_value="1" if is_ipv6_single_stack_cluster() else "0",
         )