diff options
Diffstat (limited to 'test/py/test_request_errors.py')
-rw-r--r-- | test/py/test_request_errors.py | 117 |
1 files changed, 113 insertions, 4 deletions
diff --git a/test/py/test_request_errors.py b/test/py/test_request_errors.py index a734bab..6cd50cb 100644 --- a/test/py/test_request_errors.py +++ b/test/py/test_request_errors.py @@ -27,17 +27,17 @@ # DAMAGE. # +from unittest.mock import patch from libvfio_user import * import errno import os ctx = None sock = None - argsz = len(vfio_irq_set()) -def test_request_errors_setup(): +def setup_function(function): global ctx, sock ctx = vfu_create_ctx(flags=LIBVFIO_USER_FLAG_ATTACH_NB) @@ -49,12 +49,29 @@ def test_request_errors_setup(): ret = vfu_setup_device_nr_irqs(ctx, VFU_DEV_MSIX_IRQ, 2048) assert ret == 0 + vfu_setup_device_quiesce_cb(ctx) + + ret = vfu_setup_device_reset_cb(ctx) + assert ret == 0 + + ret = vfu_setup_region(ctx, index=VFU_PCI_DEV_MIGR_REGION_IDX, size=0x2000, + flags=VFU_REGION_FLAG_RW) + assert ret == 0 + + ret = vfu_setup_device_migration_callbacks(ctx, offset=0x4000) + assert ret == 0 + ret = vfu_realize_ctx(ctx) assert ret == 0 sock = connect_client(ctx) +def teardown_function(function): + global ctx + vfu_destroy_ctx(ctx) + + def test_too_small(): # struct vfio_user_header hdr = struct.pack("HHIII", 0xbad1, VFIO_USER_DEVICE_SET_IRQS, @@ -126,5 +143,97 @@ def test_bad_request_closes_fds(): os.close(fd2) -def test_request_errors_cleanup(): - vfu_destroy_ctx(ctx) +@patch('libvfio_user.reset_cb') +@patch('libvfio_user.quiesce_cb', return_value=0) +def test_disconnected_socket(mock_quiesce, mock_reset): + """Tests that calling vfu_run_ctx on a disconnected socket results in + resetting the context and returning ENOTCONN.""" + + global ctx, sock + sock.close() + + vfu_run_ctx(ctx, errno.ENOTCONN) + + # quiesce callback gets called during reset + # FIXME how can we ensure that quiesce is called before reset? + mock_quiesce.assert_called_with(ctx) + mock_reset.assert_called_with(ctx, VFU_RESET_LOST_CONN) + + +@patch('libvfio_user.quiesce_cb', side_effect=fail_with_errno(errno.EBUSY)) +def test_disconnected_socket_quiesce_busy(mock_quiesce): + """Tests that calling vfu_run_ctx on a disconnected socket results in + resetting the context which returns EBUSY.""" + + global ctx, sock + sock.close() + + vfu_run_ctx(ctx, errno.EBUSY) + + # quiesce callback must be called during reset + mock_quiesce.assert_called_once_with(ctx) + + # device hasn't finished quiescing + for _ in range(0, 3): + vfu_run_ctx(ctx, errno.EBUSY) + + # device quiesced + ret = vfu_device_quiesced(ctx, 0) + assert ret == 0 + + vfu_run_ctx(ctx, errno.ENOTCONN) + + # no further calls to the quiesce callback should have been made + mock_quiesce.assert_called_once_with(ctx) + + +@patch('libvfio_user.reset_cb') +@patch('libvfio_user.quiesce_cb', side_effect=fail_with_errno(errno.EBUSY)) +@patch('libvfio_user.migr_get_pending_bytes_cb') +def test_reply_fail_quiesce_busy(mock_get_pending_bytes, mock_quiesce, + mock_reset): + """Tests failing to reply and the quiesce callback returning EBUSY.""" + + global ctx, sock + + def get_pending_bytes_side_effect(ctx): + sock.close() + return 0 + mock_get_pending_bytes.side_effect = get_pending_bytes_side_effect + + # read the get_pending_bytes register, it should close the socket causing + # the reply to fail + read_region(ctx, sock, VFU_PCI_DEV_MIGR_REGION_IDX, + vfio_user_migration_info.pending_bytes.offset, + vfio_user_migration_info.pending_bytes.size, rsp=False, + expect_run_ctx_errno=errno.EBUSY) + + # vfu_run_ctx will try to reset the context and to do that it needs to + # quiesce the device first + mock_quiesce.assert_called_once_with(ctx) + + # vfu_run_ctx will be returning EBUSY and nothing should have happened + # until the device quiesces + for _ in range(0, 3): + vfu_run_ctx(ctx, errno.EBUSY) + mock_quiesce.assert_called_once_with(ctx) + mock_reset.assert_not_called() + + ret = vfu_device_quiesced(ctx, 0) + assert ret == 0 + + # the device quiesced, reset should should happen now + mock_quiesce.assert_called_once_with(ctx) + mock_reset.assert_called_once_with(ctx, VFU_RESET_LOST_CONN) + + try: + get_reply(sock) + except OSError as e: + assert e.errno == errno.EBADF + else: + assert False + + vfu_run_ctx(ctx, errno.ENOTCONN) + + +# ex: set tabstop=4 shiftwidth=4 softtabstop=4 expandtab: # |