testing.py
Go to the documentation of this file.
00001 #!/usr/bin/env python
00002 """Support classes for automated testing.
00003 
00004 This module contains three parts:
00005 
00006 * `AsyncTestCase`/`AsyncHTTPTestCase`:  Subclasses of unittest.TestCase
00007   with additional support for testing asynchronous (IOLoop-based) code.
00008 
00009 * `LogTrapTestCase`:  Subclass of unittest.TestCase that discards log output
00010   from tests that pass and only produces output for failing tests.
00011 
00012 * `main()`: A simple test runner (wrapper around unittest.main()) with support
00013   for the tornado.autoreload module to rerun the tests when code changes.
00014 
00015 These components may be used together or independently.  In particular,
00016 it is safe to combine AsyncTestCase and LogTrapTestCase via multiple
00017 inheritance.  See the docstrings for each class/function below for more
00018 information.
00019 """
00020 
00021 from __future__ import absolute_import, division, with_statement
00022 
00023 from cStringIO import StringIO
00024 try:
00025     from tornado.httpclient import AsyncHTTPClient
00026     from tornado.httpserver import HTTPServer
00027     from tornado.ioloop import IOLoop
00028 except ImportError:
00029     # These modules are not importable on app engine.  Parts of this module
00030     # won't work, but e.g. LogTrapTestCase and main() will.
00031     AsyncHTTPClient = None
00032     HTTPServer = None
00033     IOLoop = None
00034 from tornado.stack_context import StackContext, NullContext
00035 from tornado.util import raise_exc_info
00036 import contextlib
00037 import logging
00038 import signal
00039 import sys
00040 import time
00041 import unittest
00042 
00043 _next_port = 10000
00044 
00045 
00046 def get_unused_port():
00047     """Returns a (hopefully) unused port number."""
00048     global _next_port
00049     port = _next_port
00050     _next_port = _next_port + 1
00051     return port
00052 
00053 
00054 class AsyncTestCase(unittest.TestCase):
00055     """TestCase subclass for testing IOLoop-based asynchronous code.
00056 
00057     The unittest framework is synchronous, so the test must be complete
00058     by the time the test method returns.  This method provides the stop()
00059     and wait() methods for this purpose.  The test method itself must call
00060     self.wait(), and asynchronous callbacks should call self.stop() to signal
00061     completion.
00062 
00063     By default, a new IOLoop is constructed for each test and is available
00064     as self.io_loop.  This IOLoop should be used in the construction of
00065     HTTP clients/servers, etc.  If the code being tested requires a
00066     global IOLoop, subclasses should override get_new_ioloop to return it.
00067 
00068     The IOLoop's start and stop methods should not be called directly.
00069     Instead, use self.stop self.wait.  Arguments passed to self.stop are
00070     returned from self.wait.  It is possible to have multiple
00071     wait/stop cycles in the same test.
00072 
00073     Example::
00074 
00075         # This test uses an asynchronous style similar to most async
00076         # application code.
00077         class MyTestCase(AsyncTestCase):
00078             def test_http_fetch(self):
00079                 client = AsyncHTTPClient(self.io_loop)
00080                 client.fetch("http://www.tornadoweb.org/", self.handle_fetch)
00081                 self.wait()
00082 
00083             def handle_fetch(self, response):
00084                 # Test contents of response (failures and exceptions here
00085                 # will cause self.wait() to throw an exception and end the
00086                 # test).
00087                 # Exceptions thrown here are magically propagated to
00088                 # self.wait() in test_http_fetch() via stack_context.
00089                 self.assertIn("FriendFeed", response.body)
00090                 self.stop()
00091 
00092         # This test uses the argument passing between self.stop and self.wait
00093         # for a simpler, more synchronous style.
00094         # This style is recommended over the preceding example because it
00095         # keeps the assertions in the test method itself, and is therefore
00096         # less sensitive to the subtleties of stack_context.
00097         class MyTestCase2(AsyncTestCase):
00098             def test_http_fetch(self):
00099                 client = AsyncHTTPClient(self.io_loop)
00100                 client.fetch("http://www.tornadoweb.org/", self.stop)
00101                 response = self.wait()
00102                 # Test contents of response
00103                 self.assertIn("FriendFeed", response.body)
00104     """
00105     def __init__(self, *args, **kwargs):
00106         super(AsyncTestCase, self).__init__(*args, **kwargs)
00107         self.__stopped = False
00108         self.__running = False
00109         self.__failure = None
00110         self.__stop_args = None
00111         self.__timeout = None
00112 
00113     def setUp(self):
00114         super(AsyncTestCase, self).setUp()
00115         self.io_loop = self.get_new_ioloop()
00116 
00117     def tearDown(self):
00118         if (not IOLoop.initialized() or
00119             self.io_loop is not IOLoop.instance()):
00120             # Try to clean up any file descriptors left open in the ioloop.
00121             # This avoids leaks, especially when tests are run repeatedly
00122             # in the same process with autoreload (because curl does not
00123             # set FD_CLOEXEC on its file descriptors)
00124             self.io_loop.close(all_fds=True)
00125         super(AsyncTestCase, self).tearDown()
00126 
00127     def get_new_ioloop(self):
00128         '''Creates a new IOLoop for this test.  May be overridden in
00129         subclasses for tests that require a specific IOLoop (usually
00130         the singleton).
00131         '''
00132         return IOLoop()
00133 
00134     @contextlib.contextmanager
00135     def _stack_context(self):
00136         try:
00137             yield
00138         except Exception:
00139             self.__failure = sys.exc_info()
00140             self.stop()
00141 
00142     def __rethrow(self):
00143         if self.__failure is not None:
00144             failure = self.__failure
00145             self.__failure = None
00146             raise_exc_info(failure)
00147 
00148     def run(self, result=None):
00149         with StackContext(self._stack_context):
00150             super(AsyncTestCase, self).run(result)
00151         # In case an exception escaped super.run or the StackContext caught
00152         # an exception when there wasn't a wait() to re-raise it, do so here.
00153         self.__rethrow()
00154 
00155     def stop(self, _arg=None, **kwargs):
00156         '''Stops the ioloop, causing one pending (or future) call to wait()
00157         to return.
00158 
00159         Keyword arguments or a single positional argument passed to stop() are
00160         saved and will be returned by wait().
00161         '''
00162         assert _arg is None or not kwargs
00163         self.__stop_args = kwargs or _arg
00164         if self.__running:
00165             self.io_loop.stop()
00166             self.__running = False
00167         self.__stopped = True
00168 
00169     def wait(self, condition=None, timeout=5):
00170         """Runs the IOLoop until stop is called or timeout has passed.
00171 
00172         In the event of a timeout, an exception will be thrown.
00173 
00174         If condition is not None, the IOLoop will be restarted after stop()
00175         until condition() returns true.
00176         """
00177         if not self.__stopped:
00178             if timeout:
00179                 def timeout_func():
00180                     try:
00181                         raise self.failureException(
00182                           'Async operation timed out after %s seconds' %
00183                           timeout)
00184                     except Exception:
00185                         self.__failure = sys.exc_info()
00186                     self.stop()
00187                 if self.__timeout is not None:
00188                     self.io_loop.remove_timeout(self.__timeout)
00189                 self.__timeout = self.io_loop.add_timeout(time.time() + timeout, timeout_func)
00190             while True:
00191                 self.__running = True
00192                 with NullContext():
00193                     # Wipe out the StackContext that was established in
00194                     # self.run() so that all callbacks executed inside the
00195                     # IOLoop will re-run it.
00196                     self.io_loop.start()
00197                 if (self.__failure is not None or
00198                     condition is None or condition()):
00199                     break
00200         assert self.__stopped
00201         self.__stopped = False
00202         self.__rethrow()
00203         result = self.__stop_args
00204         self.__stop_args = None
00205         return result
00206 
00207 
00208 class AsyncHTTPTestCase(AsyncTestCase):
00209     '''A test case that starts up an HTTP server.
00210 
00211     Subclasses must override get_app(), which returns the
00212     tornado.web.Application (or other HTTPServer callback) to be tested.
00213     Tests will typically use the provided self.http_client to fetch
00214     URLs from this server.
00215 
00216     Example::
00217 
00218         class MyHTTPTest(AsyncHTTPTestCase):
00219             def get_app(self):
00220                 return Application([('/', MyHandler)...])
00221 
00222             def test_homepage(self):
00223                 # The following two lines are equivalent to
00224                 #   response = self.fetch('/')
00225                 # but are shown in full here to demonstrate explicit use
00226                 # of self.stop and self.wait.
00227                 self.http_client.fetch(self.get_url('/'), self.stop)
00228                 response = self.wait()
00229                 # test contents of response
00230     '''
00231     def setUp(self):
00232         super(AsyncHTTPTestCase, self).setUp()
00233         self.__port = None
00234 
00235         self.http_client = AsyncHTTPClient(io_loop=self.io_loop)
00236         self._app = self.get_app()
00237         self.http_server = HTTPServer(self._app, io_loop=self.io_loop,
00238                                       **self.get_httpserver_options())
00239         self.http_server.listen(self.get_http_port(), address="127.0.0.1")
00240 
00241     def get_app(self):
00242         """Should be overridden by subclasses to return a
00243         tornado.web.Application or other HTTPServer callback.
00244         """
00245         raise NotImplementedError()
00246 
00247     def fetch(self, path, **kwargs):
00248         """Convenience method to synchronously fetch a url.
00249 
00250         The given path will be appended to the local server's host and port.
00251         Any additional kwargs will be passed directly to
00252         AsyncHTTPClient.fetch (and so could be used to pass method="POST",
00253         body="...", etc).
00254         """
00255         self.http_client.fetch(self.get_url(path), self.stop, **kwargs)
00256         return self.wait()
00257 
00258     def get_httpserver_options(self):
00259         """May be overridden by subclasses to return additional
00260         keyword arguments for HTTPServer.
00261         """
00262         return {}
00263 
00264     def get_http_port(self):
00265         """Returns the port used by the HTTPServer.
00266 
00267         A new port is chosen for each test.
00268         """
00269         if self.__port is None:
00270             self.__port = get_unused_port()
00271         return self.__port
00272 
00273     def get_url(self, path):
00274         """Returns an absolute url for the given path on the test server."""
00275         return 'http://localhost:%s%s' % (self.get_http_port(), path)
00276 
00277     def tearDown(self):
00278         self.http_server.stop()
00279         self.http_client.close()
00280         super(AsyncHTTPTestCase, self).tearDown()
00281 
00282 
00283 class LogTrapTestCase(unittest.TestCase):
00284     """A test case that captures and discards all logging output
00285     if the test passes.
00286 
00287     Some libraries can produce a lot of logging output even when
00288     the test succeeds, so this class can be useful to minimize the noise.
00289     Simply use it as a base class for your test case.  It is safe to combine
00290     with AsyncTestCase via multiple inheritance
00291     ("class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):")
00292 
00293     This class assumes that only one log handler is configured and that
00294     it is a StreamHandler.  This is true for both logging.basicConfig
00295     and the "pretty logging" configured by tornado.options.
00296     """
00297     def run(self, result=None):
00298         logger = logging.getLogger()
00299         if len(logger.handlers) > 1:
00300             # Multiple handlers have been defined.  It gets messy to handle
00301             # this, especially since the handlers may have different
00302             # formatters.  Just leave the logging alone in this case.
00303             super(LogTrapTestCase, self).run(result)
00304             return
00305         if not logger.handlers:
00306             logging.basicConfig()
00307         self.assertEqual(len(logger.handlers), 1)
00308         handler = logger.handlers[0]
00309         assert isinstance(handler, logging.StreamHandler)
00310         old_stream = handler.stream
00311         try:
00312             handler.stream = StringIO()
00313             logging.info("RUNNING TEST: " + str(self))
00314             old_error_count = len(result.failures) + len(result.errors)
00315             super(LogTrapTestCase, self).run(result)
00316             new_error_count = len(result.failures) + len(result.errors)
00317             if new_error_count != old_error_count:
00318                 old_stream.write(handler.stream.getvalue())
00319         finally:
00320             handler.stream = old_stream
00321 
00322 
00323 def main():
00324     """A simple test runner.
00325 
00326     This test runner is essentially equivalent to `unittest.main` from
00327     the standard library, but adds support for tornado-style option
00328     parsing and log formatting.
00329 
00330     The easiest way to run a test is via the command line::
00331 
00332         python -m tornado.testing tornado.test.stack_context_test
00333 
00334     See the standard library unittest module for ways in which tests can
00335     be specified.
00336 
00337     Projects with many tests may wish to define a test script like
00338     tornado/test/runtests.py.  This script should define a method all()
00339     which returns a test suite and then call tornado.testing.main().
00340     Note that even when a test script is used, the all() test suite may
00341     be overridden by naming a single test on the command line::
00342 
00343         # Runs all tests
00344         tornado/test/runtests.py
00345         # Runs one test
00346         tornado/test/runtests.py tornado.test.stack_context_test
00347 
00348     """
00349     from tornado.options import define, options, parse_command_line
00350 
00351     define('autoreload', type=bool, default=False,
00352            help="DEPRECATED: use tornado.autoreload.main instead")
00353     define('httpclient', type=str, default=None)
00354     define('exception_on_interrupt', type=bool, default=True,
00355            help=("If true (default), ctrl-c raises a KeyboardInterrupt "
00356                  "exception.  This prints a stack trace but cannot interrupt "
00357                  "certain operations.  If false, the process is more reliably "
00358                  "killed, but does not print a stack trace."))
00359     argv = [sys.argv[0]] + parse_command_line(sys.argv)
00360 
00361     if options.httpclient:
00362         from tornado.httpclient import AsyncHTTPClient
00363         AsyncHTTPClient.configure(options.httpclient)
00364 
00365     if not options.exception_on_interrupt:
00366         signal.signal(signal.SIGINT, signal.SIG_DFL)
00367 
00368     if __name__ == '__main__' and len(argv) == 1:
00369         print >> sys.stderr, "No tests specified"
00370         sys.exit(1)
00371     try:
00372         # In order to be able to run tests by their fully-qualified name
00373         # on the command line without importing all tests here,
00374         # module must be set to None.  Python 3.2's unittest.main ignores
00375         # defaultTest if no module is given (it tries to do its own
00376         # test discovery, which is incompatible with auto2to3), so don't
00377         # set module if we're not asking for a specific test.
00378         if len(argv) > 1:
00379             unittest.main(module=None, argv=argv)
00380         else:
00381             unittest.main(defaultTest="all", argv=argv)
00382     except SystemExit, e:
00383         if e.code == 0:
00384             logging.info('PASS')
00385         else:
00386             logging.error('FAIL')
00387         if not options.autoreload:
00388             raise
00389     if options.autoreload:
00390         import tornado.autoreload
00391         tornado.autoreload.wait()
00392 
00393 if __name__ == '__main__':
00394     main()


rosbridge_server
Author(s): Jonathan Mace
autogenerated on Thu Jan 2 2014 11:53:55