嘲笑芹菜任務內的呼叫

我有一個運行芹菜任務的瓶子應用程序。我試圖模擬出發生在該任務內部的單個API調用。嘲笑芹菜任務內的呼叫

views.py

from mypackage.task_module import my_task 
@app.route('/run_task') 
def run_task(): 
    task = my_task.delay() 
    return some_response

task_module.py

from mypackage.some_module import SomeClass 

@celery.task 
def my_task(): 
    return SomeClass().some_function()

some_module.py

from mypackage.xyz import external_service 
class SomeClass(object): 
    def some_function(self): 
     #do some stuff 
     result = external_service(some_param) 
     if 'x' in result: 
      #do something 
     elif 'y' in result: 
      #do something else

我想嘲笑Ø取出result = external_service()行，以便我可以觸發第一個或第二個代碼路徑。

所以這裏就是我想：

@mock.patch('mypackage.some_module.external_service', autospec=True) 
def test_x_path(my_mock): 
    my_mock.return_value = {'x': some_val} 
    #run test, expect 'x' code path to run

然而，這是不行的，因爲（我認爲）的補丁發生在瓶的Python進程，而不是芹菜是使用一個。嘲笑任務本身將不起作用，因爲我試圖測試的是當外部服務返回'x'或'y'時，該任務的行爲如何。

幫助將不勝感激。

來源

2015-05-26 Chinmay Kanchi

一個不錯的選擇是在您的測試配置中將CELERY_ALWAYS_EAGER設置爲True。這使得所有對Celery的調用都是同步的。請參閱documentation for this option。有了這個選項，你在Flask進程中設置的任何模擬都應該在Celery任務中工作。

作爲一個好處，您的測試配置被簡化了，因爲您不需要有一個芹菜工作者。

UPDATE：在評論中討論之後，看起來你不想或者無法擺脫芹菜工作者的測試配置。在這種情況下，我可以提供三種解決方案，我認爲你需要的東西：

寫remote control命令嘲笑你的芹菜任務，然後有broadcast()上所有工人的測試代碼運行。
爲您的工作人員定義custom command line option，比如--test。然後添加一個bootstep，檢查這個參數並進行模擬。
創建一個替代模塊，爲-A命令行參數中的芹菜工作者提供參考。這應該是你的原始模塊的一個完全相同的副本，但添加了模擬。然後用這個替代模塊啓動你的工作人員進行測試。

我希望你找到滿意的三個選項之一！

來源

2015-05-26 06:16:13 Miguel

有沒有一種方法來設置這個單個測試的只是時間？我的套件中已經有測試異步功能的測試，並且在運行此測試時已經有一大堆芹菜工人正在運行。 –

如果你已經有芹菜工人，那你爲什麼不在工人過程中嘲笑這項工作？ – Miguel

那麼，我該怎麼做？我不想在生產代碼中添加任何東西，它必須在測試代碼中完成。 –

創建成立了測試功能

class TestCeleryTask(TestCase): 
    def setUp(self): 
     app.config['CELERY_ALWAYS_EAGER'] = True 
     app.config['BROKER_BACKEND'] = 'memory' 
     app.config['CELERY_EAGER_PROPAGATES_EXCEPTIONS'] = True 

    def test_task(self): 
     # test it

來源

2015-05-26 12:32:23 ChillarAnand

嘲笑芹菜任務內的呼叫

回答

相關問題