Моделирование вызова в задаче сельдерея

У меня есть фляжное приложение, которое запускает задачу сельдерея. Я пытаюсь смоделировать один вызов 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 Flask, а не в том, который использует Celery. Имитация самой задачи не сработает, поскольку я пытаюсь проверить, как ведет себя задача, когда внешняя служба возвращает 'x' или 'y'.

Помощь будет высоко оценена.


flask unit-testing celery python-mock
person Chinmay Kanchi    schedule 26.05.2015    source источник

Ответы (2)


arrow_upward
7
arrow_downward

Хороший вариант — установить CELERY_ALWAYS_EAGER в True в вашей тестовой конфигурации. Это делает все вызовы Celery синхронными. См. документацию по этому параметру. С этой опцией любой макет, который вы настроили в своем процессе Flask, должен работать в задаче Celery.

В качестве дополнительного преимущества ваша тестовая конфигурация упрощается, так как вам не нужно иметь исполнителя Celery.

ОБНОВЛЕНИЕ: после обсуждения в комментариях кажется, что вы не хотите или не можете избавиться от воркеров Celery для своей тестовой конфигурации. В этом случае я могу предложить три решения, которые, как мне кажется, делают то, что вам нужно:

  1. Напишите пульт дистанционного управления, которая имитирует вашу задачу Celery, а затем пусть тестовый код запустит ее на всех ваших воркерах с помощью трансляция().

  2. Определите настраиваемый параметр командной строки для своего исполнителя. , скажем --test. Затем добавьте загрузочный шаг, который проверяет этот аргумент и выполняет насмешка.

  3. Создайте альтернативный модуль, чтобы передать рабочие процессы Celery в аргументе командной строки -A. Это должна быть идентичная копия вашего исходного модуля, но с добавлением насмешек. Затем запустите своих рабочих с этим альтернативным модулем для ваших тестов.

Я надеюсь, что вы найдете один из этих трех вариантов удовлетворительным!

person Miguel    schedule 26.05.2015
comment
Есть ли способ установить это только на время одного теста? У меня уже есть тесты в моем наборе, которые проверяют асинхронную функциональность, и поэтому уже есть куча рабочих celery, работающих при запуске этого теста. - person Chinmay Kanchi; 26.05.2015
comment
Если у вас уже есть воркеры Celery, то почему бы вам не смочить эту задачу в рабочем процессе? - person Miguel; 29.05.2015
comment
Ну, как мне это сделать? Я не хочу ничего добавлять в производственный код, это нужно делать из тестового кода. - person Chinmay Kanchi; 29.05.2015

arrow_upward
5
arrow_downward

Создайте настройку для тестовой функции

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
person ChillarAnand    schedule 26.05.2015