test_blobxfer_operations_upload.py 文件源码

def test_worker_thread_upload(ts):
    u = ops.Uploader(mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
    u._general_options.concurrency.transfer_threads = 1

    u._transfer_set = mock.MagicMock()
    u._transfer_set.__len__.side_effect = [5, 0, 0, 0]
    u._upload_queue.put(mock.MagicMock)
    u._upload_queue.put(mock.MagicMock)
    u._process_upload_descriptor = mock.MagicMock()
    u._process_upload_descriptor.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, False, False, True]
        u._worker_thread_upload()
        assert u._process_upload_descriptor.call_count == 2
        assert len(u._exceptions) == 1