net, libs: Refactor modules and add new lib helper

conftest.py

@@ -73,6 +73,11 @@
     validate_collected_tests_arch_params,
 )
 
+pytest_plugins = [
+    "tests.fixtures.network.l2_bridge",
+    "tests.fixtures.network.cluster",
+]
+
 LOGGER = logging.getLogger(__name__)
 BASIC_LOGGER = logging.getLogger("basic")
 

libs/net/cluster.py

@@ -29,3 +29,15 @@ def ipv6_supported_cluster() -> bool:
 
 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"))
+
+
+@cache
+def cluster_vlans() -> list[int]:
+    """Return VLAN IDs from test config as normalized integers.
+
+    Supports both list input and comma-separated string input from py_config["vlans"].
+    """
+    vlans = py_config["vlans"]
+    if not isinstance(vlans, list):
+        vlans = vlans.split(",")
+    return [int(vlan_id) for vlan_id in vlans]

tests/network/libs/nodenetworkconfigurationpolicy.py → libs/net/nodenetworkconfigurationpolicy.py

@@ -12,7 +12,7 @@
 from ocp_resources.resource import Resource, ResourceEditor
 from timeout_sampler import retry
 
-from tests.network.libs.apimachinery import dict_normalization_for_dataclass
+from libs.net.apimachinery import dict_normalization_for_dataclass
 
 WAIT_FOR_STATUS_TIMEOUT_SEC = 120
 WAIT_FOR_STATUS_INTERVAL_SEC = 5

libs/net/vmspec.py

@@ -2,6 +2,7 @@
 
 import ipaddress
 from collections.abc import Callable
+from copy import deepcopy
 from typing import TYPE_CHECKING, Any, Final
 
 from kubernetes.dynamic.client import ResourceField
@@ -303,3 +304,23 @@ def _vmi_condition_not_set(existing_conditions: list[ResourceField], required_co
         for cond in existing_conditions
         if cond.type == required_condition
     )
+
+
+def update_nad_references(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str]) -> None:
+    """Update secondary network NAD references and wait for the change to be fully applied.
+
+    Patches the VM spec atomically, then waits for the MigrationRequired condition to
+    appear (change detected) and disappear (migration completed).
+
+    Args:
+        vm: The virtual machine to update.
+        nad_name_by_net: Mapping of spec network name to new NAD name.
+    """
+    resource_version = vm.vmi.instance.metadata.resourceVersion
+    networks = deepcopy(vm.template_spec.networks) or []
+    for network in networks:
+        if network.name in nad_name_by_net and network.multus:
+            network.multus.networkName = nad_name_by_net[network.name]
+    vm.set_networks(networks=networks)
+    wait_for_vmi_condition_status(vm=vm, condition="MigrationRequired", resource_version=resource_version)
+    wait_for_no_vmi_condition(vm=vm, condition="MigrationRequired")

tests/fixtures/network/cluster.py

@@ -0,0 +1,10 @@
+from collections.abc import Iterator
+
+import pytest
+
+from libs.net.cluster import cluster_vlans
+
+
+@pytest.fixture(scope="module")
+def cluster_vlan_ids() -> Iterator[int]:
+    return iter(cluster_vlans())

tests/fixtures/network/l2_bridge.py

@@ -0,0 +1,76 @@
+from collections.abc import Generator, Iterator
+
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.namespace import Namespace
+
+from libs.net import nodenetworkconfigurationpolicy as libnncp
+from libs.net.netattachdef import CNIPluginBridgeConfig, NetConfig, NetworkAttachmentDefinition
+from utilities.constants.cluster import WORKER_NODE_LABEL_KEY
+from utilities.constants.networking import LINUX_BRIDGE
+
+
+@pytest.fixture(scope="package")
+def bridge_nncp(
+    nmstate_dependent_placeholder: None,
+    admin_client: DynamicClient,
+    hosts_common_available_ports: list[str],
+) -> Generator[libnncp.NodeNetworkConfigurationPolicy]:
+    with libnncp.NodeNetworkConfigurationPolicy(
+        client=admin_client,
+        name="l2-bridge-test-nncp",
+        desired_state=libnncp.DesiredState(
+            interfaces=[
+                libnncp.Interface(
+                    name="br1-test",
+                    type=LINUX_BRIDGE,
+                    state=libnncp.Resource.Interface.State.UP,
+                    bridge=libnncp.Bridge(
+                        port=[libnncp.Port(name=hosts_common_available_ports[-1])],
+                        options=libnncp.BridgeOptions(libnncp.STP(enabled=False)),
+                    ),
+                )
+            ]
+        ),
+        node_selector={WORKER_NODE_LABEL_KEY: ""},
+    ) as nncp_br:
+        nncp_br.wait_for_status_success()
+        yield nncp_br
+
+
+@pytest.fixture(scope="module")
+def bridge_nad_a(
+    admin_client: DynamicClient,
+    namespace: Namespace,
+    bridge_nncp: libnncp.NodeNetworkConfigurationPolicy,
+    cluster_vlan_ids: Iterator[int],
+) -> Generator[NetworkAttachmentDefinition]:
+    bridge = bridge_nncp.desired_state_spec.interfaces[0].name  # type: ignore
+    with NetworkAttachmentDefinition(
+        name="nad-vlan-a",
+        namespace=namespace.name,
+        config=NetConfig(
+            name="nad-vlan-a", plugins=[CNIPluginBridgeConfig(bridge=bridge, vlan=next(cluster_vlan_ids))]
+        ),
+        client=admin_client,
+    ) as nad:
+        yield nad
+
+
+@pytest.fixture(scope="module")
+def bridge_nad_b(
+    admin_client: DynamicClient,
+    namespace: Namespace,
+    bridge_nncp: libnncp.NodeNetworkConfigurationPolicy,
+    cluster_vlan_ids: Iterator[int],
+) -> Generator[NetworkAttachmentDefinition]:
+    bridge = bridge_nncp.desired_state_spec.interfaces[0].name  # type: ignore[union-attr, index]
+    with NetworkAttachmentDefinition(
+        name="nad-vlan-b",
+        namespace=namespace.name,
+        config=NetConfig(
+            name="nad-vlan-b", plugins=[CNIPluginBridgeConfig(bridge=bridge, vlan=next(cluster_vlan_ids))]
+        ),
+        client=admin_client,
+    ) as nad:
+        yield nad

tests/network/bgp/conftest.py

@@ -12,14 +12,14 @@
 from ocp_resources.node import Node
 
 from libs.net import netattachdef as libnad
+from libs.net import nodenetworkconfigurationpolicy as libnncp
 from libs.net.ip import random_ipv4_address
 from libs.net.traffic_generator import PodTcpClient as TcpClient
 from libs.net.traffic_generator import TcpServer
 from libs.net.udn import UDN_BINDING_DEFAULT_PLUGIN_NAME, create_udn_namespace
 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
 from tests.network.libs.bgp import (
     EXTERNAL_FRR_POD_LABEL,
     NET_TOOLS_CONTAINER_NAME,

tests/network/conftest.py

@@ -11,11 +11,9 @@
 from ocp_resources.network_config_openshift_io import Network
 from ocp_resources.performance_profile import PerformanceProfile
 from ocp_resources.pod import Pod
-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,
     CLUSTER_NETWORK_ADDONS_OPERATOR,
@@ -95,19 +93,6 @@ def sriov_workers_node2(sriov_workers):
     return sriov_workers[1]
 
 
-@pytest.fixture(scope="session")
-def vlans_list():
-    vlans = py_config["vlans"]
-    if not isinstance(vlans, list):
-        vlans = vlans.split(",")
-    return [int(_id) for _id in vlans]
-
-
-@pytest.fixture(scope="module")
-def vlan_index_number(vlans_list):
-    return get_vlan_index_number(vlans_list=vlans_list)
-
-
 @pytest.fixture(scope="session")
 def cluster_network_mtu(admin_client):
     network_resource = Network(name=CLUSTER, client=admin_client)

tests/network/connectivity/conftest.py

@@ -13,18 +13,18 @@ def bridge_device_name(index_number):
 
 
 @pytest.fixture(scope="module")
-def vlan_id_1(vlan_index_number):
-    return next(vlan_index_number)
+def vlan_id_1(cluster_vlan_ids):
+    return next(cluster_vlan_ids)
 
 
 @pytest.fixture(scope="module")
-def vlan_id_2(vlan_index_number):
-    return next(vlan_index_number)
+def vlan_id_2(cluster_vlan_ids):
+    return next(cluster_vlan_ids)
 
 
 @pytest.fixture(scope="module")
-def vlan_id_3(vlan_index_number):
-    return next(vlan_index_number)
+def vlan_id_3(cluster_vlan_ids):
+    return next(cluster_vlan_ids)
 
 
 @pytest.fixture(scope="class")

tests/network/l2_bridge/bandwidth/conftest.py

@@ -7,7 +7,7 @@
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.namespace import Namespace
 
-import tests.network.libs.nodenetworkconfigurationpolicy as libnncp
+from libs.net import nodenetworkconfigurationpolicy as libnncp
 from libs.net.ip import random_ip_addresses_by_family
 from libs.net.netattachdef import (
     CNIPluginBandwidthConfig,

tests/network/l2_bridge/conftest.py

@@ -5,7 +5,7 @@
 from kubernetes.dynamic import DynamicClient
 from pyhelper_utils.shell import run_ssh_commands
 
-import tests.network.libs.nodenetworkconfigurationpolicy as libnncp
+from libs.net import nodenetworkconfigurationpolicy as libnncp
 from libs.net.ip import random_ipv4_address
 from libs.net.netattachdef import CNIPluginBridgeConfig, NetConfig, NetworkAttachmentDefinition
 from tests.network.l2_bridge.libl2bridge import DHCP_INTERFACE_NAME, bridge_attached_vm
@@ -18,7 +18,6 @@
     UNIQUE_CLIENT_ID,
     verify_dhcpd_activated,
 )
-from utilities.constants import LINUX_BRIDGE, WORKER_NODE_LABEL_KEY
 from utilities.data_utils import name_prefix
 from utilities.infra import get_node_selector_dict
 from utilities.network import (
@@ -97,11 +96,11 @@ def dhcp_nad(
     namespace,
     l2_bridge_device_worker_1,
     l2_bridge_device_worker_2,
-    vlan_index_number,
+    cluster_vlan_ids,
     bridge_device_type,
     l2_bridge_device_name,
 ):
-    vlan_tag = next(vlan_index_number)
+    vlan_tag = next(cluster_vlan_ids)
     with network_nad(
         namespace=namespace,
         nad_type=bridge_device_type,
@@ -288,34 +287,6 @@ def started_vmb_dhcp_client(l2_bridge_running_vm_b, eth3_nmcli_connection_uuid):
     )
 
 
-@pytest.fixture(scope="package")
-def bridge_nncp(
-    nmstate_dependent_placeholder: None,
-    admin_client: DynamicClient,
-    hosts_common_available_ports: list[str],
-) -> Generator[libnncp.NodeNetworkConfigurationPolicy]:
-    with libnncp.NodeNetworkConfigurationPolicy(
-        client=admin_client,
-        name="l2-bridge-test-nncp",
-        desired_state=libnncp.DesiredState(
-            interfaces=[
-                libnncp.Interface(
-                    name="br1-test",
-                    type=LINUX_BRIDGE,
-                    state=libnncp.Resource.Interface.State.UP,
-                    bridge=libnncp.Bridge(
-                        port=[libnncp.Port(name=hosts_common_available_ports[-1])],
-                        options=libnncp.BridgeOptions(libnncp.STP(enabled=False)),
-                    ),
-                )
-            ]
-        ),
-        node_selector={WORKER_NODE_LABEL_KEY: ""},
-    ) as nncp_br:
-        nncp_br.wait_for_status_success()
-        yield nncp_br
-
-
 @pytest.fixture(scope="class")
 def bridge_nad(
     admin_client: DynamicClient,

tests/network/l2_bridge/migration_stuntime/conftest.py

@@ -5,8 +5,8 @@
 from kubernetes.dynamic import DynamicClient
 from ocp_resources.namespace import Namespace
 
-import tests.network.libs.nodenetworkconfigurationpolicy as libnncp
 from libs.net import netattachdef as libnad
+from libs.net import nodenetworkconfigurationpolicy as libnncp
 from libs.net.ip import random_ipv4_address, random_ipv6_address
 from libs.net.vmspec import lookup_iface_status_ip
 from libs.vm.affinity import new_pod_affinity

tests/network/l2_bridge/nad_ref_change/conftest.py

@@ -5,7 +5,7 @@
 from ocp_resources.namespace import Namespace
 
 from libs.net.ip import filter_link_local_addresses, random_cidr_addresses_by_family
-from libs.net.netattachdef import CNIPluginBridgeConfig, NetConfig, NetworkAttachmentDefinition
+from libs.net.netattachdef import NetworkAttachmentDefinition
 from libs.net.vmspec import lookup_iface_status, wait_for_ifaces_status
 from libs.vm.vm import BaseVirtualMachine
 from tests.network.l2_bridge.libl2bridge import LINUX_BRIDGE_IFACE_NAME_1, LINUX_BRIDGE_IFACE_NAME_2
@@ -15,48 +15,9 @@
     NET_SEED,
     two_secondary_bridge_vm,
 )
-from tests.network.libs import nodenetworkconfigurationpolicy as libnncp
 from tests.network.libs.connectivity import ARP_ISOLATION_SYSCTL_CMD, poll_tcp_connectivity
 
 
-@pytest.fixture(scope="module")
-def bridge_nad_a(
-    admin_client: DynamicClient,
-    namespace: Namespace,
-    bridge_nncp: libnncp.NodeNetworkConfigurationPolicy,
-    vlan_index_number: Generator[int],
-) -> Generator[NetworkAttachmentDefinition]:
-    bridge = bridge_nncp.desired_state_spec.interfaces[0].name  # type: ignore
-    with NetworkAttachmentDefinition(
-        name="nad-vlan-a",
-        namespace=namespace.name,
-        config=NetConfig(
-            name="nad-vlan-a", plugins=[CNIPluginBridgeConfig(bridge=bridge, vlan=next(vlan_index_number))]
-        ),
-        client=admin_client,
-    ) as nad:
-        yield nad
-
-
-@pytest.fixture(scope="module")
-def bridge_nad_b(
-    admin_client: DynamicClient,
-    namespace: Namespace,
-    bridge_nncp: libnncp.NodeNetworkConfigurationPolicy,
-    vlan_index_number: Generator[int],
-) -> Generator[NetworkAttachmentDefinition]:
-    bridge = bridge_nncp.desired_state_spec.interfaces[0].name  # type: ignore[union-attr, index]
-    with NetworkAttachmentDefinition(
-        name="nad-vlan-b",
-        namespace=namespace.name,
-        config=NetConfig(
-            name="nad-vlan-b", plugins=[CNIPluginBridgeConfig(bridge=bridge, vlan=next(vlan_index_number))]
-        ),
-        client=admin_client,
-    ) as nad:
-        yield nad
-
-
 @pytest.fixture(scope="module")
 def ref_vm(
     namespace: Namespace,

tests/network/l2_bridge/nad_ref_change/test_nad_ref_change.py

@@ -14,15 +14,14 @@
 import pytest
 
 from libs.net.ip import filter_link_local_addresses
-from libs.net.vmspec import lookup_iface_status
+from libs.net.vmspec import lookup_iface_status, update_nad_references
 from tests.network.l2_bridge.libl2bridge import LINUX_BRIDGE_IFACE_NAME_1, LINUX_BRIDGE_IFACE_NAME_2
 from tests.network.l2_bridge.nad_ref_change.lib_helpers import (
     GUEST_IFACE_1,
     GUEST_IFACE_2,
     assert_connectivity,
     assert_no_connectivity,
 )
-from tests.network.libs.nad_ref import update_nad_references
 
 
 @pytest.mark.usefixtures("baseline_connectivity")

tests/network/libs/cloudinit.py

@@ -3,8 +3,8 @@
 
 import yaml
 
+from libs.net.apimachinery import dict_normalization_for_dataclass
 from libs.net.cluster import ipv4_supported_cluster, ipv6_supported_cluster
-from tests.network.libs.apimachinery import dict_normalization_for_dataclass
 
 NETWORK_DATA: Final[str] = "networkData"