test_blobxfer_operations_download.py 文件源码

def test_check_for_downloads_from_md5():
    lpath = 'lpath'
    rfile = azmodels.StorageEntity('cont')
    rfile._md5 = 'abc'
    rfile._client = mock.MagicMock()
    rfile._client.primary_endpoint = 'ep'
    rfile._name = 'name'
    rfile._vio = None
    rfile._size = 256
    key = ops.Downloader.create_unique_transfer_operation_id(rfile)
    d = ops.Downloader(mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
    d._md5_map[key] = rfile
    d._transfer_set.add(key)
    d._md5_offload = mock.MagicMock()
    d._md5_offload.done_cv = multiprocessing.Condition()
    d._md5_offload.pop_done_queue.side_effect = [
        None,
        (key, lpath, rfile._size, False),
    ]
    d._add_to_download_queue = mock.MagicMock()
    d._all_remote_files_processed = False
    d._download_terminate = True
    d._check_for_downloads_from_md5()
    assert d._add_to_download_queue.call_count == 0

    with mock.patch(
            'blobxfer.operations.download.Downloader.'
            'termination_check_md5',
            new_callable=mock.PropertyMock) as patched_tc:
        d = ops.Downloader(
            mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
        d._md5_map[key] = rfile
        d._transfer_set.add(key)
        d._md5_offload = mock.MagicMock()
        d._md5_offload.done_cv = multiprocessing.Condition()
        d._md5_offload.pop_done_queue.side_effect = [
            None,
            (key, lpath, rfile._size, False),
        ]
        d._add_to_download_queue = mock.MagicMock()
        patched_tc.side_effect = [False, False, True]
        d._check_for_downloads_from_md5()
        assert d._add_to_download_queue.call_count == 1

    with mock.patch(
            'blobxfer.operations.download.Downloader.'
            'termination_check_md5',
            new_callable=mock.PropertyMock) as patched_tc:
        d = ops.Downloader(
            mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
        d._md5_map[key] = rfile
        d._transfer_set.add(key)
        d._md5_offload = mock.MagicMock()
        d._md5_offload.done_cv = multiprocessing.Condition()
        d._md5_offload.pop_done_queue.side_effect = [None]
        d._add_to_download_queue = mock.MagicMock()
        patched_tc.side_effect = [False, True, True]
        d._check_for_downloads_from_md5()
        assert d._add_to_download_queue.call_count == 0