Method test_detect_quorum_queues_negative

t/unit/worker/test_consumer.py:890–896

(self)

        assert name == "celery"

    def test_detect_quorum_queues_negative(self):
        c = self.c
        self.c.connection.transport.driver_type = 'amqp'
        c.app.amqp.queues = {"celery": Mock(queue_arguments=None)}
        result, name = detect_quorum_queues(c.app, c.connection.transport.driver_type)
        assert not result
        assert name == ""

    def test_detect_quorum_queues_not_rabbitmq(self):
        c = self.c