Django内のタスクを委譲するメソッドをどのようにテストしますか？

Question

私は2つのセロリのタスクを持って言う：

@app.task(name='process_one_line') 
def process_one_line(line): 
    do_alot(line) 

@app.task(name='process_one_file') 
def process_one_file(file_id): 
    for line in get_file_by_id(file_id): 
     process_one_line.delay(line) 

と、私はこのようになりますテストケースを持っていると言う：

def test_processing_a_file(self): 
    process_one_file(self.file_id) 

これは正常に動作しますが、委任process_one_line方法は実際に私にオフにプッシュされますテストケース外の他のセロリインスタンス。要するに、ポート6379で動作している私のRedisサーバーはタスクを受け取り、それを「メイン」ワーカーにプッシュします。したがって、do_alotの変更によって、djangoテストの実行時に作成されたテストデータベースではなく、実際のデータベースに反映されます。

私はprocess_one_lineを単独でテストすることでテストできますが、全体をすべてテストすることができればいいと思います。

Answer 1

テスト設定にCELERY_ALWAYS_EAGER=Trueを追加して、すべてのタスクがレディスに送信されないようにすることができます。