时间戳

首页 > 解决方案 > 在单元测试中获取 Celery 重试方法的参数

python-3.x - 在单元测试中获取 Celery 重试方法的参数

问题描述

我有一个 Celery 任务，我会延迟异步调用它。它在发生异常时重试，每次尝试都会增加延迟。

@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 中获取该信息。

标签： python-3.xasynchronouscelerydjango-celery

解决方案

通过执行以下操作管理模拟功能而不破坏功能：

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

推荐阅读