我目前正在为我的 celery 任务编写单元测试，并想测试我的任务是否正在重试。
注意：ALWAYS_EAGER 在测试设置中设置为 True
@app.shared_task(bind=True, soft_time_limit=600, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3},
retry_backoff=3)
def my_task(self, obj_pk):
try:
# ...
function_call_that_can_raise_exception()
except Exception as exc:
last_try = self.request.retries >= self.retry_kwargs["max_retries"]
if last_try:
# ....
else:
# ...
raise_with_context(exc)
我可以通过模拟 celery.app.task.Task.request 来测试我的最后一次运行；
@mock.patch("celery.app.task.Task.request")
def test_my_task(self):
mock_task_request.retries = 3
my_task(12)
# some asserts
如何测试我的任务是否确实自动重试？
技巧是使用 apply 而不是 delay 或 apply_async：
def test_my_task_is_retried(self):
my_task.apply(kwargs={"obj_pk": 12})
# assert what should only happen in the last run
本文内容由网友自发贡献，版权归原作者所有，本站不承担相应法律责任。如您发现有涉嫌抄袭侵权的内容，请联系:hwhale#tublm.com(使用前将#替换为@)
