Pytest, Flask, Celery – Celery NoneType Error

  celery, flask, pytest, python

I have a Flask application using Celery, and the async processing works fine when the app is running locally. However, when I try to test (pytest) a route that uses Celery tasks, I get this error:

app/bp_dir/ in <module>
    from app import db, celery_tasks
app/ in <module>
E   AttributeError: 'NoneType' object has no attribute 'task'

It seems that Celery is not being spun up correctly when I’m testing. I think the underlying question is how to set up pytest fixtures so that routes importing Celery tasks can be tested.

app/ looks like this:

from app import celeryapp

celery = celeryapp.celery

def example_task():
    return 1


from app import celeryapp

def create_app(config_class=Config):
    app = Flask(__name__)


    # Celery
    celery = celeryapp.create_celery_app(app)
    celeryapp.celery = celery


    return app


from celery import Celery


db_session = None
celery = None

def create_celery_app(_app=None):
    Create a new Celery object and tie together the Celery config to the app's config.
    Wrap all tasks in the context of the Flask application.
    :param _app: Flask app
    :return: Celery app

    from app import db

    celery = Celery(_app.import_name,

    always_eager = _app.config['TESTING'] or False
    celery.conf.update({'TASK_ALWAYS_EAGER': always_eager,
                        'CELERY_RESULT_BACKEND': 'redis'})

    TaskBase = celery.Task

    class ContextTask(TaskBase):
        abstract = True

        def __call__(self, *args, **kwargs):
            if not celery.conf.CELERY_ALWAYS_EAGER:
                with _app.app_context():
                    return TaskBase.__call__(self, *args, **kwargs)
                # special pytest setup
                db.session = db_session
                return TaskBase.__call__(self, *args, **kwargs)

        def after_return(self, status, retval, task_id, args, kwargs, einfo):
            After each Celery task, teardown our db session.
            Flask-SQLAlchemy uses create_scoped_session at startup which avoids any setup on a
            per-request basis. This means Celery can piggyback off of this initialization.
            if _app.config['SQLALCHEMY_COMMIT_ON_TEARDOWN']:
                if not isinstance(retval, Exception):

            # If we aren't in an eager request (i.e. Flask will perform teardown), then teardown
            if not celery.conf.CELERY_ALWAYS_EAGER:

    celery.Task = ContextTask

    return celery


from app import celeryapp, create_app

app = create_app()
celery = celeryapp.create_celery_app(app)
celeryapp.celery = celery

The code for the Celery set up is mostly borrowed from this [repo] The repo has tests [here], but I can’t really see what I’m missing.


import re
import flask
import pytest
from app import create_app, db

EMAIL = '[email protected]'
PASSWORD = 'Password123!'

def get_codes(data_structure):
    return [(str(x)) for x in range(len(data_structure))]

def extract_csrf_token(response: flask.Response) -> str:
    # is there better way to get CSRF token? I couldn't find one
    return'name="csrf_token" type="hidden" value="([^"]*)"', str(

def flask_app():
    app = create_app()
    app.config['SQLALCHEMY_DATABASE_URI'] = "sqlite://"
    app.config['TESTING'] = True
    app.config['SECRET_KEY'] = 'this is crucial for storing section'
    app.testing = True

    return app

def app_context(flask_app):
    with flask_app.app_context():

def setup_db(app_context):

def test_client(setup_db, flask_app):
    with flask_app.test_client() as c:
        yield c

def registered_user(test_client, flask_app):
    resp = test_client.get("/auth/register")  # to get csrf_token
    csrf_token = extract_csrf_token(resp)

    # follow_redirect=False is important - otherwise you won't be able to tell the difference between successful
    # and faulty registration
    resp ='/auth/register',
                            data={'email': EMAIL, 'password': PASSWORD, 'password2': PASSWORD,
                                  'csrf_token': csrf_token, 'is_test':True},
    assert resp.status_code == 302, "/auth/register should redirect on successful registering"

def logged_client(registered_user, test_client):
    resp = test_client.get("/auth/login")  # to get csrf_token
    csrf_token = extract_csrf_token(resp)

    resp ='/auth/login', data={'email': EMAIL, 'password': PASSWORD, 'csrf_token': csrf_token})

    assert resp.status_code == 302, "/auth/login should redirect to another page on successful login"

    yield test_client

    resp = test_client.get('/auth/logout')
    assert resp.status_code == 302

Example test:

from testing_fixtures import *

def test_example_route(logged_client):
    response = logged_client.get('/example_route')
    assert response.status_code == 200

Example route:

from app import db, celery_tasks

def example_route():

    return 1

Source: Python Questions