How to test Retry in Celery application in Python?

Question

I'm trying to test if the application is retrying.

@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)

And I'm mocking the test

@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)

And I'm getting this error

TypeError: exceptions must be old-style classes or derived from BaseException, not MagicMock

Which is obvious but I don't know how to do it otherwise to test if the method is being called.

Answer 1

I havn't gotten it to work with just using the built in retry so I have to use a mock with the side effect of the real Retry, this makes it possible to catch it in a test. I've done it like this:

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()

If anyone knows how I can get rid of the extra step in mocking the Retry "exception" I'd be happy to hear it!

Answer 2

The answers here didn't help me, so I dived even deeper into celery's code and found a hack that works for me:

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()

Answer 3

For me it worked to patch celery.app.task.Task.request. This way I could also simulate later retries (eg. to test that the task is retried multiple times).

Using pytest and unittest.mock.patch() this looks like:

@mock.patch("celery.app.task.Task.request")
def test_celery_task_retry(mock_request):
    # Override called_directly so that Task.retry() produces a Retry exception.
    mock_request.called_directly = False
    # Simulate the 42nd retry.
    mock_request.retries = 42

    with pytest.raises(celery.exceptions.Retry) as retry_exc:
        task()

    assert retry_exc.value.when > 0

Answer 4

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)

you're also missing some stuff in your task:

@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)