如何对运行芹菜任务的代码进行单元测试?
How to unit test code that runs celery tasks?
我正在开发的应用程序是高度异步的。 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 本身。有很大的不同。我在构思这个问题之前确实遇到了那个问题。
看看 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
我正在开发的应用程序是高度异步的。 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 本身。有很大的不同。我在构思这个问题之前确实遇到了那个问题。
看看 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