如何在Celery任务中触发SMTPException测试邮件发送？

核心问题分析 #

Celery任务默认异步执行时会在独立worker进程运行，你在测试进程里的mock不会作用到worker环境，这就是普通视图函数（当前进程执行）能触发异常，但Celery任务不行的根本原因。以下是具体解决步骤：

1. 强制Celery任务同步执行（最直接的测试方式） #

在测试环境的Django配置里添加：

# settings.py 测试环境专属配置
CELERY_TASK_ALWAYS_EAGER = True
CELERY_TASK_EAGER_PROPAGATES = True

这样所有Celery任务都会在当前测试进程同步执行，你的mock能直接覆盖任务里的邮件发送逻辑。

2. 精准mock任务中的邮件发送方法 #

不要mock全局的邮件方法，直接mock任务文件里用到的EmailMultiAlternatives.send，确保mock作用域正确。示例测试代码（基于pytest+mock）：

import pytest
from unittest.mock import patch
from django.core.mail import SMTPException
from my_app.mailer.tasks import send_mail

def test_send_mail_task_smtp_exception(capsys):
    # mock任务中的send方法，抛出指定的SMTPException
    with patch('my_app.mailer.tasks.EmailMultiAlternatives.send') as mock_send:
        mock_send.side_effect = SMTPException("SMTP服务器连接失败")
        
        # 同步执行Celery任务（因配置了ALWAYS_EAGER，delay也会同步执行）
        send_mail.delay()
        
        # 验证mock是否被调用
        assert mock_send.called_once()
        
        # 捕获print输出，验证异常信息是否正确打印
        captured = capsys.readouterr()
        assert "Exception  SMTP服务器连接失败" in captured.out

3. 优化任务的异常处理（可选，便于测试和生产排查） #

把任务里的print改成Celery自带的日志记录，这样测试和生产环境都能统一捕获日志：

from django.core.mail import EmailMultiAlternatives
from celery.utils.log import get_task_logger

logger = get_task_logger(__name__)

@app.task(bind=True )
def send_mail(self):
    subject, from_email, to = 'hello', 'from@example.com', 'to@example.com'
    text_content = 'This is an important message.'
    html_content = '<p>This is an <strong>important</strong> message.</p>'
    msg = EmailMultiAlternatives(subject, text_content, from_email, [to])
    msg.attach_alternative(html_content, "text/html")
    try:
        msg.send(fail_silently=False)
    except SMTPException as exc:
        logger.error('SMTP发送失败: %s', exc)  # 用日志替代print

对应的测试可以捕获日志输出：

def test_send_mail_task_logging(caplog):
    with patch('my_app.mailer.tasks.EmailMultiAlternatives.send') as mock_send:
        mock_send.side_effect = SMTPException("连接超时")
        send_mail.delay()
        assert "SMTP发送失败: 连接超时" in caplog.text

4. 手动同步执行任务（无需修改全局配置） #

如果不想修改全局Celery配置，可以用apply()方法手动同步执行任务：

result = send_mail.apply(args=(), kwargs={})
# 因为任务内部捕获了异常，所以result会标记为成功状态
assert result.successful()

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