Add test that run x-query-virtio-status QMP command, and its HMP analog. Now QEMU crashes in these scenarios due to bugs, so the test-cases are skipped. You may run them by setting QEMU_FAILING_TESTS environment variable.
Following commits will fix bug and enable the test-cases. Wrap QEMU run by "unshare" to access vhost-net without sudo. Signed-off-by: Vladimir Sementsov-Ogievskiy <[email protected]> --- MAINTAINERS | 1 + tests/functional/qemu_test/testcase.py | 9 +- tests/functional/x86_64/meson.build | 2 + tests/functional/x86_64/test_virtio_status.py | 97 +++++++++++++++++++ 4 files changed, 105 insertions(+), 4 deletions(-) create mode 100755 tests/functional/x86_64/test_virtio_status.py diff --git a/MAINTAINERS b/MAINTAINERS index 6be9c5725b9..8bd7c8472d9 100644 --- a/MAINTAINERS +++ b/MAINTAINERS @@ -2535,6 +2535,7 @@ F: include/hw/virtio/ F: docs/devel/virtio* F: docs/devel/migration/virtio.rst F: tests/functional/x86_64/test_virtio_version.py +F: tests/functional/x86_64/test_virtio_status.py virtio-balloon M: Michael S. Tsirkin <[email protected]> diff --git a/tests/functional/qemu_test/testcase.py b/tests/functional/qemu_test/testcase.py index bf2dc1dc101..d719b79eb1e 100644 --- a/tests/functional/qemu_test/testcase.py +++ b/tests/functional/qemu_test/testcase.py @@ -387,12 +387,13 @@ def require_device(self, devicename): if helptxt.find(devicename) < 0: self.skipTest('no support for device ' + devicename) - def _new_vm(self, name, monitor_address): + def _new_vm(self, name, monitor_address, wrapper): vm = QEMUMachine(self.qemu_bin, name=name, base_temp_dir=self.workdir, log_dir=self.log_file(), - monitor_address=monitor_address) + monitor_address=monitor_address, + wrapper=wrapper) self.log.debug('QEMUMachine "%s" created', name) self.log.debug('QEMUMachine "%s" temp_dir: %s', name, vm.temp_dir) @@ -407,11 +408,11 @@ def _new_vm(self, name, monitor_address): def vm(self): return self.get_vm(name='default') - def get_vm(self, name=None, monitor_address=None): + def get_vm(self, name=None, monitor_address=None, wrapper=()): if not name: name = str(uuid.uuid4()) if self._vms.get(name) is None: - self._vms[name] = self._new_vm(name, monitor_address) + self._vms[name] = self._new_vm(name, monitor_address, wrapper) if self.cpu is not None: self._vms[name].add_args('-cpu', self.cpu) if self.machine is not None: diff --git a/tests/functional/x86_64/meson.build b/tests/functional/x86_64/meson.build index 0353b2af8e1..62dedb2e203 100644 --- a/tests/functional/x86_64/meson.build +++ b/tests/functional/x86_64/meson.build @@ -7,6 +7,7 @@ test_x86_64_timeouts = { 'netdev_ethtool' : 180, 'replay' : 480, 'virtio_balloon': 120, + 'virtio_status': 120, } tests_x86_64_system_quick = [ @@ -50,5 +51,6 @@ tests_x86_64_system_thorough = [ 'vhost_user_bridge', 'virtio_balloon', 'virtio_gpu', + 'virtio_status', 'rebuild_vmfd', ] diff --git a/tests/functional/x86_64/test_virtio_status.py b/tests/functional/x86_64/test_virtio_status.py new file mode 100755 index 00000000000..c8cbdf7c42e --- /dev/null +++ b/tests/functional/x86_64/test_virtio_status.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +# +# SPDX-License-Identifier: MIT + +import os +import subprocess + +from qemu_test import ( + Asset, + LinuxKernelTest, + skipIfMissingCommands, + skipIfMissingEnv, +) + + +class VirtioStatus(LinuxKernelTest): + ASSET_KERNEL = Asset( + "https://dl-cdn.alpinelinux.org/alpine/v3.24/releases/x86_64/"; + "netboot-3.24.2/vmlinuz-virt", + "be8ae7782de532d1a791c97fbf9ddc955b9de78349382ad097c83ffe9db857f4") + ASSET_INITRD = Asset( + "https://dl-cdn.alpinelinux.org/alpine/v3.24/releases/x86_64/"; + "netboot-3.24.2/initramfs-virt", + "385418a99b17c0ef44947d88ea448d43b2df4a2c3343da72e724c6cf3b05e9f7") + + VIRTIO_PATH = "/machine/peripheral/net0/virtio-backend" + + def setUp(self): + super().setUp() + self.qemu_wrapper = [] + self.require_accelerator("kvm") + self.require_device("virtio-net-pci") + self.set_machine("q35") + + def has_hmp(self, vm): + commands = vm.cmd('query-commands') + return any(cmd['name'] == 'human-monitor-command' for cmd in commands) + + def check_commands(self, vm): + if self.has_hmp(vm): + vm.cmd( + "human-monitor-command", + command_line="info virtio-status " + self.VIRTIO_PATH, + ) + vm.cmd("x-query-virtio-status", path=self.VIRTIO_PATH) + + @skipIfMissingCommands("unshare") + @skipIfMissingEnv("QEMU_FAILING_TESTS") + def test_vhost(self): + self.require_netdev("tap") + for device in ("/dev/net/tun", "/dev/vhost-net"): + if not os.access(device, os.R_OK | os.W_OK): + self.skipTest(f"Read/write access to {device} is required") + + wrapper = ["unshare", "-Urn", "--"] + probe = subprocess.run(wrapper + ["true"], + capture_output=True, + text=True, check=False) + if probe.returncode: + self.skipTest("User/network namespaces unavailable: " + + probe.stderr.strip()) + + vm = self.get_vm("wrapped", wrapper=wrapper) + + kernel = self.ASSET_KERNEL.fetch() + initrd = self.ASSET_INITRD.fetch() + + vm.set_console() + vm.add_args("-accel", "kvm") + vm.add_args("-kernel", kernel, + "-initrd", initrd, + "-append", "console=ttyS0 modules=virtio_net", + "-netdev", + "tap,id=net,script=,downscript=,vhost=on", + "-device", + "virtio-net-pci,id=net0,netdev=net,romfile=") + vm.launch() + + # modules=virtio_net ensures the driver is loaded by this point + self.wait_for_console_pattern("Loading boot drivers: ok.", vm=vm) + vm.cmd( + "x-query-virtio-vhost-queue-status", path=self.VIRTIO_PATH, queue=0 + ) + + self.check_commands(vm) + + @skipIfMissingEnv("QEMU_FAILING_TESTS") + def test_no_vhost(self): + self.vm.add_args("-accel", "kvm", + "-S", "-net", "none", "-device", + "virtio-net-pci,id=net0,romfile=") + self.vm.launch() + self.check_commands(self.vm) + + +if __name__ == "__main__": + LinuxKernelTest.main() -- 2.43.0
