net: Drop cluster net stack fixtures

libs/net/cluster.py

@@ -9,13 +9,23 @@
 
 @cache
 def is_ipv6_single_stack_cluster() -> bool:
-    ipv4_supported = cluster_ip_family_supported(ip_family=4)
-    ipv6_supported = cluster_ip_family_supported(ip_family=6)
+    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
 
 
-def cluster_ip_family_supported(ip_family: int) -> bool:
+@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"))

libs/net/ip.py

@@ -3,6 +3,8 @@
 from functools import cache
 from typing import Final
 
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
+
 _MAX_NUM_OF_RANDOM_OCTETS_PER_SESSION: Final[int] = 16
 _MAX_NUM_OF_RANDOM_HEXTETS_PER_SESSION: Final[int] = 16
 _IPV4_ADDRESS_SUBNET_PREFIX_VMI: Final[str] = "172.16"
@@ -80,26 +82,22 @@ def _random_hextets(count: int) -> list[int]:
 
 
 def random_ip_addresses_by_family(
-    ipv4_supported: bool,
-    ipv6_supported: bool,
     net_seed: int,
     host_address: int,
 ) -> list[str]:
-    """Generate IP addresses for each supported IP family.
+    """Generate IP addresses for each IP family supported by the cluster network stack.
 
     Args:
-        ipv4_supported: Whether IPv4 is supported.
-        ipv6_supported: Whether IPv6 is supported.
         net_seed: Seed index for selecting the random network portion of the address.
         host_address: Host portion of the address, used to place VMs on the same subnet.
 
     Returns:
-        List of IP address strings, one per supported IP family.
+        List of IP address strings, one per IP family supported by the cluster.
     """
     ips = []
-    if ipv4_supported:
+    if ipv4_supported_cluster():
         ips.append(random_ipv4_address(net_seed=net_seed, host_address=host_address))
-    if ipv6_supported:
+    if ipv6_supported_cluster():
         ips.append(random_ipv6_address(net_seed=net_seed, host_address=host_address))
     return ips
 

tests/conftest.py

@@ -3,7 +3,6 @@
 """
 
 import copy
-import ipaddress
 import logging
 import os
 import os.path
@@ -44,7 +43,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
@@ -73,6 +71,7 @@
 from timeout_sampler import TimeoutSampler
 
 import utilities.hco
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from libs.net.ip import filter_link_local_addresses, random_ipv4_address, random_ipv6_address
 from libs.net.vmspec import lookup_iface_status
 from tests.utils import download_and_extract_tar
@@ -1470,8 +1469,6 @@ def cluster_info(
     hco_image,
     ocs_current_version,
     kubevirt_resource_scope_session,
-    ipv6_supported_cluster,
-    ipv4_supported_cluster,
     workers_type,
     nodes_cpu_architecture,
 ):
@@ -1491,8 +1488,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"
     )
 
@@ -1656,8 +1653,6 @@ def running_vm_upgrade_a(
     upgrade_bridge_marker_nad,
     kmp_enabled_namespace,
     upgrade_br1test_nad,
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
 ):
     name = "vm-upgrade-a"
     cloud_init_data = cloud_init_network_data(
@@ -1684,7 +1679,7 @@ def running_vm_upgrade_a(
     ) as vm:
         running_vm(vm=vm, wait_for_cloud_init=True)
         ip_families = [
-            family for family, enabled in ((4, ipv4_supported_cluster), (6, ipv6_supported_cluster)) if enabled
+            family for family, enabled in ((4, ipv4_supported_cluster()), (6, ipv6_supported_cluster())) if enabled
         ]
         lookup_iface_status(
             vm=vm,
@@ -1702,8 +1697,6 @@ def running_vm_upgrade_b(
     upgrade_bridge_marker_nad,
     kmp_enabled_namespace,
     upgrade_br1test_nad,
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
 ):
     name = "vm-upgrade-b"
     cloud_init_data = cloud_init_network_data(
@@ -1730,7 +1723,7 @@ def running_vm_upgrade_b(
     ) as vm:
         running_vm(vm=vm, wait_for_cloud_init=True)
         ip_families = [
-            family for family, enabled in ((4, ipv4_supported_cluster), (6, ipv6_supported_cluster)) if enabled
+            family for family, enabled in ((4, ipv4_supported_cluster()), (6, ipv6_supported_cluster())) if enabled
         ]
         lookup_iface_status(
             vm=vm,
@@ -1875,23 +1868,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,
@@ -2715,11 +2691,6 @@ def nmstate_dependent_placeholder():
     return
 
 
-@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,
                 },
             },
@@ -173,8 +167,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,
@@ -315,8 +307,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

@@ -261,8 +261,6 @@ def nad_ovs_bridge_vlan_3(
 
 @pytest.fixture(scope="class")
 def vm_linux_bridge_attached_vma_source(
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
     worker_node1,
     namespace,
     unprivileged_client,
@@ -279,11 +277,7 @@ def vm_linux_bridge_attached_vma_source(
 
     cloud_init_data = compose_cloud_init_data_dict(
         ipv6_network_data=ipv6_primary_interface_cloud_init_data,
-        network_data=secondary_interfaces_cloud_init_data(
-            ipv4_supported_cluster=ipv4_supported_cluster,
-            ipv6_supported_cluster=ipv6_supported_cluster,
-            host_id=1,
-        ),
+        network_data=secondary_interfaces_cloud_init_data(host_id=1),
     )
 
     yield from create_running_vm(
@@ -298,8 +292,6 @@ def vm_linux_bridge_attached_vma_source(
 
 @pytest.fixture(scope="class")
 def vm_ovs_bridge_attached_vma_source(
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
     worker_node1,
     namespace,
     unprivileged_client,
@@ -316,11 +308,7 @@ def vm_ovs_bridge_attached_vma_source(
 
     cloud_init_data = compose_cloud_init_data_dict(
         ipv6_network_data=ipv6_primary_interface_cloud_init_data,
-        network_data=secondary_interfaces_cloud_init_data(
-            ipv4_supported_cluster=ipv4_supported_cluster,
-            ipv6_supported_cluster=ipv6_supported_cluster,
-            host_id=1,
-        ),
+        network_data=secondary_interfaces_cloud_init_data(host_id=1),
     )
 
     yield from create_running_vm(
@@ -335,8 +323,6 @@ def vm_ovs_bridge_attached_vma_source(
 
 @pytest.fixture(scope="class")
 def vm_linux_bridge_attached_vmb_destination(
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
     worker_node2,
     namespace,
     unprivileged_client,
@@ -353,11 +339,7 @@ def vm_linux_bridge_attached_vmb_destination(
 
     cloud_init_data = compose_cloud_init_data_dict(
         ipv6_network_data=ipv6_primary_interface_cloud_init_data,
-        network_data=secondary_interfaces_cloud_init_data(
-            ipv4_supported_cluster=ipv4_supported_cluster,
-            ipv6_supported_cluster=ipv6_supported_cluster,
-            host_id=2,
-        ),
+        network_data=secondary_interfaces_cloud_init_data(host_id=2),
     )
 
     yield from create_running_vm(
@@ -372,8 +354,6 @@ def vm_linux_bridge_attached_vmb_destination(
 
 @pytest.fixture(scope="class")
 def vm_ovs_bridge_attached_vmb_destination(
-    ipv4_supported_cluster,
-    ipv6_supported_cluster,
     worker_node2,
     namespace,
     unprivileged_client,
@@ -390,11 +370,7 @@ def vm_ovs_bridge_attached_vmb_destination(
 
     cloud_init_data = compose_cloud_init_data_dict(
         ipv6_network_data=ipv6_primary_interface_cloud_init_data,
-        network_data=secondary_interfaces_cloud_init_data(
-            ipv4_supported_cluster=ipv4_supported_cluster,
-            ipv6_supported_cluster=ipv6_supported_cluster,
-            host_id=2,
-        ),
+        network_data=secondary_interfaces_cloud_init_data(host_id=2),
     )
 
     yield from create_running_vm(

tests/network/connectivity/utils.py

@@ -1,5 +1,6 @@
 from collections import OrderedDict
 
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
 from libs.net.ip import random_ipv4_address, random_ipv6_address
 from utilities.virt import VirtualMachineForTests, fedora_vm_body
 
@@ -33,17 +34,15 @@ def create_running_vm(
 
 
 def secondary_interfaces_cloud_init_data(
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
     host_id: int,
 ) -> dict[str, dict[str, dict[str, list[str]]]]:
     ethernets = {}
     for i in range(3):
         interface_name = f"eth{i + 1}"
         addresses = []
-        if ipv4_supported_cluster:
+        if ipv4_supported_cluster():
             addresses.append(f"{random_ipv4_address(net_seed=i, host_address=host_id)}/24")
-        if ipv6_supported_cluster:
+        if ipv6_supported_cluster():
             addresses.append(f"{random_ipv6_address(net_seed=i, host_address=host_id)}/64")
 
         ethernets[interface_name] = {"addresses": addresses}