Как обычно, у меня не так много времени, поэтому я дам код и подробно остановлюсь на нем. Однако есть кое-что, о чем стоит поговорить заранее.

Предисловие

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

Поскольку вы здесь, я предполагаю, что вы знаете, что такое Celery, используя Django, но не знаете, как тестировать свои задачи.

Пока я искал ресурсы по тестированию задач Celery, я обнаружил много устаревших ресурсов. Итак, в этой статье я использую Django 2.1 и Celery 4.2.1. Я полагаю, что использовать Django 2 и Celery 4 безопасно. В качестве брокера сообщений я использую Redis, но я не думаю, что эта зависимость может вызвать проблемы, если ее не использовать, поэтому вы можете использовать все, что захотите.

Я также использую pytest вместо классической встроенной библиотеки unittest вместе с pytest-django. Я думаю, что это не вызовет проблем с использованием встроенного модульного тестирования и, конечно же, собственного тестового модуля Django, но в любом случае, если у вас есть проблемы с тестированием, попробуйте pytest-django.

Что касается базы данных, я почти всегда стараюсь избегать использования реальной базы данных в своей среде разработки, под этим я имею в виду MySQL (и его вилку, MariaDB) и PostgreSQL. SQLite может хранить свои данные в памяти во время тестирования, что приводит к молниеносной миграции и тестированию. Вот почему я обычно пишу свои модели с учетом классического реляционного подхода, а не с использованием нового материала PostgreSQL. Итак, если у вас есть конкретное поле PostgreSQL в ваших моделях, вы можете либо перейти на классический подход, либо просто покинуть эту статью. Серьезно, миграции во время тестирования в среде разработки, даже если есть возможность оставить тестовую базу данных как есть, чертовски болезненны. С другой стороны, SQLite использует оперативную память и намного быстрее.

И, наконец, тема не сельдерея. Сельдерейный бит - другой зверь. Надеюсь, вы найдете ответ где-нибудь еще.

Код и проработка

Без лишних слов, вот как выглядит мой тестовый пример:

from django.test import TransactionTestCase
from anyapp import tasks
from celery.contrib.testing.worker import start_worker
class FooTaskTestCase(TransactionTestCase):
    @classmethod
    def setUpClass(cls):
        super().setUpClass()
        cls.celery_worker = start_worker(app)
        cls.celery_worker.__enter__()
    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()
        cls.celery_worker.__exit__(None, None, None)
    def setUp(self):
        super().setUp()
        self.task = tasks.foo.delay("bar") # whatever your method and args are
        self.results = self.task.get()
    def test_success(self):
        assert self.task.state == "SUCCESS"

Итак, что здесь происходит?

Как видите, мы запускаем рабочий процесс Celery внутри setUpClass и заканчиваем его в tearDownClass. Есть причина, по которой мы это делаем.

Запуск Celery с терминала и выполнение тестов вместе с не будет указывать на одну и ту же базу данных. Ваши тесты будут использовать тестовую базу данных, в то время как Celery знает только вашу настоящую базу данных. Это будет работать и потерпит неудачу, вероятно, скажет: «Объект не существует». или что-то еще. Чтобы тестирование и Celery указывали на одну и ту же базу данных, нам нужно создать их в одном процессе. Сначала мы создаем экземпляр рабочего Celery в setUpClass:

cls.celery_worker = start_worker(app)

Мы начинаем его с __enter__, который снова находится в setUpClass, и убиваем его с __exit__, который должен быть в tearDownClass (вы видите None, None и None?). Поскольку setUpClass и tearDownClass являются методами класса, они будут вызываться только один раз за тест, когда тест начинается и заканчивается.

В setUp мы можем вызывать наши задачи. Его не должно быть там, вы также можете поместить его в setUpClass, но это было то, что мне было нужно в моем случае. Также обратите внимание на строку self.task.get() выше, это заблокирует поток до тех пор, пока он не получит результат от Celery, что очень важно.

Вы видели, что наш тестовый пример не является обычным TestCase. На самом деле это TransactionTestCase. Я поместил его туда, чтобы он не столкнулся с блокировкой или атомарным блоком SQLite. Если вы не собираетесь менять модель, вы можете изменить ее на все, что захотите.

использованная литература

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

Удивительная статья Соланки гораздо удобнее для начинающих, и вы, возможно, захотите ее прочитать. Однако он настолько примитивен и использует чистый Python. Среда Django может сильно отличаться от этого.

Этот, опять же, использует чистый Python, а тема - HTTP.