writing tests with flask and celery

Question

The celery_worker fixture doesn't work when testing a flask app because the pytest fixtures that comes with celery doesn't run within flask app context.

# tasks.py
@current_app.task(bind=True)
def some_task(name, sha):
    return Release.query.filter_by(name=name, sha=sha).all()

# test_celery.py
def test_some_celery_task(celery_worker):
    async_result = some_task.delay(default_appname, default_sha)
    assert len(async_result.get()) == 0

The tests above will simply throw RuntimeError: No application found. And refuse to run.

Normally when using celery inside a flask project, we have to inherit celery.Celery and patch the __call__ method so that the actual celery tasks will run inside the flask app context, something like this:

def make_celery(app):
    celery = Celery(app.import_name)
    celery.config_from_object('citadel.config')

    class EruGRPCTask(Task):

        abstract = True

        def __call__(self, *args, **kwargs):
            with app.app_context():
                return super(EruGRPCTask, self).__call__(*args, **kwargs)

    celery.Task = EruGRPCTask
    celery.autodiscover_tasks(['citadel'])
    return celery

But looking at celery.contrib.pytest, I see no easy way to do the same with these fixtures, that is, to modify the base celery app so that tasks can run inside the flask app context.

Answer 1

run.py

from flask import Flask
from celery import Celery

celery = Celery()


def make_celery(app):
    celery.conf.update(app.config)

    class ContextTask(celery.Task):
        def __call__(self, *args, **kwargs):
            with app.app_context():
                return self.run(*args, **kwargs)

    celery.Task = ContextTask
    return celery


@celery.task
def add(x, y):
    return x + y


def create_app():
    app = Flask(__name__)
    # CELERY_BROKER_URL
    app.config['BROKER_URL'] = 'sqla+sqlite:///celerydb.sqlite'
    app.config['CELERY_RESULT_BACKEND'] = 'db+sqlite:///results.sqlite'
    make_celery(app)
    return app


app = create_app()

test_celery.py

import pytest

from run import app as app_, add


@pytest.fixture(scope='session')
def app(request):
    ctx = app_.app_context()
    ctx.push()

    def teardown():
        ctx.pop()

    request.addfinalizer(teardown)
    return app_


@pytest.fixture(scope='session')
def celery_app(app):
    from run import celery
    # for use celery_worker fixture
    from celery.contrib.testing import tasks  # NOQA
    return celery


def test_add(celery_app, celery_worker):
    assert add.delay(1, 2).get() == 3

I hope this may help you!

More example about Flask RESTful API, project build ... : https://github.com/TTWShell/hobbit-core

Why not use fixture celery_config, celery_app, celery_worker in celery.contrib.pytest? See celery doc: Testing with Celery.

Because this instance Celery twice, one is in run.py and the other is in celery.contrib.pytest.celery_app. When we delay a task, error occurred.

We rewrite celery_app for celery run within flask app context.

Answer 2

I didn't use celery.contrib.pytest, but I want to propose not bad solution.

First what you need is to divide celery tasks to sync and async parts. Here an example of sync_tasks.py:

def filtering_something(my_arg1):
    # do something here

def processing_something(my_arg2):
    # do something here

Example of async_tasks.py(or your celery tasks):

@current_app.task(bind=True)
def async_filtering_something(my_arg1):
    # just call sync code from celery task...
    return filtering_something(my_arg1)

@current_app.task(bind=True)
def async_processing_something(my_arg2):
    processing_something(my_arg2)
    # or one more call...
    # or one more call...

In this case you can write tests to all functionality and not depend on Celery application:

from unittest import TestCase

class SyncTasks(TestCase):

    def test_filtering_something(self):
       # ....

    def test_processing_something(self):
       # ....

What are the benefits?

1. Your tests are separated from celery app and flask app.
2. You won't problems with worker_pool, brokers, connection-pooling or something else.
3. You can write easy, clear and fast tests.
4. You don't depend on celery.contrib.pytest, but you can cover your code with tests for 100%.
5. You don't need any mocks.
6. You can prepare all necessary data(db, fixtures etc) before tests.

Hope this helps.