def test_body_encoding(self):
        unicode_body = u("\xe9")
        byte_body = binascii.a2b_hex(b"e9")

        # unicode string in body gets converted to utf8
        response = self.fetch(
            "/echopost", method="POST", body=unicode_body, headers={"Content-Type": "application/blah"}
        )
        self.assertEqual(response.headers["Content-Length"], "2")
        self.assertEqual(response.body, utf8(unicode_body))

        # byte strings pass through directly
        response = self.fetch("/echopost", method="POST", body=byte_body, headers={"Content-Type": "application/blah"})
        self.assertEqual(response.headers["Content-Length"], "1")
        self.assertEqual(response.body, byte_body)

        # Mixing unicode in headers and byte string bodies shouldn't
        # break anything
        response = self.fetch(
            "/echopost",
            method="POST",
            body=byte_body,
            headers={"Content-Type": "application/blah"},
            user_agent=u("foo"),
        )
        self.assertEqual(response.headers["Content-Length"], "1")
        self.assertEqual(response.body, byte_body)

    def setUp(self):

        self.formatter = LogFormatter(color=False)

        # Fake color support.  
# We can't guarantee anything about the $TERM
        # variable when the tests are run, so just patch in some values
        # for testing.  (testing with color off fails to expose some potential
        # encoding issues from the control characters)
        self.formatter._colors = {
            logging.ERROR: u("\u0001"),
        }

        self.formatter._normal = u("\u0002")

        # construct a Logger directly to bypass getLogger's caching
        # 定义一个logger，带上标识，标识唯一的logger对象
        #self.logger = logging.Logger('LogFormatterTest')
        self.logger = logging.Logger('test')

        self.logger.propagate = False

        self.tempdir = tempfile.mkdtemp()
        print(self.tempdir)

        # self.filename = os.path.join(self.tempdir, 'log.out')
        self.filename = './log.out'
        print(self.filename)

        self.handler = self.make_handler(self.filename)

        self.handler.setFormatter(self.formatter)

        # 添加日志处理器
        self.logger.addHandler(self.handler)

	def _on_authentication_verified(self, future, response):
		if response.error or b"is_valid:true" not in response.body:
			future.set_exception(AuthError(
				"Invalid OpenID response: %s" % (response.error or
												 response.body)))
			return

		# Make sure we got back at least an email from attribute exchange
		ax_ns = None
		for name in self.request.arguments:
			if name.startswith("openid.ns.") and \
					self.get_argument(name) == u("http://openid.net/srv/ax/1.0"):
				ax_ns = name[10:]
				break

		def get_ax_arg(uri):
			if not ax_ns:
				return u("")
			prefix = "openid." + ax_ns + ".type."
			ax_name = None
			for name in self.request.arguments.keys():
				if self.get_argument(name) == uri and name.startswith(prefix):
					part = name[len(prefix):]
					ax_name = "openid." + ax_ns + ".value." + part
					break
			if not ax_name:
				return u("")
			return self.get_argument(ax_name, u(""))

		email = get_ax_arg("http://axschema.org/contact/email")
		name = get_ax_arg("http://axschema.org/namePerson")
		first_name = get_ax_arg("http://axschema.org/namePerson/first")
		last_name = get_ax_arg("http://axschema.org/namePerson/last")
		username = get_ax_arg("http://axschema.org/namePerson/friendly")
		locale = get_ax_arg("http://axschema.org/pref/language").lower()
		user = dict()
		name_parts = []
		if first_name:
			user["first_name"] = first_name
			name_parts.append(first_name)
		if last_name:
			user["last_name"] = last_name
			name_parts.append(last_name)
		if name:
			user["name"] = name
		elif name_parts:
			user["name"] = u(" ").join(name_parts)
		elif email:
			user["name"] = email.split("@")[0]
		if email:
			user["email"] = email
		if locale:
			user["locale"] = locale
		if username:
			user["username"] = username
		claimed_id = self.get_argument("openid.claimed_id", None)
		if claimed_id:
			user["claimed_id"] = claimed_id
		future.set_result(user)

    def test_json_decode(self):
        # json_decode accepts both bytes and unicode, but strings it returns
        # are always unicode.
        self.assertEqual(json_decode(b'"foo"'), u("foo"))
        self.assertEqual(json_decode(u('"foo"')), u("foo"))

        # Non-ascii bytes are interpreted as utf8
        self.assertEqual(json_decode(utf8(u('"\u00e9"'))), u("\u00e9"))

 def test_json_encode(self):
     # json deals with strings, not bytes.  On python 2 byte strings will
     # convert automatically if they are utf8; on python 3 byte strings
     # are not allowed.
     self.assertEqual(json_decode(json_encode(u("\u00e9"))), u("\u00e9"))
     if bytes is str:
         self.assertEqual(json_decode(json_encode(utf8(u("\u00e9")))), u("\u00e9"))
         self.assertRaises(UnicodeDecodeError, json_encode, b"\xe9")

 def test_unicode_message(self):
     ws = yield websocket_connect(
         'ws://localhost:%d/echo' % self.get_http_port())
     ws.write_message(u('hello \u00e9'))
     response = yield ws.read_message()
     self.assertEqual(response, u('hello \u00e9'))
     ws.close()
     yield self.close_future

def _make_text_block(name, content, content_type=None):
    """Helper function for the builder that creates an XML text block."""
    if content_type == 'xhtml':
        return u('<%s type="xhtml"><div xmlns="%s">%s</div></%s>\n') % \
               (name, XHTML_NAMESPACE, content, name)
    if not content_type:
        return u('<%s>%s</%s>\n') % (name, escape(content), name)
    return u('<%s type="%s">%s</%s>\n') % (name, content_type,
                                         escape(content), name)

 def test_utf8_logging(self):
     self.logger.error(u("\u00e9").encode("utf8"))
     if issubclass(bytes_type, basestring_type):
         # on python 2, utf8 byte strings (and by extension ascii byte
         # strings) are passed through as-is.
         self.assertEqual(self.get_output(), utf8(u("\u00e9")))
     else:
         # on python 3, byte strings always get repr'd even if
         # they're ascii-only, so this degenerates into another
         # copy of test_bytes_logging.
         self.assertEqual(self.get_output(), utf8(repr(utf8(u("\u00e9")))))

    def test_url_escape_unicode(self):
        tests = [
            # byte strings are passed through as-is
            (u('\u00e9').encode('utf8'), '%C3%A9'),
            (u('\u00e9').encode('latin1'), '%E9'),

            # unicode strings become utf8
            (u('\u00e9'), '%C3%A9'),
        ]
        for unescaped, escaped in tests:
            self.assertEqual(url_escape(unescaped), escaped)

 def test_url_unescape_unicode(self):
     tests = [
         ('%C3%A9', u('\u00e9'), 'utf8'),
         ('%C3%A9', u('\u00c3\u00a9'), 'latin1'),
         ('%C3%A9', utf8(u('\u00e9')), None),
     ]
     for escaped, unescaped, encoding in tests:
         # input strings to url_unescape should only contain ascii
         # characters, but make sure the function accepts both byte
         # and unicode strings.
         self.assertEqual(url_unescape(to_unicode(escaped), encoding), unescaped)
         self.assertEqual(url_unescape(utf8(escaped), encoding), unescaped)

 def test_unicode_literal_expression(self):
     # Unicode literals should be usable in templates.  Note that this
     # test simulates unicode characters appearing directly in the
     # template file (with utf8 encoding), i.e. \u escapes would not
     # be used in the template file itself.
     if str is unicode_type:
         # python 3 needs a different version of this test since
         # 2to3 doesn't run on template internals
         template = Template(utf8(u('{{ "\u00e9" }}')))
     else:
         template = Template(utf8(u('{{ u"\u00e9" }}')))
     self.assertEqual(template.generate(), utf8(u("\u00e9")))

    def test_xhtml_escape(self):
        tests = [
            ("<foo>", "&lt;foo&gt;"),
            (u("<foo>"), u("&lt;foo&gt;")),
            (b"<foo>", b"&lt;foo&gt;"),

            ("<>&\"", "&lt;&gt;&amp;&quot;"),
            ("&amp;", "&amp;amp;"),
        ]
        for unescaped, escaped in tests:
            self.assertEqual(utf8(xhtml_escape(unescaped)), utf8(escaped))
            self.assertEqual(utf8(unescaped), utf8(xhtml_unescape(escaped)))

		def get_ax_arg(uri):
			if not ax_ns:
				return u("")
			prefix = "openid." + ax_ns + ".type."
			ax_name = None
			for name in self.request.arguments.keys():
				if self.get_argument(name) == uri and name.startswith(prefix):
					part = name[len(prefix):]
					ax_name = "openid." + ax_ns + ".value." + part
					break
			if not ax_name:
				return u("")
			return self.get_argument(ax_name, u(""))

    def list(self, parts):
        """返回给定列表的一个由逗号分隔的部分.

        格式是, e.g., "A, B and C", "A and B" 或者"A"当列表长度为1.
        """
        _ = self.translate
        if len(parts) == 0:
            return ""
        if len(parts) == 1:
            return parts[0]
        comma = u(' \u0648 ') if self.code.startswith("fa") else u(", ")
        return _("%(commas)s and %(last)s") % {
            "commas": comma.join(parts[:-1]),
            "last": parts[len(parts) - 1],
        }

 def test_gettext(self):
     tornado.locale.load_gettext_translations(
         os.path.join(os.path.dirname(__file__), 'gettext_translations'),
         "tornado_test")
     locale = tornado.locale.get("fr_FR")
     self.assertTrue(isinstance(locale, tornado.locale.GettextLocale))
     self.assertEqual(locale.translate("school"), u("\u00e9cole"))

    def list(self, parts):
        """Returns a comma-separated list for the given list of parts.

        The format is, e.g., "A, B and C", "A and B" or just "A" for lists
        of size 1.
        """
        _ = self.translate
        if len(parts) == 0:
            return ""
        if len(parts) == 1:
            return parts[0]
        comma = u(' \u0648 ') if self.code.startswith("fa") else u(", ")
        return _("%(commas)s and %(last)s") % {
            "commas": comma.join(parts[:-1]),
            "last": parts[len(parts) - 1],
        }

    def get_handlers(self):
        class EchoHandler(RequestHandler):
            def get(self, path):
                self.write(path)

        return [("/str/(?P<path>.*)", EchoHandler),
                (u("/unicode/(?P<path>.*)"), EchoHandler)]

 def test_multipart_form(self):
     # Encodings here are tricky:  Headers are latin1, bodies can be
     # anything (we use utf8 by default).
     response = self.raw_fetch(
         [
             b"POST /multipart HTTP/1.0",
             b"Content-Type: multipart/form-data; boundary=1234567890",
             b"X-Header-encoding-test: \xe9",
         ],
         b"\r\n".join(
             [
                 b"Content-Disposition: form-data; name=argument",
                 b"",
                 u("\u00e1").encode("utf-8"),
                 b"--1234567890",
                 u('Content-Disposition: form-data; name="files"; filename="\u00f3"').encode("utf8"),
                 b"",
                 u("\u00fa").encode("utf-8"),
                 b"--1234567890--",
                 b"",
             ]
         ),
     )
     data = json_decode(response)
     self.assertEqual(u("\u00e9"), data["header"])
     self.assertEqual(u("\u00e1"), data["argument"])
     self.assertEqual(u("\u00f3"), data["filename"])
     self.assertEqual(u("\u00fa"), data["filebody"])

    def test_chunked_request_body(self):
        # Chunked requests are not widely supported and we don't have a way
        # to generate them in AsyncHTTPClient, but HTTPServer will read them.
        self.stream.write(b"""\
POST /echo HTTP/1.1
Transfer-Encoding: chunked
Content-Type: application/x-www-form-urlencoded

4
foo=
3
bar
0

""".replace(b"\n", b"\r\n"))
        read_stream_body(self.stream, self.stop)
        headers, response = self.wait()
        self.assertEqual(json_decode(response), {u('foo'): [u('bar')]})

 def test_reverse_url(self):
     self.assertEqual(self.app.reverse_url('decode_arg', 'foo'),
                      '/decode_arg/foo')
     self.assertEqual(self.app.reverse_url('decode_arg', 42),
                      '/decode_arg/42')
     self.assertEqual(self.app.reverse_url('decode_arg', b'\xe9'),
                      '/decode_arg/%E9')
     self.assertEqual(self.app.reverse_url('decode_arg', u('\u00e9')),
                      '/decode_arg/%C3%A9')