net: non-migratable VM NAD ref change negative test

Implement test_non_migratable_vm_nad_change_not_applied (CNV-15947):
verifies that updating the NAD reference on a non-migratable VM does
not apply live — the VM reports a RestartRequired condition and retains
connectivity to the original VLAN-A.

A blank 20Gi RWO DataVolume is attached via two_secondary_bridge_vm's
new data_volume_template parameter, making the VM non-live-migratable.
Connectivity to VLAN-A is verified as a fixture precondition.

Quarantined via CNV-87878: the RestartRequired condition is not yet
wired up in KubeVirt — the wait_for_condition call will be unblocked
when the product fix lands.

Signed-off-by: Asia Khromov <azhivovk@redhat.com>
Assisted-by: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>

Author: azhivovk

tests/network/l2_bridge/nad_ref_change/conftest.py

@@ -2,7 +2,9 @@
 
 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.netattachdef import CNIPluginBridgeConfig, NetConfig, NetworkAttachmentDefinition
@@ -145,3 +147,73 @@ def baseline_connectivity(
             server_bind_dev=GUEST_IFACE_2,
             expect_connectivity=False,
         )
+
+
+@pytest.fixture()
+def non_migratable_under_test_vm(
+    namespace: Namespace,
+    unprivileged_client: DynamicClient,
+    bridge_nad_a: NetworkAttachmentDefinition,
+) -> Generator[BaseVirtualMachine]:
+    iface_a_ips = random_cidr_addresses_by_family(net_seed=NET_SEED, host_address=5)
+    blank_dv = DataVolume(
+        name="non-migratable-dv",
+        namespace=namespace.name,
+        source_dict={"blank": {}},
+        access_modes="ReadWriteOnce",
+        size="20Gi",
+        storage_class=py_config["default_storage_class"],
+        api_name="storage",
+    )
+    blank_dv.to_dict()
+    dv_template = blank_dv.res
+    with two_secondary_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],
+        runcmd=ARP_ISOLATION_SYSCTL_CMD,
+        data_volume_template=dv_template,
+    ) 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.
+    """
+    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=non_migratable_under_test_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=non_migratable_under_test_vm,
+            server_vm=ref_vm,
+            server_ip=str(server_ip),
+            server_bind_dev=GUEST_IFACE_2,
+            expect_connectivity=False,
+        )

tests/network/l2_bridge/nad_ref_change/lib_helpers.py

@@ -1,14 +1,18 @@
-from typing import Final
+from typing import Any, Final
 
 from kubernetes.dynamic import DynamicClient
 
 from libs.vm.factory import base_vmspec, fedora_vm
 from libs.vm.spec import (
     CloudInitNoCloud,
+    DataVolumeRef,
     Devices,
+    Disk,
     Interface,
     Multus,
     Network,
+    SpecDisk,
+    Volume,
 )
 from libs.vm.vm import BaseVirtualMachine, add_volume_disk, cloudinitdisk_storage
 from tests.network.libs import cloudinit
@@ -81,6 +85,7 @@ def two_secondary_bridge_vm(
     ip_addresses: list[list[str]],
     iface_names: list[str],
     runcmd: list[str] | None = None,
+    data_volume_template: dict[str, Any] | None = None,
 ) -> BaseVirtualMachine:
     """Create a Fedora VM with a masquerade primary interface and bridge-bound secondary interfaces.
 
@@ -98,6 +103,8 @@ def two_secondary_bridge_vm(
             Each inner list contains one address per supported IP family.
         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.
+        data_volume_template: When provided, embeds this DataVolume template in the VM spec and
+            attaches it as an additional disk. A RWO DataVolume makes the VM non-live-migratable.
     """
     spec = base_vmspec()
     spec.template.spec.domain.devices = Devices(
@@ -125,5 +132,15 @@ def two_secondary_bridge_vm(
             userData=cloudinit.format_cloud_config(userdata=userdata),
         )
     )
+    if data_volume_template:
+        dv_name = data_volume_template["metadata"]["name"]
+        spec.template.spec = add_volume_disk(
+            vmi_spec=spec.template.spec,
+            volume=Volume(name=dv_name, dataVolume=DataVolumeRef(name=dv_name)),
+            disk=SpecDisk(name=dv_name, disk=Disk(bus="virtio")),
+        )
     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)
+    vm = fedora_vm(namespace=namespace, name=name, client=client, spec=spec)
+    if data_volume_template:
+        vm.body["spec"].setdefault("dataVolumeTemplates", []).append(data_volume_template)
+    return vm

tests/network/l2_bridge/nad_ref_change/test_nad_ref_change.py

@@ -14,15 +14,15 @@
 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, 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,
     GUEST_IFACE_2,
     assert_connectivity,
     assert_no_connectivity,
 )
-from tests.network.libs.nad_ref import update_nad_references
+from tests.network.libs.nad_ref import patch_nad_references, update_nad_references
 
 
 @pytest.mark.usefixtures("baseline_connectivity")
@@ -212,8 +212,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.
@@ -233,6 +241,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,
+            )

tests/network/libs/nad_ref.py

@@ -15,10 +15,28 @@ def update_nad_references(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str
         nad_name_by_net: Mapping of spec network name to new NAD name.
     """
     resource_version = vm.vmi.instance.metadata.resourceVersion
+    networks = _patch_networks(vm=vm, nad_name_by_net=nad_name_by_net)
+    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 patch_nad_references(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str]) -> None:
+    """Patch secondary network NAD references without waiting for the change to be applied.
+
+    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.
+    """
+    networks = _patch_networks(vm=vm, nad_name_by_net=nad_name_by_net)
+    vm.set_networks(networks=networks)
+
+
+def _patch_networks(vm: BaseVirtualMachine, nad_name_by_net: dict[str, str]) -> list:
     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")
+    return networks