codekingpro/portable-devtools

1import asyncio
2import logging
3import os
4import signal
5import subprocess
6import sys
7import time
8import unittest
9
10from tornado.httpclient import HTTPClient, HTTPError
11from tornado.httpserver import HTTPServer
12from tornado.log import gen_log
13from tornado.process import fork_processes, task_id, Subprocess
14from tornado.simple_httpclient import SimpleAsyncHTTPClient
15from tornado.testing import bind_unused_port, ExpectLog, AsyncTestCase, gen_test
16from tornado.test.util import skipIfNonUnix
17from tornado.web import RequestHandler, Application
18
19
20# Not using AsyncHTTPTestCase because we need control over the IOLoop.
21@skipIfNonUnix
22class ProcessTest(unittest.TestCase):
23    def get_app(self):
24        class ProcessHandler(RequestHandler):
25            def get(self):
26                if self.get_argument("exit", None):
27                    # must use os._exit instead of sys.exit so unittest's
28                    # exception handler doesn't catch it
29                    os._exit(int(self.get_argument("exit")))
30                if self.get_argument("signal", None):
31                    os.kill(os.getpid(), int(self.get_argument("signal")))
32                self.write(str(os.getpid()))
33
34        return Application([("/", ProcessHandler)])
35
36    def tearDown(self):
37        if task_id() is not None:
38            # We're in a child process, and probably got to this point
39            # via an uncaught exception.  If we return now, both
40            # processes will continue with the rest of the test suite.
41            # Exit now so the parent process will restart the child
42            # (since we don't have a clean way to signal failure to
43            # the parent that won't restart)
44            logging.error("aborting child process from tearDown")
45            logging.shutdown()
46            os._exit(1)
47        # In the surviving process, clear the alarm we set earlier
48        signal.alarm(0)
49        super().tearDown()
50
51    def test_multi_process(self):
52        # This test doesn't work on twisted because we use the global
53        # reactor and don't restore it to a sane state after the fork
54        # (asyncio has the same issue, but we have a special case in
55        # place for it).
56        with ExpectLog(
57            gen_log, "(Starting .* processes|child .* exited|uncaught exception)"
58        ):
59            sock, port = bind_unused_port()
60
61            def get_url(path):
62                return "http://127.0.0.1:%d%s" % (port, path)
63
64            # ensure that none of these processes live too long
65            signal.alarm(5)  # master process
66            try:
67                id = fork_processes(3, max_restarts=3)
68                self.assertIsNotNone(id)
69                signal.alarm(5)  # child processes
70            except SystemExit as e:
71                # if we exit cleanly from fork_processes, all the child processes
72                # finished with status 0
73                self.assertEqual(e.code, 0)
74                self.assertIsNone(task_id())
75                sock.close()
76                return
77            try:
78                if id in (0, 1):
79                    self.assertEqual(id, task_id())
80
81                    async def f():
82                        server = HTTPServer(self.get_app())
83                        server.add_sockets([sock])
84                        await asyncio.Event().wait()
85
86                    asyncio.run(f())
87                elif id == 2:
88                    self.assertEqual(id, task_id())
89                    sock.close()
90                    # Always use SimpleAsyncHTTPClient here; the curl
91                    # version appears to get confused sometimes if the
92                    # connection gets closed before it's had a chance to
93                    # switch from writing mode to reading mode.
94                    client = HTTPClient(SimpleAsyncHTTPClient)
95
96                    def fetch(url, fail_ok=False):
97                        try:
98                            return client.fetch(get_url(url))
99                        except HTTPError as e:
100                            if not (fail_ok and e.code == 599):
101                                raise
102
103                    # Make two processes exit abnormally
104                    fetch("/?exit=2", fail_ok=True)
105                    fetch("/?exit=3", fail_ok=True)
106
107                    # They've been restarted, so a new fetch will work
108                    int(fetch("/").body)
109
110                    # Now the same with signals
111                    # Disabled because on the mac a process dying with a signal
112                    # can trigger an "Application exited abnormally; send error
113                    # report to Apple?" prompt.
114                    # fetch("/?signal=%d" % signal.SIGTERM, fail_ok=True)
115                    # fetch("/?signal=%d" % signal.SIGABRT, fail_ok=True)
116                    # int(fetch("/").body)
117
118                    # Now kill them normally so they won't be restarted
119                    fetch("/?exit=0", fail_ok=True)
120                    # One process left; watch it's pid change
121                    pid = int(fetch("/").body)
122                    fetch("/?exit=4", fail_ok=True)
123                    pid2 = int(fetch("/").body)
124                    self.assertNotEqual(pid, pid2)
125
126                    # Kill the last one so we shut down cleanly
127                    fetch("/?exit=0", fail_ok=True)
128
129                    os._exit(0)
130            except Exception:
131                logging.error("exception in child process %d", id, exc_info=True)
132                raise
133
134
135@skipIfNonUnix
136class SubprocessTest(AsyncTestCase):
137    def term_and_wait(self, subproc):
138        subproc.proc.terminate()
139        subproc.proc.wait()
140
141    @gen_test
142    def test_subprocess(self):
143        subproc = Subprocess(
144            [sys.executable, "-u", "-i", "-I"],
145            stdin=Subprocess.STREAM,
146            stdout=Subprocess.STREAM,
147            stderr=subprocess.STDOUT,
148        )
149        self.addCleanup(lambda: self.term_and_wait(subproc))
150        self.addCleanup(subproc.stdout.close)
151        self.addCleanup(subproc.stdin.close)
152        yield subproc.stdout.read_until(b">>> ")
153        subproc.stdin.write(b"print('hello')\n")
154        data = yield subproc.stdout.read_until(b"\n")
155        self.assertEqual(data, b"hello\n")
156
157        yield subproc.stdout.read_until(b">>> ")
158        subproc.stdin.write(b"raise SystemExit\n")
159        data = yield subproc.stdout.read_until_close()
160        self.assertEqual(data, b"")
161
162    @gen_test
163    def test_close_stdin(self):
164        # Close the parent's stdin handle and see that the child recognizes it.
165        subproc = Subprocess(
166            [sys.executable, "-u", "-i", "-I"],
167            stdin=Subprocess.STREAM,
168            stdout=Subprocess.STREAM,
169            stderr=subprocess.STDOUT,
170        )
171        self.addCleanup(lambda: self.term_and_wait(subproc))
172        yield subproc.stdout.read_until(b">>> ")
173        subproc.stdin.close()
174        data = yield subproc.stdout.read_until_close()
175        self.assertEqual(data, b"\n")
176
177    @gen_test
178    def test_stderr(self):
179        # This test is mysteriously flaky on twisted: it succeeds, but logs
180        # an error of EBADF on closing a file descriptor.
181        subproc = Subprocess(
182            [sys.executable, "-u", "-c", r"import sys; sys.stderr.write('hello\n')"],
183            stderr=Subprocess.STREAM,
184        )
185        self.addCleanup(lambda: self.term_and_wait(subproc))
186        data = yield subproc.stderr.read_until(b"\n")
187        self.assertEqual(data, b"hello\n")
188        # More mysterious EBADF: This fails if done with self.addCleanup instead of here.
189        subproc.stderr.close()
190
191    def test_sigchild(self):
192        Subprocess.initialize()
193        self.addCleanup(Subprocess.uninitialize)
194        subproc = Subprocess([sys.executable, "-c", "pass"])
195        subproc.set_exit_callback(self.stop)
196        ret = self.wait()
197        self.assertEqual(ret, 0)
198        self.assertEqual(subproc.returncode, ret)
199
200    @gen_test
201    def test_sigchild_future(self):
202        Subprocess.initialize()
203        self.addCleanup(Subprocess.uninitialize)
204        subproc = Subprocess([sys.executable, "-c", "pass"])
205        ret = yield subproc.wait_for_exit()
206        self.assertEqual(ret, 0)
207        self.assertEqual(subproc.returncode, ret)
208
209    def test_sigchild_signal(self):
210        Subprocess.initialize()
211        self.addCleanup(Subprocess.uninitialize)
212        subproc = Subprocess(
213            [sys.executable, "-c", "import time; time.sleep(30)"],
214            stdout=Subprocess.STREAM,
215        )
216        self.addCleanup(subproc.stdout.close)
217        subproc.set_exit_callback(self.stop)
218
219        # For unclear reasons, killing a process too soon after
220        # creating it can result in an exit status corresponding to
221        # SIGKILL instead of the actual signal involved. This has been
222        # observed on macOS 10.15 with Python 3.8 installed via brew,
223        # but not with the system-installed Python 3.7.
224        time.sleep(0.1)
225
226        os.kill(subproc.pid, signal.SIGTERM)
227        try:
228            ret = self.wait()
229        except AssertionError:
230            # We failed to get the termination signal. This test is
231            # occasionally flaky on pypy, so try to get a little more
232            # information: did the process close its stdout
233            # (indicating that the problem is in the parent process's
234            # signal handling) or did the child process somehow fail
235            # to terminate?
236            fut = subproc.stdout.read_until_close()
237            fut.add_done_callback(lambda f: self.stop())  # type: ignore
238            try:
239                self.wait()
240            except AssertionError:
241                raise AssertionError("subprocess failed to terminate")
242            else:
243                raise AssertionError(
244                    "subprocess closed stdout but failed to " "get termination signal"
245                )
246        self.assertEqual(subproc.returncode, ret)
247        self.assertEqual(ret, -signal.SIGTERM)
248
249    @gen_test
250    def test_wait_for_exit_raise(self):
251        Subprocess.initialize()
252        self.addCleanup(Subprocess.uninitialize)
253        subproc = Subprocess([sys.executable, "-c", "import sys; sys.exit(1)"])
254        with self.assertRaises(subprocess.CalledProcessError) as cm:
255            yield subproc.wait_for_exit()
256        self.assertEqual(cm.exception.returncode, 1)
257
258    @gen_test
259    def test_wait_for_exit_raise_disabled(self):
260        Subprocess.initialize()
261        self.addCleanup(Subprocess.uninitialize)
262        subproc = Subprocess([sys.executable, "-c", "import sys; sys.exit(1)"])
263        ret = yield subproc.wait_for_exit(raise_error=False)
264        self.assertEqual(ret, 1)
265