Как фиксировать задачи Сельдерея во время модульного тестирования?

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

Например, я хотел бы написать тест, который выглядит примерно так:

def test_add_user_avatar(): add_user_avatar(…) tasks = get_deferred_tasks(…) assert_equal(tasks[0], ResizeImageTask(…)) 

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

One Solution collect form web for “Как фиксировать задачи Сельдерея во время модульного тестирования?”

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

 from … import ResizeImageTask class NonQueuedTestCase(…): def setUp(self): """ Patch out ResizeImageTask's delay method """ super(NonQueuedTestCase, self).setUp() self.patcher = patch.object(ResizeImageTask, 'delay', autospec=True) self.m_delay = self.patcher.start() def tearDown(self): self.patcher.stop() super(NonQueuedTestCase, self).tearDown() def test_add_user_avatar(self): # Make call to front-end code add_user_avatar(…) # Check delay call was made self.m_delay.assert_called_once_with(…) 

Вы можете запускать эти тесты без поддержки (в памяти или в противном случае), поддерживать чистый разрыв между интерфейсным кодом и кодом задачи и тестировать несколько путей кода, которые обычно ставят очередь на длительную задачу без ее запуска.

  • Сельдерей прекращает выполнение цепочки
  • Сельдерей Торнадо не может использовать gen.Task или CallBack
  • Соединение AMQP сброшено одноранговым узлом, но сельдерей подключен
  • Состояние задачи сельдерея зависит от CELERY_TASK_RESULT_EXPIRES
  • Параллельно распределенная задача сельдерея с многопроцессорной обработкой
  • Как контролировать состояние очереди в сельдерее
  • Получение всех идентификаторов задач из вложенных цепочек и аккордов
  • Сельдерей. Может ли сообщение в RabbitMQ потреблять два или более рабочих одновременно?
  • Как перезапустить (отменять) отмененную задачу с помощью Celery?
  • Сельдерей и сигналы
  • Каков наилучший вариант для очереди задач (Python 3) в Windows теперь, когда Celery 4 отказался от поддержки Windows?
  • Python - лучший язык программирования в мире.