test_blobxfer_operations_upload.py 文件源码

def test_check_for_uploads_from_md5():
    u = ops.Uploader(mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
    u._md5_offload = mock.MagicMock()
    u._post_md5_skip_on_check = mock.MagicMock()

    with mock.patch(
            'blobxfer.operations.upload.Uploader.termination_check_md5',
            new_callable=mock.PropertyMock) as patched_tcm:
        patched_tcm.side_effect = [False, False, False, True, True]
        u._md5_offload.pop_done_queue.side_effect = [
            None, mock.MagicMock(), None
        ]

        u._check_for_uploads_from_md5()
        assert u._post_md5_skip_on_check.call_count == 1