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.

did you look into this? https://github.com/pytest-dev/pytest-flask — fodma1, Nov 30 '17 at 10:03
yes and I'm using this package for my testings already. I've added some of my understandings to this problem, which sorta explains why I haven't been able to find anything that could help in pytest-flask. @fodma1 — timfeirg, Nov 30 '17 at 10:23

Legolas Bloom · Answer 1 · 2018-11-09T08:19:48.870

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.

you save my day. And for me, it works after added the default celery task `ping`. https://stackoverflow.com/a/46564964/2794539 — gaozhidf, Oct 21 '19 at 08:42

score 1 · Answer 2 · answered Nov 30 '17 at 18:57

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?

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

Hope this helps.

While this looks ideal for writing tests, I don't imagine it would be applicable to my project: it requires much changes of the code structure (I would have to break functions up in order to test every subroutine separately), and when it comes to function calls that rely on the flask app context (for example I could be manipulating flask-sqlalchemy data), this testing method wouldn't work either and there's no escape around that. But this has been very educating nontheless, thank you. — timfeirg, Dec 01 '17 at 09:03
By the way, **7** - you won't problems with `flask app context` ))) You can send all the necessary request data from endpoint/view/Resource. Anyway, divide application(`flask`, `celery` and your own business logic) is a good way. Tests will be small and supportable(as and your application). If you have a time I recommend make changes in your project(or part of project/ just for fun). You will be surprised at how much easier it is to work. Good luck ;) — Danila Ganchar, Dec 01 '17 at 11:30

writing tests with flask and celery

2 Answers2