Django Pytest调用Celery遇Kombu连接拒绝，求本地及CI解决方案

一、本地测试不用启动RabbitMQ的办法 #

1. 切换Celery测试模式（最简单） #

直接在测试环境的配置文件里改Celery设置，让任务同步执行，完全绕开外部Broker依赖：

# settings/test.py
CELERY_BROKER_URL = 'memory://'  # 用内存作为临时Broker
CELERY_TASK_ALWAYS_EAGER = True  # 任务立即同步执行，不进队列
CELERY_TASK_EAGER_PROPAGATES = True  # 任务异常直接抛出，方便调试

这样跑测试时，Celery不会尝试连接RabbitMQ，任务直接在当前进程执行，完全不需要外部服务。

2. 单个测试Mock Celery任务 #

如果不想全局修改配置，针对触发任务的测试用例，直接Mock掉任务调用即可：

def test_stock_created_with_valid_data(api_client, stock_json, mocker):
    # Mock掉你的Celery任务的delay方法
    mock_task = mocker.patch('your_app.tasks.your_background_task.delay')
    
    # 执行测试动作
    res = api_client.post(reverse_lazy("core_api:stocks-list"), stock_json, format="json")
    
    # 验证任务是否被正确调用
    mock_task.assert_called_once_with(你预期的参数)

这种方法适合只需要验证任务是否被触发，不需要测试任务逻辑的场景。

3. 用pytest-celery插件管理测试环境 #

安装pytest-celery后，它会自动帮你搭建临时的Celery测试环境，不用手动配置：

pip install pytest-celery

然后在项目根目录的conftest.py里添加配置：

import pytest

@pytest.fixture(scope='session')
def celery_config():
    return {
        'broker_url': 'memory://',
        'result_backend': 'cache+memory://',
    }

跑测试时插件会自动应用这些配置，完全不需要启动RabbitMQ。

二、GitHub Actions中配置RabbitMQ服务 #

在你的CI配置文件（比如.github/workflows/tests.yml）里，通过services块启动RabbitMQ镜像，确保测试时Broker可用：

name: 运行测试

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    services:
      rabbitmq:
        image: rabbitmq:3.9-alpine  # 用轻量的Alpine镜像
        ports:
          - 5672:5672  # 映射端口到宿主机器
        # 等待RabbitMQ完全启动再执行测试
        options: >-
          --health-cmd "rabbitmq-diagnostics -q ping"
          --health-interval 10s
          --health-timeout 5s
          --health-retries 5
    steps:
      - uses: actions/checkout@v4
      - name: 设置Python环境
        uses: actions/setup-python@v5
        with:
          python-version: '3.10'
      - name: 安装依赖
        run: |
          python -m pip install --upgrade pip
          pip install -r requirements.txt
      - name: 执行pytest
        run: pytest
        env:
          # 告诉Celery连接本地的RabbitMQ服务
          CELERY_BROKER_URL: amqp://guest:guest@localhost:5672//

配置里的health-cmd会检查RabbitMQ是否就绪，避免测试因Broker未启动而失败。

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