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..90b2301e66 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 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 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: