Я только что понял, что мы используем слегка устаревшую версию django-celery, какую версию вы используете?
==========================================================================================================

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

Вот пример теста:

У меня есть два вопроса:

Как сделать так, чтобы Celery мог видеть объекты, которые тестирует Django?

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

Я готов вручную очистить объекты, если это невозможно сделать автоматически, но удаление объектов в`tearDown` даже в`APISimpleTestCase` похоже откатился назад.

ФонБаза данных Django в памяти - sqlite3. Как говорится на[страница описания для баз данных Sqlite в памяти](https://sqlite.org/inmemorydb.html), «Все подключения к базе данных, совместно использующие базу данных в памяти, должны быть в одном процессе». Это означает, что, пока Django использует тестовую базу данных в памяти, а Celery запускается в отдельном процессе, принципиально невозможно, чтобы Celery и Django совместно использовали тестовую базу данных.

Однако с`celery.contrib.testing.worker.start_worker`, можно запустить рабочий Celery в отдельном потоке в рамках одного и того же процесса. Этот работник может получить доступ к базе данных в памяти.

Это предполагает, что Celery уже настроен в[обычный способ](http://docs.celeryproject.org/en/latest/django/first-steps-with-django.html) с проектом Django.

РешениеТак как Django-Celery использует межпотоковое взаимодействие, будут работать только те тестовые случаи, которые не выполняются в изолированных транзакциях. Тестовый пример должен наследоваться напрямую от`SimpleTestCase` или его Отдых эквивалент`APISimpleTestCase` и установите атрибут класса`allow_database_queries` в`True`.

Ключ заключается в том, чтобы запустить работника сельдерея в`setUpClass` метод`TestCase` и закройте его в`tearDownClass` метод. Ключевая функция`celery.contrib.testing.worker.start_worker(app)`, для которого требуется экземпляр текущего приложения Celery, предположительно полученный из`mysite.celery.app` и возвращает питона`ContextManager`, у которого есть`__enter__` а также`__exit__` методы, которые должны быть вызваны в`setUpClass` а также`tearDownClass`соответственно. Вероятно, есть способ избежать ручного ввода и существования`ContextManager` с декоратором или что-то, но я не мог понять это. Вот пример`tests.py` файл:

По какой-то причине тестирующий пытается использовать задачу под названием`'celery.ping'`, вероятно, чтобы обеспечить лучшие сообщения об ошибках в случае сбоя работника. Даже настройка`perform_ping_check` в`False` в качестве ключевого аргумента от`start_worker` до сих пор проверяет его существование. Задача, которую он ищет,`celery.contrib.testing.tasks.ping`, Однако эта задача не установлена по умолчанию. Это можно сделать, добавив`celery.contrib.testing` в`INSTALLED_APPS` в`settings.py`, Однако это только делает его видимым для работника; не код, который генерирует рабочий. Код, который генерирует работник делает`assert 'celery.ping' in app.tasks`, который терпит неудачу. Комментирование этого заставляет все работать, но изменение установленной библиотеки не является хорошим решением. Я, вероятно, делаю что-то не так, но я решил обойти эту проблему, скопировав простую функцию, где ее можно найти`app.autodiscover_tasks()`, такие как`celery.py`:

Теперь, когда тесты запущены, нет необходимости запускать отдельный процесс Celery. Рабочий Celery будет запущен в процессе тестирования Django как отдельный поток. Этот работник может видеть любые базы данных в памяти, включая тестовую базу данных в памяти по умолчанию. Чтобы контролировать количество работников, есть варианты, доступные в`start_worker`, но, по-видимому, по умолчанию используется один рабочий.



льдерея, две следующие ссылки предоставят вам необходимую информацию для начала ваших юнит-тестов:

<http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.html><http://docs.celeryproject.org/en/latest/userguide/testing.html>Если вы действительно хотите протестировать вызовы функций сельдерея, включая очередь, я бы, вероятно, настроил dockercompose с комбинацией сервера, рабочего, очереди и расширил пользовательский CeleryTestRunner из документации django-celery. Но я не вижу в этом никакой пользы, потому что тестовая система может быть далека от производства, чтобы быть репрезентативной.