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