Тестирование задачи celery с помощью django

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

databases = '__all__'

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

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

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

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

Но я получил psycopg2.InterfaceError: connection already closed when run cls.celery_worker.enter().

Кто-нибудь может мне помочь?

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