test_info.py 1.1 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344
  1. from __future__ import absolute_import
  2. from celery import Celery
  3. from celery.utils.text import indent
  4. from celery.tests.utils import Case
  5. RANDTEXT = """\
  6. The quick brown
  7. fox jumps
  8. over the
  9. lazy dog\
  10. """
  11. RANDTEXT_RES = """\
  12. The quick brown
  13. fox jumps
  14. over the
  15. lazy dog\
  16. """
  17. QUEUES = {"queue1": {
  18. "exchange": "exchange1",
  19. "exchange_type": "type1",
  20. "binding_key": "bind1"},
  21. "queue2": {
  22. "exchange": "exchange2",
  23. "exchange_type": "type2",
  24. "binding_key": "bind2"}}
  25. QUEUE_FORMAT1 = """. queue1: exchange:exchange1 (type1) binding:bind1"""
  26. QUEUE_FORMAT2 = """. queue2: exchange:exchange2 (type2) binding:bind2"""
  27. class test_Info(Case):
  28. def test_textindent(self):
  29. self.assertEqual(indent(RANDTEXT, 4), RANDTEXT_RES)
  30. def test_format_queues(self):
  31. celery = Celery(set_as_current=False)
  32. celery.amqp.queues = celery.amqp.Queues(QUEUES)
  33. self.assertEqual(sorted(celery.amqp.queues.format().split("\n")),
  34. sorted([QUEUE_FORMAT1, QUEUE_FORMAT2]))