net: NAD live update VM localnet test (#5039)

##### What this PR does / why we need it: 
Implements the NAD reference change test for localnet: verifies that a
running VM can switch its secondary localnet network from one VLAN to
another without rebooting, and that TCP connectivity is established on
the new VLAN.

##### Which issue(s) this PR fixes: -

##### Special notes for reviewer:
Usage of shared localnet setup: NNCP and localnet CUDN (as CUDN-VLAN-A)
Includes move of shared helpers to be used in several packages like
upcoming metrics nad ref change test.

##### jira-ticket: https://redhat.atlassian.net/browse/CNV-80573
<!-- full-ticket-url needs to be provided. This would add a link to the
pull request to the jira and close it when the pull request is merged
If the task is not tracked by a Jira ticket, just write "NONE".
-->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **Tests**
* Reorganized network connectivity test helpers for better code
organization and reusability.
* Added new test fixtures for localnet network configuration and
baseline connectivity validation.
* Enhanced test coverage with configurable runtime commands and improved
VM provisioning capabilities.

* **Refactoring**
* Consolidated network interface naming constants across test modules
for consistency.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

Author: rnetser

tests/network/l2_bridge/nad_ref_change/lib_helpers.py

@@ -1,9 +1,7 @@
-from copy import deepcopy
 from typing import Final
 
 from kubernetes.dynamic import DynamicClient
 
-from libs.net.vmspec import wait_for_no_vmi_condition, wait_for_vmi_condition_status
 from libs.vm.factory import base_vmspec, fedora_vm
 from libs.vm.spec import (
     CloudInitNoCloud,
@@ -75,26 +73,6 @@ def assert_no_connectivity(
     )
 
 
-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 interface 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")
-
-
 def two_secondary_bridge_vm(
     namespace: str,
     name: str,

tests/network/l2_bridge/nad_ref_change/test_nad_ref_change.py

@@ -21,8 +21,8 @@
     GUEST_IFACE_2,
     assert_connectivity,
     assert_no_connectivity,
-    update_nad_references,
 )
+from tests.network.libs.nad_ref import update_nad_references
 
 
 @pytest.mark.usefixtures("baseline_connectivity")

tests/network/libs/nad_ref.py

@@ -0,0 +1,24 @@
+from copy import deepcopy
+
+from libs.net.vmspec import wait_for_no_vmi_condition, wait_for_vmi_condition_status
+from libs.vm.vm import BaseVirtualMachine
+
+
+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/network/localnet/liblocalnet.py

@@ -34,6 +34,10 @@
 GUEST_2ND_IFACE_NAME: Final[str] = "eth1"
 GUEST_3RD_IFACE_NAME: Final[str] = "eth2"
 
+IFACE_A_NAME: Final[str] = "localnet-vlan-a"
+IFACE_B_NAME: Final[str] = "localnet-vlan-b"
+CUDN_B_NAME: Final[str] = "cudn-localnet-vlan-b"
+
 LOGGER = logging.getLogger(__name__)
 
 
@@ -65,6 +69,7 @@ def localnet_vm(
     networks: list[Network],
     interfaces: list[Interface],
     network_data: cloudinit.NetworkData | None = None,
+    runcmd: list[str] | None = None,
     affinity: Affinity | None = None,
     vm_labels: dict[str, str] | None = None,
 ) -> BaseVirtualMachine:
@@ -85,6 +90,8 @@ def localnet_vm(
             Each Interface should have a name matching a Network, and additional configuration and state.
         network_data (cloudinit.NetworkData | None): Cloud-init NetworkData object containing the network
             configuration for the VM interfaces. If None, no network configuration is applied via cloud-init.
+        runcmd (list[str] | None): Commands to run on first boot via cloud-init runcmd. None means no
+            extra commands are injected.
         affinity (Affinity | None): Optional Affinity object for VM scheduling. Controls the VM scheduling
             location. If None, no affinity constraints are applied.
         vm_labels (dict[str, str] | None): Optional labels to apply to the VM template metadata.
@@ -124,7 +131,7 @@ def localnet_vm(
 
     if network_data is not None:
         # Prevents cloud-init from overriding the default OS user credentials
-        userdata = cloudinit.UserData(users=[])
+        userdata = cloudinit.UserData(users=[], runcmd=runcmd)
         disk, volume = cloudinitdisk_storage(
             data=CloudInitNoCloud(
                 networkData=cloudinit.asyaml(no_cloud=network_data),

tests/network/localnet/nad_ref_change/conftest.py

@@ -0,0 +1,139 @@
+from collections.abc import Generator
+from typing import Final
+
+import pytest
+from kubernetes.dynamic import DynamicClient
+from ocp_resources.namespace import Namespace
+
+from libs.net.ip import filter_link_local_addresses, random_cidr_addresses_by_family
+from libs.net.vmspec import lookup_iface_status, wait_for_ifaces_status
+from libs.vm.spec import Interface, Multus, Network
+from libs.vm.vm import BaseVirtualMachine
+from tests.network.libs import cloudinit
+from tests.network.libs import cluster_user_defined_network as libcudn
+from tests.network.libs.connectivity import ARP_ISOLATION_SYSCTL_CMD, poll_tcp_connectivity
+from tests.network.localnet.liblocalnet import (
+    CUDN_B_NAME,
+    GUEST_1ST_IFACE_NAME,
+    GUEST_2ND_IFACE_NAME,
+    IFACE_A_NAME,
+    IFACE_B_NAME,
+    LOCALNET_BR_EX_NETWORK,
+    LOCALNET_TEST_LABEL,
+    localnet_cudn,
+    localnet_vm,
+)
+
+NET_SEED: Final[int] = 0
+
+
+@pytest.fixture(scope="module")
+def cudn_nad_ref_vlan_b(
+    admin_client: DynamicClient,
+    cudn_localnet: libcudn.ClusterUserDefinedNetwork,
+    vlan_index_number: Generator[int],
+) -> Generator[libcudn.ClusterUserDefinedNetwork]:
+    with localnet_cudn(
+        name=CUDN_B_NAME,
+        match_labels=LOCALNET_TEST_LABEL,
+        vlan_id=next(vlan_index_number),
+        physical_network_name=LOCALNET_BR_EX_NETWORK,
+        client=admin_client,
+    ) as cudn:
+        cudn.wait_for_status_success()
+        yield cudn
+
+
+@pytest.fixture(scope="module")
+def ref_vm_localnet(
+    namespace_localnet_1: Namespace,
+    unprivileged_client: DynamicClient,
+    cudn_localnet: libcudn.ClusterUserDefinedNetwork,
+    cudn_nad_ref_vlan_b: libcudn.ClusterUserDefinedNetwork,
+) -> Generator[BaseVirtualMachine]:
+    iface_a_ips = random_cidr_addresses_by_family(net_seed=NET_SEED, host_address=1)
+    iface_b_ips = random_cidr_addresses_by_family(net_seed=NET_SEED, host_address=2)
+    with localnet_vm(
+        namespace=namespace_localnet_1.name,
+        name="ref-vm",
+        client=unprivileged_client,
+        networks=[
+            Network(name=IFACE_A_NAME, multus=Multus(networkName=cudn_localnet.name)),
+            Network(name=IFACE_B_NAME, multus=Multus(networkName=cudn_nad_ref_vlan_b.name)),
+        ],
+        interfaces=[
+            Interface(name=IFACE_A_NAME, bridge={}),
+            Interface(name=IFACE_B_NAME, bridge={}),
+        ],
+        network_data=cloudinit.NetworkData(
+            ethernets={
+                GUEST_1ST_IFACE_NAME: cloudinit.EthernetDevice(addresses=iface_a_ips),
+                GUEST_2ND_IFACE_NAME: cloudinit.EthernetDevice(addresses=iface_b_ips),
+            }
+        ),
+        runcmd=ARP_ISOLATION_SYSCTL_CMD,
+    ) as vm:
+        vm.start(wait=True)
+        vm.wait_for_agent_connected()
+        wait_for_ifaces_status(
+            vm=vm,
+            ip_addresses_by_spec_net_name={
+                IFACE_A_NAME: [addr.split("/")[0] for addr in iface_a_ips],
+                IFACE_B_NAME: [addr.split("/")[0] for addr in iface_b_ips],
+            },
+        )
+        yield vm
+
+
+@pytest.fixture()
+def under_test_vm_localnet(
+    namespace_localnet_1: Namespace,
+    unprivileged_client: DynamicClient,
+    cudn_localnet: libcudn.ClusterUserDefinedNetwork,
+) -> Generator[BaseVirtualMachine]:
+    iface_a_ips = random_cidr_addresses_by_family(net_seed=NET_SEED, host_address=3)
+    with localnet_vm(
+        namespace=namespace_localnet_1.name,
+        name="under-test-vm",
+        client=unprivileged_client,
+        networks=[Network(name=IFACE_A_NAME, multus=Multus(networkName=cudn_localnet.name))],
+        interfaces=[Interface(name=IFACE_A_NAME, bridge={})],
+        network_data=cloudinit.NetworkData(
+            ethernets={GUEST_1ST_IFACE_NAME: cloudinit.EthernetDevice(addresses=iface_a_ips)},
+        ),
+    ) as vm:
+        vm.start(wait=True)
+        vm.wait_for_agent_connected()
+        wait_for_ifaces_status(
+            vm=vm,
+            ip_addresses_by_spec_net_name={
+                IFACE_A_NAME: [addr.split("/")[0] for addr in iface_a_ips],
+            },
+        )
+        yield vm
+
+
+@pytest.fixture()
+def baseline_connectivity_localnet(
+    under_test_vm_localnet: BaseVirtualMachine,
+    ref_vm_localnet: BaseVirtualMachine,
+) -> None:
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm_localnet, iface_name=IFACE_A_NAME).ipAddresses
+    ):
+        poll_tcp_connectivity(
+            client_vm=under_test_vm_localnet,
+            server_vm=ref_vm_localnet,
+            server_ip=str(server_ip),
+            server_bind_dev=GUEST_1ST_IFACE_NAME,
+        )
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm_localnet, iface_name=IFACE_B_NAME).ipAddresses
+    ):
+        poll_tcp_connectivity(
+            client_vm=under_test_vm_localnet,
+            server_vm=ref_vm_localnet,
+            server_ip=str(server_ip),
+            server_bind_dev=GUEST_2ND_IFACE_NAME,
+            expect_connectivity=False,
+        )

tests/network/localnet/nad_ref_change/test_nad_ref_change.py

@@ -12,9 +12,26 @@
 
 import pytest
 
+from libs.net.ip import filter_link_local_addresses
+from libs.net.vmspec import lookup_iface_status
+from tests.network.libs.connectivity import poll_tcp_connectivity
+from tests.network.libs.nad_ref import update_nad_references
+from tests.network.localnet.liblocalnet import (
+    GUEST_1ST_IFACE_NAME,
+    GUEST_2ND_IFACE_NAME,
+    IFACE_A_NAME,
+    IFACE_B_NAME,
+)
 
+
+@pytest.mark.usefixtures("nncp_localnet", "baseline_connectivity_localnet")
 @pytest.mark.polarion("CNV-15948")
-def test_running_vm_vlan_change():
+def test_running_vm_vlan_change(
+    subtests,
+    under_test_vm_localnet,
+    ref_vm_localnet,
+    cudn_nad_ref_vlan_b,
+):
     """
     Test that a running VM can change the VLAN of its secondary localnet network, without rebooting.
     The VM should establish TCP connectivity on the new VLAN.
@@ -33,6 +50,28 @@ def test_running_vm_vlan_change():
         - Under-test VM eventually has TCP connectivity to the reference VM on NAD-VLAN-B
         - Under-test VM has no TCP connectivity to the reference VM on NAD-VLAN-A
     """
+    update_nad_references(vm=under_test_vm_localnet, nad_name_by_net={IFACE_A_NAME: cudn_nad_ref_vlan_b.name})
 
-
-test_running_vm_vlan_change.__test__ = False
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm_localnet, iface_name=IFACE_B_NAME).ipAddresses
+    ):
+        with subtests.test(msg=f"IPv{server_ip.version} connectivity on {IFACE_B_NAME}"):
+            poll_tcp_connectivity(
+                client_vm=under_test_vm_localnet,
+                server_vm=ref_vm_localnet,
+                server_ip=str(server_ip),
+                server_bind_dev=GUEST_2ND_IFACE_NAME,
+                client_bind_dev=GUEST_1ST_IFACE_NAME,
+            )
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm_localnet, iface_name=IFACE_A_NAME).ipAddresses
+    ):
+        with subtests.test(msg=f"IPv{server_ip.version} no connectivity on {IFACE_A_NAME}"):
+            poll_tcp_connectivity(
+                client_vm=under_test_vm_localnet,
+                server_vm=ref_vm_localnet,
+                server_ip=str(server_ip),
+                server_bind_dev=GUEST_1ST_IFACE_NAME,
+                client_bind_dev=GUEST_1ST_IFACE_NAME,
+                expect_connectivity=False,
+            )