在单元测试中获取芹菜重试方法的参数

Question

我有一个延迟异步调用的芹菜任务。当发生异常时，它会重试自身，延迟会增加每次尝试。

@shared_task(bind=True, max_retries=None)
def cancel_issue_request(self):
try:
raise Exception('test')
except Exception:
countdown = 5 * 60 * (2 ** self.request.retries)
raise self.retry(countdown=countdown, max_retries=5)

这在实践中可以正常工作，但我正在尝试编写一个单元测试来检查倒计时是否确实具有正确的值。

我试图做的是模拟 cancel_issue_request.retry 方法，但是一旦我这样做，方法本身就会中断并且不会重试任务。 我以为你可以模拟一种方法，它仍然使用 spec=celery.task.Task.retry 执行其所有功能，但我无法让它工作，请启发我？

或者，我想在某处阅读从 Celery 进行的重试，但我无法从 Celery 中获取该信息。

Answer 1

设法模拟函数而不破坏功能：

cancel_issue_request.retry = Mock(side_effect=cancel_issue_request.retry)