用烧瓶和芹菜编写测试

Question

测试烧瓶应用时，celery_worker夹具不起作用，因为芹菜附带的 pytest 夹具不在烧瓶应用上下文中运行。

# 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

上面的测试只会抛出RuntimeError: No application found.和 拒绝逃跑。

通常，在烧瓶项目中使用芹菜时，我们必须继承celery.Celery并修补__call__方法，以便实际的芹菜任务将在烧瓶应用上下文中运行，如下所示：

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

但是看看celery.contrib.pytest，我认为没有简单的方法可以对这些灯具做同样的事情，即修改基本的芹菜应用程序，以便任务可以在烧瓶应用程序上下文中运行。

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

我希望这对你有所帮助！

更多关于 Flask RESTful API 的示例，项目构建 ... ： https://github.com/TTWShell/hobbit-core

为什么不在celery.contrib.pytest中使用夹具celery_config，celery_app，celery_worker？请参阅芹菜文档：使用芹菜进行测试。

因为这个实例芹菜两次，一次在run.py，另一次在celery.contrib.pytest.celery_app。当我们delay任务时，发生了错误。

我们重写了在烧瓶应用上下文中运行的芹菜celery_app。

Answer 2

我没有使用celery.contrib.pytest，但我想提出还不错的解决方案。

首先，您需要将芹菜任务分为sync和async部分。 这里有一个sync_tasks.py的例子：

def filtering_something(my_arg1):
# do something here
def processing_something(my_arg2):
# do something here

async_tasks.py(或您的芹菜任务)示例：

@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...

在这种情况下，您可以为所有功能编写测试，而不依赖于Celery application：

from unittest import TestCase
class SyncTasks(TestCase):
def test_filtering_something(self):
# ....
def test_processing_something(self):
# ....

有什么好处？

1. 您的测试与celery app和flask app分开。
2. 您不会遇到worker_pool，brokers，连接池或其他问题。
3. 您可以编写简单，清晰和快速的测试。
4. 您不依赖于celery.contrib.pytest，但您可以使用 100% 的测试来覆盖您的代码。
5. 你不需要任何mocks.
6. 您可以在测试前准备所有必要的数据(数据库，夹具等)。

希望这有帮助。