Pytest测试导入Celery任务的Flask路由出现NoneType属性错误如何解决
报错原因
你的代码在模块导入阶段就会执行Celery任务的装饰逻辑:
- pytest加载测试用例时会先导入路由模块
- 路由模块顶部导入
celery_tasks celery_tasks.py顶部直接取celeryapp.celery这个全局变量,此时还没有调用create_app初始化Celery,所以这个变量是None,执行@celery.task()就会抛出AttributeError
修复步骤
1. 调整Celery任务的导入时机(可选但推荐)
不要在路由的顶部导入Celery任务,改成在视图函数内部按需导入,避免模块加载阶段就触发Celery实例的依赖:
# 路由代码修改后 from app import db ... @example_bp.route('/example_route') def example_route(): from app import celery_tasks # 放到函数内部导入 # 你的业务逻辑 return 1
2. 完善pytest fixture配置
修改testing_fixtures.py里的flask_app fixture,确保Celery实例和全局变量都被正确初始化:
import re import flask import pytest from app import create_app, db from app import celeryapp # 提前导入celeryapp模块 EMAIL = 'testemail@gmail.com' PASSWORD = 'Password123!' # 其他工具函数不变... @pytest.fixture(scope="session") # 改成session级,全局只初始化一次 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' # 明确配置Celery测试模式,任务同步执行不需要Broker app.config['CELERY_BROKER_URL'] = 'memory://' app.config['CELERY_BACKEND_URL'] = 'cache+memory://' app.testing = True # 初始化Celery实例并赋值给celeryapp的全局变量 celery = celeryapp.create_celery_app(app) celeryapp.celery = celery # 给celeryapp的全局db_session赋值,适配测试环境的上下文 celeryapp.db_session = db.session return app
3. 可选:强制提前初始化Celery
如果不想修改路由的导入逻辑,可以在测试目录下新建conftest.py文件,添加钩子在pytest加载所有测试模块前先初始化Celery:
# conftest.py import pytest from app import create_app from app import celeryapp def pytest_sessionstart(session): # 测试会话启动时先初始化app和Celery app = create_app() app.config['TESTING'] = True app.config['CELERY_BROKER_URL'] = 'memory://' app.config['CELERY_BACKEND_URL'] = 'cache+memory://' celery = celeryapp.create_celery_app(app) celeryapp.celery = celery
验证
修改完成后再运行测试,此时导入celery_tasks模块时,celeryapp.celery已经是初始化完成的实例,不会再报NoneType的错误,且任务会以TASK_ALWAYS_EAGER模式同步执行,不需要启动Celery worker也能正常测试。
内容的提问来源于stack exchange,提问作者Dr. Funkenstein
相关产品推荐
相关产品推荐

