1. This site uses cookies. By continuing to use this site, you are agreeing to our use of cookies. Learn More.
  2. Anuncie Aqui
    Anuncie aqui você Também: fdantas@4each.com.br

[Python] How to test Retry in Celery application in Python?

Discussão em 'Python' iniciado por Stack, Setembro 8, 2025.

  1. Stack

    Stack Membro Participativo

    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.

    Continue reading...

Compartilhe esta Página