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