test_text.py 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384
  1. from __future__ import absolute_import, unicode_literals
  2. import pytest
  3. from celery.utils.text import (abbr, abbrtask, ensure_newlines, indent,
  4. pretty, truncate)
  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 = {
  18. 'queue1': {
  19. 'exchange': 'exchange1',
  20. 'exchange_type': 'type1',
  21. 'routing_key': 'bind1',
  22. },
  23. 'queue2': {
  24. 'exchange': 'exchange2',
  25. 'exchange_type': 'type2',
  26. 'routing_key': 'bind2',
  27. },
  28. }
  29. QUEUE_FORMAT1 = '.> queue1 exchange=exchange1(type1) key=bind1'
  30. QUEUE_FORMAT2 = '.> queue2 exchange=exchange2(type2) key=bind2'
  31. class test_Info:
  32. def test_textindent(self):
  33. assert indent(RANDTEXT, 4) == RANDTEXT_RES
  34. def test_format_queues(self, app):
  35. app.amqp.queues = app.amqp.Queues(QUEUES)
  36. assert (sorted(app.amqp.queues.format().split('\n')) ==
  37. sorted([QUEUE_FORMAT1, QUEUE_FORMAT2]))
  38. def test_ensure_newlines(self):
  39. assert len(ensure_newlines('foo\nbar\nbaz\n').splitlines()) == 3
  40. assert len(ensure_newlines('foo\nbar').splitlines()) == 2
  41. @pytest.mark.parametrize('s,maxsize,expected', [
  42. ('ABCDEFGHI', 3, 'ABC...'),
  43. ('ABCDEFGHI', 10, 'ABCDEFGHI'),
  44. ])
  45. def test_truncate_text(s, maxsize, expected):
  46. assert truncate(s, maxsize) == expected
  47. @pytest.mark.parametrize('args,expected', [
  48. ((None, 3), '???'),
  49. (('ABCDEFGHI', 6), 'ABC...'),
  50. (('ABCDEFGHI', 20), 'ABCDEFGHI'),
  51. (('ABCDEFGHI', 6, None), 'ABCDEF'),
  52. ])
  53. def test_abbr(args, expected):
  54. assert abbr(*args) == expected
  55. @pytest.mark.parametrize('s,maxsize,expected', [
  56. (None, 3, '???'),
  57. ('feeds.tasks.refresh', 10, '[.]refresh'),
  58. ('feeds.tasks.refresh', 30, 'feeds.tasks.refresh'),
  59. ])
  60. def test_abbrtask(s, maxsize, expected):
  61. assert abbrtask(s, maxsize) == expected
  62. def test_pretty():
  63. assert pretty(('a', 'b', 'c'))