Автор оригинала: Arpit Solanki.
Недавно я столкнулся с проблемой написания модульных тестов для задач Сельдерей . Написание модульных тестов для задач celery может быть болезненным, так как они асинхронны и длительны. Я объясню, как писать модульные тесты для задач сельдерея, и попытаюсь рассмотреть различные случаи для этого. Для этого урока мы будем использовать собственную библиотеку python unit test .
Предпосылки
- Вы должны знать основы сельдерея, чтобы следовать этому учебнику.
- Если вы знаете основы написания тестов, то это потрясающе, но не обязательно для этого урока.
Настройка среды
- Этот учебник написан для python 2.x и должен работать или python 3.x также.
- У вас должна быть рабочая установка сельдерея, RabbitMQ и Redis (redis, потому что мы хотим сохранить результаты)
Вот и все давайте погрузимся в учебник
Мы создадим очень простую задачу сельдерея, которая сложит два числа
Откройте свой любимый текстовый редактор, вставьте следующий код и сохраните его как tasks.py
from celery import Celery from time import sleep
BROKER_URL = 'amqp://guest@localhost:5672//'
REDIS = 'redis://localhost/0'
app = Celery('test', broker=BROKER_URL, backend=REDIS)
@app.task(name="add") def add(x, y): return x+y
Теперь давайте напишем несколько тестов для этой задачи.
Мы должны проверить, возвращает ли наша задача правильное значение для заданных двух чисел или нет. Поэтому мы напишем один класс тестового случая, чтобы протестировать все возможные тесты для этой задачи.
Теперь создайте файл с именем tests.py в том же каталоге, что и tasks.py и вставьте следующий код.
from tasks import add, app
import unittest
class TestAddTask(unittest.TestCase):
def setUp(self):
self.task = add.apply_async(args=[3, 5])
self.results = self.task.get()
Посмотрим, что мы там натворили. Мы унаследовали unittest.TestCase class и определил setUp method. Этот метод используется для инициализации тестов и установки необходимых переменных для тестирования. Здесь мы вызвали нашу задачу add с аргументами 3, 5 и вызвали метод get для получения результата. Этот вызов метода get ожидает завершения задачи. Когда задачи завершены, мы получили результаты и инициализация теста завершена. Теперь давайте проверим результат этой задачи.
Давайте проверим состояние нашей задачи, вернула ли она УСПЕХ или НЕУДАЧУ
В вышеупомянутом классе TestAddTask мы теперь определяем метод для проверки состояния нашей задачи
def test_task_state(self): self.assertEqual(self.task.state, 'SUCCESS')
В вышеупомянутом классе TestAddTask теперь мы определяем метод для проверки наших выходных данных.
def test_addition(self): self.assertEqual(self.results, 8)
Полный класс тестового случая
from tasks import add, app
import unittest
class TestAddTask(unittest.TestCase):
def setUp(self):
self.task = add.apply_async(args=[3, 5])
self.results = self.task.get()
def test_task_state(self):
self.assertEqual(self.task.state, 'SUCCESS')
def test_addition(self):
self.assertEqual(self.results, 8)
А теперь давайте проведем тесты.
Запустите celery worker: Внутри каталога, где tasks.py выполняется следующая команда.
сельдерей -Работник задач —
Запустите тесты: Внутри каталога, где tests.py выполняется следующая команда
python -m unittest discover
Вы должны увидеть вывод примерно так, как показано ниже
➜ python -m unittest discover.. — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — — Ran 2 tests in 5.353s OK
Вуаля, мы успешно провели тесты на задачах celery.
Что делать, если мои задачи находятся на удаленной машине, и я не могу их импортировать?
Не волнуйтесь, просто замените функцию apply_async следующим оператором
self.task = app.send_task('add', args=[3, 5])
Теперь давайте возьмем пример, когда задача не возвращает никаких результатов
Давайте напишем задачу, которая загружает изображение и записывает его в файл. Обновление tasks.py со следующей задачей
import requests
@app.task(name='download-image', ignore_results=True)
def download_image(url):
r = requests.get(url)
with open('image.jpg', 'wb') as f:
f.write(r.content)
Давайте посмотрим, как мы должны проверить эту задачу. Напишите следующий код в tests.py
class TestDownloadTask(unittest.TestCase):
def setUp(self):
self.task = app.send_task('download-image',
args=['https://www.math.ust.hk/~masyleung/Teaching/CAS/MATLAB/image/images/cameraman.jpg'])
self.results = self.task.get()
def test_task_state(self):
self.assertEqual(self.task.state, 'SUCCESS')
def test_download(self):
self.assertEqual(os.path.exists('image.jpg'), True)
Здесь мы проверяем, успешно ли наша задача написала образ или нет, используя библиотеку os .
Подробнее о тестировании задач сельдерея читайте в official docs . Пожалуйста, прокомментируйте, если вы обнаружите что-то не так.