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/tcpserver_test.py (6674B)

import socket
import subprocess
import sys
import textwrap
import unittest

from tornado.escape import utf8, to_unicode
from tornado import gen
from tornado.iostream import IOStream
from tornado.log import app_log
from tornado.tcpserver import TCPServer
from tornado.test.util import skipIfNonUnix
from tornado.testing import AsyncTestCase, ExpectLog, bind_unused_port, gen_test

class TCPServerTest(AsyncTestCase):
    @gen_test
    def test_handle_stream_coroutine_logging(self):
        # handle_stream may be a coroutine and any exception in its
        # Future will be logged.
        class TestServer(TCPServer):
            @gen.coroutine
            def handle_stream(self, stream, address):
                yield stream.read_bytes(len(b"hello"))
                stream.close()
                1 / 0

server = client = None
        try:
            sock, port = bind_unused_port()
            server = TestServer()
            server.add_socket(sock)
            client = IOStream(socket.socket())
            with ExpectLog(app_log, "Exception in callback"):
                yield client.connect(("localhost", port))
                yield client.write(b"hello")
                yield client.read_until_close()
                yield gen.moment
        finally:
            if server is not None:
                server.stop()
            if client is not None:
                client.close()

@gen_test
    def test_handle_stream_native_coroutine(self):
        # handle_stream may be a native coroutine.

class TestServer(TCPServer):
            async def handle_stream(self, stream, address):
                stream.write(b"data")
                stream.close()

sock, port = bind_unused_port()
        server = TestServer()
        server.add_socket(sock)
        client = IOStream(socket.socket())
        yield client.connect(("localhost", port))
        result = yield client.read_until_close()
        self.assertEqual(result, b"data")
        server.stop()
        client.close()

def test_stop_twice(self):
        sock, port = bind_unused_port()
        server = TCPServer()
        server.add_socket(sock)
        server.stop()
        server.stop()

@gen_test
    def test_stop_in_callback(self):
        # Issue #2069: calling server.stop() in a loop callback should not
        # raise EBADF when the loop handles other server connection
        # requests in the same loop iteration

class TestServer(TCPServer):
            @gen.coroutine
            def handle_stream(self, stream, address):
                server.stop()  # type: ignore
                yield stream.read_until_close()

sock, port = bind_unused_port()
        server = TestServer()
        server.add_socket(sock)
        server_addr = ("localhost", port)
        N = 40
        clients = [IOStream(socket.socket()) for i in range(N)]
        connected_clients = []

@gen.coroutine
        def connect(c):
            try:
                yield c.connect(server_addr)
            except EnvironmentError:
                pass
            else:
                connected_clients.append(c)

yield [connect(c) for c in clients]

self.assertGreater(len(connected_clients), 0, "all clients failed connecting")
        try:
            if len(connected_clients) == N:
                # Ideally we'd make the test deterministic, but we're testing
                # for a race condition in combination with the system's TCP stack...
                self.skipTest(
                    "at least one client should fail connecting "
                    "for the test to be meaningful"
                )
        finally:
            for c in connected_clients:
                c.close()

# Here tearDown() would re-raise the EBADF encountered in the IO loop

@skipIfNonUnix
class TestMultiprocess(unittest.TestCase):
    # These tests verify that the two multiprocess examples from the
    # TCPServer docs work. Both tests start a server with three worker
    # processes, each of which prints its task id to stdout (a single
    # byte, so we don't have to worry about atomicity of the shared
    # stdout stream) and then exits.
    def run_subproc(self, code):
        proc = subprocess.Popen(
            sys.executable, stdin=subprocess.PIPE, stdout=subprocess.PIPE
        )
        proc.stdin.write(utf8(code))
        proc.stdin.close()
        proc.wait()
        stdout = proc.stdout.read()
        proc.stdout.close()
        if proc.returncode != 0:
            raise RuntimeError(
                "Process returned %d. stdout=%r" % (proc.returncode, stdout)
            )
        return to_unicode(stdout)

def test_single(self):
        # As a sanity check, run the single-process version through this test
        # harness too.
        code = textwrap.dedent(
            """
            from tornado.ioloop import IOLoop
            from tornado.tcpserver import TCPServer

server = TCPServer()
            server.listen(0, address='127.0.0.1')
            IOLoop.current().run_sync(lambda: None)
            print('012', end='')
        """
        )
        out = self.run_subproc(code)
        self.assertEqual("".join(sorted(out)), "012")

def test_simple(self):
        code = textwrap.dedent(
            """
            from tornado.ioloop import IOLoop
            from tornado.process import task_id
            from tornado.tcpserver import TCPServer

server = TCPServer()
            server.bind(0, address='127.0.0.1')
            server.start(3)
            IOLoop.current().run_sync(lambda: None)
            print(task_id(), end='')
        """
        )
        out = self.run_subproc(code)
        self.assertEqual("".join(sorted(out)), "012")

def test_advanced(self):
        code = textwrap.dedent(
            """
            from tornado.ioloop import IOLoop
            from tornado.netutil import bind_sockets
            from tornado.process import fork_processes, task_id
            from tornado.ioloop import IOLoop
            from tornado.tcpserver import TCPServer

sockets = bind_sockets(0, address='127.0.0.1')
            fork_processes(3)
            server = TCPServer()
            server.add_sockets(sockets)
            IOLoop.current().run_sync(lambda: None)
            print(task_id(), end='')
        """
        )
        out = self.run_subproc(code)
        self.assertEqual("".join(sorted(out)), "012")

cmd: