在 Flask 应用程序中与 Celery 同步测试

Question

我有一个基于 Flask 的 Web API，它使用异步任务管理器celery。我创建了一个 API 端点，它只是运行我想要的异步任务，然后添加了另一个端点来公开该任务的状态和结果。

通过此 API 端点的集成测试工作正常并通过，但是我还想使用task_always_eagercelery 中的设置同步测试这些功能。但无论我如何尝试设置此配置设置，应用程序仍然异步运行任务（我通过在测试运行时观察 celery 日志知道这一点；celery 仍在启动工作程序并将它们分配给此任务。

我尝试过修改配置变量中间测试文件

# TEST STATISTICS API
current_app.config.update(task_always_eager=True)

以及 config.py 和init .py 文件中

<config.py, next to the rest of the working config settings>
task_always_eager = True

<__init__.py, I use the "application factory" pattern>
celery = Celery(__name__, broker=Config.CELERY_BROKER_URL)
def create_app(config_name):
    app = Flask(__name__)
    celery.conf.update(app.config)

然而，芹菜仍然继续让工人执行这项任务。这是单元测试的一个问题（无法有效地测试它自己的bind = True参数的bc函数），并且因为测试的celery任务不会显示在我们带有coverage.py模块的测试覆盖率套件上。任何帮助将不胜感激！

应用程序编程接口

公开的 API 端点：

@api.route('/stats/<start_time>/<end_time>', methods=['GET'])
def stats_endpoint(start_time, end_time):
    task = async_stats.delay(start_time, end_time)
    return jsonify({}), 202, {'Location': url_for('api.stats_status', task_id=task.id)}

后端异步 API 任务：

@celery.task(bind=True)
def async_stats(self, start_time, end_time):
    <do stuff, update task state, do more stuff>
    return

异步任务状态/结果端点：

@api.route('/stats/status/<task_id>')
def stats_status(task_id):
    task = async_stats.AsyncResult(task_id)
    result = task.result['result']
    status = task.result['status']
    if task.state in ('PENDING', 'STARTED', 'PROGRESS', 'SUCCESS'):
        # task was created, in various stages of completion
        response = jsonify(result)
        response.status_code = status
    <more response handling...>
    return response

更新：CELERY_ALWAYS_EAGER在配置期间将变量设置为 True 成功同步运行我的测试脚本。然而，尝试在脚本中执行此操作current_app.config.update是行不通的，我怀疑正在运行的工作人员在测试脚本运行时已经以某种方式进行了配置。如果有人在这个领域有经验，我会喜欢对此的一些见解。并行进行单元测试和集成测试将需要按需更改配置，但这是理想的选择