test_blobxfer_operations_synccopy.py 文件源码

python
阅读 27 收藏 0 点赞 0 评论 0

项目:blobxfer 作者: Azure 项目源码 文件源码
def test_worker_thread_transfer():
    s = ops.SyncCopy(mock.MagicMock(), mock.MagicMock(), mock.MagicMock())
    s._transfer_queue.put(mock.MagicMock())
    s._transfer_queue.put(mock.MagicMock())
    s._process_synccopy_descriptor = mock.MagicMock()
    s._process_synccopy_descriptor.side_effect = [None, Exception()]

    with mock.patch(
            'blobxfer.operations.synccopy.SyncCopy.termination_check',
            new_callable=mock.PropertyMock) as patched_tc:
        patched_tc.side_effect = [False, False, True]
        s._worker_thread_transfer()
        assert s._process_synccopy_descriptor.call_count == 2
        assert len(s._exceptions) == 1
评论列表
文章目录


问题


面经


文章

微信
公众号

扫码关注公众号