Run celery task when testing (pytest) in Django

Question

I have three Celery tasks:

@celery_app.task
def load_rawdata_on_monday():
    if not load_rawdata():  # run synchronously
        notify_report_was_not_updated.delay()


@celery_app.task
def load_rawdata():
    # load and process file from FTP
    return False  # some error happened


@celery_app.task
def notify_rawdata_was_not_updated():
    pass  # send email by Django

I need to test that email was sent if load_rawdata task (function) returns False. For that I have written some test which does not work:

@override_settings(EMAIL_BACKEND='django.core.mail.backends.memcache.EmailBackend')
@override_settings(CELERY_ALWAYS_EAGER=False)
@patch('load_rawdata', MagicMock(return_value=False))
def test_load_rawdata_on_monday():
    load_rawdata_on_monday()
    assert len(mail.outbox) == 1, "Inbox is not empty"
    assert mail.outbox[0].subject == 'Subject here'
    assert mail.outbox[0].body == 'Here is the message.'
    assert mail.outbox[0].from_email == '[email protected]'
    assert mail.outbox[0].to == ['[email protected]']

It seems notify_rawdata_was_not_updated still being run asynchronously. How to write proper test?

Answer 1

It looks like two things may be happening:

• You should call your task with using the apply() method to run it synchronously.
• The CELERY_ALWAYS_EAGER setting should be active to allow subsequent task calls to be executed as well.

@override_settings(EMAIL_BACKEND='django.core.mail.backends.memcache.EmailBackend')
@override_settings(CELERY_ALWAYS_EAGER=True)
@patch('load_rawdata', MagicMock(return_value=False))
def test_load_rawdata_on_monday():
    load_rawdata_on_monday.apply()
    assert len(mail.outbox) == 1, "Inbox is not empty"
    assert mail.outbox[0].subject == 'Subject here'
    assert mail.outbox[0].body == 'Here is the message.'
    assert mail.outbox[0].from_email == '[email protected]'
    assert mail.outbox[0].to == ['[email protected]']

Answer 2

While @tinom9 is correct about using the apply() method, the issue of notify_rawdata_was_not_updated still running asynchronously has to do with your task definition:

@celery_app.task
def load_rawdata_on_monday():
    if not load_rawdata():  
        notify_report_was_not_updated.delay() # delay is an async invocation

try this:

@celery_app.task
def load_rawdata_on_monday():
    if not load_rawdata():  
        notify_report_was_not_updated.apply() # run on local thread

and for the test, calling load_rawdata_on_monday() without .delay() or .apply() should still execute the task locally and block until the task result returns. Just make sure you are handling the return values correctly, some celery invocation methods, like apply() return an celery.result.EagerResult instance compared to delay() or apply_async() which return an celery.result.AsyncResult instance, which may not give you the desired outcome if expecting False when you check if not load_rawdata() or anywhere else you try to get the return value of the function and not the task itself.