diff options
author | John Levon <john.levon@nutanix.com> | 2021-05-15 00:20:16 +0100 |
---|---|---|
committer | GitHub <noreply@github.com> | 2021-05-15 00:20:16 +0100 |
commit | 8aeb1195af036961b4c9b0dcbcc69024042f8624 (patch) | |
tree | 369a47cdfd7b59f949c816fc6f0590e7ff99a53c /test/py | |
parent | 95f16fceadba22a7730c2a9c94cfd4028ef04a02 (diff) | |
download | libvfio-user-8aeb1195af036961b4c9b0dcbcc69024042f8624.zip libvfio-user-8aeb1195af036961b4c9b0dcbcc69024042f8624.tar.gz libvfio-user-8aeb1195af036961b4c9b0dcbcc69024042f8624.tar.bz2 |
move PCI capability testing to Python (#453)
This also adds a couple of additional tests (as well as more fully testing the
region access path).
Signed-off-by: John Levon <john.levon@nutanix.com>
Reviewed-by: Swapnil Ingle <swapnil.ingle@nutanix.com>
Diffstat (limited to 'test/py')
-rw-r--r-- | test/py/libvfio_user.py | 165 | ||||
-rw-r--r-- | test/py/test_negotiate.py | 5 | ||||
-rw-r--r-- | test/py/test_pci_caps.py | 305 | ||||
-rw-r--r-- | test/py/test_pci_ext_caps.py | 302 | ||||
-rw-r--r-- | test/py/test_vfu_realize_ctx.py | 2 |
5 files changed, 755 insertions, 24 deletions
diff --git a/test/py/libvfio_user.py b/test/py/libvfio_user.py index 2d64269..60a93a8 100644 --- a/test/py/libvfio_user.py +++ b/test/py/libvfio_user.py @@ -40,7 +40,32 @@ import socket import struct import syslog -SOCK_PATH = b"/tmp/vfio-user.sock.%d" % os.getpid() +# from linux/pci_regs.h and linux/pci_defs.h + +PCI_HEADER_TYPE_NORMAL = 0 + +PCI_STD_HEADER_SIZEOF = 64 + +PCI_BARS_NR = 6 + +PCI_PM_SIZEOF = 8 + +PCI_CFG_SPACE_SIZE = 256 +PCI_CFG_SPACE_EXP_SIZE = 4096 + +PCI_CAP_LIST_NEXT = 1 + +PCI_CAP_ID_PM = 0x1 +PCI_CAP_ID_VNDR = 0x9 + +PCI_EXT_CAP_ID_DSN = 0x03 +PCI_EXT_CAP_ID_VNDR = 0x0b + +PCI_EXT_CAP_DSN_SIZEOF = 12 + +PCI_EXT_CAP_VNDR_HDR_SIZEOF = 8 + +# libvfio-user defines VFU_TRANS_SOCK = 0 LIBVFIO_USER_FLAG_ATTACH_NB = (1 << 0) @@ -94,8 +119,6 @@ VFU_REGION_FLAG_WRITE = 2 VFU_REGION_FLAG_RW = (VFU_REGION_FLAG_READ | VFU_REGION_FLAG_WRITE) VFU_REGION_FLAG_MEM = 4 -PCI_BARS_NR = 6 - # enum vfu_dev_irq_type VFU_DEV_INTX_IRQ = 0 VFU_DEV_MSI_IRQ = 1 @@ -110,13 +133,11 @@ VFU_PCI_TYPE_PCI_X_1 = 1 VFU_PCI_TYPE_PCI_X_2 = 2 VFU_PCI_TYPE_EXPRESS = 3 -PCI_HEADER_TYPE_NORMAL = 0 - -PCI_STD_HEADER_SIZEOF = 64 - -PCI_CFG_SPACE_SIZE = 256 -PCI_CAP_ID_PM = b'\1' +VFU_CAP_FLAG_EXTENDED = (1 << 0) +VFU_CAP_FLAG_CALLBACK = (1 << 1) +VFU_CAP_FLAG_READONLY = (1 << 2) +SOCK_PATH = b"/tmp/vfio-user.sock.%d" % os.getpid() topdir = os.path.realpath(os.path.dirname(__file__) + "/../..") build_type = os.getenv("BUILD_TYPE", default="dbg") @@ -130,14 +151,22 @@ lib.vfu_realize_ctx.argtypes = (c.c_void_p,) lib.vfu_attach_ctx.argtypes = (c.c_void_p,) lib.vfu_run_ctx.argtypes = (c.c_void_p,) lib.vfu_destroy_ctx.argtypes = (c.c_void_p,) -lib.vfu_setup_region.argtypes = (c.c_void_p, c.c_int, c.c_long, c.c_void_p, - c.c_int, c.c_void_p, c.c_int, c.c_int) +vfu_region_access_cb_t = c.CFUNCTYPE(c.c_int, c.c_void_p, c.POINTER(c.c_char), + c.c_ulong, c.c_long, c.c_bool) +lib.vfu_setup_region.argtypes = (c.c_void_p, c.c_int, c.c_long, + vfu_region_access_cb_t, c.c_int, c.c_void_p, + c.c_uint32, c.c_int) lib.vfu_pci_get_config_space.argtypes = (c.c_void_p,) lib.vfu_pci_get_config_space.restype = (c.c_void_p) -lib.vfu_setup_device_nr_irqs.argtypes = (c.c_void_p, c.c_int, c.c_int) +lib.vfu_setup_device_nr_irqs.argtypes = (c.c_void_p, c.c_int, c.c_uint32) lib.vfu_pci_init.argtypes = (c.c_void_p, c.c_int, c.c_int, c.c_int) -lib.vfu_pci_add_capability.argtypes = (c.c_void_p, c.c_long, c.c_int, +lib.vfu_pci_add_capability.argtypes = (c.c_void_p, c.c_ulong, c.c_int, c.POINTER(c.c_byte)) +lib.vfu_pci_find_capability.argtypes = (c.c_void_p, c.c_bool, c.c_int) +lib.vfu_pci_find_capability.restype = (c.c_ulong) +lib.vfu_pci_find_next_capability.argtypes = (c.c_void_p, c.c_bool, c.c_ulong, + c.c_int) +lib.vfu_pci_find_next_capability.restype = (c.c_ulong) msg_id = 1 @@ -187,25 +216,111 @@ class vfu_pci_hdr_t(c.Structure): # Util functions # +def to_byte(val): + """Cast an int to a byte value.""" + return val.to_bytes(1, 'little') + +def ext_cap_hdr(buf, offset): + """Read an extended cap header.""" + + # struct pcie_ext_cap_hdr + cap_id, cap_next = struct.unpack('HH', buf[offset:offset+4]) + cap_next >>= 4 + return cap_id, cap_next + +def skip(data, fmt): + """Return the data remaining after skipping the given elements.""" + return data[struct.calcsize(fmt):] + +def parse_json(json_str): + """Parse JSON into an object with attributes (instead of using a dict).""" + return json.loads(json_str, object_hook=lambda d: SimpleNamespace(**d)) + def connect_sock(): sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) sock.connect(SOCK_PATH) return sock +def connect_client(ctx): + sock = connect_sock() + + json = b'{ "capabilities": { "max_fds": 8 } }' + # struct vfio_user_version + payload = struct.pack("HH%dsc" % len(json), LIBVFIO_USER_MAJOR, + LIBVFIO_USER_MINOR, json, b'\0') + hdr = vfio_user_header(VFIO_USER_VERSION, size=len(payload)) + sock.send(hdr + payload) + vfu_attach_ctx(ctx, expect=0) + payload = get_reply(sock, expect=0) + return sock + +def disconnect_client(ctx, sock): + sock.close() + + # notice client closed connection + vfu_run_ctx(ctx) + def get_reply(sock, expect=0): buf = sock.recv(4096) (msg_id, cmd, msg_size, flags, errno) = struct.unpack("HHIII", buf[0:16]) + assert (flags & VFIO_USER_F_TYPE_REPLY) != 0 assert errno == expect return buf[16:] -def parse_json(json_str): - """Parse JSON into an object with attributes (instead of using a dict).""" - return json.loads(json_str, object_hook=lambda d: SimpleNamespace(**d)) - def get_pci_header(ctx): ptr = lib.vfu_pci_get_config_space(ctx) return c.cast(ptr, c.POINTER(vfu_pci_hdr_t)).contents +def get_pci_cfg_space(ctx): + ptr = lib.vfu_pci_get_config_space(ctx) + return c.cast(ptr, c.POINTER(c.c_char))[0:PCI_CFG_SPACE_SIZE] + +def get_pci_ext_cfg_space(ctx): + ptr = lib.vfu_pci_get_config_space(ctx) + return c.cast(ptr, c.POINTER(c.c_char))[0:PCI_CFG_SPACE_EXP_SIZE] + +def read_pci_cfg_space(ctx, buf, count, offset, extended=False): + space = get_pci_ext_cfg_space(ctx) if extended else get_pci_cfg_space(ctx) + + for i in range(count): + buf[i] = space[offset+i] + return count + +def write_pci_cfg_space(ctx, buf, count, offset, extended=False): + max_offset = PCI_CFG_SPACE_EXP_SIZE if extended else PCI_CFG_SPACE_SIZE + + assert offset + count <= max_offset + + space = c.cast(lib.vfu_pci_get_config_space(ctx), c.POINTER(c.c_char)) + + for i in range(count): + space[offset+i] = buf[i] + return count + +def access_region(ctx, sock, is_write, region, offset, count, + data=None, expect=0): + # struct vfio_user_region_access + payload = struct.pack("QII", offset, region, count) + if is_write: + payload += data + + cmd = VFIO_USER_REGION_WRITE if is_write else VFIO_USER_REGION_READ + hdr = vfio_user_header(cmd, size=len(payload)) + sock.send(hdr + payload) + vfu_run_ctx(ctx) + result = get_reply(sock, expect=expect) + + if is_write: + return None + + return skip(result, "QII") + +def write_region(ctx, sock, region, offset, count, data, expect=0): + access_region(ctx, sock, True, region, offset, count, data, expect=expect) + +def read_region(ctx, sock, region, offset, count, expect=0): + return access_region(ctx, sock, False, region, offset, count, expect=expect) + # # Library wrappers # @@ -257,9 +372,11 @@ def vfu_destroy_ctx(ctx): if os.path.exists(SOCK_PATH): os.remove(SOCK_PATH) -def vfu_setup_region(ctx, index, size, flags=0): +def vfu_setup_region(ctx, index, size, flags=0, cb=None): assert ctx != None - ret = lib.vfu_setup_region(ctx, index, size, None, flags, None, 0, -1) + ret = lib.vfu_setup_region(ctx, index, size, + c.cast(cb, vfu_region_access_cb_t), + flags, None, 0, -1) return ret def vfu_setup_device_nr_irqs(ctx, irqtype, count): @@ -276,3 +393,13 @@ def vfu_pci_add_capability(ctx, pos, flags, data): databuf = (c.c_byte * len(data)).from_buffer(bytearray(data)) return lib.vfu_pci_add_capability(ctx, pos, flags, databuf) + +def vfu_pci_find_capability(ctx, extended, cap_id): + assert ctx != None + + return lib.vfu_pci_find_capability(ctx, extended, cap_id) + +def vfu_pci_find_next_capability(ctx, extended, offset, cap_id): + assert ctx != None + + return lib.vfu_pci_find_next_capability(ctx, extended, offset, cap_id) diff --git a/test/py/test_negotiate.py b/test/py/test_negotiate.py index 09edf99..e05b450 100644 --- a/test/py/test_negotiate.py +++ b/test/py/test_negotiate.py @@ -155,10 +155,7 @@ def test_valid_negotiate_no_json(): assert json.capabilities.max_msg_size == SERVER_MAX_MSG_SIZE # FIXME: migration object checks - sock.close() - - # notice client closed connection - vfu_run_ctx(ctx) + disconnect_client(ctx, sock) def test_valid_negotiate_empty_json(): client_version_json(json=b'{}') diff --git a/test/py/test_pci_caps.py b/test/py/test_pci_caps.py new file mode 100644 index 0000000..9691fe1 --- /dev/null +++ b/test/py/test_pci_caps.py @@ -0,0 +1,305 @@ +# +# Copyright (c) 2021 Nutanix Inc. All rights reserved. +# +# Authors: John Levon <john.levon@nutanix.com> +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of Nutanix nor the names of its contributors may be +# used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY +# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES +# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH +# DAMAGE. +# + +from libvfio_user import * +import ctypes as c +import errno + +ctx = None + +def test_pci_cap_setup(): + global ctx + + ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB) + assert ctx != None + + ret = vfu_pci_init(ctx, pci_type=VFU_PCI_TYPE_CONVENTIONAL) + assert ret == 0 + + ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX, + size=PCI_CFG_SPACE_SIZE, flags=VFU_REGION_FLAG_RW) + assert ret == 0 + +def test_pci_cap_bad_flags(): + pos = vfu_pci_add_capability(ctx, pos=0, flags=999, + data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0)) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + +def test_pci_cap_no_cb(): + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_CALLBACK, + data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0)) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + +def test_pci_cap_unknown_cap(): + pos = vfu_pci_add_capability(ctx, pos=0, flags=0, + data=struct.pack("ccHH", b'\x81', b'\0', 0, 0)) + assert pos == -1 + assert c.get_errno() == errno.ENOTSUP + +def test_pci_cap_bad_pos(): + pos = vfu_pci_add_capability(ctx, pos=PCI_CFG_SPACE_SIZE, flags=0, + data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0)) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + +@c.CFUNCTYPE(c.c_int, c.c_void_p, c.POINTER(c.c_char), + c.c_long, c.c_long, c.c_int) +def pci_region_cb(ctx, buf, count, offset, is_write): + if not is_write: + return read_pci_cfg_space(ctx, buf, count, offset) + + return write_pci_cfg_space(ctx, buf, count, offset) + +def test_pci_cap_setup_cb(): + global ctx + + vfu_destroy_ctx(ctx) + + ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB) + assert ctx != None + + ret = vfu_pci_init(ctx, pci_type=VFU_PCI_TYPE_CONVENTIONAL) + assert ret == 0 + + ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX, + size=PCI_CFG_SPACE_SIZE, cb=pci_region_cb, + flags=VFU_REGION_FLAG_RW) + assert ret == 0 + +cap_offsets = ( + PCI_STD_HEADER_SIZEOF, + PCI_STD_HEADER_SIZEOF + PCI_PM_SIZEOF, + # NB: note 4-byte alignment of vsc2 + PCI_STD_HEADER_SIZEOF + PCI_PM_SIZEOF + 8, + 0x80, + 0x90 +) + +def test_add_caps(): + pos = vfu_pci_add_capability(ctx, pos=0, flags=0, + data=struct.pack("ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0)) + assert pos == cap_offsets[0] + + data = b"abc" + cap = struct.pack("ccc%ds" % len(data), to_byte(PCI_CAP_ID_VNDR), b'\0', + to_byte(3 + len(data)), data) + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_READONLY, + data=cap) + + assert pos == cap_offsets[1] + + data = b"Hello world." + cap = struct.pack("ccc%ds" % len(data), to_byte(PCI_CAP_ID_VNDR), b'\0', + to_byte(3 + len(data)), data) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_CALLBACK, + data=cap) + assert pos == cap_offsets[2] + + pos = vfu_pci_add_capability(ctx, pos=cap_offsets[3], flags=0, data=cap) + assert pos == cap_offsets[3] + + pos = vfu_pci_add_capability(ctx, pos=cap_offsets[4], flags=0, data=cap) + assert pos == cap_offsets[4] + + ret = vfu_realize_ctx(ctx) + assert ret == 0 + +def test_find_caps(): + offset = vfu_pci_find_capability(ctx, False, PCI_CAP_ID_PM) + assert offset == cap_offsets[0] + + space = get_pci_cfg_space(ctx) + + assert space[offset] == PCI_CAP_ID_PM + assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[1] + + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_PM) + assert offset == 0 + + offset = vfu_pci_find_capability(ctx, False, PCI_CAP_ID_VNDR) + assert offset == cap_offsets[1] + assert space[offset] == PCI_CAP_ID_VNDR + assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[2] + + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_PM) + assert offset == 0 + + offset = vfu_pci_find_next_capability(ctx, False, 0, PCI_CAP_ID_VNDR) + assert offset == cap_offsets[1] + assert space[offset] == PCI_CAP_ID_VNDR + assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[2] + + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR) + assert offset == cap_offsets[2] + assert space[offset] == PCI_CAP_ID_VNDR + assert space[offset + PCI_CAP_LIST_NEXT] == cap_offsets[3] + + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR) + assert offset == cap_offsets[3] + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR) + assert offset == cap_offsets[4] + offset = vfu_pci_find_next_capability(ctx, False, offset, PCI_CAP_ID_VNDR) + assert offset == 0 + + # check for invalid offsets + + offset = vfu_pci_find_next_capability(ctx, False, 8192, PCI_CAP_ID_PM) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + + offset = vfu_pci_find_next_capability(ctx, False, 256, PCI_CAP_ID_PM) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + + offset = vfu_pci_find_next_capability(ctx, False, 255, PCI_CAP_ID_PM) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + + offset = vfu_pci_find_next_capability(ctx, False, + PCI_STD_HEADER_SIZEOF + + PCI_PM_SIZEOF + 1, + PCI_CAP_ID_VNDR) + assert offset == 0 + assert c.get_errno() == errno.ENOENT + +def test_pci_cap_write_hdr(): + sock = connect_client(ctx) + + # offset of struct cap_hdr + offset=cap_offsets[0] + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + disconnect_client(ctx, sock) + +def test_pci_cap_readonly(): + sock = connect_client(ctx) + + # start of vendor payload + offset=cap_offsets[1] + 2 + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + # offsetof(struct vsc, data) + offset=cap_offsets[1] + 3 + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=3) + assert payload == b'abc' + + disconnect_client(ctx, sock) + +def test_pci_cap_callback(): + sock = connect_client(ctx) + + # offsetof(struct vsc, data) + offset=cap_offsets[2] + 3 + data = b"Hello world." + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + assert payload == data + + data = b"Bye world." + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data) + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + assert payload == data + + disconnect_client(ctx, sock) + +def test_pci_cap_write_pmcs(): + sock = connect_client(ctx) + + # struct pc + + offset=cap_offsets[0] + 3 + data=b'\x01\x02' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EINVAL) + + offset=cap_offsets[0] + 2 + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EINVAL) + + offset=cap_offsets[0] + 2 + data=b'\x01\x02' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.ENOTSUP) + + # struct pmcs + + offset=cap_offsets[0] + 5 + data=b'\x01\x02' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EINVAL) + + offset=cap_offsets[0] + 4 + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EINVAL) + + offset = cap_offsets[0] + 4 + data=b'\x01\x02' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data) + + assert get_pci_cfg_space(ctx)[offset:offset+2] == data + + # pmcsr_se + offset=cap_offsets[0] + 6 + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.ENOTSUP) + + # data + offset=cap_offsets[0] + 7 + data=b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.ENOTSUP) + + disconnect_client(ctx, sock) + +def test_pci_cap_write_px(): + # FIXME + pass + +def test_pci_cap_write_msix(): + # FIXME + pass + +def test_pci_cap_cleanup(): + vfu_destroy_ctx(ctx) diff --git a/test/py/test_pci_ext_caps.py b/test/py/test_pci_ext_caps.py new file mode 100644 index 0000000..70f8253 --- /dev/null +++ b/test/py/test_pci_ext_caps.py @@ -0,0 +1,302 @@ +# +# Copyright (c) 2021 Nutanix Inc. All rights reserved. +# +# Authors: John Levon <john.levon@nutanix.com> +# +# Redistribution and use in source and binary forms, with or without +# modification, are permitted provided that the following conditions are met: +# * Redistributions of source code must retain the above copyright +# notice, this list of conditions and the following disclaimer. +# * Redistributions in binary form must reproduce the above copyright +# notice, this list of conditions and the following disclaimer in the +# documentation and/or other materials provided with the distribution. +# * Neither the name of Nutanix nor the names of its contributors may be +# used to endorse or promote products derived from this software without +# specific prior written permission. +# +# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE +# ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY +# DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES +# (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +# SERVICESLOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +# CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT +# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY +# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH +# DAMAGE. +# + +from libvfio_user import * +import ctypes as c +import errno + +ctx = None + +def test_pci_ext_cap_conventional(): + ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB) + assert ctx != None + + ret = vfu_pci_init(ctx, pci_type=VFU_PCI_TYPE_CONVENTIONAL) + assert ret == 0 + + ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX, + size=PCI_CFG_SPACE_SIZE, + flags=VFU_REGION_FLAG_RW) + assert ret == 0 + + # struct dsncap + cap = struct.pack("HHII", PCI_EXT_CAP_ID_DSN, 0, 0, 0) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_EXTENDED, + data=cap) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + + vfu_destroy_ctx(ctx) + +def test_pci_ext_cap_setup(): + global ctx + + ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB) + assert ctx != None + + ret = vfu_pci_init(ctx) + assert ret == 0 + + ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_CFG_REGION_IDX, + size=PCI_CFG_SPACE_EXP_SIZE, cb=pci_region_cb, + flags=VFU_REGION_FLAG_RW) + assert ret == 0 + +def test_pci_ext_cap_unknown_cap(): + cap = struct.pack("HHII", PCI_EXT_CAP_ID_DSN + 99, 0, 0, 0) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_EXTENDED, + data=cap) + assert pos == -1 + assert c.get_errno() == errno.ENOTSUP + +def test_pci_ext_cap_bad_pos(): + cap = struct.pack("HHII", PCI_EXT_CAP_ID_DSN, 0, 0, 0) + + pos = vfu_pci_add_capability(ctx, pos=(PCI_CFG_SPACE_EXP_SIZE - 2), + flags=VFU_CAP_FLAG_EXTENDED, data=cap) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + + # first cap must be at 256 + pos = vfu_pci_add_capability(ctx, pos=512, + flags=VFU_CAP_FLAG_EXTENDED, data=cap) + assert pos == -1 + assert c.get_errno() == errno.EINVAL + +@c.CFUNCTYPE(c.c_int, c.c_void_p, c.POINTER(c.c_char), + c.c_long, c.c_long, c.c_int) +def pci_region_cb(ctx, buf, count, offset, is_write): + if not is_write: + return read_pci_cfg_space(ctx, buf, count, offset, extended=True) + + return write_pci_cfg_space(ctx, buf, count, offset, extended=True) + +cap_offsets = ( + PCI_CFG_SPACE_SIZE, + PCI_CFG_SPACE_SIZE + PCI_EXT_CAP_DSN_SIZEOF, + PCI_CFG_SPACE_SIZE + PCI_EXT_CAP_DSN_SIZEOF + + PCI_EXT_CAP_VNDR_HDR_SIZEOF + 8, + 512, + 600 +) + +def test_add_ext_caps(): + cap = struct.pack("HHII", PCI_EXT_CAP_ID_DSN, 0, 4, 8) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=VFU_CAP_FLAG_EXTENDED, + data=cap) + assert pos == cap_offsets[0] + + # struct pcie_ext_cap_vsc_hdr + data = b"abcde" + cap = struct.pack("HHHH%ds" % len(data), PCI_EXT_CAP_ID_VNDR, 0, 0x1a, + (len(data) + 8) << 4, data) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=(VFU_CAP_FLAG_EXTENDED | + VFU_CAP_FLAG_READONLY), data=cap) + assert pos == cap_offsets[1] + + data = b"Hello world." + cap = struct.pack("HHHH%ds" % len(data), PCI_EXT_CAP_ID_VNDR, 0, 0x1b, + (len(data) + 8) << 4, data) + + pos = vfu_pci_add_capability(ctx, pos=0, flags=(VFU_CAP_FLAG_EXTENDED | + VFU_CAP_FLAG_CALLBACK), data=cap) + assert pos == cap_offsets[2] + + cap = struct.pack("HHHH%ds" % len(data), PCI_EXT_CAP_ID_VNDR, 0, 0x1c, + (len(data) + 8) << 4, data) + + pos = vfu_pci_add_capability(ctx, pos=cap_offsets[3], + flags=(VFU_CAP_FLAG_EXTENDED | + VFU_CAP_FLAG_CALLBACK), data=cap) + assert pos == cap_offsets[3] + + cap = struct.pack("HHHH%ds" % len(data), PCI_EXT_CAP_ID_VNDR, 0, 0x1d, + (len(data) + 8) << 4, data) + + pos = vfu_pci_add_capability(ctx, pos=cap_offsets[4], + flags=(VFU_CAP_FLAG_EXTENDED | + VFU_CAP_FLAG_CALLBACK), data=cap) + assert pos == cap_offsets[4] + + ret = vfu_realize_ctx(ctx) + assert ret == 0 + +def test_find_ext_caps(): + offset = vfu_pci_find_capability(ctx, True, PCI_EXT_CAP_ID_DSN) + assert offset == cap_offsets[0] + + space = get_pci_ext_cfg_space(ctx) + + cap_id, cap_next = ext_cap_hdr(space, offset) + assert cap_id == PCI_EXT_CAP_ID_DSN + assert cap_next == cap_offsets[1] + + offset = vfu_pci_find_next_capability(ctx, True, offset, PCI_EXT_CAP_ID_DSN) + assert offset == 0 + + offset = vfu_pci_find_capability(ctx, True, PCI_EXT_CAP_ID_VNDR) + assert offset == cap_offsets[1] + cap_id, cap_next = ext_cap_hdr(space, offset) + assert cap_id == PCI_EXT_CAP_ID_VNDR + assert cap_next == cap_offsets[2] + + offset = vfu_pci_find_next_capability(ctx, True, offset, PCI_EXT_CAP_ID_DSN) + assert offset == 0 + + offset = vfu_pci_find_next_capability(ctx, True, 0, PCI_EXT_CAP_ID_VNDR) + assert offset == cap_offsets[1] + cap_id, cap_next = ext_cap_hdr(space, offset) + assert cap_id == PCI_EXT_CAP_ID_VNDR + assert cap_next == cap_offsets[2] + + offset = vfu_pci_find_next_capability(ctx, True, offset, + PCI_EXT_CAP_ID_VNDR) + assert offset == cap_offsets[2] + cap_id, cap_next = ext_cap_hdr(space, offset) + assert cap_id == PCI_EXT_CAP_ID_VNDR + assert cap_next == cap_offsets[3] + + offset = vfu_pci_find_next_capability(ctx, True, offset, + PCI_EXT_CAP_ID_VNDR) + assert offset == cap_offsets[3] + offset = vfu_pci_find_next_capability(ctx, True, offset, + PCI_EXT_CAP_ID_VNDR) + assert offset == cap_offsets[4] + offset = vfu_pci_find_next_capability(ctx, True, offset, + PCI_EXT_CAP_ID_VNDR) + assert offset == 0 + + # check for invalid offsets + + offset = vfu_pci_find_next_capability(ctx, True, 8192, PCI_EXT_CAP_ID_DSN) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + offset = vfu_pci_find_next_capability(ctx, True, 4096, PCI_EXT_CAP_ID_DSN) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + offset = vfu_pci_find_next_capability(ctx, True, 4095, PCI_EXT_CAP_ID_DSN) + assert offset == 0 + assert c.get_errno() == errno.EINVAL + + offset = vfu_pci_find_next_capability(ctx, True, cap_offsets[1] + 1, + PCI_EXT_CAP_ID_DSN) + assert offset == 0 + assert c.get_errno() == errno.ENOENT + +def test_pci_ext_cap_write_hdr(): + sock = connect_client(ctx) + + # struct pcie_ext_cap_hdr + offset = cap_offsets[0] + data = b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + # struct pcie_ext_cap_vsc_hdr also + offset = cap_offsets[1] + 4 + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + disconnect_client(ctx, sock) + +def test_pci_ext_cap_readonly(): + sock = connect_client(ctx) + + # start of vendor payload + offset = cap_offsets[1] + 8 + data = b'\x01' + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + offset = cap_offsets[1] + 8 + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=5) + assert payload == b'abcde' + + disconnect_client(ctx, sock) + +def test_pci_ext_cap_callback(): + sock = connect_client(ctx) + + # start of vendor payload + offset = cap_offsets[2] + 8 + data = b"Hello world." + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + assert payload == data + + data = b"Bye world." + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data) + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + assert payload == data + + disconnect_client(ctx, sock) + +def test_pci_ext_cap_write_dsn(): + sock = connect_client(ctx) + + data = struct.pack("II", 1, 2); + offset = cap_offsets[0] + 4 + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data, expect=errno.EPERM) + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + + # unchanged! + assert payload == struct.pack("II", 4, 8) + + disconnect_client(ctx, sock) + +def test_pci_ext_cap_write_vendor(): + sock = connect_client(ctx) + + data = struct.pack("II", 0x1, 0x2); + # start of vendor payload + offset = cap_offsets[2] + 8 + write_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data), data=data) + + payload = read_region(ctx, sock, VFU_PCI_DEV_CFG_REGION_IDX, offset=offset, + count=len(data)) + + assert payload == data + + disconnect_client(ctx, sock) + +def test_pci_ext_cap_cleanup(): + vfu_destroy_ctx(ctx) diff --git a/test/py/test_vfu_realize_ctx.py b/test/py/test_vfu_realize_ctx.py index c35612c..9b863cb 100644 --- a/test/py/test_vfu_realize_ctx.py +++ b/test/py/test_vfu_realize_ctx.py @@ -107,7 +107,7 @@ def test_vfu_realize_ctx_caps(): assert ret == 0 pos = vfu_pci_add_capability(ctx, pos=0, flags=0, data=struct.pack( - "ccHH", PCI_CAP_ID_PM, b'\0', 0, 0)) + "ccHH", to_byte(PCI_CAP_ID_PM), b'\0', 0, 0)) assert pos == PCI_STD_HEADER_SIZEOF ret = vfu_realize_ctx(ctx) |