FM / test

/opt/imh-python/lib/python3.9/site-packages/tornado/test/

/opt/imh-python/lib/python3.9/site-packages/tornado/test

Name	Size	Mode	Actions
csv_translations/	-	0755	rm
gettext_translations/	-	0755	rm
static/	-	0755	rm
templates/	-	0755	rm
__pycache__/	-	0755	rm
asyncio_test.py	7329	0644	edit dl rm
auth_test.py	24026	0644	edit dl rm
autoreload_test.py	4075	0644	edit dl rm
concurrent_test.py	6174	0644	edit dl rm
curl_httpclient_test.py	4477	0644	edit dl rm
escape_test.py	12694	0644	edit dl rm
gen_test.py	34721	0644	edit dl rm
http1connection_test.py	1972	0644	edit dl rm
httpclient_test.py	32067	0644	edit dl rm
httpserver_test.py	46332	0644	edit dl rm
httputil_test.py	19600	0644	edit dl rm
import_test.py	2046	0644	edit dl rm
ioloop_test.py	26060	0644	edit dl rm
iostream_test.py	46178	0644	edit dl rm
locale_test.py	5907	0644	edit dl rm
locks_test.py	17564	0644	edit dl rm
log_test.py	9778	0644	edit dl rm
netutil_test.py	8047	0644	edit dl rm
options_test.cfg	76	0644	edit dl rm
options_test.py	12174	0644	edit dl rm
options_test_types.cfg	277	0644	edit dl rm
options_test_types_str.cfg	158	0644	edit dl rm
process_test.py	11392	0644	edit dl rm
queues_test.py	14188	0644	edit dl rm
resolve_test_helper.py	421	0644	edit dl rm
routing_test.py	9121	0644	edit dl rm
runtests.py	8440	0644	edit dl rm
simple_httpclient_test.py	31821	0644	edit dl rm
static_foo.txt	97	0644	edit dl rm
tcpclient_test.py	17115	0644	edit dl rm
tcpserver_test.py	6674	0644	edit dl rm
template_test.py	19204	0644	edit dl rm
test.crt	1244	0644	edit dl rm
test.key	1732	0644	edit dl rm
testing_test.py	10985	0644	edit dl rm
twisted_test.py	8116	0644	edit dl rm
util.py	3768	0644	edit dl rm
util_test.py	10127	0644	edit dl rm
websocket_test.py	28331	0644	edit dl rm
web_test.py	118423	0644	edit dl rm
windows_test.py	698	0644	edit dl rm
wsgi_test.py	677	0644	edit dl rm
__init__.py	398	0644	edit dl rm
__main__.py	347	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/tornado/test/process_test.py (11392B)

import asyncio
import logging
import os
import signal
import subprocess
import sys
import unittest

from tornado.httpclient import HTTPClient, HTTPError
from tornado.httpserver import HTTPServer
from tornado.ioloop import IOLoop
from tornado.log import gen_log
from tornado.process import fork_processes, task_id, Subprocess
from tornado.simple_httpclient import SimpleAsyncHTTPClient
from tornado.testing import bind_unused_port, ExpectLog, AsyncTestCase, gen_test
from tornado.test.util import skipIfNonUnix
from tornado.web import RequestHandler, Application

# Not using AsyncHTTPTestCase because we need control over the IOLoop.
@skipIfNonUnix
class ProcessTest(unittest.TestCase):
    def get_app(self):
        class ProcessHandler(RequestHandler):
            def get(self):
                if self.get_argument("exit", None):
                    # must use os._exit instead of sys.exit so unittest's
                    # exception handler doesn't catch it
                    os._exit(int(self.get_argument("exit")))
                if self.get_argument("signal", None):
                    os.kill(os.getpid(), int(self.get_argument("signal")))
                self.write(str(os.getpid()))

return Application([("/", ProcessHandler)])

def tearDown(self):
        if task_id() is not None:
            # We're in a child process, and probably got to this point
            # via an uncaught exception.  If we return now, both
            # processes will continue with the rest of the test suite.
            # Exit now so the parent process will restart the child
            # (since we don't have a clean way to signal failure to
            # the parent that won't restart)
            logging.error("aborting child process from tearDown")
            logging.shutdown()
            os._exit(1)
        # In the surviving process, clear the alarm we set earlier
        signal.alarm(0)
        super(ProcessTest, self).tearDown()

def test_multi_process(self):
        # This test doesn't work on twisted because we use the global
        # reactor and don't restore it to a sane state after the fork
        # (asyncio has the same issue, but we have a special case in
        # place for it).
        with ExpectLog(
            gen_log, "(Starting .* processes|child .* exited|uncaught exception)"
        ):
            sock, port = bind_unused_port()

def get_url(path):
                return "http://127.0.0.1:%d%s" % (port, path)

# ensure that none of these processes live too long
            signal.alarm(5)  # master process
            try:
                id = fork_processes(3, max_restarts=3)
                self.assertTrue(id is not None)
                signal.alarm(5)  # child processes
            except SystemExit as e:
                # if we exit cleanly from fork_processes, all the child processes
                # finished with status 0
                self.assertEqual(e.code, 0)
                self.assertTrue(task_id() is None)
                sock.close()
                return
            try:
                if asyncio is not None:
                    # Reset the global asyncio event loop, which was put into
                    # a broken state by the fork.
                    asyncio.set_event_loop(asyncio.new_event_loop())
                if id in (0, 1):
                    self.assertEqual(id, task_id())
                    server = HTTPServer(self.get_app())
                    server.add_sockets([sock])
                    IOLoop.current().start()
                elif id == 2:
                    self.assertEqual(id, task_id())
                    sock.close()
                    # Always use SimpleAsyncHTTPClient here; the curl
                    # version appears to get confused sometimes if the
                    # connection gets closed before it's had a chance to
                    # switch from writing mode to reading mode.
                    client = HTTPClient(SimpleAsyncHTTPClient)

def fetch(url, fail_ok=False):
                        try:
                            return client.fetch(get_url(url))
                        except HTTPError as e:
                            if not (fail_ok and e.code == 599):
                                raise

# Make two processes exit abnormally
                    fetch("/?exit=2", fail_ok=True)
                    fetch("/?exit=3", fail_ok=True)

# They've been restarted, so a new fetch will work
                    int(fetch("/").body)

# Now the same with signals
                    # Disabled because on the mac a process dying with a signal
                    # can trigger an "Application exited abnormally; send error
                    # report to Apple?" prompt.
                    # fetch("/?signal=%d" % signal.SIGTERM, fail_ok=True)
                    # fetch("/?signal=%d" % signal.SIGABRT, fail_ok=True)
                    # int(fetch("/").body)

# Now kill them normally so they won't be restarted
                    fetch("/?exit=0", fail_ok=True)
                    # One process left; watch it's pid change
                    pid = int(fetch("/").body)
                    fetch("/?exit=4", fail_ok=True)
                    pid2 = int(fetch("/").body)
                    self.assertNotEqual(pid, pid2)

# Kill the last one so we shut down cleanly
                    fetch("/?exit=0", fail_ok=True)

os._exit(0)
            except Exception:
                logging.error("exception in child process %d", id, exc_info=True)
                raise

@skipIfNonUnix
class SubprocessTest(AsyncTestCase):
    def term_and_wait(self, subproc):
        subproc.proc.terminate()
        subproc.proc.wait()

@gen_test
    def test_subprocess(self):
        if IOLoop.configured_class().__name__.endswith("LayeredTwistedIOLoop"):
            # This test fails non-deterministically with LayeredTwistedIOLoop.
            # (the read_until('\n') returns '\n' instead of 'hello\n')
            # This probably indicates a problem with either TornadoReactor
            # or TwistedIOLoop, but I haven't been able to track it down
            # and for now this is just causing spurious travis-ci failures.
            raise unittest.SkipTest(
                "Subprocess tests not compatible with " "LayeredTwistedIOLoop"
            )
        subproc = Subprocess(
            [sys.executable, "-u", "-i"],
            stdin=Subprocess.STREAM,
            stdout=Subprocess.STREAM,
            stderr=subprocess.STDOUT,
        )
        self.addCleanup(lambda: self.term_and_wait(subproc))
        self.addCleanup(subproc.stdout.close)
        self.addCleanup(subproc.stdin.close)
        yield subproc.stdout.read_until(b">>> ")
        subproc.stdin.write(b"print('hello')\n")
        data = yield subproc.stdout.read_until(b"\n")
        self.assertEqual(data, b"hello\n")

yield subproc.stdout.read_until(b">>> ")
        subproc.stdin.write(b"raise SystemExit\n")
        data = yield subproc.stdout.read_until_close()
        self.assertEqual(data, b"")

@gen_test
    def test_close_stdin(self):
        # Close the parent's stdin handle and see that the child recognizes it.
        subproc = Subprocess(
            [sys.executable, "-u", "-i"],
            stdin=Subprocess.STREAM,
            stdout=Subprocess.STREAM,
            stderr=subprocess.STDOUT,
        )
        self.addCleanup(lambda: self.term_and_wait(subproc))
        yield subproc.stdout.read_until(b">>> ")
        subproc.stdin.close()
        data = yield subproc.stdout.read_until_close()
        self.assertEqual(data, b"\n")

@gen_test
    def test_stderr(self):
        # This test is mysteriously flaky on twisted: it succeeds, but logs
        # an error of EBADF on closing a file descriptor.
        subproc = Subprocess(
            [sys.executable, "-u", "-c", r"import sys; sys.stderr.write('hello\n')"],
            stderr=Subprocess.STREAM,
        )
        self.addCleanup(lambda: self.term_and_wait(subproc))
        data = yield subproc.stderr.read_until(b"\n")
        self.assertEqual(data, b"hello\n")
        # More mysterious EBADF: This fails if done with self.addCleanup instead of here.
        subproc.stderr.close()

def test_sigchild(self):
        Subprocess.initialize()
        self.addCleanup(Subprocess.uninitialize)
        subproc = Subprocess([sys.executable, "-c", "pass"])
        subproc.set_exit_callback(self.stop)
        ret = self.wait()
        self.assertEqual(ret, 0)
        self.assertEqual(subproc.returncode, ret)

@gen_test
    def test_sigchild_future(self):
        Subprocess.initialize()
        self.addCleanup(Subprocess.uninitialize)
        subproc = Subprocess([sys.executable, "-c", "pass"])
        ret = yield subproc.wait_for_exit()
        self.assertEqual(ret, 0)
        self.assertEqual(subproc.returncode, ret)

def test_sigchild_signal(self):
        Subprocess.initialize()
        self.addCleanup(Subprocess.uninitialize)
        subproc = Subprocess(
            [sys.executable, "-c", "import time; time.sleep(30)"],
            stdout=Subprocess.STREAM,
        )
        self.addCleanup(subproc.stdout.close)
        subproc.set_exit_callback(self.stop)
        os.kill(subproc.pid, signal.SIGTERM)
        try:
            ret = self.wait(timeout=1.0)
        except AssertionError:
            # We failed to get the termination signal. This test is
            # occasionally flaky on pypy, so try to get a little more
            # information: did the process close its stdout
            # (indicating that the problem is in the parent process's
            # signal handling) or did the child process somehow fail
            # to terminate?
            fut = subproc.stdout.read_until_close()
            fut.add_done_callback(lambda f: self.stop())  # type: ignore
            try:
                self.wait(timeout=1.0)
            except AssertionError:
                raise AssertionError("subprocess failed to terminate")
            else:
                raise AssertionError(
                    "subprocess closed stdout but failed to " "get termination signal"
                )
        self.assertEqual(subproc.returncode, ret)
        self.assertEqual(ret, -signal.SIGTERM)

@gen_test
    def test_wait_for_exit_raise(self):
        Subprocess.initialize()
        self.addCleanup(Subprocess.uninitialize)
        subproc = Subprocess([sys.executable, "-c", "import sys; sys.exit(1)"])
        with self.assertRaises(subprocess.CalledProcessError) as cm:
            yield subproc.wait_for_exit()
        self.assertEqual(cm.exception.returncode, 1)

@gen_test
    def test_wait_for_exit_raise_disabled(self):
        Subprocess.initialize()
        self.addCleanup(Subprocess.uninitialize)
        subproc = Subprocess([sys.executable, "-c", "import sys; sys.exit(1)"])
        ret = yield subproc.wait_for_exit(raise_error=False)
        self.assertEqual(ret, 1)

cmd: