如何用Mock/Monkeypatch/Patch将Celery的apply_async转为.s().apply()？

完全可以通过mock/patch/monkeypatch实现这个需求，核心思路是替换send_notification的apply_async方法，让它在被调用时转而执行同步的send_notification.s(...).apply()。

下面是两种常用实现方式：

1. 使用unittest.mock.patch（适用于unittest框架） #

from unittest.mock import patch
from your_module import abc, send_notification

def test_abc():
    def mock_apply_async(*args, **kwargs):
        # 提取传给apply_async的任务参数，转换为signature格式
        task_args = kwargs.pop('args', ()) + args
        task_sig = send_notification.s(*task_args, **kwargs)
        # 同步执行任务并返回结果
        return task_sig.apply()

    # 替换send_notification的apply_async方法
    with patch.object(send_notification, 'apply_async', side_effect=mock_apply_async):
        # 调用待测试的abc函数
        abc(...)
        # 这里可以添加断言逻辑，比如验证任务参数是否正确、执行结果是否符合预期

2. 使用pytest的monkeypatch（适用于pytest框架） #

import pytest
from your_module import abc, send_notification

def test_abc(monkeypatch):
    def mock_apply_async(*args, **kwargs):
        task_args = kwargs.pop('args', ()) + args
        task_sig = send_notification.s(*task_args, **kwargs)
        return task_sig.apply()

    # 替换apply_async方法
    monkeypatch.setattr(send_notification, 'apply_async', mock_apply_async)
    # 执行测试
    abc(...)
    # 添加自定义断言

注意事项 #

• 要注意apply_async的参数格式：比如apply_async(args=(1,2), kwargs={'key': 'val'})和send_notification.s(1,2, key='val')的参数对应关系，上面的代码已经处理了args关键字参数和位置参数的合并
• 由于你的任务使用了bind=True，同步执行apply()时会自动处理self参数，无需额外配置
• 如果需要验证任务是否被调用，可以给mock方法添加调用记录，或者直接断言任务执行后的副作用（比如数据库变更、通知内容等）

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