codekingpro/portable-devtools

1from functools import reduce
2import gc
3import io
4import locale  # system locale module, not tornado.locale
5import logging
6import operator
7import textwrap
8import sys
9import unittest
10import warnings
11
12from tornado.httpclient import AsyncHTTPClient
13from tornado.httpserver import HTTPServer
14from tornado.netutil import Resolver
15from tornado.options import define, add_parse_callback, options
16from tornado.test.util import ABT_SKIP_MESSAGE
17
18
19TEST_MODULES = [
20    "tornado.httputil.doctests",
21    "tornado.iostream.doctests",
22    "tornado.util.doctests",
23    "tornado.test.asyncio_test",
24    "tornado.test.auth_test",
25    "tornado.test.autoreload_test",
26    "tornado.test.circlerefs_test",
27    "tornado.test.concurrent_test",
28    "tornado.test.curl_httpclient_test",
29    "tornado.test.escape_test",
30    "tornado.test.gen_test",
31    "tornado.test.http1connection_test",
32    "tornado.test.httpclient_test",
33    "tornado.test.httpserver_test",
34    "tornado.test.httputil_test",
35    "tornado.test.import_test",
36    "tornado.test.ioloop_test",
37    "tornado.test.iostream_test",
38    "tornado.test.locale_test",
39    "tornado.test.locks_test",
40    "tornado.test.netutil_test",
41    "tornado.test.log_test",
42    "tornado.test.options_test",
43    "tornado.test.process_test",
44    "tornado.test.queues_test",
45    "tornado.test.routing_test",
46    "tornado.test.simple_httpclient_test",
47    "tornado.test.tcpclient_test",
48    "tornado.test.tcpserver_test",
49    "tornado.test.template_test",
50    "tornado.test.testing_test",
51    "tornado.test.twisted_test",
52    "tornado.test.util_test",
53    "tornado.test.web_test",
54    "tornado.test.websocket_test",
55    "tornado.test.wsgi_test",
56]
57
58
59def all():
60    return unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULES)
61
62
63def test_runner_factory(stderr):
64
65    class TornadoTextTestResult(unittest.TextTestResult):
66        def addSkip(self, test, reason):
67            if reason == ABT_SKIP_MESSAGE:
68                # Don't report abstract base tests as skips in our own tooling.
69                #
70                # See tornado.test.util.abstract_base_test.
71                return
72            super().addSkip(test, reason)
73
74    class TornadoTextTestRunner(unittest.TextTestRunner):
75        def __init__(self, *args, **kwargs):
76            kwargs["stream"] = stderr
77            kwargs["resultclass"] = TornadoTextTestResult
78            super().__init__(*args, **kwargs)
79
80        def run(self, test):
81            result = super().run(test)
82            if result.skipped:
83                skip_reasons = {reason for (test, reason) in result.skipped}
84                self.stream.write(  # type: ignore
85                    textwrap.fill(
86                        "Some tests were skipped because: %s"
87                        % ", ".join(sorted(skip_reasons))
88                    )
89                )
90                self.stream.write("\n")  # type: ignore
91            return result
92
93    return TornadoTextTestRunner
94
95
96class LogCounter(logging.Filter):
97    """Counts the number of WARNING or higher log records."""
98
99    def __init__(self, *args, **kwargs):
100        super().__init__(*args, **kwargs)
101        self.info_count = self.warning_count = self.error_count = 0
102
103    def filter(self, record):
104        if record.levelno >= logging.ERROR:
105            self.error_count += 1
106        elif record.levelno >= logging.WARNING:
107            self.warning_count += 1
108        elif record.levelno >= logging.INFO:
109            self.info_count += 1
110        return True
111
112
113class CountingStderr(io.IOBase):
114    def __init__(self, real):
115        self.real = real
116        self.byte_count = 0
117
118    def write(self, data):
119        self.byte_count += len(data)
120        return self.real.write(data)
121
122    def flush(self):
123        return self.real.flush()
124
125
126def main():
127    # Be strict about most warnings (This is set in our test running
128    # scripts to catch import-time warnings, but set it again here to
129    # be sure). This also turns on warnings that are ignored by
130    # default, including DeprecationWarnings and python 3.2's
131    # ResourceWarnings.
132    warnings.filterwarnings("error")
133    # setuptools sometimes gives ImportWarnings about things that are on
134    # sys.path even if they're not being used.
135    warnings.filterwarnings("ignore", category=ImportWarning)
136    # Tornado generally shouldn't use anything deprecated, but some of
137    # our dependencies do (last match wins).
138    warnings.filterwarnings("ignore", category=DeprecationWarning)
139    warnings.filterwarnings("error", category=DeprecationWarning, module=r"tornado\..*")
140    warnings.filterwarnings("ignore", category=PendingDeprecationWarning)
141    warnings.filterwarnings(
142        "error", category=PendingDeprecationWarning, module=r"tornado\..*"
143    )
144
145    logging.getLogger("tornado.access").setLevel(logging.CRITICAL)
146
147    define(
148        "httpclient",
149        type=str,
150        default=None,
151        callback=lambda s: AsyncHTTPClient.configure(
152            s, defaults=dict(allow_ipv6=False)
153        ),
154    )
155    define("httpserver", type=str, default=None, callback=HTTPServer.configure)
156    define("resolver", type=str, default=None, callback=Resolver.configure)
157    define(
158        "debug_gc",
159        type=str,
160        multiple=True,
161        help="A comma-separated list of gc module debug constants, "
162        "e.g. DEBUG_STATS or DEBUG_COLLECTABLE,DEBUG_OBJECTS",
163        callback=lambda values: gc.set_debug(
164            reduce(operator.or_, (getattr(gc, v) for v in values))
165        ),
166    )
167    define(
168        "fail-if-logs",
169        default=True,
170        help="If true, fail the tests if any log output is produced (unless captured by ExpectLog)",
171    )
172
173    def set_locale(x):
174        locale.setlocale(locale.LC_ALL, x)
175
176    define("locale", type=str, default=None, callback=set_locale)
177
178    log_counter = LogCounter()
179    add_parse_callback(lambda: logging.getLogger().handlers[0].addFilter(log_counter))
180
181    # Certain errors (especially "unclosed resource" errors raised in
182    # destructors) go directly to stderr instead of logging. Count
183    # anything written by anything but the test runner as an error.
184    orig_stderr = sys.stderr
185    counting_stderr = CountingStderr(orig_stderr)
186    sys.stderr = counting_stderr  # type: ignore
187
188    import tornado.testing
189
190    kwargs = {}
191
192    # HACK:  unittest.main will make its own changes to the warning
193    # configuration, which may conflict with the settings above
194    # or command-line flags like -bb.  Passing warnings=False
195    # suppresses this behavior, although this looks like an implementation
196    # detail.  http://bugs.python.org/issue15626
197    kwargs["warnings"] = False
198
199    kwargs["testRunner"] = test_runner_factory(orig_stderr)
200    try:
201        tornado.testing.main(**kwargs)
202    finally:
203        # The tests should run clean; consider it a failure if they
204        # logged anything at info level or above.
205        if (
206            log_counter.info_count > 0
207            or log_counter.warning_count > 0
208            or log_counter.error_count > 0
209            or counting_stderr.byte_count > 0
210        ):
211            logging.error(
212                "logged %d infos, %d warnings, %d errors, and %d bytes to stderr",
213                log_counter.info_count,
214                log_counter.warning_count,
215                log_counter.error_count,
216                counting_stderr.byte_count,
217            )
218            if options.fail_if_logs:
219                sys.exit(1)
220
221
222if __name__ == "__main__":
223    main()
224