无法测试由不同Celery应用执行的任务链，该如何解决？

问题根源 #

开启CELERY_TASK_ALWAYS_EAGER=True后，所有任务会在当前测试进程内同步执行，但当前进程仅加载了app1的任务注册表，app2的task2未被注册到当前进程的Celery实例中，因此触发NotRegistered错误。

可行解决方案 #

方案1：Mock跨应用任务调用 #

既然两个应用属于不同环境，测试时无需真的执行task2，只需验证task1的结果是否正确传递给task2的调用逻辑即可。用unittest.mock模拟app2的任务执行：

from unittest.mock import patch

def test_chain(self):
    # Mock app2的任务发送方法，模拟执行结果
    with patch.object(app2, 'send_task') as mock_send_task:
        mock_send_task.return_value.get.return_value = "value"
        
        # 执行任务链
        final_result = task_chain.apply_async().get()
        
        # 验证最终结果符合预期
        self.assertEqual(final_result, "value")
        # 验证task2被正确调用，参数为task1的返回值
        mock_send_task.assert_called_once_with('app2.task2', args=({"key": "value"},))

方案2：拆分测试逻辑 #

将任务链拆分为两部分测试：

1. 单独验证task1的功能正确性
2. 验证任务链是否正确触发task2的调用

def test_task1(self):
    # 单独测试task1的执行结果
    result = task1.apply_async().get()
    self.assertEqual(result, {"key": "value"})

def test_chain_trigger_logic(self):
    # 验证任务链是否正确触发task2的签名调用
    with patch.object(app2.signature('app2.task2'), 'apply_async') as mock_apply:
        task_chain.apply_async()
        # 确认task2的签名被触发
        mock_apply.assert_called_once()
        # 检查传递给task2的参数是否为task1的返回值
        call_args = mock_apply.call_args
        self.assertEqual(call_args[1]['args'], ({"key": "value"},))

方案3：端到端测试（需启动对应Worker） #

如果必须进行完整的端到端测试，需要关闭CELERY_TASK_ALWAYS_EAGER=True，同时确保app1和app2的Worker已分别启动并连接到各自的Broker：

# 注意：不要设置CELERY_TASK_ALWAYS_EAGER=True
def test_end_to_end_chain(self):
    # 确保app1、app2的Worker已启动并正常监听任务
    chain_result = task_chain.apply_async()
    # 等待任务执行完成，设置合理超时
    final_result = chain_result.get(timeout=10)
    self.assertEqual(final_result, "value")

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