test_retry.py 文件源码

def test_enforcing_wait_time_sleeps(mocker):
    f = concurrent.Future()
    f.set_result(None)
    mock_gen_sleep = mocker.patch.object(retry.gen, 'sleep')
    mock_gen_sleep.return_value = f

    def in_the_future(_):
        return 60

    request = Mock()
    policy = retry.RetryPolicy(try_limit=3, sleep_func=in_the_future)

    yield policy.enforce(request)
    assert mock_gen_sleep.called is False

    yield policy.enforce(request)
    mock_gen_sleep.assert_called_once_with(60)