python类url()的实例源码

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 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_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 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 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 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_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 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 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)