Re: [PATCH 1/3] tests/funtional: add test_virtio_status.py

["Michael S. Tsirkin" <mst@redhat.com>]

Typo in the subject

On Wed, Sep 23, 2026 at 12:32:08AM +0300, Vladimir Sementsov-Ogievskiy wrote:
> 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 <vsementsov@yandex-team.ru>
> ---
>  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 <mst@redhat.com>
> 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

QEMU uses GPL v2.0 or later.

> +
> +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