我正在使用 tenacity 库来使用它的
@retry
装饰器。
我正在使用它来创建一个函数,在失败的情况下使 HTTP 请求“重复”多次。
这是一个简单的代码片段:
@retry(stop=stop_after_attempt(7), wait=wait_random_exponential(multiplier=1, max=60))
def func():
...
requests.post(...)
该函数使用 tenacity
wait
参数在调用之间等待一段时间。
该功能与
@retry
装饰器一起似乎工作得很好。
但我还有一个单元测试,它检查函数在失败时是否确实被调用了 7 次。由于尝试之间的
wait
,此测试需要花费大量时间。
是否可以仅在单元测试中以某种方式禁用等待时间?
解决方案来自 tenacity 的维护者本人在这个 Github 问题中:https://github.com/jd/tenacity/issues/106
您可以简单地暂时更改单元测试的等待函数:
from tenacity import wait_none
func.retry.wait = wait_none()
阅读完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
特定功能。也许,它对于某人来说可能是有用的例子,至少对于未来的我来说。
感谢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)
模拟基类等待函数:
mock.patch('tenacity.BaseRetrying.wait', side_effect=lambda *args, **kwargs: 0)
总是等不及
您可以使用unittest.mock模块来模拟tentacity库的一些元素。 在您的情况下,您使用的所有装饰器都是类,例如
retry
是here定义的装饰器类。所以这可能有点棘手,但我认为尝试
mock.patch('tentacity.wait.wait_random_exponential.__call__', ...)
可能有帮助。
我想重写
retry
属性的 retry
函数,虽然这听起来很明显,但如果你是第一次使用它,它看起来不太正确,但事实确实如此。
sut.my_func.retry.retry = retry_if_not_result(lambda x: True)
感谢其他人为我指明了正确的方向。
您可以在单元测试根文件夹中的conftest.py中模拟
tenacity.nap.time
。
@pytest.fixture(autouse=True)
def tenacity_wait(mocker):
mocker.patch('tenacity.nap.time')
使用@patch(“tenacity.nap.time.sleep”)解决了我的问题。这是我的代码示例
@patch("tenacity.nap.time.sleep")
def test_retry_logic(self, mock_wait):
retry_attempts = 0
try:
.......
except tenacity.RetryError as retry_err:
retry_attempts = retry_err.last_attempt.attempt_number
self.assertEqual(retry_attempts, 4)