正确地模拟在另一个芹菜任务中被调用的芹菜任务

Question

如何正确地模拟在另一个芹菜任务内调用的芹菜任务？ （下面的虚拟代码）

@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中有完全相同的代码结构。 PROJ/CEL/my_module.py 我想在凸出编写测试/测试/ 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) 

Answer 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 =========================== 

Answer 2

要启动，测试芹菜任务可真难。我通常把我所有的逻辑放到一个不是任务的函数中，然后创建一个调用该函数的任务，以便正确测试逻辑。其次，我不认为你想要在任务内部调用任务（不确定，但我相信这通常不被推荐）。相反，根据您的需要，你可能应该链接或分组：

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

最后，回答您的实际问题，你想修补delay方法正是它在你的代码中出现，如描述在this post。