如何在 Python 中测试 Celery 应用程序中的重试？答案

【问题标题】：How to test Retry in Celery application in Python?如何在 Python 中测试 Celery 应用程序中的重试？
【发布时间】：2015-06-10 00:21:24
【问题描述】：

我正在尝试测试应用程序是否正在重试。

@celery.task(bind=False, default_retry_delay=30)
def convert_video(gif_url, webhook):
    // doing something
    VideoManager().convert(gif_url)
    return
    except Exception as exc:
         raise convert_video.retry(exc=exc)

我在嘲笑测试

@patch('src.video_manager.VideoManager.convert')
@patch('requests.post')
def test_retry_failed_task(self, mock_video_manager, mock_requests):
    mock_video_manager.return_value= {'webm':'file.webm', 'mp4':'file.mp4', 'ogv' : 'file.ogv', 'snapshot':'snapshot.png'}
    mock_video_manager.side_effect = Exception('some error')

    server.convert_video.retry = MagicMock()

    server.convert_video('gif_url', 'http://www.company.com/webhook?attachment_id=1234')
    server.convert_video.retry.assert_called_with(ANY)

我收到了这个错误

TypeError：异常必须是旧式类或派生自 BaseException，而不是 MagicMock

这很明显，但我不知道该怎么做才能测试是否正在调用该方法。

【问题讨论】：

这个答案显示了一个更简单的选择：stackoverflow.com/a/69099778/1199807

标签： python celery python-mock

【解决方案1】：

我还没有让它与仅使用内置重试一起工作，所以我必须使用具有真正重试副作用的模拟，这使得在测试中捕获它成为可能。我是这样做的：

from celery.exceptions import Retry
from mock import MagicMock
from nose.plugins.attrib import attr

# Set it for for every task-call (or per task below with @patch)
task.retry = MagicMock(side_effect=Retry)

#@patch('task.retry', MagicMock(side_effect=Retry)
def test_task(self):
    with assert_raises(Retry):
        task() # Note, no delay or things like that

# and the task, I don't know if it works without bind.
@Celery.task(bind=True)
def task(self):
   raise self.retry()

如果有人知道我可以如何摆脱模拟重试“异常”的额外步骤，我会很高兴听到它！

【讨论】：

【解决方案2】：

from mock import patch

import pytest    


@patch('tasks.convert_video.retry')
@patch('tasks.VideoManager')
def test_retry_on_exception(mock_video_manger, mock_retry):
    mock_video_manger.convert.side_effect = error = Exception()
    with pytest.raises(Exception):
        tasks.convert_video('foo', 'bar')
    mock_retry.assert_called_with(exc=error)

你的任务中还缺少一些东西：

@celery.task(bind=False, default_retry_delay=30)
def convert_video(gif_url, webhook):
    try:
        return VideoManager().convert(gif_url)
    except Exception as exc:
         convert_video.retry(exc=exc)

【讨论】：

【解决方案3】：

这里的答案对我没有帮助，所以我更深入地研究了celery's code 并找到了一个适合我的 hack：

def test_celery_retry(monkeypatch):
    # so the retry will be eager
    monkeypatch.setattr(celery_app.conf, 'task_always_eager', True)
    # so celery won't try to raise an error and actually retry
    monkeypatch.setattr(celery.app.task.Context, 'called_directly', False)
    task.delay()

【讨论】：