codekingpro/portable-devtools

1#
2# Copyright 2012 Facebook
3#
4# Licensed under the Apache License, Version 2.0 (the "License"); you may
5# not use this file except in compliance with the License. You may obtain
6# a copy of the License at
7#
8#     http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13# License for the specific language governing permissions and limitations
14# under the License.
15import contextlib
16import glob
17import logging
18import os
19import re
20import subprocess
21import sys
22import tempfile
23import unittest
24import warnings
25
26from tornado.escape import utf8
27from tornado.log import LogFormatter, define_logging_options, enable_pretty_logging
28from tornado.options import OptionParser
29from tornado.util import basestring_type
30
31
32@contextlib.contextmanager
33def ignore_bytes_warning():
34    with warnings.catch_warnings():
35        warnings.simplefilter("ignore", category=BytesWarning)
36        yield
37
38
39class LogFormatterTest(unittest.TestCase):
40    # Matches the output of a single logging call (which may be multiple lines
41    # if a traceback was included, so we use the DOTALL option)
42    LINE_RE = re.compile(
43        b"(?s)\x01\\[E [0-9]{6} [0-9]{2}:[0-9]{2}:[0-9]{2} log_test:[0-9]+\\]\x02 (.*)"
44    )
45
46    def setUp(self):
47        self.formatter = LogFormatter(color=False)
48        # Fake color support.  We can't guarantee anything about the $TERM
49        # variable when the tests are run, so just patch in some values
50        # for testing.  (testing with color off fails to expose some potential
51        # encoding issues from the control characters)
52        self.formatter._colors = {logging.ERROR: "\u0001"}
53        self.formatter._normal = "\u0002"
54        # construct a Logger directly to bypass getLogger's caching
55        self.logger = logging.Logger("LogFormatterTest")
56        self.logger.propagate = False
57        self.tempdir = tempfile.mkdtemp()
58        self.filename = os.path.join(self.tempdir, "log.out")
59        self.handler = self.make_handler(self.filename)
60        self.handler.setFormatter(self.formatter)
61        self.logger.addHandler(self.handler)
62
63    def tearDown(self):
64        self.handler.close()
65        os.unlink(self.filename)
66        os.rmdir(self.tempdir)
67
68    def make_handler(self, filename):
69        return logging.FileHandler(filename, encoding="utf-8")
70
71    def get_output(self):
72        with open(self.filename, "rb") as f:
73            line = f.read().strip()
74            m = LogFormatterTest.LINE_RE.match(line)
75            if m:
76                return m.group(1)
77            else:
78                raise Exception("output didn't match regex: %r" % line)
79
80    def test_basic_logging(self):
81        self.logger.error("foo")
82        self.assertEqual(self.get_output(), b"foo")
83
84    def test_bytes_logging(self):
85        with ignore_bytes_warning():
86            # This will be "\xe9" on python 2 or "b'\xe9'" on python 3
87            self.logger.error(b"\xe9")
88            self.assertEqual(self.get_output(), utf8(repr(b"\xe9")))
89
90    def test_utf8_logging(self):
91        with ignore_bytes_warning():
92            self.logger.error("\u00e9".encode())
93        if issubclass(bytes, basestring_type):
94            # on python 2, utf8 byte strings (and by extension ascii byte
95            # strings) are passed through as-is.
96            self.assertEqual(self.get_output(), utf8("\u00e9"))
97        else:
98            # on python 3, byte strings always get repr'd even if
99            # they're ascii-only, so this degenerates into another
100            # copy of test_bytes_logging.
101            self.assertEqual(self.get_output(), utf8(repr(utf8("\u00e9"))))
102
103    def test_bytes_exception_logging(self):
104        try:
105            raise Exception(b"\xe9")
106        except Exception:
107            self.logger.exception("caught exception")
108        # This will be "Exception: \xe9" on python 2 or
109        # "Exception: b'\xe9'" on python 3.
110        output = self.get_output()
111        self.assertRegex(output, rb"Exception.*\\xe9")
112        # The traceback contains newlines, which should not have been escaped.
113        self.assertNotIn(rb"\n", output)
114
115    def test_unicode_logging(self):
116        self.logger.error("\u00e9")
117        self.assertEqual(self.get_output(), utf8("\u00e9"))
118
119
120class EnablePrettyLoggingTest(unittest.TestCase):
121    def setUp(self):
122        super().setUp()
123        self.options = OptionParser()
124        define_logging_options(self.options)
125        self.logger = logging.Logger("tornado.test.log_test.EnablePrettyLoggingTest")
126        self.logger.propagate = False
127
128    def test_log_file(self):
129        tmpdir = tempfile.mkdtemp()
130        try:
131            self.options.log_file_prefix = tmpdir + "/test_log"
132            enable_pretty_logging(options=self.options, logger=self.logger)
133            self.assertEqual(1, len(self.logger.handlers))
134            self.logger.error("hello")
135            self.logger.handlers[0].flush()
136            filenames = glob.glob(tmpdir + "/test_log*")
137            self.assertEqual(1, len(filenames))
138            with open(filenames[0], encoding="utf-8") as f:
139                self.assertRegex(f.read(), r"^\[E [^]]*\] hello$")
140        finally:
141            for handler in self.logger.handlers:
142                handler.flush()
143                handler.close()
144            for filename in glob.glob(tmpdir + "/test_log*"):
145                os.unlink(filename)
146            os.rmdir(tmpdir)
147
148    def test_log_file_with_timed_rotating(self):
149        tmpdir = tempfile.mkdtemp()
150        try:
151            self.options.log_file_prefix = tmpdir + "/test_log"
152            self.options.log_rotate_mode = "time"
153            enable_pretty_logging(options=self.options, logger=self.logger)
154            self.logger.error("hello")
155            self.logger.handlers[0].flush()
156            filenames = glob.glob(tmpdir + "/test_log*")
157            self.assertEqual(1, len(filenames))
158            with open(filenames[0], encoding="utf-8") as f:
159                self.assertRegex(f.read(), r"^\[E [^]]*\] hello$")
160        finally:
161            for handler in self.logger.handlers:
162                handler.flush()
163                handler.close()
164            for filename in glob.glob(tmpdir + "/test_log*"):
165                os.unlink(filename)
166            os.rmdir(tmpdir)
167
168    def test_wrong_rotate_mode_value(self):
169        try:
170            self.options.log_file_prefix = "some_path"
171            self.options.log_rotate_mode = "wrong_mode"
172            self.assertRaises(
173                ValueError,
174                enable_pretty_logging,
175                options=self.options,
176                logger=self.logger,
177            )
178        finally:
179            for handler in self.logger.handlers:
180                handler.flush()
181                handler.close()
182
183
184class LoggingOptionTest(unittest.TestCase):
185    """Test the ability to enable and disable Tornado's logging hooks."""
186
187    def logs_present(self, statement, args=None):
188        # Each test may manipulate and/or parse the options and then logs
189        # a line at the 'info' level.  This level is ignored in the
190        # logging module by default, but Tornado turns it on by default
191        # so it is the easiest way to tell whether tornado's logging hooks
192        # ran.
193        IMPORT = "from tornado.options import options, parse_command_line"
194        LOG_INFO = 'import logging; logging.info("hello")'
195        program = ";".join([IMPORT, statement, LOG_INFO])
196        proc = subprocess.Popen(
197            [sys.executable, "-c", program] + (args or []),
198            stdout=subprocess.PIPE,
199            stderr=subprocess.STDOUT,
200        )
201        stdout, stderr = proc.communicate()
202        self.assertEqual(proc.returncode, 0, "process failed: %r" % stdout)
203        return b"hello" in stdout
204
205    def test_default(self):
206        self.assertFalse(self.logs_present("pass"))
207
208    def test_tornado_default(self):
209        self.assertTrue(self.logs_present("parse_command_line()"))
210
211    def test_disable_command_line(self):
212        self.assertFalse(self.logs_present("parse_command_line()", ["--logging=none"]))
213
214    def test_disable_command_line_case_insensitive(self):
215        self.assertFalse(self.logs_present("parse_command_line()", ["--logging=None"]))
216
217    def test_disable_code_string(self):
218        self.assertFalse(
219            self.logs_present('options.logging = "none"; parse_command_line()')
220        )
221
222    def test_disable_code_none(self):
223        self.assertFalse(
224            self.logs_present("options.logging = None; parse_command_line()")
225        )
226
227    def test_disable_override(self):
228        # command line trumps code defaults
229        self.assertTrue(
230            self.logs_present(
231                "options.logging = None; parse_command_line()", ["--logging=info"]
232            )
233        )
234