不能直接用celery.task.delay()在pytest中测异步任务，因为delay()会将任务发给真实Broker（如Redis/RabbitMQ），依赖外部服务、变慢、难断言且可能污染队列；应启用task_always_eager=True使delay()同步执行并捕获结果。
为什么不能直接用celery.task.delay()在pytest里测异步任务
因为delay()会把任务发给真实Broker（比如Redis/RabbitMQ），再由Worker执行——这会让测试变慢、依赖外部服务、难以断言结果，还可能污染生产队列。你真正需要的是「不发消息、不启Worker、立刻执行、能捕获返回值」的测试模式。
关键不是“怎么测异步”，而是“怎么让Celery在测试时同步执行”。核心开关是task_always_eager=True，它会让apply_async()和delay()跳过Broker，直接调用run()方法。
- 必须在测试配置中启用：
CELERY_TASK_ALWAYS_EAGER = True（Django）或app.conf.task_always_eager = True（纯Celery） - 同时设
task_eager_propagates = True，否则异常不会冒泡到测试进程，assert会失效 - 不要用
apply()（它本来就是同步的），而要用delay()或apply_async()——这样才能验证你的业务代码是否正确调用了Celery接口
如何在pytest中安全地配置Celery测试App
别复用生产Celery实例，也别改全局配置。最干净的做法是：每个测试函数用@pytest.fixture构造一个隔离的celery_app，并预设好测试所需参数。
示例fixture：
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
立即学习“Python免费学习笔记（深入）”；
@pytest.fixture
def celery_app():
from celery import Celery
app = Celery("test")
app.conf.update(
task_always_eager=True,
task_eager_propagates=True,
broker_url="memory://", # 防止误连真实Broker
result_backend="cache+memory://",
)
return app
-
broker_url="memory://"是Celery内置的内存Broker，比redis://更轻量，且不启动任何服务 - 如果任务用了
@shared_task（Django场景），需确保fixture返回的app被shared_task._app识别，可在fixture末尾加app.set_default() - 避免在
conftest.py里全局修改current_app，否则多个测试并发时可能互相干扰
测试带参数和重试逻辑的Celery任务时要注意什么
参数传递本身没问题，但重试（autoretry_for、retry_kwargs）和task_always_eager=True共存时，行为容易误解：重试不会真正延迟或换Worker，而是立刻循环调用run()，直到成功或达到max_retries。
- 若想验证重试逻辑是否注册正确，检查
my_task.retry_kwargs属性，而不是观察是否真的“重试了” - 要测失败后是否触发重试，得让
run()抛出指定异常（如ValueError），并确认my_task.apply_async(...).get()最终返回预期结果或抛出MaxRetriesExceededError - 注意
countdown和eta参数在task_always_eager=True下会被忽略——它们只对Broker生效，测试中无意义
如何断言Celery任务是否被正确调用（而非执行）
有些场景你并不关心任务执行结果，只关心它有没有被调度出去，比如发通知、触发下游流程。这时该用celery.contrib.testing.mocks或unittest.mock.patch打桩。
- 用
@patch("myapp.tasks.send_email.delay")，然后检查mock_delay.assert_called_once_with("user@example.com", "Hi") - 不要patch
send_email.run——那是在测执行逻辑；要patch.delay或.apply_async，才对应“调度动作” - 如果任务在类方法里被调用（如
self.send_notification.delay()），patch位置必须精确到实例属性或绑定方法，否则mock无效
真实任务链（group/chord）的测试更复杂，需要额外启用task_inherit_parent_priority等配置，且memory:// Broker对复杂工作流支持有限——这时候建议退回到单元级，单独测每个子任务的逻辑，用mock绕过链式调度。
