Я пытаюсь использовать библиотеку python mock для исправления задачи Celery, которая запускается, когда модель сохраняется в моем приложении django, чтобы убедиться, что она правильно называется.
В принципе задача задается внутри myapp.tasks и импортируется в верхней части моего model.py файла следующим образом:
from .tasks import mytask
... и затем пробегает save() внутри модели, используя mytask.delay(foo, bar). Пока все хорошо - отлично работает, когда я на самом деле запускаю Celerydd и т.д.
Я хочу построить unit test, который издевается над задачей, просто чтобы проверить, что она вызвана с правильными аргументами, и на самом деле не пытается запустить задачу Celery когда-либо.
Итак, в тестовом файле у меня есть что-то вроде этого внутри стандартной TestCase:
from mock import patch # at the top of the file
# ...then later
def test_celery_task(self):
with patch('myapp.models.mytask.delay') as mock_task:
# ...create an instance of the model and save it etc
self.assertTrue(mock_task.called)
... но он никогда не вызывается/всегда false. Я попробовал различные инкарнации (вместо исправления myapp.models.mytask и проверки, если вместо этого был вызван mock_task.delay. Я собрал из mock docs, что путь импорта имеет решающее значение, а googling говорит мне, что это должен быть путь, как он отображается внутри тестируемого модуля (который будет myapp.models.mytask.delay, а не myapp.tasks.mytask.delay, если я его правильно пойму).
Где я здесь ошибаюсь? Есть ли определенные трудности в исправлении задач Сельдерея? Могу ли я запланировать celery.task (который используется как декоратор для mytask) вместо?
Проблема, с которой вы сталкиваетесь, не связана с тем, что это задача сельдерея. Вы просто ошибаетесь.;)
В частности, вам нужно выяснить, какой вид или другой файл импортирует "mytask" и исправляет его там, поэтому соответствующая строка будет выглядеть так:
with patch('myapp.myview.mytask.delay') as mock_task:
Здесь есть еще несколько привкус:
http://www.voidspace.org.uk/python/mock/patch.html#where-to-patch
Декоратор @task заменяет функцию объектом Task (см. документация). Если вы издеваетесь над самой задачей, вы замените (несколько волшебный) объект Task на MagicMock, и он не будет планировать задачу вообще. Вместо этого издевайтесь над методом Task object run(), например:
@override_settings(CELERY_ALWAYS_EAGER=True)
@patch('monitor.tasks.monitor_user.run')
def test_monitor_all(self, monitor_user):
"""
Test monitor.all task
"""
user = ApiUserFactory()
tasks.monitor_all.delay()
monitor_user.assert_called_once_with(user.key)