How to check retry in a Celery Python app?

Question

I am trying to check if the application is repeating.

@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 mock 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 get this error

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

This is obvious, but I don’t know how to do it otherwise to check if the method is called.

Answer 1

I didn’t get the job using only the built-in attempt, so I need to use a layout with a side effect of real repetition, which allows me to catch it in the test. I did 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 someone knows how I can get rid of the extra step in mocking the “exception” Retry, I would be glad to hear it!

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

you also miss some things 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) 

Answer 3

The answers here didn't help me, so I delved into the celery code and found a way that suits 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()