test_blobxfer_operations_upload.py 文件源码

def test_worker_thread_transfer():
    u = ops.Uploader(mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
    u._transfer_queue.put(
        (mock.MagicMock, mock.MagicMock, mock.MagicMock, mock.MagicMock)
    )
    u._transfer_queue.put(
        (mock.MagicMock, mock.MagicMock, mock.MagicMock, mock.MagicMock)
    )
    u._process_transfer = mock.MagicMock()
    u._process_transfer.side_effect = [None, Exception()]

    with mock.patch(
            'blobxfer.operations.upload.Uploader.termination_check',
            new_callable=mock.PropertyMock) as patched_tc:
        patched_tc.side_effect = [False, False, True]
        u._worker_thread_transfer()
        assert u._process_transfer.call_count == 2
        assert len(u._exceptions) == 1