text: """Support classes for automated testing.
  
  * `AsyncTestCase` and `AsyncHTTPTestCase`:  Subclasses of unittest.TestCase
    with additional support for testing asynchronous (`.IOLoop`-based) code.
  
  * `ExpectLog`: Make test logs less spammy.
  
  * `main()`: A simple test runner (wrapper around unittest.main()) with support
    for the tornado.autoreload module to rerun the tests when code changes.
  """
  
  from __future__ import absolute_import, division, print_function
  
  try:
      from tornado import gen
      from tornado.httpclient import AsyncHTTPClient
      from tornado.httpserver import HTTPServer
      from tornado.simple_httpclient import SimpleAsyncHTTPClient
      from tornado.ioloop import IOLoop, TimeoutError
      from tornado import netutil
      from tornado.process import Subprocess
  except ImportError:
      # These modules are not importable on app engine.  Parts of this module
      # won't work, but e.g. main() will.
      AsyncHTTPClient = None  # type: ignore
      gen = None  # type: ignore
      HTTPServer = None  # type: ignore
      IOLoop = None  # type: ignore
      netutil = None  # type: ignore
      SimpleAsyncHTTPClient = None  # type: ignore
      Subprocess = None  # type: ignore
  from tornado.log import app_log
  from tornado.stack_context import ExceptionStackContext
  from tornado.util import raise_exc_info, basestring_type, PY3
  import functools
  import inspect
  import logging
  import os
  import re
  import signal
  import socket
  import sys
  
  try:
      import asyncio
  except ImportError:
      asyncio = None
  
  
  try:
      from collections.abc import Generator as GeneratorType  # type: ignore
  except ImportError:
      from types import GeneratorType  # type: ignore
  
  if sys.version_info >= (3, 5):
      iscoroutine = inspect.iscoroutine  # type: ignore
      iscoroutinefunction = inspect.iscoroutinefunction  # type: ignore
  else:
      iscoroutine = iscoroutinefunction = lambda f: False
  
  # Tornado's own test suite requires the updated unittest module
  # (either py27+ or unittest2) so tornado.test.util enforces
  # this requirement, but for other users of tornado.testing we want
  # to allow the older version if unitest2 is not available.
  if PY3:
      # On python 3, mixing unittest2 and unittest (including doctest)
      # doesn't seem to work, so always use unittest.
      import unittest
  else:
      # On python 2, prefer unittest2 when available.
      try:
          import unittest2 as unittest  # type: ignore
      except ImportError:
          import unittest  # type: ignore
  
  
  def bind_unused_port(reuse_port=False):
      """Binds a server socket to an available port on localhost.
  
      Returns a tuple (socket, port).
  
      .. versionchanged:: 4.4
         Always binds to ``127.0.0.1`` without resolving the name
         ``localhost``.
      """
      sock = netutil.bind_sockets(None, '127.0.0.1', family=socket.AF_INET,
                                  reuse_port=reuse_port)[0]
      port = sock.getsockname()[1]
      return sock, port
  
  
  def get_async_test_timeout():
      """Get the global timeout setting for async tests.
  
      Returns a float, the timeout in seconds.
  
      .. versionadded:: 3.1
      """
      try:
          return float(os.environ.get('ASYNC_TEST_TIMEOUT'))
      except (ValueError, TypeError):
          return 5
  
  
  class _TestMethodWrapper(object):
      """Wraps a test method to raise an error if it returns a value.
  
      This is mainly used to detect undecorated generators (if a test
      method yields it must use a decorator to consume the generator),
      but will also detect other kinds of return values (these are not
      necessarily errors, but we alert anyway since there is no good
      reason to return a value from a test).
      """
      def __init__(self, orig_method):
          self.orig_method = orig_method
  
      def __call__(self, *args, **kwargs):
          result = self.orig_method(*args, **kwargs)
          if isinstance(result, GeneratorType) or iscoroutine(result):
              raise TypeError("Generator and coroutine test methods should be"
                              " decorated with tornado.testing.gen_test")
          elif result is not None:
              raise ValueError("Return value from test method ignored: %r" %
                               result)
  
      def __getattr__(self, name):
          """Proxy all unknown attributes to the original method.
  
          This is important for some of the decorators in the `unittest`
          module, such as `unittest.skipIf`.
          """
          return getattr(self.orig_method, name)
  
  
  class AsyncTestCase(unittest.TestCase):
      """`~unittest.TestCase` subclass for testing `.IOLoop`-based
      asynchronous code.
  
      The unittest framework is synchronous, so the test must be
      complete by the time the test method returns.  This means that
      asynchronous code cannot be used in quite the same way as usual.
      To write test functions that use the same ``yield``-based patterns
      used with the `tornado.gen` module, decorate your test methods
      with `tornado.testing.gen_test` instead of
      `tornado.gen.coroutine`.  This class also provides the `stop()`
      and `wait()` methods for a more manual style of testing.  The test
      method itself must call ``self.wait()``, and asynchronous
      callbacks should call ``self.stop()`` to signal completion.
  
      By default, a new `.IOLoop` is constructed for each test and is available
      as ``self.io_loop``.  If the code being tested requires a
      global `.IOLoop`, subclasses should override `get_new_ioloop` to return it.
  
      The `.IOLoop`'s ``start`` and ``stop`` methods should not be
      called directly.  Instead, use `self.stop <stop>` and `self.wait
      <wait>`.  Arguments passed to ``self.stop`` are returned from
      ``self.wait``.  It is possible to have multiple ``wait``/``stop``
      cycles in the same test.
  
      Example::
  
          # This test uses coroutine style.
          class MyTestCase(AsyncTestCase):
              @tornado.testing.gen_test
              def test_http_fetch(self):
                  client = AsyncHTTPClient()
                  response = yield client.fetch("http://www.tornadoweb.org")
                  # Test contents of response
                  self.assertIn("FriendFeed", response.body)
  
          # This test uses argument passing between self.stop and self.wait.
          class MyTestCase2(AsyncTestCase):
              def test_http_fetch(self):
                  client = AsyncHTTPClient()
                  client.fetch("http://www.tornadoweb.org/", self.stop)
                  response = self.wait()
                  # Test contents of response
                  self.assertIn("FriendFeed", response.body)
  
          # This test uses an explicit callback-based style.
          class MyTestCase3(AsyncTestCase):
              def test_http_fetch(self):
                  client = AsyncHTTPClient()
                  client.fetch("http://www.tornadoweb.org/", self.handle_fetch)
                  self.wait()
  
              def handle_fetch(self, response):
                  # Test contents of response (failures and exceptions here
                  # will cause self.wait() to throw an exception and end the
                  # test).
                  # Exceptions thrown here are magically propagated to
                  # self.wait() in test_http_fetch() via stack_context.
                  self.assertIn("FriendFeed", response.body)
                  self.stop()
      """
      def __init__(self, methodName='runTest'):
          super(AsyncTestCase, self).__init__(methodName)
          self.__stopped = False
          self.__running = False
          self.__failure = None
          self.__stop_args = None
          self.__timeout = None
  
          # It's easy to forget the @gen_test decorator, but if you do
          # the test will silently be ignored because nothing will consume
          # the generator.  Replace the test method with a wrapper that will
          # make sure it's not an undecorated generator.
          setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName)))
  
      def setUp(self):
          super(AsyncTestCase, self).setUp()
          self.io_loop = self.get_new_ioloop()
          self.io_loop.make_current()
  
      def tearDown(self):
          # Clean up Subprocess, so it can be used again with a new ioloop.
          Subprocess.uninitialize()
          self.io_loop.clear_current()
          # Try to clean up any file descriptors left open in the ioloop.
          # This avoids leaks, especially when tests are run repeatedly
          # in the same process with autoreload (because curl does not
          # set FD_CLOEXEC on its file descriptors)
          self.io_loop.close(all_fds=True)
          super(AsyncTestCase, self).tearDown()
          # In case an exception escaped or the StackContext caught an exception
          # when there wasn't a wait() to re-raise it, do so here.
          # This is our last chance to raise an exception in a way that the
          # unittest machinery understands.
          self.__rethrow()
  
      def get_new_ioloop(self):
          """Creates a new `.IOLoop` for this test.  May be overridden in
          subclasses for tests that require a specific `.IOLoop` (usually
          the singleton `.IOLoop.instance()`).
          """
          return IOLoop()
  
      def _handle_exception(self, typ, value, tb):
          if self.__failure is None:
              self.__failure = (typ, value, tb)
          else:
              app_log.error("multiple unhandled exceptions in test",
                            exc_info=(typ, value, tb))
          self.stop()
          return True
  
      def __rethrow(self):
          if self.__failure is not None:
              failure = self.__failure
              self.__failure = None
              raise_exc_info(failure)
  
      def run(self, result=None):
          with ExceptionStackContext(self._handle_exception):
              super(AsyncTestCase, self).run(result)
          # As a last resort, if an exception escaped super.run() and wasn't
          # re-raised in tearDown, raise it here.  This will cause the
          # unittest run to fail messily, but that's better than silently
          # ignoring an error.
          self.__rethrow()
  
      def stop(self, _arg=None, **kwargs):
          """Stops the `.IOLoop`, causing one pending (or future) call to `wait()`
          to return.
  
          Keyword arguments or a single positional argument passed to `stop()` are
          saved and will be returned by `wait()`.
          """
          assert _arg is None or not kwargs
          self.__stop_args = kwargs or _arg
          if self.__running:
              self.io_loop.stop()
              self.__running = False
          self.__stopped = True
  
      def wait(self, condition=None, timeout=None):
          """Runs the `.IOLoop` until stop is called or timeout has passed.
  
          In the event of a timeout, an exception will be thrown. The
          default timeout is 5 seconds; it may be overridden with a
          ``timeout`` keyword argument or globally with the
          ``ASYNC_TEST_TIMEOUT`` environment variable.
  
          If ``condition`` is not None, the `.IOLoop` will be restarted
          after `stop()` until ``condition()`` returns true.
  
          .. versionchanged:: 3.1
             Added the ``ASYNC_TEST_TIMEOUT`` environment variable.
          """
          if timeout is None:
              timeout = get_async_test_timeout()
  
          if not self.__stopped:
              if timeout:
                  def timeout_func():
                      try:
                          raise self.failureException(
                              'Async operation timed out after %s seconds' %
                              timeout)
                      except Exception:
                          self.__failure = sys.exc_info()
                      self.stop()
                  self.__timeout = self.io_loop.add_timeout(self.io_loop.time() + timeout,
                                                            timeout_func)
              while True:
                  self.__running = True
                  self.io_loop.start()
                  if (self.__failure is not None or
                          condition is None or condition()):
                      break
              if self.__timeout is not None:
                  self.io_loop.remove_timeout(self.__timeout)
                  self.__timeout = None
          assert self.__stopped
          self.__stopped = False
          self.__rethrow()
          result = self.__stop_args
          self.__stop_args = None
          return result
  
  
  class AsyncHTTPTestCase(AsyncTestCase):
      """A test case that starts up an HTTP server.
  
      Subclasses must override `get_app()`, which returns the
      `tornado.web.Application` (or other `.HTTPServer` callback) to be tested.
      Tests will typically use the provided ``self.http_client`` to fetch
      URLs from this server.
  
      Example, assuming the "Hello, world" example from the user guide is in
      ``hello.py``::
  
          import hello
  
          class TestHelloApp(AsyncHTTPTestCase):
              def get_app(self):
                  return hello.make_app()
  
              def test_homepage(self):
                  response = self.fetch('/')
                  self.assertEqual(response.code, 200)
                  self.assertEqual(response.body, 'Hello, world')
  
      That call to ``self.fetch()`` is equivalent to ::
  
          self.http_client.fetch(self.get_url('/'), self.stop)
          response = self.wait()
  
      which illustrates how AsyncTestCase can turn an asynchronous operation,
      like ``http_client.fetch()``, into a synchronous operation. If you need
      to do other asynchronous operations in tests, you'll probably need to use
      ``stop()`` and ``wait()`` yourself.
      """
      def setUp(self):
          super(AsyncHTTPTestCase, self).setUp()
          sock, port = bind_unused_port()
          self.__port = port
  
          self.http_client = self.get_http_client()
          self._app = self.get_app()
          self.http_server = self.get_http_server()
          self.http_server.add_sockets([sock])
  
      def get_http_client(self):
          return AsyncHTTPClient()
  
      def get_http_server(self):
          return HTTPServer(self._app, **self.get_httpserver_options())
  
      def get_app(self):
          """Should be overridden by subclasses to return a
          `tornado.web.Application` or other `.HTTPServer` callback.
          """
          raise NotImplementedError()
  
      def fetch(self, path, **kwargs):
          """Convenience method to synchronously fetch a URL.
  
          The given path will be appended to the local server's host and
          port.  Any additional kwargs will be passed directly to
          `.AsyncHTTPClient.fetch` (and so could be used to pass
          ``method="POST"``, ``body="..."``, etc).
  
          If the path begins with http:// or https://, it will be treated as a
          full URL and will be fetched as-is.
  
          .. versionchanged:: 5.0
             Added support for absolute URLs.
          """
          if path.lower().startswith(('http://', 'https://')):
              self.http_client.fetch(path, self.stop, **kwargs)
          else:
              self.http_client.fetch(self.get_url(path), self.stop, **kwargs)
          return self.wait()
  
      def get_httpserver_options(self):
          """May be overridden by subclasses to return additional
          keyword arguments for the server.
          """
          return {}
  
      def get_http_port(self):
          """Returns the port used by the server.
  
          A new port is chosen for each test.
          """
          return self.__port
  
      def get_protocol(self):
          return 'http'
  
      def get_url(self, path):
          """Returns an absolute url for the given path on the test server."""
          return '%s://127.0.0.1:%s%s' % (self.get_protocol(),
                                          self.get_http_port(), path)
  
      def tearDown(self):
          self.http_server.stop()
          self.io_loop.run_sync(self.http_server.close_all_connections,
                                timeout=get_async_test_timeout())
          self.http_client.close()
          super(AsyncHTTPTestCase, self).tearDown()
  
  
  class AsyncHTTPSTestCase(AsyncHTTPTestCase):
      """A test case that starts an HTTPS server.
  
      Interface is generally the same as `AsyncHTTPTestCase`.
      """
      def get_http_client(self):
          return AsyncHTTPClient(force_instance=True,
                                 defaults=dict(validate_cert=False))
  
      def get_httpserver_options(self):
          return dict(ssl_options=self.get_ssl_options())
  
      def get_ssl_options(self):
          """May be overridden by subclasses to select SSL options.
  
          By default includes a self-signed testing certificate.
          """
          # Testing keys were generated with:
          # openssl req -new -keyout tornado/test/test.key \
          #                     -out tornado/test/test.crt -nodes -days 3650 -x509
          module_dir = os.path.dirname(__file__)
          return dict(
              certfile=os.path.join(module_dir, 'test', 'test.crt'),
              keyfile=os.path.join(module_dir, 'test', 'test.key'))
  
      def get_protocol(self):
          return 'https'
  
  
  def gen_test(func=None, timeout=None):
      """Testing equivalent of ``@gen.coroutine``, to be applied to test methods.
  
      ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not
      already running.  ``@gen_test`` should be applied to test methods
      on subclasses of `AsyncTestCase`.
  
      Example::
  
          class MyTest(AsyncHTTPTestCase):
              @gen_test
              def test_something(self):
                  response = yield self.http_client.fetch(self.get_url('/'))
  
      By default, ``@gen_test`` times out after 5 seconds. The timeout may be
      overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable,
      or for each test with the ``timeout`` keyword argument::
  
          class MyTest(AsyncHTTPTestCase):
              @gen_test(timeout=10)
              def test_something_slow(self):
                  response = yield self.http_client.fetch(self.get_url('/'))
  
      Note that ``@gen_test`` is incompatible with `AsyncTestCase.stop`,
      `AsyncTestCase.wait`, and `AsyncHTTPTestCase.fetch`. Use ``yield
      self.http_client.fetch(self.get_url())`` as shown above instead.
  
      .. versionadded:: 3.1
         The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment
         variable.
  
      .. versionchanged:: 4.0
         The wrapper now passes along ``*args, **kwargs`` so it can be used
         on functions with arguments.
  
      """
      if timeout is None:
          timeout = get_async_test_timeout()
  
      def wrap(f):
          # Stack up several decorators to allow us to access the generator
          # object itself.  In the innermost wrapper, we capture the generator
          # and save it in an attribute of self.  Next, we run the wrapped
          # function through @gen.coroutine.  Finally, the coroutine is
          # wrapped again to make it synchronous with run_sync.
          #
          # This is a good case study arguing for either some sort of
          # extensibility in the gen decorators or cancellation support.
          @functools.wraps(f)
          def pre_coroutine(self, *args, **kwargs):
              result = f(self, *args, **kwargs)
              if isinstance(result, GeneratorType) or iscoroutine(result):
                  self._test_generator = result
              else:
                  self._test_generator = None
              return result
  
          if iscoroutinefunction(f):
              coro = pre_coroutine
          else:
              coro = gen.coroutine(pre_coroutine)
  
          @functools.wraps(coro)
          def post_coroutine(self, *args, **kwargs):
              try:
                  return self.io_loop.run_sync(
                      functools.partial(coro, self, *args, **kwargs),
                      timeout=timeout)
              except TimeoutError as e:
                  # run_sync raises an error with an unhelpful traceback.
                  # If the underlying generator is still running, we can throw the
                  # exception back into it so the stack trace is replaced by the
                  # point where the test is stopped. The only reason the generator
                  # would not be running would be if it were cancelled, which means
                  # a native coroutine, so we can rely on the cr_running attribute.
                  if getattr(self._test_generator, 'cr_running', True):
                      self._test_generator.throw(e)
                      # In case the test contains an overly broad except
                      # clause, we may get back here.
                  # Coroutine was stopped or didn't raise a useful stack trace,
                  # so re-raise the original exception which is better than nothing.
                  raise
          return post_coroutine
  
      if func is not None:
          # Used like:
          #     @gen_test
          #     def f(self):
          #         pass
          return wrap(func)
      else:
          # Used like @gen_test(timeout=10)
          return wrap
  
  
  # Without this attribute, nosetests will try to run gen_test as a test
  # anywhere it is imported.
  gen_test.__test__ = False  # type: ignore
  
  
  class ExpectLog(logging.Filter):
      """Context manager to capture and suppress expected log output.
  
      Useful to make tests of error conditions less noisy, while still
      leaving unexpected log entries visible.  *Not thread safe.*
  
      The attribute ``logged_stack`` is set to true if any exception
      stack trace was logged.
  
      Usage::
  
          with ExpectLog('tornado.application', "Uncaught exception"):
              error_response = self.fetch("/some_page")
  
      .. versionchanged:: 4.3
         Added the ``logged_stack`` attribute.
      """
      def __init__(self, logger, regex, required=True):
          """Constructs an ExpectLog context manager.
  
          :param logger: Logger object (or name of logger) to watch.  Pass
              an empty string to watch the root logger.
          :param regex: Regular expression to match.  Any log entries on
              the specified logger that match this regex will be suppressed.
          :param required: If true, an exception will be raised if the end of
              the ``with`` statement is reached without matching any log entries.
          """
          if isinstance(logger, basestring_type):
              logger = logging.getLogger(logger)
          self.logger = logger
          self.regex = re.compile(regex)
          self.required = required
          self.matched = False
          self.logged_stack = False
  
      def filter(self, record):
          if record.exc_info:
              self.logged_stack = True
          message = record.getMessage()
          if self.regex.match(message):
              self.matched = True
              return False
          return True
  
      def __enter__(self):
          self.logger.addFilter(self)
          return self
  
      def __exit__(self, typ, value, tb):
          self.logger.removeFilter(self)
          if not typ and self.required and not self.matched:
              raise Exception("did not get expected log message")
  
  
  def main(**kwargs):
      """A simple test runner.
  
      This test runner is essentially equivalent to `unittest.main` from
      the standard library, but adds support for tornado-style option
      parsing and log formatting. It is *not* necessary to use this
      `main` function to run tests using `AsyncTestCase`; these tests
      are self-contained and can run with any test runner.
  
      The easiest way to run a test is via the command line::
  
          python -m tornado.testing tornado.test.stack_context_test
  
      See the standard library unittest module for ways in which tests can
      be specified.
  
      Projects with many tests may wish to define a test script like
      ``tornado/test/runtests.py``.  This script should define a method
      ``all()`` which returns a test suite and then call
      `tornado.testing.main()`.  Note that even when a test script is
      used, the ``all()`` test suite may be overridden by naming a
      single test on the command line::
  
          # Runs all tests
          python -m tornado.test.runtests
          # Runs one test
          python -m tornado.test.runtests tornado.test.stack_context_test
  
      Additional keyword arguments passed through to ``unittest.main()``.
      For example, use ``tornado.testing.main(verbosity=2)``
      to show many test details as they are run.
      See http://docs.python.org/library/unittest.html#unittest.main
      for full argument list.
  
      .. versionchanged:: 5.0
  
         This function produces no output of its own; only that produced
         by the `unittest` module (Previously it would add a PASS or FAIL
         log message).
      """
      from tornado.options import define, options, parse_command_line
  
      define('exception_on_interrupt', type=bool, default=True,
             help=("If true (default), ctrl-c raises a KeyboardInterrupt "
                   "exception.  This prints a stack trace but cannot interrupt "
                   "certain operations.  If false, the process is more reliably "
                   "killed, but does not print a stack trace."))
  
      # support the same options as unittest's command-line interface
      define('verbose', type=bool)
      define('quiet', type=bool)
      define('failfast', type=bool)
      define('catch', type=bool)
      define('buffer', type=bool)
  
      argv = [sys.argv[0]] + parse_command_line(sys.argv)
  
      if not options.exception_on_interrupt:
          signal.signal(signal.SIGINT, signal.SIG_DFL)
  
      if options.verbose is not None:
          kwargs['verbosity'] = 2
      if options.quiet is not None:
          kwargs['verbosity'] = 0
      if options.failfast is not None:
          kwargs['failfast'] = True
      if options.catch is not None:
          kwargs['catchbreak'] = True
      if options.buffer is not None:
          kwargs['buffer'] = True
  
      if __name__ == '__main__' and len(argv) == 1:
          print("No tests specified", file=sys.stderr)
          sys.exit(1)
      # In order to be able to run tests by their fully-qualified name
      # on the command line without importing all tests here,
      # module must be set to None.  Python 3.2's unittest.main ignores
      # defaultTest if no module is given (it tries to do its own
      # test discovery, which is incompatible with auto2to3), so don't
      # set module if we're not asking for a specific test.
      if len(argv) > 1:
          unittest.main(module=None, argv=argv, **kwargs)
      else:
          unittest.main(defaultTest="all", argv=argv, **kwargs)
  
  
  if __name__ == '__main__':
      main()
  