| #!/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() |
