你如何測試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) 

，並說我有一個TestCase，看起來像這樣：

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添加到您的測試設置，以防止將所有任務發送到redis。