Go to the documentation of this file.00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016 from __future__ import absolute_import, division, print_function, with_statement
00017
00018 import contextlib
00019 import glob
00020 import logging
00021 import os
00022 import re
00023 import subprocess
00024 import sys
00025 import tempfile
00026 import warnings
00027
00028 from tornado.escape import utf8
00029 from tornado.log import LogFormatter, define_logging_options, enable_pretty_logging
00030 from tornado.options import OptionParser
00031 from tornado.test.util import unittest
00032 from tornado.util import u, bytes_type, basestring_type
00033
00034
00035 @contextlib.contextmanager
00036 def ignore_bytes_warning():
00037 with warnings.catch_warnings():
00038 warnings.simplefilter('ignore', category=BytesWarning)
00039 yield
00040
00041
00042 class LogFormatterTest(unittest.TestCase):
00043
00044
00045 LINE_RE = re.compile(b"(?s)\x01\\[E [0-9]{6} [0-9]{2}:[0-9]{2}:[0-9]{2} log_test:[0-9]+\\]\x02 (.*)")
00046
00047 def setUp(self):
00048 self.formatter = LogFormatter(color=False)
00049
00050
00051
00052
00053 self.formatter._colors = {
00054 logging.ERROR: u("\u0001"),
00055 }
00056 self.formatter._normal = u("\u0002")
00057
00058 self.logger = logging.Logger('LogFormatterTest')
00059 self.logger.propagate = False
00060 self.tempdir = tempfile.mkdtemp()
00061 self.filename = os.path.join(self.tempdir, 'log.out')
00062 self.handler = self.make_handler(self.filename)
00063 self.handler.setFormatter(self.formatter)
00064 self.logger.addHandler(self.handler)
00065
00066 def tearDown(self):
00067 self.handler.close()
00068 os.unlink(self.filename)
00069 os.rmdir(self.tempdir)
00070
00071 def make_handler(self, filename):
00072
00073
00074
00075
00076 return logging.FileHandler(filename)
00077
00078 def get_output(self):
00079 with open(self.filename, "rb") as f:
00080 line = f.read().strip()
00081 m = LogFormatterTest.LINE_RE.match(line)
00082 if m:
00083 return m.group(1)
00084 else:
00085 raise Exception("output didn't match regex: %r" % line)
00086
00087 def test_basic_logging(self):
00088 self.logger.error("foo")
00089 self.assertEqual(self.get_output(), b"foo")
00090
00091 def test_bytes_logging(self):
00092 with ignore_bytes_warning():
00093
00094 self.logger.error(b"\xe9")
00095 self.assertEqual(self.get_output(), utf8(repr(b"\xe9")))
00096
00097 def test_utf8_logging(self):
00098 self.logger.error(u("\u00e9").encode("utf8"))
00099 if issubclass(bytes_type, basestring_type):
00100
00101
00102 self.assertEqual(self.get_output(), utf8(u("\u00e9")))
00103 else:
00104
00105
00106
00107 self.assertEqual(self.get_output(), utf8(repr(utf8(u("\u00e9")))))
00108
00109 def test_bytes_exception_logging(self):
00110 try:
00111 raise Exception(b'\xe9')
00112 except Exception:
00113 self.logger.exception('caught exception')
00114
00115
00116 output = self.get_output()
00117 self.assertRegexpMatches(output, br'Exception.*\\xe9')
00118
00119 self.assertNotIn(br'\n', output)
00120
00121
00122 class UnicodeLogFormatterTest(LogFormatterTest):
00123 def make_handler(self, filename):
00124
00125
00126
00127 return logging.FileHandler(filename, encoding="utf8")
00128
00129 def test_unicode_logging(self):
00130 self.logger.error(u("\u00e9"))
00131 self.assertEqual(self.get_output(), utf8(u("\u00e9")))
00132
00133
00134 class EnablePrettyLoggingTest(unittest.TestCase):
00135 def setUp(self):
00136 super(EnablePrettyLoggingTest, self).setUp()
00137 self.options = OptionParser()
00138 define_logging_options(self.options)
00139 self.logger = logging.Logger('tornado.test.log_test.EnablePrettyLoggingTest')
00140 self.logger.propagate = False
00141
00142 def test_log_file(self):
00143 tmpdir = tempfile.mkdtemp()
00144 try:
00145 self.options.log_file_prefix = tmpdir + '/test_log'
00146 enable_pretty_logging(options=self.options, logger=self.logger)
00147 self.assertEqual(1, len(self.logger.handlers))
00148 self.logger.error('hello')
00149 self.logger.handlers[0].flush()
00150 filenames = glob.glob(tmpdir + '/test_log*')
00151 self.assertEqual(1, len(filenames))
00152 with open(filenames[0]) as f:
00153 self.assertRegexpMatches(f.read(), r'^\[E [^]]*\] hello$')
00154 finally:
00155 for handler in self.logger.handlers:
00156 handler.flush()
00157 handler.close()
00158 for filename in glob.glob(tmpdir + '/test_log*'):
00159 os.unlink(filename)
00160 os.rmdir(tmpdir)
00161
00162
00163 class LoggingOptionTest(unittest.TestCase):
00164 """Test the ability to enable and disable Tornado's logging hooks."""
00165 def logs_present(self, statement, args=None):
00166
00167
00168
00169
00170
00171 IMPORT = 'from tornado.options import options, parse_command_line'
00172 LOG_INFO = 'import logging; logging.info("hello")'
00173 program = ';'.join([IMPORT, statement, LOG_INFO])
00174 proc = subprocess.Popen(
00175 [sys.executable, '-c', program] + (args or []),
00176 stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
00177 stdout, stderr = proc.communicate()
00178 self.assertEqual(proc.returncode, 0, 'process failed: %r' % stdout)
00179 return b'hello' in stdout
00180
00181 def test_default(self):
00182 self.assertFalse(self.logs_present('pass'))
00183
00184 def test_tornado_default(self):
00185 self.assertTrue(self.logs_present('parse_command_line()'))
00186
00187 def test_disable_command_line(self):
00188 self.assertFalse(self.logs_present('parse_command_line()',
00189 ['--logging=none']))
00190
00191 def test_disable_command_line_case_insensitive(self):
00192 self.assertFalse(self.logs_present('parse_command_line()',
00193 ['--logging=None']))
00194
00195 def test_disable_code_string(self):
00196 self.assertFalse(self.logs_present(
00197 'options.logging = "none"; parse_command_line()'))
00198
00199 def test_disable_code_none(self):
00200 self.assertFalse(self.logs_present(
00201 'options.logging = None; parse_command_line()'))
00202
00203 def test_disable_override(self):
00204
00205 self.assertTrue(self.logs_present(
00206 'options.logging = None; parse_command_line()',
00207 ['--logging=info']))