Run celery task when testing (pytest) in Django

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]']

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.