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/routing_test.py (9121B)

# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

from tornado.httputil import (
    HTTPHeaders,
    HTTPMessageDelegate,
    HTTPServerConnectionDelegate,
    ResponseStartLine,
)
from tornado.routing import (
    HostMatches,
    PathMatches,
    ReversibleRouter,
    Router,
    Rule,
    RuleRouter,
)
from tornado.testing import AsyncHTTPTestCase
from tornado.web import Application, HTTPError, RequestHandler
from tornado.wsgi import WSGIContainer

import typing  # noqa: F401

class BasicRouter(Router):
    def find_handler(self, request, **kwargs):
        class MessageDelegate(HTTPMessageDelegate):
            def __init__(self, connection):
                self.connection = connection

def finish(self):
                self.connection.write_headers(
                    ResponseStartLine("HTTP/1.1", 200, "OK"),
                    HTTPHeaders({"Content-Length": "2"}),
                    b"OK",
                )
                self.connection.finish()

return MessageDelegate(request.connection)

class BasicRouterTestCase(AsyncHTTPTestCase):
    def get_app(self):
        return BasicRouter()

def test_basic_router(self):
        response = self.fetch("/any_request")
        self.assertEqual(response.body, b"OK")

resources = {}  # type: typing.Dict[str, bytes]

class GetResource(RequestHandler):
    def get(self, path):
        if path not in resources:
            raise HTTPError(404)

self.finish(resources[path])

class PostResource(RequestHandler):
    def post(self, path):
        resources[path] = self.request.body

class HTTPMethodRouter(Router):
    def __init__(self, app):
        self.app = app

def find_handler(self, request, **kwargs):
        handler = GetResource if request.method == "GET" else PostResource
        return self.app.get_handler_delegate(request, handler, path_args=[request.path])

class HTTPMethodRouterTestCase(AsyncHTTPTestCase):
    def get_app(self):
        return HTTPMethodRouter(Application())

def test_http_method_router(self):
        response = self.fetch("/post_resource", method="POST", body="data")
        self.assertEqual(response.code, 200)

response = self.fetch("/get_resource")
        self.assertEqual(response.code, 404)

response = self.fetch("/post_resource")
        self.assertEqual(response.code, 200)
        self.assertEqual(response.body, b"data")

def _get_named_handler(handler_name):
    class Handler(RequestHandler):
        def get(self, *args, **kwargs):
            if self.application.settings.get("app_name") is not None:
                self.write(self.application.settings["app_name"] + ": ")

self.finish(handler_name + ": " + self.reverse_url(handler_name))

return Handler

FirstHandler = _get_named_handler("first_handler")
SecondHandler = _get_named_handler("second_handler")

class CustomRouter(ReversibleRouter):
    def __init__(self):
        super(CustomRouter, self).__init__()
        self.routes = {}  # type: typing.Dict[str, typing.Any]

def add_routes(self, routes):
        self.routes.update(routes)

def find_handler(self, request, **kwargs):
        if request.path in self.routes:
            app, handler = self.routes[request.path]
            return app.get_handler_delegate(request, handler)

def reverse_url(self, name, *args):
        handler_path = "/" + name
        return handler_path if handler_path in self.routes else None

class CustomRouterTestCase(AsyncHTTPTestCase):
    def get_app(self):
        router = CustomRouter()

class CustomApplication(Application):
            def reverse_url(self, name, *args):
                return router.reverse_url(name, *args)

app1 = CustomApplication(app_name="app1")
        app2 = CustomApplication(app_name="app2")

router.add_routes(
            {
                "/first_handler": (app1, FirstHandler),
                "/second_handler": (app2, SecondHandler),
                "/first_handler_second_app": (app2, FirstHandler),
            }
        )

return router

def test_custom_router(self):
        response = self.fetch("/first_handler")
        self.assertEqual(response.body, b"app1: first_handler: /first_handler")
        response = self.fetch("/second_handler")
        self.assertEqual(response.body, b"app2: second_handler: /second_handler")
        response = self.fetch("/first_handler_second_app")
        self.assertEqual(response.body, b"app2: first_handler: /first_handler")

class ConnectionDelegate(HTTPServerConnectionDelegate):
    def start_request(self, server_conn, request_conn):
        class MessageDelegate(HTTPMessageDelegate):
            def __init__(self, connection):
                self.connection = connection

def finish(self):
                response_body = b"OK"
                self.connection.write_headers(
                    ResponseStartLine("HTTP/1.1", 200, "OK"),
                    HTTPHeaders({"Content-Length": str(len(response_body))}),
                )
                self.connection.write(response_body)
                self.connection.finish()

return MessageDelegate(request_conn)

class RuleRouterTest(AsyncHTTPTestCase):
    def get_app(self):
        app = Application()

def request_callable(request):
            request.connection.write_headers(
                ResponseStartLine("HTTP/1.1", 200, "OK"),
                HTTPHeaders({"Content-Length": "2"}),
            )
            request.connection.write(b"OK")
            request.connection.finish()

router = CustomRouter()
        router.add_routes(
            {"/nested_handler": (app, _get_named_handler("nested_handler"))}
        )

app.add_handlers(
            ".*",
            [
                (
                    HostMatches("www.example.com"),
                    [
                        (
                            PathMatches("/first_handler"),
                            "tornado.test.routing_test.SecondHandler",
                            {},
                            "second_handler",
                        )
                    ],
                ),
                Rule(PathMatches("/.*handler"), router),
                Rule(PathMatches("/first_handler"), FirstHandler, name="first_handler"),
                Rule(PathMatches("/request_callable"), request_callable),
                ("/connection_delegate", ConnectionDelegate()),
            ],
        )

return app

def test_rule_based_router(self):
        response = self.fetch("/first_handler")
        self.assertEqual(response.body, b"first_handler: /first_handler")

response = self.fetch("/first_handler", headers={"Host": "www.example.com"})
        self.assertEqual(response.body, b"second_handler: /first_handler")

response = self.fetch("/nested_handler")
        self.assertEqual(response.body, b"nested_handler: /nested_handler")

response = self.fetch("/nested_not_found_handler")
        self.assertEqual(response.code, 404)

response = self.fetch("/connection_delegate")
        self.assertEqual(response.body, b"OK")

response = self.fetch("/request_callable")
        self.assertEqual(response.body, b"OK")

response = self.fetch("/404")
        self.assertEqual(response.code, 404)

class WSGIContainerTestCase(AsyncHTTPTestCase):
    def get_app(self):
        wsgi_app = WSGIContainer(self.wsgi_app)

class Handler(RequestHandler):
            def get(self, *args, **kwargs):
                self.finish(self.reverse_url("tornado"))

return RuleRouter(
            [
                (
                    PathMatches("/tornado.*"),
                    Application([(r"/tornado/test", Handler, {}, "tornado")]),
                ),
                (PathMatches("/wsgi"), wsgi_app),
            ]
        )

def wsgi_app(self, environ, start_response):
        start_response("200 OK", [])
        return [b"WSGI"]

def test_wsgi_container(self):
        response = self.fetch("/tornado/test")
        self.assertEqual(response.body, b"/tornado/test")

response = self.fetch("/wsgi")
        self.assertEqual(response.body, b"WSGI")

def test_delegate_not_found(self):
        response = self.fetch("/404")
        self.assertEqual(response.code, 404)

cmd: