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

发布时间：2020-12-16 22:59:02 所属栏目：Python 来源：网络整理

导读：我正在处理的应用程序非常异步. 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 mock

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

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!