Add vnic_info metric test for NAD reference change

tests/observability/metrics/conftest.py

@@ -15,7 +15,12 @@
 from pytest_testconfig import py_config
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
+from libs.net.vmspec import update_nad_references
+from libs.vm.factory import base_vmspec, fedora_vm
+from libs.vm.spec import Devices, Interface, Multus, Network
 from tests.observability.metrics.constants import (
+    BINDING_NAME,
+    BINDING_TYPE,
     GUEST_LOAD_TIME_PERIODS,
     KUBEVIRT_CONSOLE_ACTIVE_CONNECTIONS_BY_VMI,
     KUBEVIRT_VM_CREATED_BY_POD_TOTAL,
@@ -27,6 +32,7 @@
 )
 from tests.observability.metrics.utils import (
     SINGLE_VM,
+    binding_name_and_type_from_vm_or_vmi,
     create_windows11_wsl2_vm,
     disk_file_system_info,
     enable_swap_fedora_vm,
@@ -100,6 +106,7 @@
     KUBEVIRT_VMI_MEMORY_PGMINFAULT_TOTAL,
 ]
 MINIMUM_QEMU_GUEST_AGENT_VERSION_FOR_GUEST_LOAD_METRICS = "9.6"
+_NAD_SWAP_SECONDARY_IFACE = "secondary"
 
 
 @pytest.fixture(scope="module")
@@ -661,3 +668,54 @@ def expected_cpu_affinity_metric_value(admin_client, vm_with_cpu_spec):
 
     # return multiplication for multi-CPU VMs
     return str(cpu_count_from_vm_node * cpu_count_from_vm)
+
+
+@pytest.fixture(scope="class")
+def vm_for_nad_swap_test(
+    unprivileged_client,
+    namespace,
+    bridge_nad_a,
+):
+    vm_name = "vm-nad-swap-vnic-info"
+    spec = base_vmspec()
+    spec.template.spec.domain.devices = Devices(
+        interfaces=[
+            Interface(name="default", masquerade={}),
+            Interface(name=_NAD_SWAP_SECONDARY_IFACE, bridge={}),
+        ]
+    )
+    spec.template.spec.networks = [
+        Network(name="default", pod={}),
+        Network(name=_NAD_SWAP_SECONDARY_IFACE, multus=Multus(networkName=bridge_nad_a.name)),
+    ]
+    with fedora_vm(namespace=namespace.name, name=vm_name, client=unprivileged_client, spec=spec) as vm:
+        vm.start(wait=True)
+        yield vm
+
+
+@pytest.fixture(scope="class")
+def post_nad_swap_vm(
+    vm_for_nad_swap_test,
+    bridge_nad_b,
+):
+    update_nad_references(
+        vm=vm_for_nad_swap_test,
+        nad_name_by_net={_NAD_SWAP_SECONDARY_IFACE: bridge_nad_b.name},
+    )
+    yield vm_for_nad_swap_test
+
+
+@pytest.fixture(scope="class")
+def expected_vnic_info_after_swap(
+    post_nad_swap_vm,
+    bridge_nad_b,
+):
+    vm_interfaces = post_nad_swap_vm.instance.spec.template.spec.domain.devices.interfaces
+    secondary_interface = next(iface for iface in vm_interfaces if iface["name"] == _NAD_SWAP_SECONDARY_IFACE)
+    binding_info = binding_name_and_type_from_vm_or_vmi(vm_interface=secondary_interface)
+    return {
+        "vnic_name": _NAD_SWAP_SECONDARY_IFACE,
+        BINDING_NAME: binding_info[BINDING_NAME],
+        BINDING_TYPE: binding_info[BINDING_TYPE],
+        "network": bridge_nad_b.name,
+    }

tests/observability/metrics/test_vms_metrics.py

@@ -461,7 +461,9 @@ def test_metric_kubevirt_vmi_vnic_info_windows(self, prometheus, windows_vm_for_
             ),
         ],
     )
-    def test_metric_kubevirt_vm_vnic_info_after_nad_swap(self, query):
+    def test_metric_kubevirt_vm_vnic_info_after_nad_swap(
+        self, prometheus, post_nad_swap_vm, expected_vnic_info_after_swap, query
+    ):
         """
         Test that vnic_info metric updates the network label after a NAD swap.
 
@@ -474,13 +476,17 @@ def test_metric_kubevirt_vm_vnic_info_after_nad_swap(self, query):
 
         Steps:
             1. Swap the VM secondary network reference from NAD-A to NAD-B
-            2. Query vnic_info metric for the secondary interface
+            2. Wait for the live migration triggered by the swap to complete
+            3. Query vnic_info metric for the secondary interface
 
         Expected:
             - vnic_info labels match the VM spec after NAD swap
         """
-
-    test_metric_kubevirt_vm_vnic_info_after_nad_swap.__test__ = False
+        validate_vnic_info(
+            prometheus=prometheus,
+            vnic_info_to_compare=expected_vnic_info_after_swap,
+            metric_name=query.format(vm_name=post_nad_swap_vm.name),
+        )
 
 
 class TestVmiPhaseTransitionFromDeletion:

tests/observability/metrics/utils.py

@@ -595,21 +595,21 @@ def validate_vnic_info(prometheus: Prometheus, vnic_info_to_compare: dict[str, s
         func=prometheus.query,
         query=metric_name,
     )
-    sample = None
+    mismatch_vnic_info = None
     try:
         for sample in samples:
             if sample and (result := sample.get("data", {}).get("result")):
                 vnic_info_metric_result = result[0].get("metric")
-                break
+                mismatch_vnic_info = {}
+                for info, expected_value in vnic_info_to_compare.items():
+                    actual_value = vnic_info_metric_result.get(info)
+                    if actual_value != expected_value:
+                        mismatch_vnic_info[info] = {f"Expected: {expected_value}", f"Actual: {actual_value}"}
+                if not mismatch_vnic_info:
+                    return
     except TimeoutExpiredError:
-        LOGGER.error(f"Metric value of: {metric_name} is: {sample}, should not be empty.")
+        LOGGER.error(f"There is a mismatch between expected and actual results:\n {mismatch_vnic_info}")
         raise
-    mismatch_vnic_info = {}
-    for info, expected_value in vnic_info_to_compare.items():
-        actual_value = vnic_info_metric_result.get(info)
-        if actual_value != expected_value:
-            mismatch_vnic_info[info] = {f"Expected: {expected_value}", f"Actual: {actual_value}"}
-    assert not mismatch_vnic_info, f"There is a mismatch between expected and actual results:\n {mismatch_vnic_info}"
 
 
 def get_metric_labels_non_empty_value(prometheus: Prometheus, metric_name: str) -> dict[str, str]: