如何在测试中触发Celery task_failure信号并排查接收异常？

看起来你在测试Celery任务失败信号时遇到了匹配问题，我来帮你拆解一下问题并给出解决方案。

问题分析 #

先看你的代码里几个关键问题：

1. 重复且错误的信号注册：你已经用@task_failure.connect(sender='simple_task')装饰器注册了处理函数，又在celeryd_init里重复调用task_failure.connect，而且这里还犯了语法错误——sender='simple_task'前面少了逗号，这会导致无效的注册。
2. Sender匹配错误：Celery的task_failure信号触发时，默认sender是任务对象（比如simple_task这个函数对象），而不是你指定的字符串'simple_task'。这就是为什么调试时_live_receivers(sender)返回空——注册的sender是字符串，实际触发的sender是任务对象，两者不匹配。
3. 测试模式问题：如果不设置Celery为同步执行模式，apply()会异步提交任务，不会立即执行，异常也不会同步触发信号，导致测试时信号没被调用。

解决方案 #

第一步：修复任务代码中的信号注册 #

先把重复且错误的信号注册代码删掉，并且改用任务对象作为sender来注册信号：

import logging
from celery import Celery, signals

logger = logging.getLogger(__name__)
celery = Celery('tasks', broker='pyamqp://guest@localhost//')

def do_something(): 
    print("hello")

@celery.task(name='simple_task')
def simple_task(x, y): 
    do_something()
    return x - y

# 用任务对象作为sender，确保触发时能匹配上
@signals.task_failure.connect(sender=simple_task)
def task_failure_handler(args=None, **kwargs): 
    logger.exception("Task failed")

# 删掉这段错误的重复注册代码
# @celeryd_init.connect
# def init_signals(*args, **kwargs): 
#     task_failure.connect(task_failure_handler sender='simple_task')

第二步：调整测试代码，确保信号触发 #

有两种方式可以测试：一种是让Celery同步执行任务，自然触发信号；另一种是手动触发信号来验证处理函数。

方式一：同步执行任务，自然触发信号

设置Celery的task_always_eager=True，这样任务会同步执行，异常会立即触发task_failure信号：

from unittest.mock import patch
import tasks

def test_signal_handler_with_eager_execution():
    # 开启eager模式，任务同步执行
    tasks.celery.conf.update(task_always_eager=True)
    
    with patch('tasks.task_failure_handler') as signal_mock:
        with patch('tasks.do_something') as task_mock:
            task_mock.side_effect = Exception("Test failure")
            
            # 执行任务时会抛出异常，触发信号
            try:
                tasks.simple_task.apply((1, 2))
            except Exception:
                # 捕获任务执行失败的异常，不影响信号触发
                pass
            
            # 断言信号处理函数被调用
            assert signal_mock.called

方式二：手动触发信号

如果你只想测试信号处理逻辑，不想依赖任务执行流程，可以直接手动调用信号的send方法：

from unittest.mock import patch
from celery import signals
import tasks

def test_manual_signal_trigger():
    with patch('tasks.task_failure_handler') as signal_mock:
        # 手动模拟task_failure信号触发，传入正确的sender和参数
        signals.task_failure.send(
            sender=tasks.simple_task,
            args=(1, 2),
            kwargs={},
            exception=Exception("Test exception"),
            traceback=None,
            einfo=None
        )
        
        # 断言处理函数被调用
        assert signal_mock.called

为什么之前的代码不生效？ #

• 你用字符串'simple_task'作为sender注册信号，但Celery触发task_failure时，sender是任务对象，两者类型不匹配，导致接收者没被选中。
• 重复的信号注册不仅多余，还因为语法错误导致无效，进一步干扰了信号的匹配逻辑。
• 默认异步模式下，apply()只是提交任务，不会立即执行，所以测试时异常没发生，信号也没触发。

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