FastAPI中Celery shared_task无法通过pytest测试问题求助

场景说明 #

测试包含Celery任务的FastAPI接口时，尝试Mock send_email.delay 任务，但运行测试时触发Kombu连接错误。环境为Docker化部署，非测试环境运行无异常。

相关代码 #

业务代码中的Celery任务调用 #

send_email.delay(
    user.email,
    "Subject goes here",
    html_content,
)

Mock Fixture定义 #

@pytest.fixture
def mocked_send_email(mocker):
    mocker.patch('app.views.user.send_email.delay', return_value=None)

测试用例代码 #

def test_create_user(client, mocked_send_email):
    payload = {}

    response = client.post(
        f"{base_url}/user",
        json=payload,
        headers=admin_headers
    )

    mocked_send_email.assert_called_once_with(
        "email@example.com",
        "Subject goes here",
        mock.ANY
    )

conftest.py中的Celery配置 #

@pytest.fixture(scope="session")
def celery_config():
    print('running celery config')
    return {"broker_url": REDIS, "result_backend": REDIS}


@pytest.fixture(scope='session', autouse=True)
def configure_celery(celery_config):
    from app.celery import celery
    celery.conf.update(celery_config)
    yield
    celery.conf.update({
        'broker_url': 'memory://',
        'result_backend': 'rpc://'
    })

错误信息 #

usr/local/lib/python3.12/site-packages/kombu/connection.py:476: OperationalError
test_1           | ------------------------------ Captured log call -------------------------------
test_1           | WARNING  kombu.connection:connection.py:669 _info - _info - No hostname was supplied. Reverting to default 'localhost'

补充日志 #

configure_celery fixture的打印内容：

Before update:redis://redis-cache:6379/1 redis://redis-cache:6379/1
After update:redis://redis-cache:6379/5 redis://redis-cache:6379/5
After reset: memory:// rpc://

• 修正Mock路径与返回值
  你当前的Mock fixture没有返回被Mock的对象，导致后续断言会失败。同时要确保Mock路径是任务被调用时的导入路径（而非任务定义路径），调整后的fixture如下：

  @pytest.fixture
  def mocked_send_email(mocker):
      mock_obj = mocker.patch('app.views.user.send_email.delay', return_value=None)
      return mock_obj
  

• 禁用测试时的实际Celery连接
  既然要Mock任务，完全不需要连接真实Redis。修改自动启用的configure_celery fixture，直接用内存Broker避免实际连接：

  @pytest.fixture(scope='session', autouse=True)
  def configure_celery():
      from app.celery import celery
      # 保存原配置，测试后恢复
      original_broker = celery.conf.broker_url
      original_backend = celery.conf.result_backend
      # 测试时用内存Broker
      celery.conf.update({
          'broker_url': 'memory://',
          'result_backend': 'cache+memory://'
      })
      yield
      celery.conf.update({
          'broker_url': original_broker,
          'result_backend': original_backend
      })
  

• 排查Docker网络解析问题（可选）
  错误日志显示Kombu尝试连接localhost，但配置中是redis-cache。如果确实需要在测试中连接真实Redis（不推荐），要确保测试容器能通过Docker网络解析redis-cache服务名，检查Docker Compose的服务名和网络配置是否正常。

• 借助pytest-celery简化测试
  如果使用pytest-celery插件，可以直接用mock_task装饰器或celery_worker fixture，无需手动配置就能快速Mock Celery任务，减少手动配置的出错概率。

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