正确模拟在另一个 celery 任务中调用的 celery 任务答案

【问题标题】：properly mock celery task that is being called inside another celery task正确模拟在另一个 celery 任务中调用的 celery 任务
【发布时间】：2018-01-20 00:41:51
【问题描述】：

如何正确模拟在另一个 celery 任务中调用的 celery 任务？（下面的虚拟代码）

@app.task
def task1(smthg):
    do_so_basic_stuff_1
    do_so_basic_stuff_2
    other_thing(smthg)

@app.task
def task2(smthg):
    if condition:
        task1.delay(smthg[1])
    else:
        task1.delay(smthg)

我在 my_module 中确实有完全相同的代码结构。项目/cel/my_module.py 我正在尝试在 proj/tests/cel_test/test.py 中编写测试

测试功能：

def test_this_thing(self):
    # firs I want to mock task1
    # i've tried to import it from my_module.py to test.py and then mock it from test.py namespace 
    # i've tried to import it from my_module.py and mock it
    # nothing worked for me

    # what I basically want to do 
    # mock task1 here
    # and then run task 2 (synchronous)
    task2.apply()
    # and then I want to check if task one was called 
    self.assertTrue(mocked_task1.called)

【问题讨论】：

标签： python python-2.7 mocking

【解决方案1】：

您调用的不是task1() 或task2()，而是它们的方法：delay() 和apply() - 因此您需要测试这些方法是否被调用。

这是我刚刚根据您的代码编写的一个工作示例：

tasks.py

from celery import Celery

app = Celery('tasks', broker='amqp://guest@localhost//')

@app.task
def task1():
    return 'task1'

@app.task
def task2():
    task1.delay()

test.py

from tasks import task2

def test_task2(mocker):
    mocked_task1 = mocker.patch('tasks.task1')
    task2.apply()
    assert mocked_task1.delay.called

测试结果：

$ pytest -vvv test.py
============================= test session starts ==============================
platform linux -- Python 3.5.2, pytest-3.2.1, py-1.4.34, pluggy-0.4.0 -- /home/kris/.virtualenvs/3/bin/python3
cachedir: .cache
rootdir: /home/kris/projects/tmp, inifile:
plugins: mock-1.6.2, celery-4.1.0
collected 1 item                                                                

test.py::test_task2 PASSED

=========================== 1 passed in 0.02 seconds ===========================

【讨论】：

为了更完整的测试，也可以直接mockdelay或apply方法。

【解决方案2】：

首先，测试 Celery 任务可能非常困难。我一般把我所有的逻辑都放在一个不是任务的函数中，然后做一个只调用那个函数的任务，这样你就可以正确地测试逻辑了。

其次，我认为您不想在任务中调用任务（不确定，但我相信通常不建议这样做）。相反，根据您的需要，您可能应该进行链接或分组：

http://docs.celeryproject.org/en/latest/userguide/canvas.html#the-primitives

最后，要回答您的实际问题，您可能需要在代码中的确切位置修补 delay 方法，如 this post 中所述。

【讨论】：

只要你不等待它（使用它的结果），在其他任务中调用任务没有错。