From 96af76391f7c85315ad45a2d7ff630bbffd01773 Mon Sep 17 00:00:00 2001 From: Sergei Volkov Date: Sat, 16 May 2026 01:35:32 +0200 Subject: [PATCH 1/3] net: replace IfaceNotFound with correct spec/status exceptions IfaceNotFound was a single exception incorrectly covering both spec and status lookup failures with a misleading "Interface not found for NAD" message. Replace each usage with the semantically correct exception: - VMInterfaceSpecNotFoundError for spec.domain.devices.interfaces lookups (libs/vm/vm.py) - VMInterfaceStatusNotFoundError for vmi.status.interfaces lookups (utilities/network.py, tests/network/l2_bridge/) This also eliminates the only libs/ -> utilities/network import, breaking the circular dependency that prevented utilities/network.py from importing libs/net/vmspec. Signed-off-by: Sergei Volkov Assisted-by: Claude Opus 4.6 --- libs/net/vmspec.py | 8 ++++++-- libs/vm/vm.py | 6 +++--- tests/network/l2_bridge/libl2bridge.py | 12 ++++++++---- tests/network/l2_bridge/test_bridge_nic_hot_plug.py | 7 +++---- utilities/network.py | 11 ++--------- 5 files changed, 22 insertions(+), 22 deletions(-) diff --git a/libs/net/vmspec.py b/libs/net/vmspec.py index adff3acf3d..7a8eee4708 100644 --- a/libs/net/vmspec.py +++ b/libs/net/vmspec.py @@ -1,6 +1,8 @@ +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 @@ -8,7 +10,9 @@ 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 diff --git a/libs/vm/vm.py b/libs/vm/vm.py index 18756f7f2c..a93ce83bf9 100644 --- a/libs/vm/vm.py +++ b/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 = { diff --git a/tests/network/l2_bridge/libl2bridge.py b/tests/network/l2_bridge/libl2bridge.py index 93f8b05684..58564992ba 100644 --- a/tests/network/l2_bridge/libl2bridge.py +++ b/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( diff --git a/tests/network/l2_bridge/test_bridge_nic_hot_plug.py b/tests/network/l2_bridge/test_bridge_nic_hot_plug.py index 18511fa6ff..ae2eb35c84 100644 --- a/tests/network/l2_bridge/test_bridge_nic_hot_plug.py +++ b/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, diff --git a/utilities/network.py b/utilities/network.py index 605bf7f2ca..dfdbccb888 100644 --- a/utilities/network.py +++ b/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 VMInterfaceStatusNotFoundError 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): From 97bebd9dbafe3abafc22fbfc3b2fc3fcc70397a8 Mon Sep 17 00:00:00 2001 From: Sergei Volkov Date: Sat, 16 May 2026 01:35:54 +0200 Subject: [PATCH 2/3] net: strip OSC escape sequences from VM console output MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Fedora 43 (systemd 256+) emits OSC 8003 session-tracking escape sequences in the terminal. The existing regex in vm_console_run_commands only stripped CSI sequences (ESC[…), leaving OSC sequences (ESC]…) in the parsed output. This broke any caller that parses console output, e.g. json.loads() in ip_specification tests. Extend the regex to also strip OSC sequences terminated by BEL or ST, per ECMA-48 specification. Signed-off-by: Sergei Volkov Assisted-by: Claude Opus 4.6 --- utilities/virt.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/utilities/virt.py b/utilities/virt.py index 9fbc66267a..6caee75ad6 100644 --- a/utilities/virt.py +++ b/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: From 339b773a72445f365de3229b284e6d15e7d20164 Mon Sep 17 00:00:00 2001 From: Sergei Volkov Date: Sat, 16 May 2026 01:36:17 +0200 Subject: [PATCH 3/3] net: fix race condition in get_ip_from_vm_or_virt_handler_pod The function read vm.vmi.interfaces[0]["ipAddresses"] once without waiting. With Fedora 43 the guest agent reports IPs slower than Fedora 41, so ipAddresses is still None right after AgentConnected becomes True, causing TypeError on iteration. Delegate the VM branch to lookup_iface_status_ip which polls the VMI with a watcher until IPs are reported. Signed-off-by: Sergei Volkov Assisted-by: Claude Opus 4.6 --- utilities/network.py | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/utilities/network.py b/utilities/network.py index dfdbccb888..90b2301e66 100644 --- a/utilities/network.py +++ b/utilities/network.py @@ -28,7 +28,7 @@ import utilities.infra from libs.net.ip import ICMP_HEADER_SIZE, ip_header_size -from libs.net.vmspec import VMInterfaceStatusNotFoundError +from libs.net.vmspec import IpNotFound, VMInterfaceStatusNotFoundError, lookup_iface_status_ip from utilities.constants import ( ACTIVE_BACKUP, FLAT_OVERLAY_STR, @@ -753,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