4.20 manual cherry-pick: Explicit ip family lookup RedHatQE#3619 (RedHatQE#3796)

4.20 manual cherry-pick: Explicit ip family lookup
[RedHatQE#3619](RedHatQE#3619)
The backporting requires the addition of helpers which were created on
cnv-4.21 branch in `libs/net/vmspec.py`

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

Author: azhivovk

libs/net/vmspec.py

@@ -1,11 +1,14 @@
+import ipaddress
 from collections.abc import Callable
 from typing import Any, Final
 
 from kubernetes.dynamic.client import ResourceField
+from ocp_resources.virtual_machine import VirtualMachine
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler, retry
 
 from libs.vm.spec import Devices, Interface, Network, SpecDisk, VMISpec, Volume
 from libs.vm.vm import BaseVirtualMachine
+from utilities.network import IpNotFound
 
 LOOKUP_IFACE_STATUS_TIMEOUT_SEC: Final[int] = 30
 WAIT_FOR_MISSING_IFACE_STATUS_TIMEOUT_SEC: Final[int] = 120
@@ -140,3 +143,42 @@ def add_volume_disk(vmi_spec: VMISpec, volume: Volume, disk: SpecDisk) -> VMISpe
     vmi_spec.domain.devices.disks = vmi_spec.domain.devices.disks or []
     vmi_spec.domain.devices.disks.append(disk)
     return vmi_spec
+
+
+def lookup_iface_status_ip(
+    vm: VirtualMachine, iface_name: str, ip_family: int
+) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    """
+    Return the IP address of the specified family for a VM interface.
+
+    Args:
+        vm: The virtual machine to query.
+        iface_name: The name of the network interface.
+        ip_family: The IP version (4 for IPv4, 6 for IPv6).
+
+    Returns:
+        The IP address matching the specified family.
+
+    Raises:
+        IpNotFound: If no IP address of the specified family is found.
+    """
+    try:
+        iface = lookup_iface_status(
+            vm=vm,
+            iface_name=iface_name,
+            predicate=lambda iface_status: bool(
+                _lookup_first_ip_address(ip_addresses=iface_status.get("ipAddresses", []), ip_family=ip_family)
+            ),
+            timeout=120,
+        )
+    except VMInterfaceStatusNotFoundError:
+        raise IpNotFound(f"IPv{ip_family} address not found for interface {iface_name} on VM {vm.name}.")
+
+    return _lookup_first_ip_address(ip_addresses=iface["ipAddresses"], ip_family=ip_family)
+
+
+def _lookup_first_ip_address(
+    ip_addresses: list[str],
+    ip_family: int,
+) -> ipaddress.IPv4Address | ipaddress.IPv6Address | None:
+    return next((ip for ip_addr in ip_addresses if (ip := ipaddress.ip_address(ip_addr)).version == ip_family), None)

tests/network/bgp/conftest.py

@@ -15,7 +15,7 @@
 from libs.net.traffic_generator import PodTcpClient as TcpClient
 from libs.net.traffic_generator import TcpServer
 from libs.net.udn import create_udn_namespace
-from libs.net.vmspec import IP_ADDRESS, lookup_iface_status, lookup_primary_network
+from libs.net.vmspec import lookup_iface_status_ip, lookup_primary_network
 from libs.vm.vm import BaseVirtualMachine
 from tests.network.libs import cluster_user_defined_network as libcudn
 from tests.network.libs import nodenetworkconfigurationpolicy as libnncp
@@ -231,7 +231,9 @@ def tcp_client_external_network(
 ) -> Generator[TcpClient]:
     with TcpClient(
         pod=frr_external_pod.pod,
-        server_ip=lookup_iface_status(vm=vm_cudn, iface_name=lookup_primary_network(vm=vm_cudn).name)[IP_ADDRESS],
+        server_ip=str(
+            lookup_iface_status_ip(vm=vm_cudn, iface_name=lookup_primary_network(vm=vm_cudn).name, ip_family=4)
+        ),
         server_port=IPERF3_SERVER_PORT,
         bind_interface=EXTERNAL_PROVIDER_IP_V4.split("/")[0],
     ) as client:

tests/network/localnet/liblocalnet.py

@@ -7,7 +7,11 @@
 
 from libs.net.traffic_generator import TcpServer
 from libs.net.traffic_generator import VMTcpClient as TcpClient
-from libs.net.vmspec import IP_ADDRESS, add_network_interface, add_volume_disk, lookup_iface_status
+from libs.net.vmspec import (
+    add_network_interface,
+    add_volume_disk,
+    lookup_iface_status_ip,
+)
 from libs.vm.affinity import new_pod_anti_affinity
 from libs.vm.factory import base_vmspec, fedora_vm
 from libs.vm.spec import CloudInitNoCloud, Interface, Metadata, Multus, Network
@@ -48,7 +52,7 @@ def create_traffic_client(
 ) -> TcpClient:
     return TcpClient(
         vm=client_vm,
-        server_ip=lookup_iface_status(vm=server_vm, iface_name=spec_logical_network)[IP_ADDRESS],
+        server_ip=str(lookup_iface_status_ip(vm=server_vm, iface_name=spec_logical_network, ip_family=4)),
         server_port=_IPERF_SERVER_PORT,
     )
 
@@ -151,11 +155,12 @@ def client_server_active_connection(
     server_vm: BaseVirtualMachine,
     spec_logical_network: str,
     port: int = _IPERF_SERVER_PORT,
+    ip_family: int = 4,
 ) -> Generator[tuple[TcpClient, TcpServer], None, None]:
     with TcpServer(vm=server_vm, port=port) as server:
         with TcpClient(
             vm=client_vm,
-            server_ip=lookup_iface_status(vm=server_vm, iface_name=spec_logical_network)[IP_ADDRESS],
+            server_ip=str(lookup_iface_status_ip(vm=server_vm, iface_name=spec_logical_network, ip_family=ip_family)),
             server_port=port,
         ) as client:
             yield client, server

tests/network/provider_migration/test_ip_persistence.py

@@ -3,7 +3,7 @@
 import pytest
 
 from libs.net.traffic_generator import is_tcp_connection
-from libs.net.vmspec import IP_ADDRESS, lookup_iface_status, lookup_primary_network
+from libs.net.vmspec import lookup_iface_status, lookup_iface_status_ip, lookup_primary_network
 from tests.network.localnet.liblocalnet import client_server_active_connection
 from utilities.constants import PUBLIC_DNS_SERVER_IP
 from utilities.virt import migrate_vm_and_verify
@@ -20,7 +20,12 @@ def test_mac_and_ip_preserved_after_vm_import(source_vm_network_data, imported_c
     target_vm_iface = lookup_iface_status(
         vm=imported_cudn_vm, iface_name=lookup_primary_network(vm=imported_cudn_vm).name
     )
-    target_vm_mac, target_vm_ip = target_vm_iface.get("mac", None), target_vm_iface.get(IP_ADDRESS, None)
+    target_vm_mac = target_vm_iface.get("mac", None)
+    target_vm_ip = str(
+        lookup_iface_status_ip(
+            vm=imported_cudn_vm, iface_name=lookup_primary_network(vm=imported_cudn_vm).name, ip_family=4
+        )
+    )
 
     with subtests.test("MAC preserved"):
         assert source_vm_mac == target_vm_mac, (

tests/network/user_defined_network/test_user_defined_network.py

@@ -6,7 +6,7 @@
 
 from libs.net.traffic_generator import TcpServer, is_tcp_connection
 from libs.net.traffic_generator import VMTcpClient as TcpClient
-from libs.net.vmspec import lookup_iface_status, lookup_primary_network
+from libs.net.vmspec import lookup_iface_status_ip, lookup_primary_network
 from libs.vm import affinity
 from tests.network.libs.ip import random_ipv4_address
 from tests.network.libs.vm_factory import udn_vm
@@ -85,7 +85,9 @@ def server(vmb_udn):
 def client(vma_udn, vmb_udn):
     with TcpClient(
         vm=vma_udn,
-        server_ip=lookup_iface_status(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name)[IP_ADDRESS],
+        server_ip=str(
+            lookup_iface_status_ip(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name, ip_family=4)
+        ),
         server_port=SERVER_PORT,
     ) as client:
         assert client.is_running()
@@ -98,7 +100,7 @@ class TestPrimaryUdn:
     @pytest.mark.polarion("CNV-11624")
     @pytest.mark.single_nic
     def test_ip_address_in_running_vm_matches_udn_subnet(self, namespaced_layer2_user_defined_network, vma_udn):
-        ip = lookup_iface_status(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name)[IP_ADDRESS]
+        ip = str(lookup_iface_status_ip(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name, ip_family=4))
         (subnet,) = namespaced_layer2_user_defined_network.subnets
         assert ipaddress.ip_address(ip) in ipaddress.ip_network(subnet), (
             f"The VM's primary network IP address ({ip}) is not in the UDN defined subnet ({subnet})"
@@ -107,14 +109,14 @@ def test_ip_address_in_running_vm_matches_udn_subnet(self, namespaced_layer2_use
     @pytest.mark.polarion("CNV-11674")
     @pytest.mark.single_nic
     def test_ip_address_is_preserved_after_live_migration(self, vma_udn):
-        ip_before_migration = lookup_iface_status(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name)[
-            IP_ADDRESS
-        ]
+        ip_before_migration = str(
+            lookup_iface_status_ip(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name, ip_family=4)
+        )
         assert ip_before_migration
         migrate_vm_and_verify(vm=vma_udn)
-        ip_after_migration = lookup_iface_status(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name)[
-            IP_ADDRESS
-        ]
+        ip_after_migration = str(
+            lookup_iface_status_ip(vm=vma_udn, iface_name=lookup_primary_network(vm=vma_udn).name, ip_family=4)
+        )
         assert ip_before_migration == ip_after_migration, (
             f"The IP address {ip_before_migration} was not preserved during live migration. "
             f"IP after migration: {ip_after_migration}."
@@ -123,13 +125,15 @@ def test_ip_address_is_preserved_after_live_migration(self, vma_udn):
     @pytest.mark.polarion("CNV-11434")
     @pytest.mark.single_nic
     def test_vm_egress_connectivity(self, vmb_udn):
-        assert lookup_iface_status(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name)[IP_ADDRESS]
+        assert str(lookup_iface_status_ip(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name, ip_family=4))
         vmb_udn.console(commands=[f"ping -c 3 {PUBLIC_DNS_SERVER_IP}"], timeout=TIMEOUT_1MINUTE)
 
     @pytest.mark.polarion("CNV-11418")
     @pytest.mark.single_nic
     def test_basic_connectivity_between_udn_vms(self, vma_udn, vmb_udn):
-        target_vm_ip = lookup_iface_status(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name)[IP_ADDRESS]
+        target_vm_ip = str(
+            lookup_iface_status_ip(vm=vmb_udn, iface_name=lookup_primary_network(vm=vmb_udn).name, ip_family=4)
+        )
         vma_udn.console(commands=[f"ping -c 3 {target_vm_ip}"], timeout=TIMEOUT_1MIN)
 
     @pytest.mark.polarion("CNV-11427")

utilities/network.py

@@ -511,10 +511,10 @@ def _get_interface_ips():
 
 
 class IpNotFound(Exception):
-    def __init__(self, name):
+    def __init__(self, name: str) -> None:
         self.name = name
 
-    def __str__(self):
+    def __str__(self) -> str:
         return f"IP address not found for interface {self.name}"