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

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

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

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(…)

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