[4.21] net, l2 bridge: Verify iface looping bug fix

libs/vm/vm.py

@@ -4,6 +4,7 @@
 from dataclasses import asdict
 from typing import Any
 
+import yaml
 from dacite import from_dict
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.node import Node
@@ -102,6 +103,19 @@ def update_template_annotations(self, template_annotations: dict[str, str]) -> N
         }
         ResourceEditor(patches=patches).update()
 
+    @property
+    def cloud_init_network_data(self) -> cloudinit.NetworkData:
+        """Return the parsed cloud-init network data configured for this VM.
+
+        Returns:
+            NetworkData: The cloud-init network data as a dataclass.
+        """
+        volumes = {vol.name: vol for vol in self.instance.spec.template.spec.volumes}
+        return from_dict(
+            data_class=cloudinit.NetworkData,
+            data=yaml.safe_load(volumes[CLOUD_INIT_DISK_NAME].cloudInitNoCloud.networkData),
+        )
+
     def add_cloud_init(self, netdata: cloudinit.NetworkData) -> None:
         # Prevents cloud-init from overriding the default OS user credentials
         userdata = cloudinit.UserData(users=[])

tests/network/l2_bridge/vmi_interfaces_stability/conftest.py

@@ -32,11 +32,7 @@ def running_linux_bridge_vm(
     ) as vm:
         vm.start(wait=True)
         vm.wait_for_agent_connected()
-        wait_for_stable_ifaces(
-            vm=vm,
-            ipv4_supported_cluster=ipv4_supported_cluster,
-            ipv6_supported_cluster=ipv6_supported_cluster,
-        )
+        wait_for_stable_ifaces(vm=vm)
         yield vm
 
 

tests/network/l2_bridge/vmi_interfaces_stability/lib_helpers.py

@@ -13,10 +13,12 @@
 from libs.vm.spec import CloudInitNoCloud, Interface, Multus, Network
 from libs.vm.vm import BaseVirtualMachine, add_volume_disk, cloudinitdisk_storage
 from tests.network.libs import cloudinit
+from tests.network.localnet.liblocalnet import GUEST_1ST_IFACE_NAME, GUEST_3RD_IFACE_NAME
 
 LOGGER = logging.getLogger(__name__)
 
-LINUX_BRIDGE_IFACE_NAME: Final[str] = "linux-bridge"
+LINUX_BRIDGE_IFACE_NAME_1: Final[str] = "linux-bridge-1"
+LINUX_BRIDGE_IFACE_NAME_2: Final[str] = "linux-bridge-2"
 
 
 def secondary_network_vm(
@@ -29,12 +31,14 @@ def secondary_network_vm(
 ) -> BaseVirtualMachine:
     spec = base_vmspec()
     spec.template.spec.domain.devices.interfaces = [  # type: ignore
+        Interface(name=LINUX_BRIDGE_IFACE_NAME_1, bridge={}),
         Interface(name="default", masquerade={}),
-        Interface(name=LINUX_BRIDGE_IFACE_NAME, bridge={}),
+        Interface(name=LINUX_BRIDGE_IFACE_NAME_2, bridge={}),
     ]
     spec.template.spec.networks = [
+        Network(name=LINUX_BRIDGE_IFACE_NAME_1, multus=Multus(networkName=bridge_network_name)),
         Network(name="default", pod={}),
-        Network(name=LINUX_BRIDGE_IFACE_NAME, multus=Multus(networkName=bridge_network_name)),
+        Network(name=LINUX_BRIDGE_IFACE_NAME_2, multus=Multus(networkName=bridge_network_name)),
     ]
 
     ethernets = {}
@@ -43,11 +47,18 @@ def secondary_network_vm(
         ipv6_supported_cluster=ipv6_supported_cluster,
     )
     if primary:
-        ethernets["eth0"] = primary
+        ethernets["eth1"] = primary
 
-    ethernets["eth1"] = secondary_iface_cloud_init(
+    ethernets["eth0"] = secondary_iface_cloud_init(
         ipv4_supported_cluster=ipv4_supported_cluster,
         ipv6_supported_cluster=ipv6_supported_cluster,
+        host_address=1,
+    )
+
+    ethernets["eth2"] = secondary_iface_cloud_init(
+        ipv4_supported_cluster=ipv4_supported_cluster,
+        ipv6_supported_cluster=ipv6_supported_cluster,
+        host_address=2,
     )
 
     userdata = cloudinit.UserData(users=[])
@@ -79,33 +90,46 @@ def primary_iface_cloud_init(
 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
+        ipv4_supported_cluster=ipv4_supported_cluster,
+        ipv6_supported_cluster=ipv6_supported_cluster,
+        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) -> list[str]:
+def secondary_iface_ips(
+    ipv4_supported_cluster: bool,
+    ipv6_supported_cluster: bool,
+    host_address: int,
+) -> list[str]:
     ips = []
     if ipv4_supported_cluster:
-        ips.append(random_ipv4_address(net_seed=0, host_address=1))
+        ips.append(random_ipv4_address(net_seed=0, host_address=host_address))
     if ipv6_supported_cluster:
-        ips.append(random_ipv6_address(net_seed=0, host_address=1))
+        ips.append(random_ipv6_address(net_seed=0, host_address=host_address))
     return ips
 
 
 def wait_for_stable_ifaces(
     vm: BaseVirtualMachine,
-    ipv4_supported_cluster: bool,
-    ipv6_supported_cluster: bool,
 ) -> None:
     primary_network = lookup_primary_network(vm=vm)
-    secondary_ips = secondary_iface_ips(
-        ipv4_supported_cluster=ipv4_supported_cluster,
-        ipv6_supported_cluster=ipv6_supported_cluster,
-    )
+
+    secondary_iface_to_ips = {
+        LINUX_BRIDGE_IFACE_NAME_1: [
+            str(ipaddress.ip_interface(addr).ip)
+            for addr in vm.cloud_init_network_data.ethernets[GUEST_1ST_IFACE_NAME].addresses  # type: ignore[union-attr]
+        ],
+        LINUX_BRIDGE_IFACE_NAME_2: [
+            str(ipaddress.ip_interface(addr).ip)
+            for addr in vm.cloud_init_network_data.ethernets[GUEST_3RD_IFACE_NAME].addresses  # type: ignore[union-attr]
+        ],
+    }
+
     spec_interfaces = vm.instance.spec.template.spec.domain.devices.interfaces
     for iface in spec_interfaces:
         if iface.name == primary_network.name:
@@ -116,7 +140,7 @@ def wait_for_stable_ifaces(
                 iface_name=iface.name,
                 predicate=lambda iface_status: (
                     "guest-agent" in iface_status["infoSource"]
-                    and all(ip in iface_status.get("ipAddresses", []) for ip in secondary_ips)
+                    and all(ip in iface_status.get("ipAddresses", []) for ip in secondary_iface_to_ips[iface.name])
                 ),
             )
 

tests/network/l2_bridge/vmi_interfaces_stability/test_interfaces_stability.py

@@ -16,10 +16,10 @@ class TestInterfacesStability:
     @pytest.mark.polarion("CNV-14339")
     def test_interfaces_stability(self, running_linux_bridge_vm, stable_ips):
         for vmi_obj in monitor_vmi_events(vm=running_linux_bridge_vm, timeout=STABILITY_PERIOD_IN_SECONDS):
-            assert_interfaces_stable(stable_ips=stable_ips, vmi=vmi_obj, expected_num_ifaces=2)
+            assert_interfaces_stable(stable_ips=stable_ips, vmi=vmi_obj, expected_num_ifaces=len(stable_ips))
 
     @pytest.mark.polarion("CNV-14340")
     def test_interfaces_stability_after_migration(self, running_linux_bridge_vm, stable_ips):
         migrate_vm_and_verify(vm=running_linux_bridge_vm)
         for vmi_obj in monitor_vmi_events(vm=running_linux_bridge_vm, timeout=STABILITY_PERIOD_IN_SECONDS):
-            assert_interfaces_stable(stable_ips=stable_ips, vmi=vmi_obj, expected_num_ifaces=2)
+            assert_interfaces_stable(stable_ips=stable_ips, vmi=vmi_obj, expected_num_ifaces=len(stable_ips))

tests/network/localnet/liblocalnet.py

@@ -1,7 +1,7 @@
 import contextlib
 import logging
 import uuid
-from typing import Generator
+from typing import Final, Generator
 
 from kubernetes.client import ApiException
 from kubernetes.dynamic import DynamicClient
@@ -30,6 +30,11 @@
 LINK_STATE_UP = "up"
 LINK_STATE_DOWN = "down"
 NNCP_INTERFACE_TYPE_ETHERNET = "ethernet"
+GUEST_1ST_IFACE_NAME: Final[str] = "eth0"
+GUEST_2ND_IFACE_NAME: Final[str] = "eth1"
+GUEST_3RD_IFACE_NAME: Final[str] = "eth2"
+
+
 LOGGER = logging.getLogger(__name__)