別のセロリのタスクの中で呼び出されているセロリのタスクを正しく模倣する

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 私がテストを書くことをしようとしているがproj /テスト/ 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

セロリのタスクをテストすることは本当に難しいことができ、開始します。私は一般に自分のロジックをすべてタスクではない関数に置き、その関数を単に呼び出すタスクを作って、ロジックを適切にテストできるようにします。

第2に、タスクの中でタスクを呼び出すことは考えていませんが（確かではありませんが、一般的には推奨されません）代わりに、あなたのニーズに応じて、あなたはおそらく、連鎖またはグループ化する必要があります。

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

最後に、あなたの実際の質問に答えるために、それはあなたのコードで発生しdelay方法場所を正確に説明するように、あなたは、パッチを適用したいと思いますthis postにあります。