003 File Manager
Current Path:
/usr/local/lib/python3.8/site-packages/salt/ext/tornado/test
usr
/
local
/
lib
/
python3.8
/
site-packages
/
salt
/
ext
/
tornado
/
test
/
📁
..
📄
__init__.py
(20 B)
📄
__main__.py
(430 B)
📁
__pycache__
📄
asyncio_test.py
(4.78 KB)
📄
auth_test.py
(22.19 KB)
📄
concurrent_test.py
(13.66 KB)
📄
curl_httpclient_test.py
(4.83 KB)
📄
escape_test.py
(10.9 KB)
📄
gen_test.py
(46.36 KB)
📄
http1connection_test.py
(2.02 KB)
📄
httpclient_test.py
(26.79 KB)
📄
httpserver_test.py
(41.43 KB)
📄
httputil_test.py
(17.17 KB)
📄
import_test.py
(1.73 KB)
📄
ioloop_test.py
(24.55 KB)
📄
iostream_test.py
(42.53 KB)
📄
locale_test.py
(5.96 KB)
📄
locks_test.py
(15.74 KB)
📄
log_test.py
(9.51 KB)
📄
netutil_test.py
(7.6 KB)
📄
options_test.py
(10.83 KB)
📄
process_test.py
(11.44 KB)
📄
queues_test.py
(12.79 KB)
📄
resolve_test_helper.py
(514 B)
📄
routing_test.py
(7.48 KB)
📄
runtests.py
(7.59 KB)
📄
simple_httpclient_test.py
(30.33 KB)
📄
stack_context_test.py
(10.9 KB)
📄
tcpclient_test.py
(11.31 KB)
📄
tcpserver_test.py
(2.32 KB)
📄
template_test.py
(18.63 KB)
📄
testing_test.py
(8.57 KB)
📄
twisted_test.py
(27.33 KB)
📄
util.py
(4.65 KB)
📄
util_test.py
(7.29 KB)
📄
web_test.py
(111.58 KB)
📄
websocket_test.py
(22.37 KB)
📄
windows_test.py
(877 B)
📄
wsgi_test.py
(3.71 KB)
Editing: routing_test.py
# 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. # pylint: skip-file from __future__ import absolute_import, division, print_function from salt.ext.tornado.httputil import HTTPHeaders, HTTPMessageDelegate, HTTPServerConnectionDelegate, ResponseStartLine from salt.ext.tornado.routing import HostMatches, PathMatches, ReversibleRouter, Router, Rule, RuleRouter from salt.ext.tornado.testing import AsyncHTTPTestCase from salt.ext.tornado.web import Application, HTTPError, RequestHandler from salt.ext.tornado.wsgi import WSGIContainer 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 = {} 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 = {} 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): class CustomApplication(Application): def reverse_url(self, name, *args): return router.reverse_url(name, *args) router = CustomRouter() 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.write(b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK") request.finish() app.add_handlers(".*", [ (HostMatches("www.example.com"), [ (PathMatches("/first_handler"), "tornado.test.routing_test.SecondHandler", {}, "second_handler") ]), 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("/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")
Upload File
Create Folder