main_test.py 3.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. from __future__ import absolute_import
  2. from __future__ import unicode_literals
  3. import logging
  4. from compose import container
  5. from compose.cli.errors import UserError
  6. from compose.cli.formatter import ConsoleWarningFormatter
  7. from compose.cli.main import build_log_printer
  8. from compose.cli.main import convergence_strategy_from_opts
  9. from compose.cli.main import setup_console_handler
  10. from compose.service import ConvergenceStrategy
  11. from tests import mock
  12. from tests import unittest
  13. def mock_container(service, number):
  14. return mock.create_autospec(
  15. container.Container,
  16. service=service,
  17. number=number,
  18. name_without_project='{0}_{1}'.format(service, number))
  19. class CLIMainTestCase(unittest.TestCase):
  20. def test_build_log_printer(self):
  21. containers = [
  22. mock_container('web', 1),
  23. mock_container('web', 2),
  24. mock_container('db', 1),
  25. mock_container('other', 1),
  26. mock_container('another', 1),
  27. ]
  28. service_names = ['web', 'db']
  29. log_printer = build_log_printer(containers, service_names, True, False, {'follow': True})
  30. self.assertEqual(log_printer.containers, containers[:3])
  31. def test_build_log_printer_all_services(self):
  32. containers = [
  33. mock_container('web', 1),
  34. mock_container('db', 1),
  35. mock_container('other', 1),
  36. ]
  37. service_names = []
  38. log_printer = build_log_printer(containers, service_names, True, False, {'follow': True})
  39. self.assertEqual(log_printer.containers, containers)
  40. class SetupConsoleHandlerTestCase(unittest.TestCase):
  41. def setUp(self):
  42. self.stream = mock.Mock()
  43. self.stream.isatty.return_value = True
  44. self.handler = logging.StreamHandler(stream=self.stream)
  45. def test_with_tty_verbose(self):
  46. setup_console_handler(self.handler, True)
  47. assert type(self.handler.formatter) == ConsoleWarningFormatter
  48. assert '%(name)s' in self.handler.formatter._fmt
  49. assert '%(funcName)s' in self.handler.formatter._fmt
  50. def test_with_tty_not_verbose(self):
  51. setup_console_handler(self.handler, False)
  52. assert type(self.handler.formatter) == ConsoleWarningFormatter
  53. assert '%(name)s' not in self.handler.formatter._fmt
  54. assert '%(funcName)s' not in self.handler.formatter._fmt
  55. def test_with_not_a_tty(self):
  56. self.stream.isatty.return_value = False
  57. setup_console_handler(self.handler, False)
  58. assert type(self.handler.formatter) == logging.Formatter
  59. class ConvergeStrategyFromOptsTestCase(unittest.TestCase):
  60. def test_invalid_opts(self):
  61. options = {'--force-recreate': True, '--no-recreate': True}
  62. with self.assertRaises(UserError):
  63. convergence_strategy_from_opts(options)
  64. def test_always(self):
  65. options = {'--force-recreate': True, '--no-recreate': False}
  66. self.assertEqual(
  67. convergence_strategy_from_opts(options),
  68. ConvergenceStrategy.always
  69. )
  70. def test_never(self):
  71. options = {'--force-recreate': False, '--no-recreate': True}
  72. self.assertEqual(
  73. convergence_strategy_from_opts(options),
  74. ConvergenceStrategy.never
  75. )
  76. def test_changed(self):
  77. options = {'--force-recreate': False, '--no-recreate': False}
  78. self.assertEqual(
  79. convergence_strategy_from_opts(options),
  80. ConvergenceStrategy.changed
  81. )