爲什麼Celery任務測試結果不一致？

我已經爲兩個芹菜任務編寫了兩個簡單的集成測試，但是當我運行它們時，我收到了，結果不一致。我可以運行他們一分鐘和一個或兩個將通過，然後運行他們幾秒鐘和一個或兩者都會失敗。爲什麼這些結果從一個測試運行到下一個測試運行不一致？此外，這些測試是否真的測試Celery任務是否正在發送到隊列並由工作人員執行？爲什麼Celery任務測試結果不一致？

謝謝！

下面是任務：

# apps/photos/tasks.py 
from __future__ import absolute_import 
from conf.celeryapp import app 

@app.task 
def hello(): 
    return 'Hello world!' 

@app.task 
def add(x, y): 
    return x + y

下面是測試：

# apps/photos/tests/task_tests.py 
from django.test import TestCase 
from django.test.utils import override_settings 
from apps.photos.tasks import hello, add 

class TaskTestIT(TestCase): 
    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True, 
         CELERY_ALWAYS_EAGER=True, 
         BROKER_BACKEND='memory') 

    def test_hello(self): 
     result = hello.delay() 
     self.assertTrue(result.successful()) 
     self.assertEqual(str(result.result), 'Hello world!') 

    def test_add(self): 
     result = add.delay(1, 1) 
     self.assertTrue(result.successful()) 
     self.assertEquals(result.get(), 2)

我跑我的測試使用下面的命令：

./manage.py test -s

我使用Django的鼻子我測試跑步者：

# conf/settings/base.py 
USE_DJANGO_NOSE = True 
if USE_DJANGO_NOSE: 
    INSTALLED_APPS += ('django_nose',) 
    TEST_RUNNER = 'django_nose.NoseTestSuiteRunner'

這裏是我的芹菜的應用程序和配置文件：

# conf/celeryapp.py 
from celery import Celery 

app = Celery('celeryapp') 
app.config_from_object('conf.celeryconfig') 
app.autodiscover_tasks(['apps.photos']) 

# conf/celeryconfig.py 
from kombu import Queue, Exchange 

BROKER_URL = 'amqp://' 
CELERY_RESULT_BACKEND = 'amqp://' 
CELERY_DEFAULT_QUEUE = 'default' 
CELERY_QUEUES = (
    Queue('default', Exchange('default'), routing_key='default'), 
    Queue('photos', Exchange('photos'), routing_key='photos'), 
    Queue('mail', Exchange('mail'), routing_key='mail'), 
) 
CELERY_ROUTES = (
    {'apps.photos.tasks.hello': {'queue': 'default', 'routing_key': 'default'}}, 
    {'apps.photos.tasks.add': {'queue': 'photos', 'routing_key': 'photos'}}, 
    {'apps.photos.tasks.send_email': {'queue': 'mail', 'routing_key': 'mail'}}, 
)

來源

2015-05-04 William

Task.delay()不返回任務的實際結果，則返回AsyncResult對象將包含什麼時候會被執行的任務的結果。不同的結果是由於這樣的事實，即有時任務的執行速度比測試開始檢查結果的速度快，有時需要比此更長的時間。這取決於你的系統的負荷等

你應該做的是調用result.get()其中第一個等待任務執行完畢，只有經過你應該檢查它是否是.successful()等

例如，以下應產生一致的結果：

def test_hello(self): 
     result = hello.delay() 
     result.get() 
     self.assertTrue(result.successful()) 
     self.assertEqual(str(result.result), 'Hello world!')

來源

2015-05-04 16:50:34

謝謝瓦西里。你的解釋「非常清晰」，現在我的測試正在產生一致的結果。 – William

爲什麼Celery任務測試結果不一致？

回答

相關問題