從另一個模塊檢查芹菜任務狀態

我試圖測試一個芹菜任務是否在請求到我的Django服務器後啓動。我有這樣的事情：從另一個模塊檢查芹菜任務狀態

# tasks.py 
def add(x, y): 
    return x + y 

# views.py 
def home(request): # respond to request at root url 
    tasks.add.delay(1,2) 
    return HttpResponse('hello world') 

# tests.py 
class MyTest(TestCase): 
    def test_task_triggered(self): 
     self.client.get('/') 

     # XXXX HOW TO TEST THAT TASK HAS BEEN TRIGGERED?

我該如何測試任務是否在我的單元測試中開始？很明顯，我沒有直接訪問任務ID，否則like this would work。

更一般地說，如何檢測從各個函數，類或模塊觸發的芹菜任務？

感謝您的幫助。

來源

2012-03-01 dino

您可以使用Mock（單元測試時）。補充您的任務中的delay屬性，然後檢查模擬的called屬性。在你的榜樣，你可以試試：

# tests.py 
from mock import Mock, patch 

class MyTest(TestCase): 
    def test_task_triggered(self): 
     with patch('views.tasks.add.delay') as patch_mock: 
      self.client.get('/') 
     self.assertTrue(patch_mock.called)

需要注意的是，一旦你做到了這一點，tasks.add實際上不會在這個測試用例調用，因爲我們已經有一個模擬取而代之。這個測試只是斷言它被調用（並且你可以檢查patch_mock.call_args_list來聲明它被用正確的參數調用）

如果這是一個交易斷路器 - 你想在一個測試案例中斷言一個任務被調用，以及任務的影響/副作用 - 你可以用wraps屬性mock，這可能工作實驗：

# tests.py 
from mock import Mock, patch 

class MyTest(TestCase): 
    def test_task_triggered(self): 
     add_mock = Mock(wraps=tasks.add) 
     with patch('views.tasks.add.delay', new=add_mock): 
      self.client.get('/') 
     self.assertTrue(add_mock.called)

但是其他的方式可能會更好，因爲它更好地隔離開什麼正在測試。

來源

2012-03-04 04:52:55 Dan

我還發現'patch_mock.call_args'可用於解包傳遞給任務的'args'和'kwargs'。 – Matt 2015-03-07 01:27:42

Celery不允許這種交互，但您可以使用緩存或甚至會話來監視它的狀態。當然，您需要從外部傳遞一些ID以進一步獲得該價值。

但是，如果您僅需要此測試，則可以使用CELERY_ALWAYS_EAGER設置，該設置告訴Celery以簡單的程序執行所有任務，而無需運行celeryd。

來源

2012-03-01 01:35:21 ilvar

不是必須有一種方法來訪問不同模塊之間的芹菜任務ID？畢竟，celeryctl可以列出當前正在運行的任務的任務ID。 – dino 2012-03-01 22:26:33

我發現他們有兩種方法，可以使任務同步運行在我的測試

1.使用CELERY_ALWAYS_EAGER =真。

在您的測試文件的頂部，你可以將它設置這樣

from django.conf import settings 
settings.CELERY_ALWAYS_EAGER = True 

from django.test.client import Client 

class MyTest(unittest.TestCase): 
    def setUp(self): 
     self.client = Client() 

    def test_sample(self): 
     self.client = client.get('/')

2.芹菜2.5，我使用mock嘲笑delay方法

這裏是我的榜樣

import mock 

def mock_delay(task, *args, **kwargs): 
    task.run(args, kwargs) 

@mock.patch('celery.app.task.BaseTask.delay', mock_delay) 
class MyTest(unittest.TestCase): 
    def setUp(self): 
     self.client = Client() 

    def test_sample(self): 
     self.client = client.get('/')

該解決方案適用於celery2.5.3，但是當我升級到cel ery3.0.12，它不再工作。

看起來像芹菜2.5.3改變class BaseTask到class Task在celery.app.task模塊。

我修改我的代碼到，但無法解決它。仍然試圖找出它

來源

2012-11-10 04:05:42 kiennt

從另一個模塊檢查芹菜任務狀態

回答

相關問題