FastAPI中无法启用Celery的task_always_eager模式问题

我正在FastAPI应用里实现队列功能，手动验证过Broker连接和消息发送都正常。但为了测试需求，想通过设置task_always_eager选项，不用启动Broker也能跑测试。环境变量已经正确配置（调试显示值为true），但运行pytest时还是会尝试连接Broker，报连接拒绝错误。

项目结构 #

$ tree -I '*pyc|venv' src tests
src
├── celery_app.py
├── celeryconf.py
├── db.py
├── main.py
├── __pycache__
├── red_link
│	├── connectors.py
│	├── exceptions.py
│	├── __init__.py
│	├── models.py
│	├── __pycache__
│	├── services.py
│	├── tasks.py
│	└── views.py
├── settings.py
└── throttling
	├── __init__.py
	├── middleware.py
	└── __pycache__
tests
├── conftest.py
├── fixtures
│	└── casettes
│		├── test_sends_sms_fails.yml
│		└── test_sends_sms.yml
├── __pycache__
└── test_redlink_sms_sending.py

8 directories, 18 files

celery_app.py 配置 #

from celery import Celery
from settings import get_settings


settings = get_settings()


def create_celery_app() -> Celery:
    celery = Celery(
        "sms_sender",
        broker=settings.celery_broker_url,
        include=["red_link.tasks"],
    )
    celery.config_from_object("celeryconf")
    return celery


app = create_celery_app()

celeryconf.py 配置 #

from settings import get_settings


settings = get_settings()

task_always_eager = settings.celery_task_always_eager

测试报错信息 #

=============================================================================================== short test summary info ===============================================================================================
FAILED tests/test_redlink_sms_sending.py::test_sends_sms - kombu.exceptions.OperationalError: [Errno 111] Connection refused
FAILED tests/test_redlink_sms_sending.py::test_sends_sms_fails - kombu.exceptions.OperationalError: [Errno 111] Connection refused
FAILED tests/test_redlink_sms_sending.py::test_send_sms_fails_with_error_connection - kombu.exceptions.OperationalError: [Errno 111] Connection refused

内容的提问来源于stack exchange，提问作者gonczor