проверить состояние задачи сельдерея из другого модуля - PullRequest
Новая
       14

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

0 голосов
/ 01 марта 2012

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

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

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

Спасибо за вашу помощь.

Ответы [ 3 ]

2 голосов
/ 04 марта 2012

Для этого можно использовать Макет (при модульном тестировании). Извлеките атрибут 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)

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

1 голос
/ 10 ноября 2012

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

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

Это решение хорошо работает с celery2.5.3, но когда я обновляюсь до celery3.0.12, оно больше не работает.

Похоже, что celery2.5.3 меняет class BaseTask на class Task в celery.app.task модуле.

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

0 голосов
/ 01 марта 2012

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

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

...