net: adapt test helpers to Fedora 43

libs/net/vmspec.py

@@ -1,14 +1,18 @@
+from __future__ import annotations
+
 import ipaddress
 from collections.abc import Callable
-from typing import Any, Final
+from typing import TYPE_CHECKING, Any, Final
 
 from kubernetes.dynamic.client import ResourceField
 from ocp_resources.utils.resource_constants import ResourceConstants
 from ocp_resources.virtual_machine import VirtualMachine
 from timeout_sampler import retry
 
 from libs.vm.spec import Network
-from libs.vm.vm import BaseVirtualMachine
+
+if TYPE_CHECKING:
+    from libs.vm.vm import BaseVirtualMachine
 
 LOOKUP_IFACE_STATUS_TIMEOUT_SEC: Final[int] = 30
 WAIT_FOR_MISSING_IFACE_STATUS_TIMEOUT_SEC: Final[int] = 120

libs/vm/vm.py

@@ -11,6 +11,7 @@
 from ocp_resources.virtual_machine_instance import VirtualMachineInstance
 from pytest_testconfig import config as py_config
 
+from libs.net.vmspec import VMInterfaceSpecNotFoundError
 from libs.vm.spec import (
     Affinity,
     CloudInitNoCloud,
@@ -27,7 +28,6 @@
 from tests.network.libs import cloudinit
 from utilities import infra
 from utilities.constants import CLOUD_INIT_DISK_NAME
-from utilities.network import IfaceNotFound
 from utilities.virt import get_oc_image_info, vm_console_run_commands
 
 if TYPE_CHECKING:
@@ -89,13 +89,13 @@ def wait_for_agent_connected(self) -> None:
 
     def set_interface_state(self, network_name: str, state: str) -> None:
         if not self._spec.template.spec.domain.devices:
-            raise IfaceNotFound(name=network_name)
+            raise VMInterfaceSpecNotFoundError(f"Interface {network_name} not found in VM {self.name} spec")
         for interface in self._spec.template.spec.domain.devices.interfaces or []:
             if interface.name == network_name:
                 interface.state = state
                 break
         else:
-            raise IfaceNotFound(name=network_name)
+            raise VMInterfaceSpecNotFoundError(f"Interface {network_name} not found in VM {self.name} spec")
 
         devices = asdict(obj=self._spec.template.spec.domain.devices, dict_factory=self._filter_out_none_values)
         patches = {

tests/network/l2_bridge/libl2bridge.py

@@ -10,7 +10,12 @@
 from timeout_sampler import TimeoutExpiredError, TimeoutSampler
 
 from libs.net.ip import random_ipv4_address
-from libs.net.vmspec import lookup_iface_status, lookup_iface_status_ip, wait_for_missing_iface_status
+from libs.net.vmspec import (
+    VMInterfaceStatusNotFoundError,
+    lookup_iface_status,
+    lookup_iface_status_ip,
+    wait_for_missing_iface_status,
+)
 from libs.vm.factory import base_vmspec, fedora_vm
 from libs.vm.spec import Affinity, CloudInitNoCloud, Interface, Metadata, Multus, Network
 from libs.vm.vm import BaseVirtualMachine, add_volume_disk, cloudinitdisk_storage
@@ -31,7 +36,6 @@
 )
 from utilities.infra import get_pod_by_name_prefix
 from utilities.network import (
-    IfaceNotFound,
     cloud_init_network_data,
     compose_cloud_init_data_dict,
     network_device,
@@ -251,7 +255,7 @@ def get_guest_vm_interface_name_by_vmi_interface_name(vm, vm_interface_name):
     for interface in vmi_interfaces:
         if interface["name"] == vm_interface_name:
             return interface["interfaceName"]
-    raise IfaceNotFound(name=vm_interface_name)
+    raise VMInterfaceStatusNotFoundError(f"Interface {vm_interface_name} not found in VM {vm.name} status")
 
 
 @contextlib.contextmanager
@@ -294,7 +298,7 @@ def search_hot_plugged_interface_in_vmi(vm, interface_name):
     try:
         return wait_for_interface_hot_plug_completion(vmi=vm.vmi, interface_name=interface_name)
     except TimeoutExpiredError:
-        raise IfaceNotFound(name=interface_name)
+        raise VMInterfaceStatusNotFoundError(f"Interface {interface_name} not found in VM {vm.name} status")
 
 
 def get_kubemacpool_controller_log(

tests/network/l2_bridge/test_bridge_nic_hot_plug.py

@@ -4,7 +4,7 @@
 
 from libs.net import netattachdef
 from libs.net.ip import random_ipv4_address
-from libs.net.vmspec import lookup_iface_status_ip
+from libs.net.vmspec import VMInterfaceStatusNotFoundError, lookup_iface_status_ip
 from tests.network.l2_bridge.libl2bridge import (
     check_mac_released,
     create_bridge_interface_for_hot_plug,
@@ -22,7 +22,6 @@
 )
 from utilities.constants import FLAT_OVERLAY_STR, QUARANTINED, SRIOV
 from utilities.network import (
-    IfaceNotFound,
     assert_ping_successful,
     network_nad,
 )
@@ -634,7 +633,7 @@ def test_hot_unplugged_interface_removed_from_vmi_spec(
         hot_unplugged_additional_interface,
         running_vm_with_secondary_and_hot_plugged_interfaces,
     ):
-        with pytest.raises(IfaceNotFound):
+        with pytest.raises(VMInterfaceStatusNotFoundError):
             search_hot_plugged_interface_in_vmi(
                 vm=running_vm_with_secondary_and_hot_plugged_interfaces,
                 interface_name=hot_unplugged_additional_interface.name,
@@ -688,7 +687,7 @@ def test_hot_unplug_secondary_interface_from_setup(
         hot_unplug_secondary_interface_from_setup,
         network_attachment_definition_for_hot_plug,
     ):
-        with pytest.raises(IfaceNotFound):
+        with pytest.raises(VMInterfaceStatusNotFoundError):
             search_hot_plugged_interface_in_vmi(
                 vm=running_vm_with_secondary_and_hot_plugged_interfaces,
                 interface_name=network_attachment_definition_for_hot_plug.name,

utilities/network.py

@@ -28,6 +28,7 @@
 
 import utilities.infra
 from libs.net.ip import ICMP_HEADER_SIZE, ip_header_size
+from libs.net.vmspec import IpNotFound, VMInterfaceStatusNotFoundError, lookup_iface_status_ip
 from utilities.constants import (
     ACTIVE_BACKUP,
     FLAT_OVERLAY_STR,
@@ -652,19 +653,11 @@ def mac_is_within_range(self, mac):
         return self.mac_to_int(mac) in self.pool
 
 
-class IfaceNotFound(Exception):
-    def __init__(self, name: str) -> None:
-        self.name = name
-
-    def __str__(self) -> str:
-        return f"Interface not found for NAD {self.name}"
-
-
 def get_vmi_mac_address_by_iface_name(vmi, iface_name):
     for iface in vmi.interfaces:
         if iface.name == iface_name:
             return iface.mac
-    raise IfaceNotFound(name=iface_name)
+    raise VMInterfaceStatusNotFoundError(f"Interface {iface_name} not found in VMI {vmi.name} status")
 
 
 def cloud_init_network_data(data):
@@ -760,10 +753,15 @@ def get_ip_from_vm_or_virt_handler_pod(family, vm=None, virt_handler_pod=None):
         raise ValueError("must send VM or virt-handler pod")
 
     if vm:
-        addr_list = vm.vmi.interfaces[0]["ipAddresses"]
-    else:
-        addr_list = [ip_addr["ip"] for ip_addr in virt_handler_pod.instance.status.podIPs]
+        iface_name = vm.vmi.interfaces[0]["name"]
+        ip_family = 4 if family == IPV4_STR else 6
+        try:
+            ip = lookup_iface_status_ip(vm=vm, iface_name=iface_name, ip_family=ip_family)
+        except IpNotFound:
+            return None
+        return str(ip) if ip else None
 
+    addr_list = [ip_addr["ip"] for ip_addr in virt_handler_pod.instance.status.podIPs]
     ip_list = [ip for ip in addr_list if get_valid_ip_address(dst_ip=ip, family=family)]
     return ip_list[0] if ip_list else None
 

utilities/virt.py

@@ -1495,8 +1495,8 @@ def vm_console_run_commands(
         Dict of the commands outputs, where the key is the command and the value is the output as a list of lines.
     """
     output = {}
-    # Source: https://www.tutorialspoint.com/how-can-i-remove-the-ansi-escape-sequences-from-a-string-in-python
-    ansi_escape = re.compile(r"(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]")
+    # Strip CSI (ESC[…) and OSC (ESC]…BEL/ST) terminal escape sequences
+    ansi_escape = re.compile(r"(\x9B|\x1B\[)[0-?]*[ -\/]*[@-~]|\x1B\][^\x07\x1B]*(?:\x07|\x1B\\)")
     prompt = r"\$ "
     with Console(vm=vm, prompt=prompt) as vmc:
         for command in commands: