Издевательство над вызовом в задаче сельдерея

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

Помощь была бы очень признательна.
2 3
unit-testing flask celery python-mock

2 ответа:

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

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

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

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

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

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

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

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

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
5