проверить статус задачи сельдерея с другого модуля

Question

Я пытаюсь проверить, была ли запущена задача сельдерея после запроса на мой сервер 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? 

Как я могу проверить, работает ли или не задача в моих модульных тестов? Очевидно, у меня нет прямого доступа к идентификатору задачи, иначе что-то like this would work.

В более общем плане, как вы можете выявлять задачи сельдерея, которые запускаются из всех функций, классов или модулей?

Благодарим за помощь.

Answer 1

Для этого (при модульном тестировании) вы можете использовать Mock. Исправьте атрибут delay вашей задачи, а затем проверьте атрибутмакета. В вашем примере, попробовать что-то вроде:

# 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, чтобы утверждать, что он вызывается с правильными аргументами)

Если это разблокировка сделки - вы хотите утверждать в одном тестовом примере, что вызов вызывается , а также воздействие/побочных эффектов задачи - вы можете поэкспериментировать с атрибутом mockwraps, которые могли бы работать:

# 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) 

Но другой способ, вероятно, лучше, так как он лучше изолирует, что испытывается.

Answer 2

Сельдерей не допускает такого взаимодействия, но вы можете использовать кеш или даже сеанс, чтобы контролировать его состояние. Конечно, вам нужно будет передать некоторый идентификатор извне, чтобы получить это значение дальше.

Но если вам это нужно только для тестов, вы можете использовать настройку CELERY_ALWAYS_EAGER, которая сообщает Celery выполнять все задачи как простые процедуры, даже если не требуется запуск celeryd.

Answer 3

я нашел их два способа, которые могут сделать запуск задачи синхронно в моем тесте

1. Используйте CELERY_ALWAYS_EAGER = True.

В верхней части тестового файла, вы можете установить его как этот

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('/') 

Это решение хорошо работает с сельдереем2.5.3, но когда я обновляюсь до cel ery3.0.12, он больше не работает.

Кажется, сельдерей2.5.3 изменение class BaseTask до class Task в celery.app.task модуль.

Я изменяю свой код до @mock.patch('celery.app.task.Task.delay', mock_delay), но не могу его решить. Попытайтесь выяснить это