如何对运行芹菜任务的代码进行单元测试？

概述我正在处理的应用程序非常异步. Web应用程序根据用户 *** 作通过芹菜运行许多任务.芹菜任务本身能够启动进一步的任务. 下面显示的代码经常出现在我们的代码库中. def do_sth(): logic(); if condition: function1.apply_async(*args) else: function2.apply_asy 我正在处理的应用程序非常异步. Web应用程序根据用户 *** 作通过芹菜运行许多任务.芹菜任务本身能够启动进一步的任务.

下面显示的代码经常出现在我们的代码库中.

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本身.有一个主要的区别.在构建这个问题之前,我确实遇到了这个问题.

解决方法 看一下 Mock库,它允许用“mocks”替换你不想调用的函数,这些函数看起来与原始函数非常相似,以说服调用代码它正在调用“真实的东西”.然后,您可以检查是否实际调用了模拟以及使用了哪些参数.例：

import mockdef test_do_sth():    with mock.patch('function1.apply_async') as function1_mock:        do_sth()        assert function1_mock.called

总结

以上是内存溢出为你收集整理的如何对运行芹菜任务的代码进行单元测试？全部内容，希望文章能够帮你解决如何对运行芹菜任务的代码进行单元测试？所遇到的程序开发问题。

如果觉得内存溢出网站内容还不错，欢迎将内存溢出网站推荐给程序员好友。