正確地模擬在另一個芹菜任務中被調用的芹菜任務

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。