test_celeryd_detach.py 4.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129
  1. from __future__ import absolute_import, unicode_literals
  2. from celery.platforms import IS_WINDOWS
  3. from celery.bin.celeryd_detach import (
  4. detach,
  5. detached_celeryd,
  6. main,
  7. )
  8. from celery.tests.case import AppCase, Mock, override_stdouts, patch
  9. if not IS_WINDOWS:
  10. class test_detached(AppCase):
  11. @patch('celery.bin.celeryd_detach.detached')
  12. @patch('os.execv')
  13. @patch('celery.bin.celeryd_detach.logger')
  14. @patch('celery.app.log.Logging.setup_logging_subsystem')
  15. def test_execs(self, setup_logs, logger, execv, detached):
  16. context = detached.return_value = Mock()
  17. context.__enter__ = Mock()
  18. context.__exit__ = Mock()
  19. detach('/bin/boo', ['a', 'b', 'c'], logfile='/var/log',
  20. pidfile='/var/pid', hostname='foo@example.com')
  21. detached.assert_called_with(
  22. '/var/log', '/var/pid', None, None, None, None, False,
  23. after_forkers=False,
  24. )
  25. execv.assert_called_with('/bin/boo', ['/bin/boo', 'a', 'b', 'c'])
  26. r = detach('/bin/boo', ['a', 'b', 'c'],
  27. logfile='/var/log', pidfile='/var/pid',
  28. executable='/bin/foo', app=self.app)
  29. execv.assert_called_with('/bin/foo', ['/bin/foo', 'a', 'b', 'c'])
  30. execv.side_effect = Exception('foo')
  31. r = detach(
  32. '/bin/boo', ['a', 'b', 'c'],
  33. logfile='/var/log', pidfile='/var/pid',
  34. hostname='foo@example.com', app=self.app)
  35. context.__enter__.assert_called_with()
  36. self.assertTrue(logger.critical.called)
  37. setup_logs.assert_called_with(
  38. 'ERROR', '/var/log', hostname='foo@example.com')
  39. self.assertEqual(r, 1)
  40. self.patch('celery.current_app')
  41. from celery import current_app
  42. r = detach(
  43. '/bin/boo', ['a', 'b', 'c'],
  44. logfile='/var/log', pidfile='/var/pid',
  45. hostname='foo@example.com', app=None)
  46. current_app.log.setup_logging_subsystem.assert_called_with(
  47. 'ERROR', '/var/log', hostname='foo@example.com',
  48. )
  49. class test_PartialOptionParser(AppCase):
  50. def test_parser(self):
  51. x = detached_celeryd(self.app)
  52. p = x.create_parser('celeryd_detach')
  53. options, values = p.parse_args([
  54. '--logfile=foo', '--fake', '--enable',
  55. 'a', 'b', '-c1', '-d', '2',
  56. ])
  57. self.assertEqual(options.logfile, 'foo')
  58. self.assertEqual(values, ['a', 'b'])
  59. self.assertEqual(p.leftovers, ['--enable', '-c1', '-d', '2'])
  60. options, values = p.parse_args([
  61. '--fake', '--enable',
  62. '--pidfile=/var/pid/foo.pid',
  63. 'a', 'b', '-c1', '-d', '2',
  64. ])
  65. self.assertEqual(options.pidfile, '/var/pid/foo.pid')
  66. with override_stdouts():
  67. with self.assertRaises(SystemExit):
  68. p.parse_args(['--logfile'])
  69. p.get_option('--logfile').nargs = 2
  70. with self.assertRaises(SystemExit):
  71. p.parse_args(['--logfile=a'])
  72. with self.assertRaises(SystemExit):
  73. p.parse_args(['--fake=abc'])
  74. assert p.get_option('--logfile').nargs == 2
  75. p.parse_args(['--logfile=a', 'b'])
  76. p.get_option('--logfile').nargs = 1
  77. class test_Command(AppCase):
  78. argv = ['--autoscale=10,2', '-c', '1',
  79. '--logfile=/var/log', '-lDEBUG',
  80. '--', '.disable_rate_limits=1']
  81. def test_parse_options(self):
  82. x = detached_celeryd(app=self.app)
  83. o, v, l = x.parse_options('cd', self.argv)
  84. self.assertEqual(o.logfile, '/var/log')
  85. self.assertEqual(l, ['--autoscale=10,2', '-c', '1',
  86. '-lDEBUG', '--logfile=/var/log',
  87. '--pidfile=celeryd.pid'])
  88. x.parse_options('cd', []) # no args
  89. @patch('sys.exit')
  90. @patch('celery.bin.celeryd_detach.detach')
  91. def test_execute_from_commandline(self, detach, exit):
  92. x = detached_celeryd(app=self.app)
  93. x.execute_from_commandline(self.argv)
  94. self.assertTrue(exit.called)
  95. detach.assert_called_with(
  96. path=x.execv_path, uid=None, gid=None,
  97. umask=None, fake=False, logfile='/var/log', pidfile='celeryd.pid',
  98. working_directory=None, executable=None, hostname=None,
  99. argv=x.execv_argv + [
  100. '-c', '1', '-lDEBUG',
  101. '--logfile=/var/log', '--pidfile=celeryd.pid',
  102. '--', '.disable_rate_limits=1'
  103. ],
  104. app=self.app,
  105. )
  106. @patch('celery.bin.celeryd_detach.detached_celeryd')
  107. def test_main(self, command):
  108. c = command.return_value = Mock()
  109. main(self.app)
  110. c.execute_from_commandline.assert_called_with()