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

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

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

import json
from rest_framework.test import APITestCase
from myapp.models import MyModel
from myapp.util import get_result_from_response

class MyTestCase(APITestCase):
    @classmethod
    def setUpTestData(cls):
        # This object is not visible to Celery
        MyModel(id='test_object').save()

    def test_celery_integration(self):
        # This view spawns a Celery task
        # Task should see MyModel.objects.get(id='test_object'), but can't
        http_response = self.client.post('/', 'test_data', format='json')

        result = get_result_from_response(http_response)
        result.get()  # Wait for task to finish before ending test case
        # Objects saved by Celery task should be deleted, but persist

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

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

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

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

Ответы на вопрос(2)

Решение Вопроса

Фон

База данных Django в памяти - sqlite3. Как говорится настраница описания для баз данных Sqlite в памяти, «Все подключения к базе данных, совместно использующие базу данных в памяти, должны быть в одном процессе». Это означает, что, пока Django использует тестовую базу данных в памяти, а Celery запускается в отдельном процессе, принципиально невозможно, чтобы Celery и Django совместно использовали тестовую базу данных.

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

Это предполагает, что Celery уже настроен вобычный способ с проектом 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 файл:

from celery.contrib.testing.worker import start_worker
from django.test import SimpleTestCase

from mysite.celery import app

class BatchSimulationTestCase(SimpleTestCase):
    allow_database_queries = True

    @classmethod
    def setUpClass(cls):
        super().setUpClass()

        # Start up celery worker
        cls.celery_worker = start_worker(app)
        cls.celery_worker.__enter__()

    @classmethod
    def tearDownClass(cls):
        super().tearDownClass()

        # Close worker
        cls.celery_worker.__exit__(None, None, None)

    def test_my_function(self):
        # my_task.delay() or something

По какой-то причине тестирующий пытается использовать задачу под названием'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:

@app.task(name='celery.ping')
def ping():
    # type: () -> str
    """Simple task that just returns 'pong'."""
    return 'pong'

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

 Tirzono25 авг. 2018 г., 22:13
Обходной путь для решения проблемы ping - добавлениеapp.loader.import_module('celery.contrib.testing.tasks') доcls.celery_worker = start_worker(app) или добавив'celery.contrib.testing.tasks' вINSTALLED_APPS (Обратите внимание.tasks в конце).
 Marviel07 нояб. 2017 г., 13:53
Привет, этот подход определенно кажется интересным, но кажется, что я не могу общаться с работником сельдерея, когда я использую этот метод. Я поищу это, если у меня будет свободное время.

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

http://docs.celeryproject.org/projects/django-celery/en/2.4/cookbook/unit-testing.htmlhttp://docs.celeryproject.org/en/latest/userguide/testing.html

Если вы действительно хотите протестировать вызовы функций сельдерея, включая очередь, я бы, вероятно, настроил dockercompose с комбинацией сервера, рабочего, очереди и расширил пользовательский CeleryTestRunner из документации django-celery. Но я не вижу в этом никакой пользы, потому что тестовая система может быть далека от производства, чтобы быть репрезентативной.

 Johannes Reichard02 окт. 2017 г., 21:43
Я только что понял, что мы используем слегка устаревшую версию django-celery, какую версию вы используете?
 Johannes Reichard02 окт. 2017 г., 21:27
Мы используем Testrunner, наследующий отCeleryTestSuiteRunner, Но единственное, что мы делаем, - это загрузка некоторых пользовательских приборов.
 Johannes Reichard02 окт. 2017 г., 21:13
Они фактически говорят вам создать новый Testrunner и запустить ваши тесты с этим testrunner, в результате чего тесты пропускают асинхронную часть сельдерея. При этом вы сможете обрабатывать свои тесты так, как если бы они были синхронными и приводили к тому, что обе части (прежних) задач сельдерея выполняли одну и ту же тестовую базу данных.
 Johannes Reichard02 окт. 2017 г., 21:21
хм, очень странно Вы запускали свои тесты с новым тестраннером, я забыл об этом в начале, и мне потребовалось некоторое время, чтобы распознать мою ошибку ^^
 Johannes Reichard02 окт. 2017 г., 21:38

Ваш ответ на вопрос