python顽强地重试，禁用unittest的`wait`答案

【问题标题】：python retry with tenacity, disable `wait` for unittestpython顽强地重试，禁用unittest的`wait`
【发布时间】：2017-12-20 13:00:45
【问题描述】：

我正在使用tenacity 库来使用它的@retry 装饰器。

我正在使用它来创建一个函数，该函数会在失败的情况下多次“重复”HTTP 请求。

这是一个简单的代码sn-p：

@retry(stop=stop_after_attempt(7), wait=wait_random_exponential(multiplier=1, max=60))
def func():
   ...
   requests.post(...)

该函数使用坚韧wait-argument 在调用之间等待一段时间。

该函数与@retry-decorator 一起似乎工作正常。

但我也有一个单元测试，它检查函数在失败的情况下确实被调用了 7 次。由于此wait in beetween 尝试，此测试需要大量时间。

是否可以仅在单元测试中以某种方式禁用等待时间？

【问题讨论】：

我真的对tenacy一无所知，但是您可以将时间定义为局部变量并在测试中将其值替换为wait=0

标签： python unit-testing python-tenacity

【解决方案1】：

解决方案来自坚韧的维护者本人在这个 Github 问题中：https://github.com/jd/tenacity/issues/106

您可以简单地为您的单元测试临时更改等待功能：

from tenacity import wait_none

func.retry.wait = wait_none()

【讨论】：

我们如何在这里模拟停止值？
另外，如果您使用的是 stop_after_delay（取决于时间），您可以将其替换为 func.retry.stop = stop_after_attempt(3)（为 3 选择合适的数字）

【解决方案2】：

感谢here 的讨论，我发现了一种基于@steveb 的代码的优雅方法：

from tenacity import retry, stop_after_attempt, wait_exponential


@retry(reraise=True, stop=stop_after_attempt(5), wait=wait_exponential(multiplier=1, min=4, max=10))
def do_something_flaky(succeed):
    print('Doing something flaky')
    if not succeed:
        print('Failed!')
        raise Exception('Failed!')

和测试：

from unittest import TestCase, mock, skip
from main import do_something_flaky


class TestFlakyRetry(TestCase):
    def test_succeeds_instantly(self):
        try:
            do_something_flaky(True)
        except Exception:
            self.fail('Flaky function should not have failed.')

    def test_raises_exception_immediately_with_direct_mocking(self):
        do_something_flaky.retry.sleep = mock.Mock()
        with self.assertRaises(Exception):
            do_something_flaky(False)

    def test_raises_exception_immediately_with_indirect_mocking(self):
        with mock.patch('main.do_something_flaky.retry.sleep'):
            with self.assertRaises(Exception):
                do_something_flaky(False)

    @skip('Takes way too long to run!')
    def test_raises_exception_after_full_retry_period(self):
        with self.assertRaises(Exception):
            do_something_flaky(False)

【讨论】：

【解决方案3】：

模拟基类等待函数：

mock.patch('tenacity.BaseRetrying.wait', side_effect=lambda *args, **kwargs: 0)

它总是不等待

【讨论】：

【解决方案4】：

在阅读the thread in tenacity repo（感谢@DanEEStar 启动它！）后，我想出了以下代码：

@retry(
    stop=stop_after_delay(20.0),
    wait=wait_incrementing(
        start=0,
        increment=0.25,
    ),
    retry=retry_if_exception_type(SomeExpectedException),
    reraise=True,
)
def func() -> None:
    raise SomeExpectedException()


def test_func_should_retry(monkeypatch: MonkeyPatch) -> None:
    # Use monkeypatch to patch retry behavior.
    # It will automatically revert patches when test finishes.
    # Also, it doesn't create nested blocks as `unittest.mock.patch` does.

    # Originally, it was `stop_after_delay` but the test could be
    # unreasonably slow this way. After all, I don't care so much
    # about which policy is applied exactly in this test.
    monkeypatch.setattr(
        func.retry, "stop", stop_after_attempt(3)
    )

    # Disable pauses between retries.
    monkeypatch.setattr(func.retry, "wait", wait_none())

    with pytest.raises(SomeExpectedException):
        func()

    # Ensure that there were retries.
    stats: Dict[str, Any] = func.retry.statistics
    assert "attempt_number" in stats
    assert stats["attempt_number"] == 3

我在此测试中使用pytest 特有的功能。也许，它可以作为一个例子对某人有用，至少对未来的我来说是这样。

【讨论】：

【解决方案5】：

你可以使用unittest.mock 模块来模拟tentacity 库的一些元素。在您的情况下，您使用的所有装饰器都是类，例如retry 是定义在 here 的装饰器类。所以这可能有点棘手，但我认为尝试

mock.patch('tentacity.wait.wait_random_exponential.__call__', ...)

可能有帮助。

【讨论】：

【解决方案6】：

我想覆盖 retry 属性的 retry 函数，虽然这听起来很明显，但如果您是第一次使用它，它看起来并不正确，但确实如此。

sut.my_func.retry.retry = retry_if_not_result(lambda x: True)

感谢其他人为我指明正确的方向。

【讨论】：