codekingpro/portable-devtools
115k
1"""Support classes for automated testing.
2
3* `AsyncTestCase` and `AsyncHTTPTestCase`: Subclasses of unittest.TestCase
4 with additional support for testing asynchronous (`.IOLoop`-based) code.
5
6* `ExpectLog`: Make test logs less spammy.
7
8* `main()`: A simple test runner (wrapper around unittest.main()) with support
9 for the tornado.autoreload module to rerun the tests when code changes.
10"""
11
12import asyncio
13from collections.abc import Generator
14import functools
15import inspect
16import logging
17import os
18import re
19import signal
20import socket
21import sys
22import unittest
23import warnings
24
25from tornado import gen
26from tornado.httpclient import AsyncHTTPClient, HTTPResponse
27from tornado.httpserver import HTTPServer
28from tornado.ioloop import IOLoop, TimeoutError
29from tornado import netutil
30from tornado.platform.asyncio import AsyncIOMainLoop
31from tornado.process import Subprocess
32from tornado.log import app_log
33from tornado.util import raise_exc_info, basestring_type
34from tornado.web import Application
35
36import typing
37from typing import Tuple, Any, Callable, Type, Dict, Union, Optional, Coroutine
38from types import TracebackType
39
40if typing.TYPE_CHECKING:
41 _ExcInfoTuple = Tuple[
42 Optional[Type[BaseException]], Optional[BaseException], Optional[TracebackType]
43 ]
44
45
46_NON_OWNED_IOLOOPS = AsyncIOMainLoop
47
48
49def bind_unused_port(
50 reuse_port: bool = False, address: str = "127.0.0.1"
51) -> Tuple[socket.socket, int]:
52 """Binds a server socket to an available port on localhost.
53
54 Returns a tuple (socket, port).
55
56 .. versionchanged:: 4.4
57 Always binds to ``127.0.0.1`` without resolving the name
58 ``localhost``.
59
60 .. versionchanged:: 6.2
61 Added optional ``address`` argument to
62 override the default "127.0.0.1".
63 """
64 sock = netutil.bind_sockets(
65 0, address, family=socket.AF_INET, reuse_port=reuse_port
66 )[0]
67 port = sock.getsockname()[1]
68 return sock, port
69
70
71def get_async_test_timeout() -> float:
72 """Get the global timeout setting for async tests.
73
74 Returns a float, the timeout in seconds.
75
76 .. versionadded:: 3.1
77 """
78 env = os.environ.get("ASYNC_TEST_TIMEOUT")
79 if env is not None:
80 try:
81 return float(env)
82 except ValueError:
83 pass
84 return 5
85
86
87class AsyncTestCase(unittest.TestCase):
88 """`~unittest.TestCase` subclass for testing `.IOLoop`-based
89 asynchronous code.
90
91 The unittest framework is synchronous, so the test must be
92 complete by the time the test method returns. This means that
93 asynchronous code cannot be used in quite the same way as usual
94 and must be adapted to fit. To write your tests with coroutines,
95 decorate your test methods with `tornado.testing.gen_test` instead
96 of `tornado.gen.coroutine`.
97
98 This class also provides the (deprecated) `stop()` and `wait()`
99 methods for a more manual style of testing. The test method itself
100 must call ``self.wait()``, and asynchronous callbacks should call
101 ``self.stop()`` to signal completion.
102
103 By default, a new `.IOLoop` is constructed for each test and is available
104 as ``self.io_loop``. If the code being tested requires a
105 reused global `.IOLoop`, subclasses should override `get_new_ioloop` to return it,
106 although this is deprecated as of Tornado 6.3.
107
108 The `.IOLoop`'s ``start`` and ``stop`` methods should not be
109 called directly. Instead, use `self.stop <stop>` and `self.wait
110 <wait>`. Arguments passed to ``self.stop`` are returned from
111 ``self.wait``. It is possible to have multiple ``wait``/``stop``
112 cycles in the same test.
113
114 Example::
115
116 # This test uses coroutine style.
117 class MyTestCase(AsyncTestCase):
118 @tornado.testing.gen_test
119 def test_http_fetch(self):
120 client = AsyncHTTPClient()
121 response = yield client.fetch("http://www.tornadoweb.org")
122 # Test contents of response
123 self.assertIn("FriendFeed", response.body)
124
125 # This test uses argument passing between self.stop and self.wait.
126 class MyTestCase2(AsyncTestCase):
127 def test_http_fetch(self):
128 client = AsyncHTTPClient()
129 client.fetch("http://www.tornadoweb.org/", self.stop)
130 response = self.wait()
131 # Test contents of response
132 self.assertIn("FriendFeed", response.body)
133 """
134
135 def __init__(self, methodName: str = "runTest") -> None:
136 super().__init__(methodName)
137 self.__stopped = False
138 self.__running = False
139 self.__failure = None # type: Optional[_ExcInfoTuple]
140 self.__stop_args = None # type: Any
141 self.__timeout = None # type: Optional[object]
142
143 # Not used in this class itself, but used by @gen_test
144 self._test_generator = None # type: Optional[Union[Generator, Coroutine]]
145
146 def setUp(self) -> None:
147 py_ver = sys.version_info
148 if ((3, 10, 0) <= py_ver < (3, 10, 9)) or ((3, 11, 0) <= py_ver <= (3, 11, 1)):
149 # Early releases in the Python 3.10 and 3.1 series had deprecation
150 # warnings that were later reverted; we must suppress them here.
151 setup_with_context_manager(self, warnings.catch_warnings())
152 warnings.filterwarnings(
153 "ignore",
154 message="There is no current event loop",
155 category=DeprecationWarning,
156 module=r"tornado\..*",
157 )
158 super().setUp()
159 if type(self).get_new_ioloop is not AsyncTestCase.get_new_ioloop:
160 warnings.warn("get_new_ioloop is deprecated", DeprecationWarning)
161 self.io_loop = self.get_new_ioloop()
162 asyncio.set_event_loop(self.io_loop.asyncio_loop) # type: ignore[attr-defined]
163
164 def tearDown(self) -> None:
165 # Native coroutines tend to produce warnings if they're not
166 # allowed to run to completion. It's difficult to ensure that
167 # this always happens in tests, so cancel any tasks that are
168 # still pending by the time we get here.
169 asyncio_loop = self.io_loop.asyncio_loop # type: ignore
170 tasks = asyncio.all_tasks(asyncio_loop)
171 # Tasks that are done may still appear here and may contain
172 # non-cancellation exceptions, so filter them out.
173 tasks = [t for t in tasks if not t.done()] # type: ignore
174 for t in tasks:
175 t.cancel()
176 # Allow the tasks to run and finalize themselves (which means
177 # raising a CancelledError inside the coroutine). This may
178 # just transform the "task was destroyed but it is pending"
179 # warning into a "uncaught CancelledError" warning, but
180 # catching CancelledErrors in coroutines that may leak is
181 # simpler than ensuring that no coroutines leak.
182 if tasks:
183 done, pending = self.io_loop.run_sync(lambda: asyncio.wait(tasks))
184 assert not pending
185 # If any task failed with anything but a CancelledError, raise it.
186 for f in done:
187 try:
188 f.result()
189 except asyncio.CancelledError:
190 pass
191
192 # Clean up Subprocess, so it can be used again with a new ioloop.
193 Subprocess.uninitialize()
194 asyncio.set_event_loop(None)
195 if not isinstance(self.io_loop, _NON_OWNED_IOLOOPS):
196 # Try to clean up any file descriptors left open in the ioloop.
197 # This avoids leaks, especially when tests are run repeatedly
198 # in the same process with autoreload (because curl does not
199 # set FD_CLOEXEC on its file descriptors)
200 self.io_loop.close(all_fds=True)
201 super().tearDown()
202 # In case an exception escaped or the StackContext caught an exception
203 # when there wasn't a wait() to re-raise it, do so here.
204 # This is our last chance to raise an exception in a way that the
205 # unittest machinery understands.
206 self.__rethrow()
207
208 def get_new_ioloop(self) -> IOLoop:
209 """Returns the `.IOLoop` to use for this test.
210
211 By default, a new `.IOLoop` is created for each test.
212 Subclasses may override this method to return
213 `.IOLoop.current()` if it is not appropriate to use a new
214 `.IOLoop` in each tests (for example, if there are global
215 singletons using the default `.IOLoop`) or if a per-test event
216 loop is being provided by another system (such as
217 ``pytest-asyncio``).
218
219 .. deprecated:: 6.3
220 This method will be removed in Tornado 7.0.
221 """
222 return IOLoop(make_current=False)
223
224 def _handle_exception(
225 self, typ: Type[Exception], value: Exception, tb: TracebackType
226 ) -> bool:
227 if self.__failure is None:
228 self.__failure = (typ, value, tb)
229 else:
230 app_log.error(
231 "multiple unhandled exceptions in test", exc_info=(typ, value, tb)
232 )
233 self.stop()
234 return True
235
236 def __rethrow(self) -> None:
237 if self.__failure is not None:
238 failure = self.__failure
239 self.__failure = None
240 raise_exc_info(failure)
241
242 def run(
243 self, result: Optional[unittest.TestResult] = None
244 ) -> Optional[unittest.TestResult]:
245 ret = super().run(result)
246 # As a last resort, if an exception escaped super.run() and wasn't
247 # re-raised in tearDown, raise it here. This will cause the
248 # unittest run to fail messily, but that's better than silently
249 # ignoring an error.
250 self.__rethrow()
251 return ret
252
253 def _callTestMethod(self, method: Callable) -> None:
254 """Run the given test method, raising an error if it returns non-None.
255
256 Failure to decorate asynchronous test methods with ``@gen_test`` can lead to tests
257 incorrectly passing.
258
259 Remove this override when Python 3.10 support is dropped. This check (in the form of a
260 DeprecationWarning) became a part of the standard library in 3.11.
261
262 Note that ``_callTestMethod`` is not documented as a public interface. However, it is
263 present in all supported versions of Python (3.8+), and if it goes away in the future that's
264 OK because we can just remove this override as noted above.
265 """
266 # Calling super()._callTestMethod would hide the return value, even in python 3.8-3.10
267 # where the check isn't being done for us.
268 result = method()
269 if isinstance(result, Generator) or inspect.iscoroutine(result):
270 raise TypeError(
271 "Generator and coroutine test methods should be"
272 " decorated with tornado.testing.gen_test"
273 )
274 elif result is not None:
275 raise ValueError("Return value from test method ignored: %r" % result)
276
277 def stop(self, _arg: Any = None, **kwargs: Any) -> None:
278 """Stops the `.IOLoop`, causing one pending (or future) call to `wait()`
279 to return.
280
281 Keyword arguments or a single positional argument passed to `stop()` are
282 saved and will be returned by `wait()`.
283
284 .. deprecated:: 5.1
285
286 `stop` and `wait` are deprecated; use ``@gen_test`` instead.
287 """
288 assert _arg is None or not kwargs
289 self.__stop_args = kwargs or _arg
290 if self.__running:
291 self.io_loop.stop()
292 self.__running = False
293 self.__stopped = True
294
295 def wait(
296 self,
297 condition: Optional[Callable[..., bool]] = None,
298 timeout: Optional[float] = None,
299 ) -> Any:
300 """Runs the `.IOLoop` until stop is called or timeout has passed.
301
302 In the event of a timeout, an exception will be thrown. The
303 default timeout is 5 seconds; it may be overridden with a
304 ``timeout`` keyword argument or globally with the
305 ``ASYNC_TEST_TIMEOUT`` environment variable.
306
307 If ``condition`` is not ``None``, the `.IOLoop` will be restarted
308 after `stop()` until ``condition()`` returns ``True``.
309
310 .. versionchanged:: 3.1
311 Added the ``ASYNC_TEST_TIMEOUT`` environment variable.
312
313 .. deprecated:: 5.1
314
315 `stop` and `wait` are deprecated; use ``@gen_test`` instead.
316 """
317 if timeout is None:
318 timeout = get_async_test_timeout()
319
320 if not self.__stopped:
321 if timeout:
322
323 def timeout_func() -> None:
324 try:
325 raise self.failureException(
326 "Async operation timed out after %s seconds" % timeout
327 )
328 except Exception:
329 self.__failure = sys.exc_info()
330 self.stop()
331
332 self.__timeout = self.io_loop.add_timeout(
333 self.io_loop.time() + timeout, timeout_func
334 )
335 while True:
336 self.__running = True
337 self.io_loop.start()
338 if self.__failure is not None or condition is None or condition():
339 break
340 if self.__timeout is not None:
341 self.io_loop.remove_timeout(self.__timeout)
342 self.__timeout = None
343 assert self.__stopped
344 self.__stopped = False
345 self.__rethrow()
346 result = self.__stop_args
347 self.__stop_args = None
348 return result
349
350
351class AsyncHTTPTestCase(AsyncTestCase):
352 """A test case that starts up an HTTP server.
353
354 Subclasses must override `get_app()`, which returns the
355 `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.
356 Tests will typically use the provided ``self.http_client`` to fetch
357 URLs from this server.
358
359 Example, assuming the "Hello, world" example from the user guide is in
360 ``hello.py``::
361
362 import hello
363
364 class TestHelloApp(AsyncHTTPTestCase):
365 def get_app(self):
366 return hello.make_app()
367
368 def test_homepage(self):
369 response = self.fetch('/')
370 self.assertEqual(response.code, 200)
371 self.assertEqual(response.body, 'Hello, world')
372
373 That call to ``self.fetch()`` is equivalent to ::
374
375 self.http_client.fetch(self.get_url('/'), self.stop)
376 response = self.wait()
377
378 which illustrates how AsyncTestCase can turn an asynchronous operation,
379 like ``http_client.fetch()``, into a synchronous operation. If you need
380 to do other asynchronous operations in tests, you'll probably need to use
381 ``stop()`` and ``wait()`` yourself.
382 """
383
384 def setUp(self) -> None:
385 super().setUp()
386 sock, port = bind_unused_port()
387 self.__port = port
388
389 self.http_client = self.get_http_client()
390 self._app = self.get_app()
391 self.http_server = self.get_http_server()
392 self.http_server.add_sockets([sock])
393
394 def get_http_client(self) -> AsyncHTTPClient:
395 return AsyncHTTPClient()
396
397 def get_http_server(self) -> HTTPServer:
398 return HTTPServer(self._app, **self.get_httpserver_options())
399
400 def get_app(self) -> Application:
401 """Should be overridden by subclasses to return a
402 `tornado.web.Application` or other `.HTTPServer` callback.
403 """
404 raise NotImplementedError()
405
406 def fetch(
407 self, path: str, raise_error: bool = False, **kwargs: Any
408 ) -> HTTPResponse:
409 """Convenience method to synchronously fetch a URL.
410
411 The given path will be appended to the local server's host and
412 port. Any additional keyword arguments will be passed directly to
413 `.AsyncHTTPClient.fetch` (and so could be used to pass
414 ``method="POST"``, ``body="..."``, etc).
415
416 If the path begins with http:// or https://, it will be treated as a
417 full URL and will be fetched as-is.
418
419 If ``raise_error`` is ``True``, a `tornado.httpclient.HTTPError` will
420 be raised if the response code is not 200. This is the same behavior
421 as the ``raise_error`` argument to `.AsyncHTTPClient.fetch`, but
422 the default is ``False`` here (it's ``True`` in `.AsyncHTTPClient`)
423 because tests often need to deal with non-200 response codes.
424
425 .. versionchanged:: 5.0
426 Added support for absolute URLs.
427
428 .. versionchanged:: 5.1
429
430 Added the ``raise_error`` argument.
431
432 .. deprecated:: 5.1
433
434 This method currently turns any exception into an
435 `.HTTPResponse` with status code 599. In Tornado 6.0,
436 errors other than `tornado.httpclient.HTTPError` will be
437 passed through, and ``raise_error=False`` will only
438 suppress errors that would be raised due to non-200
439 response codes.
440
441 """
442 if path.lower().startswith(("http://", "https://")):
443 url = path
444 else:
445 url = self.get_url(path)
446 return self.io_loop.run_sync(
447 lambda: self.http_client.fetch(url, raise_error=raise_error, **kwargs),
448 timeout=get_async_test_timeout(),
449 )
450
451 def get_httpserver_options(self) -> Dict[str, Any]:
452 """May be overridden by subclasses to return additional
453 keyword arguments for the server.
454 """
455 return {}
456
457 def get_http_port(self) -> int:
458 """Returns the port used by the server.
459
460 A new port is chosen for each test.
461 """
462 return self.__port
463
464 def get_protocol(self) -> str:
465 return "http"
466
467 def get_url(self, path: str) -> str:
468 """Returns an absolute url for the given path on the test server."""
469 return f"{self.get_protocol()}://127.0.0.1:{self.get_http_port()}{path}"
470
471 def tearDown(self) -> None:
472 self.http_server.stop()
473 self.io_loop.run_sync(
474 self.http_server.close_all_connections, timeout=get_async_test_timeout()
475 )
476 self.http_client.close()
477 del self.http_server
478 del self._app
479 super().tearDown()
480
481
482class AsyncHTTPSTestCase(AsyncHTTPTestCase):
483 """A test case that starts an HTTPS server.
484
485 Interface is generally the same as `AsyncHTTPTestCase`.
486 """
487
488 def get_http_client(self) -> AsyncHTTPClient:
489 return AsyncHTTPClient(force_instance=True, defaults=dict(validate_cert=False))
490
491 def get_httpserver_options(self) -> Dict[str, Any]:
492 return dict(ssl_options=self.get_ssl_options())
493
494 def get_ssl_options(self) -> Dict[str, Any]:
495 """May be overridden by subclasses to select SSL options.
496
497 By default includes a self-signed testing certificate.
498 """
499 return AsyncHTTPSTestCase.default_ssl_options()
500
501 @staticmethod
502 def default_ssl_options() -> Dict[str, Any]:
503 # Testing keys were generated with:
504 # openssl req -new -keyout tornado/test/test.key \
505 # -out tornado/test/test.crt \
506 # -nodes -days 3650 -x509 \
507 # -subj "/CN=foo.example.com" -addext "subjectAltName = DNS:foo.example.com"
508 module_dir = os.path.dirname(__file__)
509 return dict(
510 certfile=os.path.join(module_dir, "test", "test.crt"),
511 keyfile=os.path.join(module_dir, "test", "test.key"),
512 )
513
514 def get_protocol(self) -> str:
515 return "https"
516
517
518@typing.overload
519def gen_test(
520 *, timeout: Optional[float] = None
521) -> Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]]:
522 pass
523
524
525@typing.overload # noqa: F811
526def gen_test(func: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
527 pass
528
529
530def gen_test( # noqa: F811
531 func: Optional[Callable[..., Union[Generator, "Coroutine"]]] = None,
532 timeout: Optional[float] = None,
533) -> Union[
534 Callable[..., None],
535 Callable[[Callable[..., Union[Generator, "Coroutine"]]], Callable[..., None]],
536]:
537 """Testing equivalent of ``@gen.coroutine``, to be applied to test methods.
538
539 ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not
540 already running. ``@gen_test`` should be applied to test methods
541 on subclasses of `AsyncTestCase`.
542
543 Example::
544
545 class MyTest(AsyncHTTPTestCase):
546 @gen_test
547 def test_something(self):
548 response = yield self.http_client.fetch(self.get_url('/'))
549
550 By default, ``@gen_test`` times out after 5 seconds. The timeout may be
551 overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,
552 or for each test with the ``timeout`` keyword argument::
553
554 class MyTest(AsyncHTTPTestCase):
555 @gen_test(timeout=10)
556 def test_something_slow(self):
557 response = yield self.http_client.fetch(self.get_url('/'))
558
559 Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,
560 `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield
561 self.http_client.fetch(self.get_url())`` as shown above instead.
562
563 .. versionadded:: 3.1
564 The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment
565 variable.
566
567 .. versionchanged:: 4.0
568 The wrapper now passes along ``*args, **kwargs`` so it can be used
569 on functions with arguments.
570
571 """
572 if timeout is None:
573 timeout = get_async_test_timeout()
574
575 def wrap(f: Callable[..., Union[Generator, "Coroutine"]]) -> Callable[..., None]:
576 # Stack up several decorators to allow us to access the generator
577 # object itself. In the innermost wrapper, we capture the generator
578 # and save it in an attribute of self. Next, we run the wrapped
579 # function through @gen.coroutine. Finally, the coroutine is
580 # wrapped again to make it synchronous with run_sync.
581 #
582 # This is a good case study arguing for either some sort of
583 # extensibility in the gen decorators or cancellation support.
584 @functools.wraps(f)
585 def pre_coroutine(self, *args, **kwargs):
586 # type: (AsyncTestCase, *Any, **Any) -> Union[Generator, Coroutine]
587 # Type comments used to avoid pypy3 bug.
588 result = f(self, *args, **kwargs)
589 if isinstance(result, Generator) or inspect.iscoroutine(result):
590 self._test_generator = result
591 else:
592 self._test_generator = None
593 return result
594
595 if inspect.iscoroutinefunction(f):
596 coro = pre_coroutine
597 else:
598 coro = gen.coroutine(pre_coroutine) # type: ignore[assignment]
599
600 @functools.wraps(coro)
601 def post_coroutine(self, *args, **kwargs):
602 # type: (AsyncTestCase, *Any, **Any) -> None
603 try:
604 return self.io_loop.run_sync(
605 functools.partial(coro, self, *args, **kwargs), timeout=timeout
606 )
607 except TimeoutError as e:
608 # run_sync raises an error with an unhelpful traceback.
609 # If the underlying generator is still running, we can throw the
610 # exception back into it so the stack trace is replaced by the
611 # point where the test is stopped. The only reason the generator
612 # would not be running would be if it were cancelled, which means
613 # a native coroutine, so we can rely on the cr_running attribute.
614 if self._test_generator is not None and getattr(
615 self._test_generator, "cr_running", True
616 ):
617 self._test_generator.throw(e)
618 # In case the test contains an overly broad except
619 # clause, we may get back here.
620 # Coroutine was stopped or didn't raise a useful stack trace,
621 # so re-raise the original exception which is better than nothing.
622 raise
623
624 return post_coroutine
625
626 if func is not None:
627 # Used like:
628 # @gen_test
629 # def f(self):
630 # pass
631 return wrap(func)
632 else:
633 # Used like @gen_test(timeout=10)
634 return wrap
635
636
637# Without this attribute, nosetests will try to run gen_test as a test
638# anywhere it is imported.
639gen_test.__test__ = False # type: ignore
640
641
642class ExpectLog(logging.Filter):
643 """Context manager to capture and suppress expected log output.
644
645 Useful to make tests of error conditions less noisy, while still
646 leaving unexpected log entries visible. *Not thread safe.*
647
648 The attribute ``logged_stack`` is set to ``True`` if any exception
649 stack trace was logged.
650
651 Usage::
652
653 with ExpectLog('tornado.application', "Uncaught exception"):
654 error_response = self.fetch("/some_page")
655
656 .. versionchanged:: 4.3
657 Added the ``logged_stack`` attribute.
658 """
659
660 def __init__(
661 self,
662 logger: Union[logging.Logger, basestring_type],
663 regex: str,
664 required: bool = True,
665 level: Optional[int] = None,
666 ) -> None:
667 """Constructs an ExpectLog context manager.
668
669 :param logger: Logger object (or name of logger) to watch. Pass an
670 empty string to watch the root logger.
671 :param regex: Regular expression to match. Any log entries on the
672 specified logger that match this regex will be suppressed.
673 :param required: If true, an exception will be raised if the end of the
674 ``with`` statement is reached without matching any log entries.
675 :param level: A constant from the ``logging`` module indicating the
676 expected log level. If this parameter is provided, only log messages
677 at this level will be considered to match. Additionally, the
678 supplied ``logger`` will have its level adjusted if necessary (for
679 the duration of the ``ExpectLog`` to enable the expected message.
680
681 .. versionchanged:: 6.1
682 Added the ``level`` parameter.
683
684 .. deprecated:: 6.3
685 In Tornado 7.0, only ``WARNING`` and higher logging levels will be
686 matched by default. To match ``INFO`` and lower levels, the ``level``
687 argument must be used. This is changing to minimize differences
688 between ``tornado.testing.main`` (which enables ``INFO`` logs by
689 default) and most other test runners (including those in IDEs)
690 which have ``INFO`` logs disabled by default.
691 """
692 if isinstance(logger, basestring_type):
693 logger = logging.getLogger(logger)
694 self.logger = logger
695 self.regex = re.compile(regex)
696 self.required = required
697 # matched and deprecated_level_matched are a counter for the respective event.
698 self.matched = 0
699 self.deprecated_level_matched = 0
700 self.logged_stack = False
701 self.level = level
702 self.orig_level = None # type: Optional[int]
703
704 def filter(self, record: logging.LogRecord) -> bool:
705 if record.exc_info:
706 self.logged_stack = True
707 message = record.getMessage()
708 if self.regex.match(message):
709 if self.level is None and record.levelno < logging.WARNING:
710 # We're inside the logging machinery here so generating a DeprecationWarning
711 # here won't be reported cleanly (if warnings-as-errors is enabled, the error
712 # just gets swallowed by the logging module), and even if it were it would
713 # have the wrong stack trace. Just remember this fact and report it in
714 # __exit__ instead.
715 self.deprecated_level_matched += 1
716 if self.level is not None and record.levelno != self.level:
717 app_log.warning(
718 "Got expected log message %r at unexpected level (%s vs %s)"
719 % (message, logging.getLevelName(self.level), record.levelname)
720 )
721 return True
722 self.matched += 1
723 return False
724 return True
725
726 def __enter__(self) -> "ExpectLog":
727 if self.level is not None and self.level < self.logger.getEffectiveLevel():
728 self.orig_level = self.logger.level
729 self.logger.setLevel(self.level)
730 self.logger.addFilter(self)
731 return self
732
733 def __exit__(
734 self,
735 typ: "Optional[Type[BaseException]]",
736 value: Optional[BaseException],
737 tb: Optional[TracebackType],
738 ) -> None:
739 if self.orig_level is not None:
740 self.logger.setLevel(self.orig_level)
741 self.logger.removeFilter(self)
742 if not typ and self.required and not self.matched:
743 raise Exception("did not get expected log message")
744 if (
745 not typ
746 and self.required
747 and (self.deprecated_level_matched >= self.matched)
748 ):
749 warnings.warn(
750 "ExpectLog matched at INFO or below without level argument",
751 DeprecationWarning,
752 )
753
754
755# From https://nedbatchelder.com/blog/201508/using_context_managers_in_test_setup.html
756def setup_with_context_manager(testcase: unittest.TestCase, cm: Any) -> Any:
757 """Use a context manager to setUp a test case.
758
759 Example::
760
761 def setUp(self):
762 setup_with_context_manager(self, warnings.catch_warnings())
763 warnings.filterwarnings("ignore", category=DeprecationWarning)
764 # The catch_warnings context manager will be deactivated
765 # automatically in tearDown.
766
767 """
768 val = cm.__enter__()
769 testcase.addCleanup(cm.__exit__, None, None, None)
770 return val
771
772
773def main(**kwargs: Any) -> None:
774 """A simple test runner.
775
776 This test runner is essentially equivalent to `unittest.main` from
777 the standard library, but adds support for Tornado-style option
778 parsing and log formatting. It is *not* necessary to use this
779 `main` function to run tests using `AsyncTestCase`; these tests
780 are self-contained and can run with any test runner.
781
782 The easiest way to run a test is via the command line::
783
784 python -m tornado.testing tornado.test.web_test
785
786 See the standard library ``unittest`` module for ways in which
787 tests can be specified.
788
789 Projects with many tests may wish to define a test script like
790 ``tornado/test/runtests.py``. This script should define a method
791 ``all()`` which returns a test suite and then call
792 `tornado.testing.main()`. Note that even when a test script is
793 used, the ``all()`` test suite may be overridden by naming a
794 single test on the command line::
795
796 # Runs all tests
797 python -m tornado.test.runtests
798 # Runs one test
799 python -m tornado.test.runtests tornado.test.web_test
800
801 Additional keyword arguments passed through to ``unittest.main()``.
802 For example, use ``tornado.testing.main(verbosity=2)``
803 to show many test details as they are run.
804 See http://docs.python.org/library/unittest.html#unittest.main
805 for full argument list.
806
807 .. versionchanged:: 5.0
808
809 This function produces no output of its own; only that produced
810 by the `unittest` module (previously it would add a PASS or FAIL
811 log message).
812 """
813 from tornado.options import define, options, parse_command_line
814
815 define(
816 "exception_on_interrupt",
817 type=bool,
818 default=True,
819 help=(
820 "If true (default), ctrl-c raises a KeyboardInterrupt "
821 "exception. This prints a stack trace but cannot interrupt "
822 "certain operations. If false, the process is more reliably "
823 "killed, but does not print a stack trace."
824 ),
825 )
826
827 # support the same options as unittest's command-line interface
828 define("verbose", type=bool)
829 define("quiet", type=bool)
830 define("failfast", type=bool)
831 define("catch", type=bool)
832 define("buffer", type=bool)
833
834 argv = [sys.argv[0]] + parse_command_line(sys.argv)
835
836 if not options.exception_on_interrupt:
837 signal.signal(signal.SIGINT, signal.SIG_DFL)
838
839 if options.verbose is not None:
840 kwargs["verbosity"] = 2
841 if options.quiet is not None:
842 kwargs["verbosity"] = 0
843 if options.failfast is not None:
844 kwargs["failfast"] = True
845 if options.catch is not None:
846 kwargs["catchbreak"] = True
847 if options.buffer is not None:
848 kwargs["buffer"] = True
849
850 if __name__ == "__main__" and len(argv) == 1:
851 print("No tests specified", file=sys.stderr)
852 sys.exit(1)
853 # In order to be able to run tests by their fully-qualified name
854 # on the command line without importing all tests here,
855 # module must be set to None. Python 3.2's unittest.main ignores
856 # defaultTest if no module is given (it tries to do its own
857 # test discovery, which is incompatible with auto2to3), so don't
858 # set module if we're not asking for a specific test.
859 if len(argv) > 1:
860 unittest.main(module=None, argv=argv, **kwargs) # type: ignore
861 else:
862 unittest.main(defaultTest="all", argv=argv, **kwargs)
863
864
865if __name__ == "__main__":
866 main()
867