Как протестировать приложение Django + Celery с несколькими очередями - PullRequest
Новая
       122

Как протестировать приложение Django + Celery с несколькими очередями

0 голосов
/ 13 апреля 2020

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

Для одной очереди я использовал решение, объясненное здесь , и это работало отлично. В основном я хотел бы сделать следующее: 

from celery.contrib.testing.worker import start_worker
from django.contrib.staticfiles.testing import StaticLiveServerTestCase

from mysite.celery import app

class MyTestCase(StaticLiveServerTestCase):

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

        cls.celery_worker1 = start_worker(app, perform_ping_check=False, queue="myqueue1")
        cls.celery_worker1.__enter__()

        cls.celery_worker2 = start_worker(app, perform_ping_check=False, queue="myqueue2")
        cls.celery_worker2.__enter__()

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

        cls.celery_worker1.__exit__(None, None, None)
        cls.celery_worker2.__exit__(None, None, None)

У этого кода есть две проблемы:

  1. Я получаю сообщение

    / home / user / .local / lib / python3 .7 / site-packages / kombu / pidbox.py: 74: Предупреждение пользователя: Узел с именем gen18391@Machine уже использует этот почтовый ящик процесса! что заставляет меня думать, что несколько рабочих не обрабатываются должным образом.

  2. Аргумент "queue" отсутствует в документации и, вероятно, вообще не анализируется. Следовательно, задача, отправленная во вторую очередь, не выполняется.

...