我们从Python开源项目中，提取了以下48个代码示例，用于说明如何使用tornado.web.url()。
def get_handlers(self): urls = [ url("/typecheck/(.*)", TypeCheckHandler, name='typecheck'), url("/decode_arg/(.*)", DecodeArgHandler, name='decode_arg'), url("/decode_arg_kw/(?P<arg>.*)", DecodeArgHandler), url("/linkify", LinkifyHandler), url("/uimodule_resources", UIModuleResourceHandler), url("/optional_path/(.+)?", OptionalPathHandler), url("/multi_header", MultiHeaderHandler), url("/redirect", RedirectHandler), url("/web_redirect_permanent", WebRedirectHandler, {"url": "/web_redirect_newpath"}), url("/web_redirect", WebRedirectHandler, {"url": "/web_redirect_newpath", "permanent": False}), url("//web_redirect_double_slash", WebRedirectHandler, {"url": '/web_redirect_newpath'}), url("/header_injection", HeaderInjectionHandler), url("/get_argument", GetArgumentHandler), url("/get_arguments", GetArgumentsHandler), ] return urls
def get_app(self): # callable objects to finish pending /trigger requests self.triggers = collections.deque() return Application([ url("/trigger", TriggerHandler, dict(queue=self.triggers, wake_callback=self.stop)), url("/chunk", ChunkHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/hang", HangHandler), url("/hello", HelloWorldHandler), url("/content_length", ContentLengthHandler), url("/head", HeadHandler), url("/options", OptionsHandler), url("/no_content", NoContentHandler), url("/see_other_post", SeeOtherPostHandler), url("/see_other_get", SeeOtherGetHandler), url("/host_echo", HostEchoHandler), url("/no_content_length", NoContentLengthHandler), url("/echo_post", EchoPostHandler), url("/respond_in_prepare", RespondInPrepareHandler), url("/redirect", RedirectHandler), ], gzip=True)
def test_ipv6(self): try: [sock] = bind_sockets(None, '::1', family=socket.AF_INET6) port = sock.getsockname()[1] self.http_server.add_socket(sock) except socket.gaierror as e: if e.args[0] == socket.EAI_ADDRFAMILY: # python supports ipv6, but it's not configured on the network # interface, so skip this test. return raise url = '%s://[::1]:%d/hello' % (self.get_protocol(), port) # ipv6 is currently enabled by default but can be disabled self.http_client.fetch(url, self.stop, allow_ipv6=False) response = self.wait() self.assertEqual(response.code, 599) self.http_client.fetch(url, self.stop) response = self.wait() self.assertEqual(response.body, b"Hello world!")
def test_streaming_follow_redirects(self): # When following redirects, header and streaming callbacks # should only be called for the final result. # TODO(bdarnell): this test belongs in httpclient_test instead of # simple_httpclient_test, but it fails with the version of libcurl # available on travis-ci. Move it when that has been upgraded # or we have a better framework to skip tests based on curl version. headers = [] chunks = [] self.fetch("/redirect?url=/hello", header_callback=headers.append, streaming_callback=chunks.append) chunks = list(map(to_unicode, chunks)) self.assertEqual(chunks, ['Hello world!']) # Make sure we only got one set of headers. num_start_lines = len([h for h in headers if h.startswith("HTTP/")]) self.assertEqual(num_start_lines, 1)
def test_decode_argument(self): # These urls all decode to the same thing urls = ["/decode_arg/%C3%A9?foo=%C3%A9&encoding=utf-8", "/decode_arg/%E9?foo=%E9&encoding=latin1", "/decode_arg_kw/%E9?foo=%E9&encoding=latin1", ] for url in urls: response = self.fetch(url) response.rethrow() data = json_decode(response.body) self.assertEqual(data, {u'path': [u'unicode', u'\u00e9'], u'query': [u'unicode', u'\u00e9'], }) response = self.fetch("/decode_arg/%C3%A9?foo=%C3%A9") response.rethrow() data = json_decode(response.body) self.assertEqual(data, {u'path': [u'bytes', u'c3a9'], u'query': [u'bytes', u'c3a9'], })
def test_ipv6(self): if not socket.has_ipv6: # python compiled without ipv6 support, so skip this test return try: self.http_server.listen(self.get_http_port(), address='::1') except socket.gaierror, e: if e.errno == socket.EAI_ADDRFAMILY: # python supports ipv6, but it's not configured on the network # interface, so skip this test. return raise url = self.get_url("/hello").replace("localhost", "[::1]") # ipv6 is currently disabled by default and must be explicitly requested self.http_client.fetch(url, self.stop) response = self.wait() self.assertEqual(response.code, 599) self.http_client.fetch(url, self.stop, allow_ipv6=True) response = self.wait() self.assertEqual(response.body, b("Hello world!"))
def test_decode_argument(self): # These urls all decode to the same thing urls = ["/decode_arg/%C3%A9?foo=%C3%A9&encoding=utf-8", "/decode_arg/%E9?foo=%E9&encoding=latin1", "/decode_arg_kw/%E9?foo=%E9&encoding=latin1", ] for url in urls: response = self.fetch(url) response.rethrow() data = json_decode(response.body) self.assertEqual(data, {u('path'): [u('unicode'), u('\u00e9')], u('query'): [u('unicode'), u('\u00e9')], }) response = self.fetch("/decode_arg/%C3%A9?foo=%C3%A9") response.rethrow() data = json_decode(response.body) self.assertEqual(data, {u('path'): [u('bytes'), u('c3a9')], u('query'): [u('bytes'), u('c3a9')], })
def get_app(self): # callable objects to finish pending /trigger requests self.triggers = collections.deque() return Application([ url("/trigger", TriggerHandler, dict(queue=self.triggers, wake_callback=self.stop)), url("/chunk", ChunkHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/hang", HangHandler), url("/hello", HelloWorldHandler), url("/content_length", ContentLengthHandler), url("/head", HeadHandler), url("/options", OptionsHandler), url("/no_content", NoContentHandler), url("/see_other_post", SeeOtherPostHandler), url("/see_other_get", SeeOtherGetHandler), url("/host_echo", HostEchoHandler), ], gzip=True)
def test_ipv6(self): try: self.http_server.listen(self.get_http_port(), address='::1') except socket.gaierror as e: if e.args[0] == socket.EAI_ADDRFAMILY: # python supports ipv6, but it's not configured on the network # interface, so skip this test. return raise url = self.get_url("/hello").replace("localhost", "[::1]") # ipv6 is currently disabled by default and must be explicitly requested self.http_client.fetch(url, self.stop) response = self.wait() self.assertEqual(response.code, 599) self.http_client.fetch(url, self.stop, allow_ipv6=True) response = self.wait() self.assertEqual(response.body, b"Hello world!")
def get_app(self): # callable objects to finish pending /trigger requests self.triggers = collections.deque() return Application([ url("/trigger", TriggerHandler, dict(queue=self.triggers, wake_callback=self.stop)), url("/chunk", ChunkHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/hang", HangHandler), url("/hello", HelloWorldHandler), url("/content_length", ContentLengthHandler), url("/head", HeadHandler), url("/options", OptionsHandler), url("/no_content", NoContentHandler), url("/see_other_post", SeeOtherPostHandler), url("/see_other_get", SeeOtherGetHandler), url("/host_echo", HostEchoHandler), url("/no_content_length", NoContentLengthHandler), url("/echo_post", EchoPostHandler), url("/respond_in_prepare", RespondInPrepareHandler), ], gzip=True)
def test_ipv6(self): try: self.http_server.listen(self.get_http_port(), address='::1') except socket.gaierror, e: if e.errno == socket.EAI_ADDRFAMILY: # ipv6 is not configured on this system, so skip this test return raise url = self.get_url("/hello").replace("localhost", "[::1]") # ipv6 is currently disabled by default and must be explicitly requested self.http_client.fetch(url, self.stop) response = self.wait() self.assertEqual(response.code, 599) self.http_client.fetch(url, self.stop, allow_ipv6=True) response = self.wait() self.assertEqual(response.body, b("Hello world!"))
def test_ipv6(self): try: self.http_server.listen(self.get_http_port(), address='::1') except socket.gaierror as e: if e.args[0] == socket.EAI_ADDRFAMILY: # python supports ipv6, but it's not configured on the network # interface, so skip this test. return raise url = self.get_url("/hello").replace("localhost", "[::1]") # ipv6 is currently disabled by default and must be explicitly requested with ExpectLog(gen_log, "uncaught exception"): self.http_client.fetch(url, self.stop) response = self.wait() self.assertEqual(response.code, 599) self.http_client.fetch(url, self.stop, allow_ipv6=True) response = self.wait() self.assertEqual(response.body, b"Hello world!")
def test_non_reversible(self): # URLSpecs are non-reversible if they include non-constant # regex features outside capturing groups. Currently, this is # only strictly enforced for backslash-escaped character # classes. paths = [ r'^/api/v\d+/foo/(\w+)$', ] for path in paths: # A URLSpec can still be created even if it cannot be reversed. url_spec = url(path, None) try: result = url_spec.reverse() self.fail("did not get expected exception when reversing %s. " "result: %s" % (path, result)) except ValueError: pass
def get_app(self): return Application([ url("/hello", HelloWorldHandler), url("/post", PostHandler), url("/put", PutHandler), url("/redirect", RedirectHandler), url("/chunk", ChunkHandler), url("/auth", AuthHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/echopost", EchoPostHandler), url("/user_agent", UserAgentHandler), url("/304_with_content_length", ContentLength304Handler), url("/all_methods", AllMethodsHandler), url('/patch', PatchHandler), url('/set_header', SetHeaderHandler), ], gzip=True)
def test_group_question_mark(self): # Ensure that url-encoded question marks are handled properly self.assertEqual(self.fetch_json('/group/%3F'), dict(path='/group/%3F', path_args=['?'], args={})) self.assertEqual(self.fetch_json('/group/%3F?%3F=%3F'), dict(path='/group/%3F', path_args=['?'], args={'?': ['?']}))
def get_handlers(self): class DefaultHandler(RequestHandler): def get(self): if self.get_argument("status", None): raise HTTPError(int(self.get_argument("status"))) 1 / 0 class WriteErrorHandler(RequestHandler): def get(self): if self.get_argument("status", None): self.send_error(int(self.get_argument("status"))) else: 1 / 0 def write_error(self, status_code, **kwargs): self.set_header("Content-Type", "text/plain") if "exc_info" in kwargs: self.write("Exception: %s" % kwargs["exc_info"][0].__name__) else: self.write("Status: %d" % status_code) class FailedWriteErrorHandler(RequestHandler): def get(self): 1 / 0 def write_error(self, status_code, **kwargs): raise Exception("exception in write_error") return [url("/default", DefaultHandler), url("/write_error", WriteErrorHandler), url("/failed_write_error", FailedWriteErrorHandler), ]
def get_handlers(self): # All three are equivalent. return [('/hello1', HelloHandler), ('/hello2', 'tornado.test.web_test.HelloHandler'), url('/hello3', 'tornado.test.web_test.HelloHandler'), ]
def connect(self, url, connection_close): # Use a raw connection so we can control the sending of data. s = socket.socket(socket.AF_INET, socket.SOCK_STREAM, 0) s.connect(("127.0.0.1", self.get_http_port())) stream = IOStream(s, io_loop=self.io_loop) stream.write(b"GET " + url + b" HTTP/1.1\r\n") if connection_close: stream.write(b"Connection: close\r\n") stream.write(b"Transfer-Encoding: chunked\r\n\r\n") return stream
def prepare(self): self.write('redirects can have bodies too') self.redirect(self.get_argument("url"), status=int(self.get_argument("status", "302")))
def get_app(self): return Application([ url("/hello", HelloWorldHandler), url("/post", PostHandler), url("/put", PutHandler), url("/redirect", RedirectHandler), url("/chunk", ChunkHandler), url("/auth", AuthHandler), url("/countdown/([0-9]+)", CountdownHandler, name="countdown"), url("/echopost", EchoPostHandler), url("/user_agent", UserAgentHandler), url("/304_with_content_length", ContentLength304Handler), url("/all_methods", AllMethodsHandler), url('/patch', PatchHandler), ], gzip=True)
def test_reuse_request_from_response(self): # The response.request attribute should be an HTTPRequest, not # a _RequestProxy. # This test uses self.http_client.fetch because self.fetch calls # self.get_url on the input unconditionally. url = self.get_url('/hello') response = yield self.http_client.fetch(url) self.assertEqual(response.request.url, url) self.assertTrue(isinstance(response.request, HTTPRequest)) response2 = yield self.http_client.fetch(response.request) self.assertEqual(response2.body, b'Hello world!')
def test_max_redirects(self): response = self.fetch("/countdown/5", max_redirects=3) self.assertEqual(302, response.code) # We requested 5, followed three redirects for 4, 3, 2, then the last # unfollowed redirect is to 1. self.assertTrue(response.request.url.endswith("/countdown/5")) self.assertTrue(response.effective_url.endswith("/countdown/2")) self.assertTrue(response.headers["Location"].endswith("/countdown/1"))
def test_see_other_redirect(self): for code in (302, 303): response = self.fetch("/see_other_post", method="POST", body="%d" % code) self.assertEqual(200, response.code) self.assertTrue(response.request.url.endswith("/see_other_post")) self.assertTrue(response.effective_url.endswith("/see_other_get")) # request is the original request, is a POST still self.assertEqual("POST", response.request.method)
