Celery任务测试求助：如何测试含短信发送的API代码

嘿，别慌！新手碰到代码测试问题太正常了，我来帮你拆解这段结合了Celery异步任务和数据库操作的代码该怎么测～

把数据库逻辑和Celery任务分开测，能更快定位问题：

1. 测试视图里的数据库操作逻辑 #

先验证mydef()里的筛选、更新逻辑是否正确——毕竟这是整个流程的基础：

• 用Django自带的TestCase（从MyModel的写法来看大概率是Django项目），先造测试数据，调用mydef()后检查数据库状态。
• 示例代码：

from django.test import TestCase
from myapp.models import MyModel
from myapp.views import mydef

class MyDefDatabaseTest(TestCase):
    def setUp(self):
        # 造测试数据：2个需要处理的，1个已经处理过的
        MyModel.objects.create(my_flag=False, phone="13800138000")
        MyModel.objects.create(my_flag=False, phone="13900139000")
        MyModel.objects.create(my_flag=True, phone="13700137000")

    def test_mydef_updates_flags_correctly(self):
        mydef()
        # 检查原本未处理的都被标记为已处理
        self.assertEqual(MyModel.objects.filter(my_flag=False).count(), 0)
        # 原本已处理的状态不变
        self.assertEqual(MyModel.objects.filter(my_flag=True).count(), 3)

2. 测试Celery异步任务send_sms #

异步任务的测试核心是验证任务被正确触发、参数传递无误，同时别真的发短信（不然测试成本太高）：

• 用unittest.mock或者pytest-mock把实际的短信发送逻辑mock掉，只检查调用情况。
• 开启Celery的eager模式，让异步任务同步执行，方便测试时直接捕获结果。
• 示例代码：

from django.test import TestCase
from unittest.mock import patch
from myapp.tasks import send_sms

class SendSmsTaskTest(TestCase):
    # 临时开启Celery eager模式
    def setUp(self):
        from django.conf import settings
        settings.CELERY_TASK_ALWAYS_EAGER = True

    @patch('myapp.tasks.send_sms')  # 替换实际的短信发送逻辑
    def test_send_sms_task_triggered(self, mock_send):
        test_addr = "13800138000"
        test_text = "您的对象已保存成功"
        send_sms.delay(test_addr, test_text)
        
        # 验证任务被调用，参数完全匹配
        mock_send.assert_called_once_with(test_addr, test_text)

单独模块没问题后，再测整个流程是否顺畅——确保mydef()执行时，所有符合条件的对象都触发了短信任务，同时数据库状态正确更新：

from django.test import TestCase
from unittest.mock import patch
from myapp.models import MyModel
from myapp.views import mydef

class MyDefFullFlowTest(TestCase):
    def setUp(self):
        # 造3个需要处理的测试对象
        for i in range(3):
            MyModel.objects.create(my_flag=False, phone=f"1380013800{i}")

    @patch('myapp.tasks.send_sms')
    def test_full_flow_works(self, mock_send):
        mydef()
        
        # 检查数据库全部更新完成
        self.assertEqual(MyModel.objects.filter(my_flag=True).count(), 3)
        # 检查短信任务被调用了3次，每次参数对应正确的手机号
        expected_calls = [
            mock.call("13800138000", "你的短信内容"),
            mock.call("13800138001", "你的短信内容"),
            mock.call("13800138002", "你的短信内容")
        ]
        mock_send.assert_has_calls(expected_calls, any_order=False)

• 一定要mock第三方服务：如果send_sms里调用了实际的短信API，绝对要mock掉，避免测试依赖外部服务、产生不必要的费用。
• 灵活设置Celery模式：除了在测试里临时开eager模式，也可以在settings.py的测试配置段统一设置：CELERY_TASK_ALWAYS_EAGER = True。
• 用工具简化测试：如果习惯用pytest，可以搭配pytest-django和pytest-celery插件，写测试会更简洁高效。

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