Celery unit test retrying

Question

I am currently writing unit tests for my celery tasks and would like to test that my task is being retried.

Note: ALWAYS_EAGER is set to True in test settings

@app.shared_task(bind=True, soft_time_limit=600, autoretry_for=(Exception,), retry_kwargs={'max_retries': 3},
                 retry_backoff=3)
def my_task(self, obj_pk):
    try:
        # ...
        function_call_that_can_raise_exception()
    except Exception as exc:
        last_try = self.request.retries >= self.retry_kwargs["max_retries"]

        if last_try:
            # ....
        else:
            # ...
        
        raise_with_context(exc)

I can test my last run by mocking celery.app.task.Task.request;

@mock.patch("celery.app.task.Task.request")
def test_my_task(self):
    mock_task_request.retries = 3
    my_task(12)
    # some asserts

How can I test that my task is actually retried automatically?

Answer 1

The trick is to use apply and not delay or apply_async:

def test_my_task_is_retried(self):
    my_task.apply(kwargs={"obj_pk": 12})
    # assert what should only happen in the last run