Тесты Django зависают при выполнении асинхронных задач Celery

Я пытаюсь создать модульные тесты в Django, которые проверяют задачи Celery и то, создает ли задача объекты из данных CSV, которые я ей передаю.

По какой-то странной причине мои тесты зависают при вызове .get() на AsyncResult -objects.

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

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

Вот соответствующий код:

stations/test.py

class StationsTests(TestCase):
    
    @classmethod
    def setUpTestData(cls):
        super(StationsTests, cls).setUpTestData()
        dirname = os.path.abspath(os.path.join(os.path.dirname( __file__ ), '..', 'csvimport'))

        cls.csv_data_type_station = 'station'
        cls.upload_type = 'safe_create'
        cls.file_station = os.path.join(dirname, 'CSVFiles/Station_test_csv.csv')


    @override_settings(CELERY_EAGER_PROPAGATES_EXCEPTIONS=True,
                    CELERY_ALWAYS_EAGER=True)
    def test_station_geoJSON_data(self):
        self.result_upload_csv = upload_csv.delay(
            self.file_station, self.csv_data_type_station, self.upload_type).get(timeout=10)

        self.assertEqual(self.result_upload_csv, '10 stations uploaded successfully')
        self.assertEqual(len(Station.objects.all()), 10)

settings.py:

DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql_psycopg2',
        'NAME': 'db_hcbapp',
        'USER': 'postgres',
        'PASSWORD': 'postgres',
        'HOST': 'localhost',
        'PORT': '5432'
    },
}
.....

CELERY_BROKER_URL = 'redis://127.0.0.1:6379'
CELERY_IMPORTS = ('csvimport.tasks')
CELERY_ACCEPT_CONTENT = ['application/json']
CELERY_RESULT_SERIALIZER = 'json'
CELERY_TASK_SERIALIZER = 'json'
CELERY_RESULT_BACKEND = 'django-db'
CELERY_RESULT_EXTENDED = True

celery.py:

from __future__ import absolute_import, unicode_literals
import os
from celery import Celery
from django.conf import settings

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myapp.settings')
app = Celery('myapp')
app.conf.enable_utc = False
app.config_from_object(settings, namespace='CELERY')

app.autodiscover_tasks()

@app.task(bind=True)
def debug_task(self):
    print('Request: {0!r}'.format(self.request))

tasks.py:

Как я уже сказал, вне тестирования задача и база данных работают отлично. Redis и Celery работают, и результаты задачи добавляются в базу данных PostgreSQL.

Есть идеи, что может быть причиной зависания?

• Я пробовал подражать тестовому методу, используя @patch('myapp.tasks.upload_csv.delay'), но это, похоже, не создает никаких объектов в моей реальной базе данных. Следовательно, с помощью self.assertEqual(len(Station.objects.all()), 10) я хочу проверить, что задача создает объекты в моей базе данных.

• Я пробовал изменить мою базу данных обратно на SQLite, но она все равно зависает.

• Я пробовал создавать другие простые временные задачи в tasks.py. Но даже они замирают.

Есть идеи?

Вернуться на верх