如何对运行 celery 任务的代码进行单元测试？答案

【问题标题】：How to unit test code that runs celery tasks?如何对运行 celery 任务的代码进行单元测试？
【发布时间】：2015-10-22 17:22:42
【问题描述】：

我正在开发的应用程序严重异步。 Web 应用程序根据用户操作通过 celery 运行许多任务。 celery 任务本身能够启动进一步的任务。

如下所示的代码经常出现在我们的代码库中。

def do_sth():
    logic();
    if condition:
         function1.apply_async(*args)
    else:
         function2.apply_asynch(*args)

现在我们要开始对我们编写的任何新代码进行单元测试，但我们不知道该怎么做。我们想在pytest 单元测试中断言的是，我们想看看 function1 是否真的被调用了。我们不希望自己运行function1，因为我们将对function1 进行单元测试。

我不想将 celery 作为进程运行，也不想在单元测试会话期间运行任何 AMQP 代理。

这可以实现吗？

编辑

有人指出这是How do you unit test a Celery task?的复制品

事实并非如此。想想看。我要问的是如何通过apply_async 测试函数是否调用了function1。这个问题是关于我如何测试 function1 本身。有一个主要区别。在提出这个问题之前，我确实遇到了这个问题。

【问题讨论】：

How do you unit test a Celery task?的可能重复
不是重复的
我已经看到了这个问题，这不是我要问的。如果不清楚，请提出更多问题

标签： python unit-testing celery pyramid pytest

【解决方案1】：

查看unittest.mock 模块，该模块允许用“模拟”替换您不想被调用的函数，这些“模拟”看起来与原始函数非常相似，可以让调用代码相信它正在调用“真实的东西”。然后，您可以检查模拟是否实际被调用以及使用了哪些参数。示例：

from unittest.mock import patch

def test_do_sth():
    with patch('function1.apply_async') as function1_mock:
        do_sth()
        assert function1_mock.called

【讨论】：

请注意，您还可以使用function1_mock.assert_called_once_with(*args) 来测试调用次数以及是否使用正确的参数调用。
你也可以使用patch作为测试函数的装饰器（在本例中为test_do_sth），我更喜欢这个。