qemu/tests/functional/x86_64/test_vfio_user_client.py

#!/usr/bin/env python3
#
# Copyright (c) 2025 Nutanix, Inc.
#
# Author:
#  Mark Cave-Ayland <mark.caveayland@nutanix.com>
#  John Levon <john.levon@nutanix.com>
#
# SPDX-License-Identifier: GPL-2.0-or-later
"""
Check basic vfio-user-pci client functionality. The test starts two VMs:

    - the server VM runs the libvfio-user "gpio" example server inside it,
      piping vfio-user traffic between a local UNIX socket and a virtio-serial
      port. On the host, the virtio-serial port is backed by a local socket.

    - the client VM loads the gpio-pci-idio-16 kernel module, with the
      vfio-user client connecting to the above local UNIX socket.

This way, we don't depend on trying to run a vfio-user server on the host
itself.

Once both VMs are running, we run some basic configuration on the gpio device
and verify that the server is logging the expected out. As this is consistent
given the same VM images, we just do a simple direct comparison.
"""

import os

from qemu_test import Asset
from qemu_test import QemuSystemTest
from qemu_test import exec_command_and_wait_for_pattern
from qemu_test import wait_for_console_pattern

# Exact output can vary, so we just sample for some expected lines.
EXPECTED_SERVER_LINES = [
    "gpio: adding DMA region [0, 0xc0000) offset=0 flags=0x3",
    "gpio: devinfo flags 0x3, num_regions 9, num_irqs 5",
    "gpio: region_info[0] offset 0 flags 0 size 0 argsz 32",
    "gpio: region_info[1] offset 0 flags 0 size 0 argsz 32",
    "gpio: region_info[2] offset 0 flags 0x3 size 256 argsz 32",
    "gpio: region_info[3] offset 0 flags 0 size 0 argsz 32",
    "gpio: region_info[4] offset 0 flags 0 size 0 argsz 32",
    "gpio: region_info[5] offset 0 flags 0 size 0 argsz 32",
    "gpio: region_info[7] offset 0 flags 0x3 size 256 argsz 32",
    "gpio: region7: read 256 bytes at 0",
    "gpio: region7: read 0 from (0x30:4)",
    "gpio: cleared EROM",
    "gpio: I/O space enabled",
    "gpio: memory space enabled",
    "gpio: SERR# enabled",
    "gpio: region7: wrote 0x103 to (0x4:2)",
    "gpio: I/O space enabled",
    "gpio: memory space enabled",
]

class VfioUserClient(QemuSystemTest):
    """vfio-user testing class."""

    ASSET_REPO = 'https://github.com/mcayland-ntx/libvfio-user-test'

    ASSET_KERNEL = Asset(
        f'{ASSET_REPO}/raw/refs/heads/main/images/bzImage',
        '40292fa6ce95d516e26bccf5974e138d0db65a6de0bc540cabae060fe9dea605'
    )

    ASSET_ROOTFS = Asset(
        f'{ASSET_REPO}/raw/refs/heads/main/images/rootfs.ext2',
        'e1e3abae8aebb8e6e77f08b1c531caeacf46250c94c815655c6bbea59fc3d1c1'
    )

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)

        self.kernel_path = None
        self.rootfs_path = None

    def configure_server_vm_args(self, server_vm, sock_path):
        """
        Configuration for the server VM. Set up virtio-serial device backed by
        the given socket path.
        """
        server_vm.add_args('-kernel', self.kernel_path)
        server_vm.add_args('-append', 'console=ttyS0 root=/dev/sda')
        server_vm.add_args('-drive',
            f"file={self.rootfs_path},if=ide,format=raw,id=drv0")
        server_vm.add_args('-snapshot')
        server_vm.add_args('-chardev',
            f"socket,id=sock0,path={sock_path},telnet=off,server=on,wait=off")
        server_vm.add_args('-device', 'virtio-serial')
        server_vm.add_args('-device',
            'virtserialport,chardev=sock0,name=org.fedoraproject.port.0')

    def configure_client_vm_args(self, client_vm, sock_path):
        """
        Configuration for the client VM. Point the vfio-user-pci device to the
        socket path configured above.
        """

        client_vm.add_args('-kernel', self.kernel_path)
        client_vm.add_args('-append', 'console=ttyS0 root=/dev/sda')
        client_vm.add_args('-drive',
            f'file={self.rootfs_path},if=ide,format=raw,id=drv0')
        client_vm.add_args('-snapshot')
        client_vm.add_args('-device',
            '{"driver":"vfio-user-pci",' +
            '"socket":{"path": "%s", "type": "unix"}}' % sock_path)

    def setup_vfio_user_pci_server(self, server_vm):
        """
        Start the libvfio-user server within the server VM, and arrange
        for data to shuttle between its socket and the virtio serial port.
        """
        wait_for_console_pattern(self, 'login:', None, server_vm)
        exec_command_and_wait_for_pattern(self, 'root', '#', None, server_vm)

        exec_command_and_wait_for_pattern(self,
            'gpio-pci-idio-16 -v /tmp/vfio-user.sock >/var/tmp/gpio.out 2>&1 &',
            '#', None, server_vm)

        # wait for libvfio-user socket to appear
        while True:
            out = exec_command_and_wait_for_pattern(self,
                 'ls --color=no /tmp/vfio-user.sock', '#', None, server_vm)
            ls_out = out.decode().splitlines()[1].strip()
            if ls_out == "/tmp/vfio-user.sock":
                break

        exec_command_and_wait_for_pattern(self,
            'socat UNIX-CONNECT:/tmp/vfio-user.sock /dev/vport0p1,ignoreeof ' +
            ' &', '#', None, server_vm)

    def test_vfio_user_pci(self):
        """Run basic sanity test."""

        self.set_machine('pc')
        self.require_device('virtio-serial')
        self.require_device('vfio-user-pci')

        self.kernel_path = self.ASSET_KERNEL.fetch()
        self.rootfs_path = self.ASSET_ROOTFS.fetch()

        sock_dir = self.socket_dir()
        socket_path = os.path.join(sock_dir.name, 'vfio-user.sock')

        server_vm = self.get_vm(name='server')
        server_vm.set_console()
        self.configure_server_vm_args(server_vm, socket_path)

        server_vm.launch()

        self.log.debug('starting libvfio-user server')

        self.setup_vfio_user_pci_server(server_vm)

        client_vm = self.get_vm(name="client")
        client_vm.set_console()
        self.configure_client_vm_args(client_vm, socket_path)

        try:
            client_vm.launch()
        except:
            self.log.error('client VM failed to start, dumping server logs')
            exec_command_and_wait_for_pattern(self, 'cat /var/tmp/gpio.out',
                '#', None, server_vm)
            raise

        self.log.debug('waiting for client VM boot')

        wait_for_console_pattern(self, 'login:', None, client_vm)
        exec_command_and_wait_for_pattern(self, 'root', '#', None, client_vm)

        #
        # Here, we'd like to actually interact with the gpio device a little
        # more as described at:
        #
        # https://github.com/nutanix/libvfio-user/blob/master/docs/qemu.md
        #
        # Unfortunately, the buildroot Linux kernel has some undiagnosed issue
        # so we don't get /sys/class/gpio. Nonetheless just the basic
        # initialization and setup is enough for basic testing of vfio-user.
        #

        self.log.debug('collecting libvfio-user server output')

        out = exec_command_and_wait_for_pattern(self,
            'cat /var/tmp/gpio.out',
            'gpio: region2: wrote 0 to (0x1:1)',
            None, server_vm)

        gpio_server_out = [s for s in out.decode().splitlines()
                                   if s.startswith("gpio:")]

        for line in EXPECTED_SERVER_LINES:
            if line not in gpio_server_out:
                self.log.error(f'Missing server debug line: {line}')
                self.fail(False)


if __name__ == '__main__':
    QemuSystemTest.main()