net: Nad live-update non-migratable VM test

libs/net/vmspec.py

@@ -306,21 +306,33 @@ def _vmi_condition_not_set(existing_conditions: list[ResourceField], required_co
     )
 
 
-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.
+def patch_nad_references(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str]) -> None:
+    """Patch network NAD references without waiting for the change to be applied.
 
-    Patches the VM spec atomically, then waits for the MigrationRequired condition to
-    appear (change detected) and disappear (migration completed).
+    The caller is responsible for waiting on the appropriate VMI condition after this call.
 
     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)
+
+
+def update_nad_references(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str]) -> None:
+    """Update 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
+    patch_nad_references(vm=vm, nad_name_by_net=nad_name_by_net)
     wait_for_vmi_condition_status(vm=vm, condition="MigrationRequired", resource_version=resource_version)
     wait_for_no_vmi_condition(vm=vm, condition="MigrationRequired")

libs/vm/spec.py

@@ -156,6 +156,12 @@ class Volume:
     name: str
     containerDisk: ContainerDisk | None = None  # noqa: N815
     cloudInitNoCloud: CloudInitNoCloud | None = None  # noqa: N815
+    dataVolume: DataVolumeRef | None = None  # noqa: N815
+
+
+@dataclass
+class DataVolumeRef:
+    name: str
 
 
 @dataclass

libs/vm/vm.py

@@ -16,6 +16,7 @@
     Affinity,
     CloudInitNoCloud,
     ContainerDisk,
+    DataVolumeRef,
     Devices,
     Disk,
     Metadata,
@@ -248,6 +249,10 @@ def cloudinitdisk_storage(data: CloudInitNoCloud) -> tuple[SpecDisk, Volume]:
     )
 
 
+def data_volume_storage(name: str) -> tuple[SpecDisk, Volume]:
+    return SpecDisk(name=name, disk=Disk(bus="virtio")), Volume(name=name, dataVolume=DataVolumeRef(name=name))
+
+
 def add_volume_disk(vmi_spec: VMISpec, volume: Volume, disk: SpecDisk) -> VMISpec:
     vmi_spec.volumes = vmi_spec.volumes or []
     vmi_spec.volumes.append(volume)

tests/network/l2_bridge/nad_ref_change/conftest.py

@@ -2,20 +2,22 @@
 
 import pytest
 from kubernetes.dynamic import DynamicClient
+from ocp_resources.datavolume import DataVolume
 from ocp_resources.namespace import Namespace
+from pytest_testconfig import config as py_config
 
-from libs.net.ip import filter_link_local_addresses, random_cidr_addresses_by_family
+from libs.net.ip import random_cidr_addresses_by_family
 from libs.net.netattachdef import NetworkAttachmentDefinition
-from libs.net.vmspec import lookup_iface_status, wait_for_ifaces_status
+from libs.net.vmspec import 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
 from tests.network.l2_bridge.nad_ref_change.lib_helpers import (
-    GUEST_IFACE_1,
-    GUEST_IFACE_2,
     NET_SEED,
+    assert_baseline_connectivity,
+    non_migratable_bridge_vm,
     two_secondary_bridge_vm,
 )
-from tests.network.libs.connectivity import ARP_ISOLATION_SYSCTL_CMD, poll_tcp_connectivity
+from tests.network.libs.connectivity import ARP_ISOLATION_SYSCTL_CMD
 
 
 @pytest.fixture(scope="module")
@@ -87,22 +89,65 @@ def baseline_connectivity(
     Asserts that the under-test VM can reach the reference VM on VLAN-A (iface-1)
     and cannot reach it on VLAN-B (iface-2) before any NAD update is applied.
     """
-    for server_ip in filter_link_local_addresses(
-        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_1).ipAddresses
-    ):
-        poll_tcp_connectivity(
-            client_vm=under_test_vm_two_ifaces,
-            server_vm=ref_vm,
-            server_ip=str(server_ip),
-            server_bind_dev=GUEST_IFACE_1,
-        )
-    for server_ip in filter_link_local_addresses(
-        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_2).ipAddresses
-    ):
-        poll_tcp_connectivity(
-            client_vm=under_test_vm_two_ifaces,
-            server_vm=ref_vm,
-            server_ip=str(server_ip),
-            server_bind_dev=GUEST_IFACE_2,
-            expect_connectivity=False,
+    assert_baseline_connectivity(client_vm=under_test_vm_two_ifaces, ref_vm=ref_vm)
+
+
+@pytest.fixture()
+def non_migratable_dv(
+    namespace: Namespace,
+    unprivileged_client: DynamicClient,
+) -> Generator[DataVolume]:
+    with DataVolume(
+        name="non-migratable-dv",
+        namespace=namespace.name,
+        source_dict={"blank": {}},
+        access_modes=DataVolume.AccessMode.RWO,
+        size="20Gi",
+        storage_class=py_config["default_storage_class"],
+        api_name="storage",
+        client=unprivileged_client,
+    ) as dv:
+        dv.wait_for_dv_success()
+        yield dv
+
+
+@pytest.fixture()
+def non_migratable_under_test_vm(
+    namespace: Namespace,
+    unprivileged_client: DynamicClient,
+    bridge_nad_a: NetworkAttachmentDefinition,
+    non_migratable_dv: DataVolume,
+) -> Generator[BaseVirtualMachine]:
+    iface_a_ips = random_cidr_addresses_by_family(net_seed=NET_SEED, host_address=5)
+    with non_migratable_bridge_vm(
+        namespace=namespace.name,
+        name="non-migratable-under-test-vm",
+        client=unprivileged_client,
+        nad_names=[bridge_nad_a.name],
+        ip_addresses=[iface_a_ips],
+        iface_names=[LINUX_BRIDGE_IFACE_NAME_1],
+        data_volume=non_migratable_dv,
+        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={
+                LINUX_BRIDGE_IFACE_NAME_1: [addr.split("/")[0] for addr in iface_a_ips],
+            },
         )
+        yield vm
+
+
+@pytest.fixture()
+def non_migratable_baseline_connectivity(
+    ref_vm: BaseVirtualMachine,
+    non_migratable_under_test_vm: BaseVirtualMachine,
+) -> None:
+    """Verify baseline connectivity before the NAD reference change.
+
+    Asserts that the non-migratable under-test VM can reach the reference VM on VLAN-A
+    and cannot reach it on VLAN-B before any NAD update is applied.
+    """
+    assert_baseline_connectivity(client_vm=non_migratable_under_test_vm, ref_vm=ref_vm)

tests/network/l2_bridge/nad_ref_change/lib_helpers.py

@@ -1,16 +1,21 @@
 from typing import Final
 
 from kubernetes.dynamic import DynamicClient
+from ocp_resources.datavolume import DataVolume
 
+from libs.net.ip import filter_link_local_addresses
+from libs.net.vmspec import lookup_iface_status
 from libs.vm.factory import base_vmspec, fedora_vm
 from libs.vm.spec import (
     CloudInitNoCloud,
     Devices,
     Interface,
     Multus,
     Network,
+    VMSpec,
 )
-from libs.vm.vm import BaseVirtualMachine, add_volume_disk, cloudinitdisk_storage
+from libs.vm.vm import BaseVirtualMachine, add_volume_disk, cloudinitdisk_storage, data_volume_storage
+from tests.network.l2_bridge.libl2bridge import LINUX_BRIDGE_IFACE_NAME_1, LINUX_BRIDGE_IFACE_NAME_2
 from tests.network.libs import cloudinit
 from tests.network.libs.cloudinit import primary_iface_cloud_init
 from tests.network.libs.connectivity import poll_tcp_connectivity
@@ -73,6 +78,37 @@ def assert_no_connectivity(
     )
 
 
+def assert_baseline_connectivity(
+    client_vm: BaseVirtualMachine,
+    ref_vm: BaseVirtualMachine,
+) -> None:
+    """Assert baseline connectivity: client reaches ref on VLAN-A, not on VLAN-B.
+
+    Args:
+        client_vm: VM initiating the connections.
+        ref_vm: Reference VM with interfaces on both VLANs.
+    """
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_1).ipAddresses
+    ):
+        poll_tcp_connectivity(
+            client_vm=client_vm,
+            server_vm=ref_vm,
+            server_ip=str(server_ip),
+            server_bind_dev=GUEST_IFACE_1,
+        )
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_2).ipAddresses
+    ):
+        poll_tcp_connectivity(
+            client_vm=client_vm,
+            server_vm=ref_vm,
+            server_ip=str(server_ip),
+            server_bind_dev=GUEST_IFACE_2,
+            expect_connectivity=False,
+        )
+
+
 def two_secondary_bridge_vm(
     namespace: str,
     name: str,
@@ -99,6 +135,62 @@ def two_secondary_bridge_vm(
         iface_names: Logical interface names for the VM spec, aligned with nad_names.
         runcmd: Commands to run on first boot via cloud-init runcmd. None means no extra commands.
     """
+    return fedora_vm(
+        namespace=namespace,
+        name=name,
+        client=client,
+        spec=_bridge_vm_spec(
+            nad_names=nad_names,
+            ip_addresses=ip_addresses,
+            iface_names=iface_names,
+            runcmd=runcmd,
+        ),
+    )
+
+
+def non_migratable_bridge_vm(
+    namespace: str,
+    name: str,
+    client: DynamicClient,
+    nad_names: list[str],
+    ip_addresses: list[list[str]],
+    iface_names: list[str],
+    data_volume: DataVolume,
+    runcmd: list[str] | None = None,
+) -> BaseVirtualMachine:
+    """Create a Fedora VM with secondary bridge interfaces backed by an existing RWO DataVolume.
+
+    The VM references the provided, already-deployed DataVolume by name via a dataVolume
+    volume reference. The RWO access mode of the backing PVC makes the VM non-live-migratable.
+
+    Args:
+        namespace: Namespace to deploy the VM in.
+        name: VM name.
+        client: Kubernetes dynamic client.
+        nad_names: NAD names for the secondary interfaces, in spec order.
+        ip_addresses: Per-interface CIDR address lists, aligned with nad_names.
+        iface_names: Logical interface names for the VM spec, aligned with nad_names.
+        data_volume: Existing deployed DataVolume referenced by name; its RWO access mode
+            makes the VM non-migratable.
+        runcmd: Commands to run on first boot via cloud-init runcmd. None means no extra commands.
+    """
+    spec = _bridge_vm_spec(
+        nad_names=nad_names,
+        ip_addresses=ip_addresses,
+        iface_names=iface_names,
+        runcmd=runcmd,
+    )
+    disk, volume = data_volume_storage(name=data_volume.name)
+    spec.template.spec = add_volume_disk(vmi_spec=spec.template.spec, volume=volume, disk=disk)
+    return fedora_vm(namespace=namespace, name=name, client=client, spec=spec)
+
+
+def _bridge_vm_spec(
+    nad_names: list[str],
+    ip_addresses: list[list[str]],
+    iface_names: list[str],
+    runcmd: list[str] | None = None,
+) -> VMSpec:
     spec = base_vmspec()
     spec.template.spec.domain.devices = Devices(
         interfaces=[
@@ -126,4 +218,4 @@ def two_secondary_bridge_vm(
         )
     )
     spec.template.spec = add_volume_disk(vmi_spec=spec.template.spec, volume=volume, disk=disk)
-    return fedora_vm(namespace=namespace, name=name, client=client, spec=spec)
+    return spec

tests/network/l2_bridge/nad_ref_change/test_nad_ref_change.py

@@ -14,7 +14,12 @@
 import pytest
 
 from libs.net.ip import filter_link_local_addresses
-from libs.net.vmspec import lookup_iface_status, update_nad_references
+from libs.net.vmspec import (
+    lookup_iface_status,
+    patch_nad_references,
+    update_nad_references,
+    wait_for_vmi_condition_status,
+)
 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,
@@ -211,8 +216,16 @@ def test_two_networks(
                 )
 
 
+@pytest.mark.jira("CNV-87878", run=False)
+@pytest.mark.usefixtures("non_migratable_baseline_connectivity")
 @pytest.mark.polarion("CNV-15947")
-def test_non_migratable_vm_nad_change_not_applied():
+def test_non_migratable_vm_nad_change_not_applied(
+    subtests,
+    non_migratable_under_test_vm,
+    ref_vm,
+    bridge_nad_a,
+    bridge_nad_b,
+):
     """
     [NEGATIVE] Test that changing the NAD reference on a non-migratable VM does not
     silently succeed — the VM remains connected to the original network.
@@ -232,6 +245,29 @@ def test_non_migratable_vm_nad_change_not_applied():
         - Non-migratable under-test VM retains connectivity to the reference VM on NAD-VLAN-A
         - Non-migratable under-test VM has no connectivity to the reference VM on NAD-VLAN-B
     """
-
-
-test_non_migratable_vm_nad_change_not_applied.__test__ = False
+    patch_nad_references(
+        vm=non_migratable_under_test_vm, nad_name_by_net={LINUX_BRIDGE_IFACE_NAME_1: bridge_nad_b.name}
+    )
+    wait_for_vmi_condition_status(vm=non_migratable_under_test_vm, condition="RestartRequired")
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_1).ipAddresses
+    ):
+        with subtests.test(msg=f"IPv{server_ip.version} on {bridge_nad_a.name}"):
+            assert_connectivity(
+                client_vm=non_migratable_under_test_vm,
+                server_vm=ref_vm,
+                server_ip=str(server_ip),
+                server_bind_dev=GUEST_IFACE_1,
+                client_bind_dev=GUEST_IFACE_1,
+            )
+    for server_ip in filter_link_local_addresses(
+        ip_addresses=lookup_iface_status(vm=ref_vm, iface_name=LINUX_BRIDGE_IFACE_NAME_2).ipAddresses
+    ):
+        with subtests.test(msg=f"IPv{server_ip.version} on {bridge_nad_b.name}"):
+            assert_no_connectivity(
+                client_vm=non_migratable_under_test_vm,
+                server_vm=ref_vm,
+                server_ip=str(server_ip),
+                server_bind_dev=GUEST_IFACE_2,
+                client_bind_dev=GUEST_IFACE_1,
+            )