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!