On Fri, Sep 05, 2025 at 04:50:39PM +0300, Vladimir Sementsov-Ogievskiy wrote: > Add test for a new feature of local TAP migration with fd passing > through unix socket. > > Signed-off-by: Vladimir Sementsov-Ogievskiy <[email protected]> > --- > .../test_x86_64_tap_fd_migration.py | 345 ++++++++++++++++++ > 1 file changed, 345 insertions(+) > create mode 100644 tests/functional/test_x86_64_tap_fd_migration.py > > diff --git a/tests/functional/test_x86_64_tap_fd_migration.py > b/tests/functional/test_x86_64_tap_fd_migration.py > new file mode 100644 > index 0000000000..a38dba39fe > --- /dev/null > +++ b/tests/functional/test_x86_64_tap_fd_migration.py > @@ -0,0 +1,345 @@ > +#!/usr/bin/env python3 > +# > +# Functional test that tests TAP local migration > +# with fd passing > +# > +# Copyright (c) Yandex > +# > +# SPDX-License-Identifier: GPL-2.0-or-later > + > +import os > +import time > +import subprocess > +from subprocess import run > +import signal > +from typing import Tuple > + > +from qemu_test import ( > + LinuxKernelTest, > + Asset, > + exec_command_and_wait_for_pattern, > +) > +from qemu_test.decorators import skipUnlessPasswordlessSudo > + > +GUEST_IP = "10.0.1.2" > +GUEST_IP_MASK = f"{GUEST_IP}/24" > +GUEST_MAC = "d6:0d:75:f8:0f:b7" > +HOST_IP = "10.0.1.1" > +HOST_IP_MASK = f"{HOST_IP}/24" > +TAP_ID = "tap0" > +TAP_MAC = "e6:1d:44:b5:03:5d" > + > + > +def del_tap() -> None: > + run( > + ["sudo", "ip", "tuntap", "del", TAP_ID, "mode", "tap", > "multi_queue"], > + check=True, > + ) > + > + > +def init_tap() -> None: > + run( > + ["sudo", "ip", "tuntap", "add", "dev", TAP_ID, "mode", "tap", > "multi_queue"], > + check=True, > + ) > + run(["sudo", "ip", "link", "set", "dev", TAP_ID, "address", TAP_MAC], > check=True) > + run(["sudo", "ip", "addr", "add", HOST_IP_MASK, "dev", TAP_ID], > check=True) > + run(["sudo", "ip", "link", "set", TAP_ID, "up"], check=True) > + > + > +def parse_ping_line(line: str) -> float: > + # suspect lines like > + # [1748524876.590509] 64 bytes from 94.245.155.3 \ > + # (94.245.155.3): icmp_seq=1 ttl=250 time=101 ms > + spl = line.split() > + return float(spl[0][1:-1]) > + > + > +def parse_ping_output(out) -> Tuple[bool, float, float]: > + lines = [x for x in out.split("\n") if x.startswith("[")] > + > + try: > + first_no_ans = next( > + (ind for ind in range(len(lines)) if lines[ind][20:26] == "no > ans") > + ) > + except StopIteration: > + return False, parse_ping_line(lines[0]), parse_ping_line(lines[-1]) > + > + last_no_ans = next( > + (ind for ind in range(len(lines) - 1, -1, -1) if lines[ind][20:26] > == "no ans") > + ) > + > + return ( > + True, > + parse_ping_line(lines[first_no_ans]), > + parse_ping_line(lines[last_no_ans]), > + ) > + > + > +def wait_migration_finish(source_vm, target_vm): > + migr_events = ( > + ("MIGRATION", {"data": {"status": "completed"}}), > + ("MIGRATION", {"data": {"status": "failed"}}), > + ) > + > + source_e = source_vm.events_wait(migr_events)["data"] > + target_e = target_vm.events_wait(migr_events)["data"] > + > + source_s = source_vm.cmd("query-status")["status"] > + target_s = target_vm.cmd("query-status")["status"] > + > + assert ( > + source_e["status"] == "completed" > + and target_e["status"] == "completed" > + and source_s == "postmigrate" > + and target_s == "paused" > + ), f"""Migration failed: > + SRC status: {source_s} > + SRC event: {source_e} > + TGT status: {target_s} > + TGT event:{target_e}""" > + > + > +@skipUnlessPasswordlessSudo() > +class VhostUserBlkFdMigration(LinuxKernelTest): > + > + ASSET_KERNEL = Asset( > + ( > + > "https://archives.fedoraproject.org/pub/archive/fedora/linux/releases"; > + "/31/Server/x86_64/os/images/pxeboot/vmlinuz" > + ), > + "d4738d03dbbe083ca610d0821d0a8f1488bebbdccef54ce33e3adb35fda00129", > + ) > + > + ASSET_INITRD = Asset( > + ( > + > "https://archives.fedoraproject.org/pub/archive/fedora/linux/releases"; > + "/31/Server/x86_64/os/images/pxeboot/initrd.img" > + ), > + "277cd6c7adf77c7e63d73bbb2cded8ef9e2d3a2f100000e92ff1f8396513cd8b", > + ) > + > + ASSET_ALPINE_ISO = Asset( > + ( > + "https://dl-cdn.alpinelinux.org/"; > + "alpine/v3.22/releases/x86_64/alpine-standard-3.22.1-x86_64.iso" > + ), > + "96d1b44ea1b8a5a884f193526d92edb4676054e9fa903ad2f016441a0fe13089", > + ) > + > + def setUp(self): > + super().setUp() > + > + init_tap() > + > + self.outer_ping_proc = None > + > + def tearDown(self): > + del_tap() > + > + if self.outer_ping_proc: > + self.stop_outer_ping()
Wrap the del_tap() and onwards in a 'try' clause and then use 'finally' to run the parent tearDown, so we're more robust if anything throws an exception > + > + super().tearDown() With regards, Daniel -- |: https://berrange.com -o- https://www.flickr.com/photos/dberrange :| |: https://libvirt.org -o- https://fstop138.berrange.com :| |: https://entangle-photo.org -o- https://www.instagram.com/dberrange :|
