net: Filter IPv4 on IPv6-only clusters

libs/net/cluster.py

@@ -7,6 +7,12 @@
 LOGGER = logging.getLogger(__name__)
 
 
+@cache
+def supported_cluster_ip_versions() -> set[int]:
+    """Return the set of IP versions (4, 6) supported by the cluster."""
+    return {version for version, enabled in ((4, ipv4_supported_cluster()), (6, ipv6_supported_cluster())) if enabled}
+
+
 @cache
 def is_ipv6_single_stack_cluster() -> bool:
     ipv4_supported = ipv4_supported_cluster()

libs/net/ip.py

@@ -3,7 +3,7 @@
 from functools import cache
 from typing import Final
 
-from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster, supported_cluster_ip_versions
 
 _MAX_NUM_OF_RANDOM_OCTETS_PER_SESSION: Final[int] = 16
 _MAX_NUM_OF_RANDOM_HEXTETS_PER_SESSION: Final[int] = 16
@@ -139,6 +139,20 @@ def filter_link_local_addresses(ip_addresses: list[str]) -> list[ipaddress.IPv4A
     return [ip for addr in ip_addresses if not (ip := ipaddress.ip_interface(address=addr).ip).is_link_local]
 
 
+def filter_cluster_unsupported_addresses(
+    ip_addresses: list[ipaddress.IPv4Address | ipaddress.IPv6Address],
+) -> list[ipaddress.IPv4Address | ipaddress.IPv6Address]:
+    """Filter out IP addresses whose family is not supported by the cluster.
+
+    Args:
+        ip_addresses: List of IP address objects to filter.
+
+    Returns:
+        List containing only addresses whose IP version is supported by the cluster.
+    """
+    return [ip for ip in ip_addresses if ip.version in supported_cluster_ip_versions()]
+
+
 def ip_header_size(ip: ipaddress.IPv4Address | ipaddress.IPv6Address | str) -> int:
     addr = ipaddress.ip_address(ip) if isinstance(ip, str) else ip
     return _IPV4_HEADER_SIZE if addr.version == 4 else _IPV6_HEADER_SIZE

tests/conftest.py

@@ -70,7 +70,7 @@
 from timeout_sampler import TimeoutSampler
 
 import utilities.hco
-from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
+from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster, supported_cluster_ip_versions
 from libs.net.ip import filter_link_local_addresses, random_cidr_addresses_by_family
 from libs.net.vmspec import lookup_iface_status
 from tests.utils import download_and_extract_tar
@@ -1695,9 +1695,7 @@ def running_vm_upgrade_a(
         eviction_strategy=ES_NONE,
     ) 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
-        ]
+        ip_families = supported_cluster_ip_versions()
         lookup_iface_status(
             vm=vm,
             iface_name=upgrade_bridge_marker_nad.name,
@@ -1730,9 +1728,7 @@ def running_vm_upgrade_b(
         eviction_strategy=ES_NONE,
     ) 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
-        ]
+        ip_families = supported_cluster_ip_versions()
         lookup_iface_status(
             vm=vm,
             iface_name=upgrade_bridge_marker_nad.name,

tests/network/localnet/conftest.py

@@ -5,8 +5,13 @@
 from ocp_resources.namespace import Namespace
 
 from libs.net import nodenetworkconfigurationpolicy as libnncp
-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.cluster import supported_cluster_ip_versions
+from libs.net.ip import (
+    filter_cluster_unsupported_addresses,
+    filter_link_local_addresses,
+    random_ipv4_address,
+    random_ipv6_address,
+)
 from libs.net.traffic_generator import TcpServer, VMTcpClient, active_tcp_connections
 from libs.net.vmspec import lookup_iface_status
 from libs.vm.oper import run_vms
@@ -210,15 +215,18 @@ def localnet_running_vms(
     vm_localnet_2: BaseVirtualMachine,
 ) -> tuple[BaseVirtualMachine, BaseVirtualMachine]:
     vm1, vm2 = run_vms(vms=(vm_localnet_1, vm_localnet_2))
-    ip_families = [
-        ip_family for ip_family, enabled in ((4, ipv4_supported_cluster()), (6, ipv6_supported_cluster())) if enabled
-    ]
+    ip_families = supported_cluster_ip_versions()
     for vm in (vm1, vm2):
         lookup_iface_status(
             vm=vm,
             iface_name=LOCALNET_BR_EX_INTERFACE,
             predicate=lambda interface: (
-                len(filter_link_local_addresses(ip_addresses=interface.get("ipAddresses", []))) == len(ip_families)
+                len(
+                    filter_cluster_unsupported_addresses(
+                        ip_addresses=filter_link_local_addresses(ip_addresses=interface.get("ipAddresses", []))
+                    )
+                )
+                == len(ip_families)
             ),
         )
     return vm1, vm2
@@ -355,15 +363,18 @@ def ovs_bridge_localnet_running_vms(
     vm_ovs_bridge_localnet_2: BaseVirtualMachine,
 ) -> Generator[tuple[BaseVirtualMachine, BaseVirtualMachine]]:
     vm1, vm2 = run_vms(vms=(vm_ovs_bridge_localnet_1, vm_ovs_bridge_localnet_2))
-    ip_families = [
-        ip_family for ip_family, enabled in ((4, ipv4_supported_cluster()), (6, ipv6_supported_cluster())) if enabled
-    ]
+    ip_families = supported_cluster_ip_versions()
     for vm in (vm1, vm2):
         lookup_iface_status(
             vm=vm,
             iface_name=LOCALNET_OVS_BRIDGE_INTERFACE,
             predicate=lambda interface: (
-                len(filter_link_local_addresses(ip_addresses=interface.get("ipAddresses", []))) == len(ip_families)
+                len(
+                    filter_cluster_unsupported_addresses(
+                        ip_addresses=filter_link_local_addresses(ip_addresses=interface.get("ipAddresses", []))
+                    )
+                )
+                == len(ip_families)
             ),
         )
     yield vm1, vm2

tests/network/localnet/test_default_bridge.py

@@ -6,7 +6,7 @@
 
 import pytest
 
-from libs.net.ip import filter_link_local_addresses, have_same_ip_families
+from libs.net.ip import filter_cluster_unsupported_addresses, filter_link_local_addresses, have_same_ip_families
 from libs.net.traffic_generator import client_server_active_connection, is_tcp_connection
 from libs.net.vmspec import lookup_iface_status
 from tests.network.localnet.liblocalnet import (
@@ -65,7 +65,6 @@ def test_connectivity_post_migration_between_localnet_vms(
                 assert is_tcp_connection(server=server, client=client)
 
 
-@pytest.mark.jira("CNV-90600", run=False)
 @pytest.mark.single_nic
 @pytest.mark.s390x
 @pytest.mark.usefixtures("nncp_localnet")
@@ -86,16 +85,20 @@ def test_vmi_reports_ip_on_secondary_interface_without_vlan(
         vm=vm,
         iface_name=LOCALNET_BR_EX_INTERFACE_NO_VLAN,
         predicate=lambda interface: have_same_ip_families(
-            actual_ips=filter_link_local_addresses(
-                ip_addresses=[str(ip_interface(addr).ip) for addr in interface["ipAddresses"]]
+            actual_ips=filter_cluster_unsupported_addresses(
+                ip_addresses=filter_link_local_addresses(
+                    ip_addresses=[str(ip_interface(addr).ip) for addr in interface["ipAddresses"]]
+                )
             ),
             expected_ips=expected_ips,
         ),
     )
-    reported_ips = filter_link_local_addresses(
-        ip_addresses=[str(ip_interface(addr).ip) for addr in iface_status.ipAddresses]
+    reported_ips = filter_cluster_unsupported_addresses(
+        ip_addresses=filter_link_local_addresses(
+            ip_addresses=[str(ip_interface(addr).ip) for addr in iface_status.ipAddresses]
+        )
     )
     assert set(reported_ips) == set(expected_ips), (
-        f"IP addresses mismatch for interface {LOCALNET_BR_EX_INTERFACE_NO_VLAN} on VM {vm.name},"
+        f"IP addresses mismatch for interface {LOCALNET_BR_EX_INTERFACE_NO_VLAN} on VM {vm.name}, "
         f"Reported: {reported_ips}, Expected: {expected_ips}"
     )