1#!/usr/bin/env python 2"""Support classes for automated testing. 3 4* `AsyncTestCase` and `AsyncHTTPTestCase`: Subclasses of unittest.TestCase 5 with additional support for testing asynchronous (`.IOLoop`-based) code. 6 7* `ExpectLog` and `LogTrapTestCase`: Make test logs less spammy. 8 9* `main()`: A simple test runner (wrapper around unittest.main()) with support 10 for the tornado.autoreload module to rerun the tests when code changes. 11""" 12 13from __future__ import absolute_import, division, print_function 14 15try: 16 from tornado import gen 17 from tornado.httpclient import AsyncHTTPClient 18 from tornado.httpserver import HTTPServer 19 from tornado.simple_httpclient import SimpleAsyncHTTPClient 20 from tornado.ioloop import IOLoop, TimeoutError 21 from tornado import netutil 22 from tornado.process import Subprocess 23except ImportError: 24 # These modules are not importable on app engine. Parts of this module 25 # won't work, but e.g. LogTrapTestCase and main() will. 26 AsyncHTTPClient = None # type: ignore 27 gen = None # type: ignore 28 HTTPServer = None # type: ignore 29 IOLoop = None # type: ignore 30 netutil = None # type: ignore 31 SimpleAsyncHTTPClient = None # type: ignore 32 Subprocess = None # type: ignore 33from tornado.log import gen_log, app_log 34from tornado.stack_context import ExceptionStackContext 35from tornado.util import raise_exc_info, basestring_type, PY3 36import functools 37import inspect 38import logging 39import os 40import re 41import signal 42import socket 43import sys 44 45if PY3: 46 from io import StringIO 47else: 48 from cStringIO import StringIO 49 50try: 51 from collections.abc import Generator as GeneratorType # type: ignore 52except ImportError: 53 from types import GeneratorType # type: ignore 54 55if sys.version_info >= (3, 5): 56 iscoroutine = inspect.iscoroutine # type: ignore 57 iscoroutinefunction = inspect.iscoroutinefunction # type: ignore 58else: 59 iscoroutine = iscoroutinefunction = lambda f: False 60 61# Tornado's own test suite requires the updated unittest module 62# (either py27+ or unittest2) so tornado.test.util enforces 63# this requirement, but for other users of tornado.testing we want 64# to allow the older version if unitest2 is not available. 65if PY3: 66 # On python 3, mixing unittest2 and unittest (including doctest) 67 # doesn't seem to work, so always use unittest. 68 import unittest 69else: 70 # On python 2, prefer unittest2 when available. 71 try: 72 import unittest2 as unittest # type: ignore 73 except ImportError: 74 import unittest # type: ignore 75 76_next_port = 10000 77 78 79def get_unused_port(): 80 """Returns a (hopefully) unused port number. 81 82 This function does not guarantee that the port it returns is available, 83 only that a series of get_unused_port calls in a single process return 84 distinct ports. 85 86 .. deprecated:: 87 Use bind_unused_port instead, which is guaranteed to find an unused port. 88 """ 89 global _next_port 90 port = _next_port 91 _next_port = _next_port + 1 92 return port 93 94 95def bind_unused_port(reuse_port=False): 96 """Binds a server socket to an available port on localhost. 97 98 Returns a tuple (socket, port). 99 100 .. versionchanged:: 4.4 101 Always binds to ``127.0.0.1`` without resolving the name 102 ``localhost``. 103 """ 104 sock = netutil.bind_sockets(None, '127.0.0.1', family=socket.AF_INET, 105 reuse_port=reuse_port)[0] 106 port = sock.getsockname()[1] 107 return sock, port 108 109 110def get_async_test_timeout(): 111 """Get the global timeout setting for async tests. 112 113 Returns a float, the timeout in seconds. 114 115 .. versionadded:: 3.1 116 """ 117 try: 118 return float(os.environ.get('ASYNC_TEST_TIMEOUT')) 119 except (ValueError, TypeError): 120 return 5 121 122 123class _TestMethodWrapper(object): 124 """Wraps a test method to raise an error if it returns a value. 125 126 This is mainly used to detect undecorated generators (if a test 127 method yields it must use a decorator to consume the generator), 128 but will also detect other kinds of return values (these are not 129 necessarily errors, but we alert anyway since there is no good 130 reason to return a value from a test). 131 """ 132 def __init__(self, orig_method): 133 self.orig_method = orig_method 134 135 def __call__(self, *args, **kwargs): 136 result = self.orig_method(*args, **kwargs) 137 if isinstance(result, GeneratorType) or iscoroutine(result): 138 raise TypeError("Generator and coroutine test methods should be" 139 " decorated with tornado.testing.gen_test") 140 elif result is not None: 141 raise ValueError("Return value from test method ignored: %r" % 142 result) 143 144 def __getattr__(self, name): 145 """Proxy all unknown attributes to the original method. 146 147 This is important for some of the decorators in the `unittest` 148 module, such as `unittest.skipIf`. 149 """ 150 return getattr(self.orig_method, name) 151 152 153class AsyncTestCase(unittest.TestCase): 154 """`~unittest.TestCase` subclass for testing `.IOLoop`-based 155 asynchronous code. 156 157 The unittest framework is synchronous, so the test must be 158 complete by the time the test method returns. This means that 159 asynchronous code cannot be used in quite the same way as usual. 160 To write test functions that use the same ``yield``-based patterns 161 used with the `tornado.gen` module, decorate your test methods 162 with `tornado.testing.gen_test` instead of 163 `tornado.gen.coroutine`. This class also provides the `stop()` 164 and `wait()` methods for a more manual style of testing. The test 165 method itself must call ``self.wait()``, and asynchronous 166 callbacks should call ``self.stop()`` to signal completion. 167 168 By default, a new `.IOLoop` is constructed for each test and is available 169 as ``self.io_loop``. This `.IOLoop` should be used in the construction of 170 HTTP clients/servers, etc. If the code being tested requires a 171 global `.IOLoop`, subclasses should override `get_new_ioloop` to return it. 172 173 The `.IOLoop`'s ``start`` and ``stop`` methods should not be 174 called directly. Instead, use `self.stop <stop>` and `self.wait 175 <wait>`. Arguments passed to ``self.stop`` are returned from 176 ``self.wait``. It is possible to have multiple ``wait``/``stop`` 177 cycles in the same test. 178 179 Example:: 180 181 # This test uses coroutine style. 182 class MyTestCase(AsyncTestCase): 183 @tornado.testing.gen_test 184 def test_http_fetch(self): 185 client = AsyncHTTPClient(self.io_loop) 186 response = yield client.fetch("http://www.tornadoweb.org") 187 # Test contents of response 188 self.assertIn("FriendFeed", response.body) 189 190 # This test uses argument passing between self.stop and self.wait. 191 class MyTestCase2(AsyncTestCase): 192 def test_http_fetch(self): 193 client = AsyncHTTPClient(self.io_loop) 194 client.fetch("http://www.tornadoweb.org/", self.stop) 195 response = self.wait() 196 # Test contents of response 197 self.assertIn("FriendFeed", response.body) 198 199 # This test uses an explicit callback-based style. 200 class MyTestCase3(AsyncTestCase): 201 def test_http_fetch(self): 202 client = AsyncHTTPClient(self.io_loop) 203 client.fetch("http://www.tornadoweb.org/", self.handle_fetch) 204 self.wait() 205 206 def handle_fetch(self, response): 207 # Test contents of response (failures and exceptions here 208 # will cause self.wait() to throw an exception and end the 209 # test). 210 # Exceptions thrown here are magically propagated to 211 # self.wait() in test_http_fetch() via stack_context. 212 self.assertIn("FriendFeed", response.body) 213 self.stop() 214 """ 215 def __init__(self, methodName='runTest'): 216 super(AsyncTestCase, self).__init__(methodName) 217 self.__stopped = False 218 self.__running = False 219 self.__failure = None 220 self.__stop_args = None 221 self.__timeout = None 222 223 # It's easy to forget the @gen_test decorator, but if you do 224 # the test will silently be ignored because nothing will consume 225 # the generator. Replace the test method with a wrapper that will 226 # make sure it's not an undecorated generator. 227 setattr(self, methodName, _TestMethodWrapper(getattr(self, methodName))) 228 229 def setUp(self): 230 super(AsyncTestCase, self).setUp() 231 self.io_loop = self.get_new_ioloop() 232 self.io_loop.make_current() 233 234 def tearDown(self): 235 # Clean up Subprocess, so it can be used again with a new ioloop. 236 Subprocess.uninitialize() 237 self.io_loop.clear_current() 238 if (not IOLoop.initialized() or 239 self.io_loop is not IOLoop.instance()): 240 # Try to clean up any file descriptors left open in the ioloop. 241 # This avoids leaks, especially when tests are run repeatedly 242 # in the same process with autoreload (because curl does not 243 # set FD_CLOEXEC on its file descriptors) 244 self.io_loop.close(all_fds=True) 245 super(AsyncTestCase, self).tearDown() 246 # In case an exception escaped or the StackContext caught an exception 247 # when there wasn't a wait() to re-raise it, do so here. 248 # This is our last chance to raise an exception in a way that the 249 # unittest machinery understands. 250 self.__rethrow() 251 252 def get_new_ioloop(self): 253 """Creates a new `.IOLoop` for this test. May be overridden in 254 subclasses for tests that require a specific `.IOLoop` (usually 255 the singleton `.IOLoop.instance()`). 256 """ 257 return IOLoop() 258 259 def _handle_exception(self, typ, value, tb): 260 if self.__failure is None: 261 self.__failure = (typ, value, tb) 262 else: 263 app_log.error("multiple unhandled exceptions in test", 264 exc_info=(typ, value, tb)) 265 self.stop() 266 return True 267 268 def __rethrow(self): 269 if self.__failure is not None: 270 failure = self.__failure 271 self.__failure = None 272 raise_exc_info(failure) 273 274 def run(self, result=None): 275 with ExceptionStackContext(self._handle_exception): 276 super(AsyncTestCase, self).run(result) 277 # As a last resort, if an exception escaped super.run() and wasn't 278 # re-raised in tearDown, raise it here. This will cause the 279 # unittest run to fail messily, but that's better than silently 280 # ignoring an error. 281 self.__rethrow() 282 283 def stop(self, _arg=None, **kwargs): 284 """Stops the `.IOLoop`, causing one pending (or future) call to `wait()` 285 to return. 286 287 Keyword arguments or a single positional argument passed to `stop()` are 288 saved and will be returned by `wait()`. 289 """ 290 assert _arg is None or not kwargs 291 self.__stop_args = kwargs or _arg 292 if self.__running: 293 self.io_loop.stop() 294 self.__running = False 295 self.__stopped = True 296 297 def wait(self, condition=None, timeout=None): 298 """Runs the `.IOLoop` until stop is called or timeout has passed. 299 300 In the event of a timeout, an exception will be thrown. The 301 default timeout is 5 seconds; it may be overridden with a 302 ``timeout`` keyword argument or globally with the 303 ``ASYNC_TEST_TIMEOUT`` environment variable. 304 305 If ``condition`` is not None, the `.IOLoop` will be restarted 306 after `stop()` until ``condition()`` returns true. 307 308 .. versionchanged:: 3.1 309 Added the ``ASYNC_TEST_TIMEOUT`` environment variable. 310 """ 311 if timeout is None: 312 timeout = get_async_test_timeout() 313 314 if not self.__stopped: 315 if timeout: 316 def timeout_func(): 317 try: 318 raise self.failureException( 319 'Async operation timed out after %s seconds' % 320 timeout) 321 except Exception: 322 self.__failure = sys.exc_info() 323 self.stop() 324 self.__timeout = self.io_loop.add_timeout(self.io_loop.time() + timeout, timeout_func) 325 while True: 326 self.__running = True 327 self.io_loop.start() 328 if (self.__failure is not None or 329 condition is None or condition()): 330 break 331 if self.__timeout is not None: 332 self.io_loop.remove_timeout(self.__timeout) 333 self.__timeout = None 334 assert self.__stopped 335 self.__stopped = False 336 self.__rethrow() 337 result = self.__stop_args 338 self.__stop_args = None 339 return result 340 341 342class AsyncHTTPTestCase(AsyncTestCase): 343 """A test case that starts up an HTTP server. 344 345 Subclasses must override `get_app()`, which returns the 346 `tornado.web.Application` (or other `.HTTPServer` callback) to be tested. 347 Tests will typically use the provided ``self.http_client`` to fetch 348 URLs from this server. 349 350 Example, assuming the "Hello, world" example from the user guide is in 351 ``hello.py``:: 352 353 import hello 354 355 class TestHelloApp(AsyncHTTPTestCase): 356 def get_app(self): 357 return hello.make_app() 358 359 def test_homepage(self): 360 response = self.fetch('/') 361 self.assertEqual(response.code, 200) 362 self.assertEqual(response.body, 'Hello, world') 363 364 That call to ``self.fetch()`` is equivalent to :: 365 366 self.http_client.fetch(self.get_url('/'), self.stop) 367 response = self.wait() 368 369 which illustrates how AsyncTestCase can turn an asynchronous operation, 370 like ``http_client.fetch()``, into a synchronous operation. If you need 371 to do other asynchronous operations in tests, you'll probably need to use 372 ``stop()`` and ``wait()`` yourself. 373 """ 374 def setUp(self): 375 super(AsyncHTTPTestCase, self).setUp() 376 sock, port = bind_unused_port() 377 self.__port = port 378 379 self.http_client = self.get_http_client() 380 self._app = self.get_app() 381 self.http_server = self.get_http_server() 382 self.http_server.add_sockets([sock]) 383 384 def get_http_client(self): 385 return AsyncHTTPClient(io_loop=self.io_loop) 386 387 def get_http_server(self): 388 return HTTPServer(self._app, io_loop=self.io_loop, 389 **self.get_httpserver_options()) 390 391 def get_app(self): 392 """Should be overridden by subclasses to return a 393 `tornado.web.Application` or other `.HTTPServer` callback. 394 """ 395 raise NotImplementedError() 396 397 def fetch(self, path, **kwargs): 398 """Convenience method to synchronously fetch a url. 399 400 The given path will be appended to the local server's host and 401 port. Any additional kwargs will be passed directly to 402 `.AsyncHTTPClient.fetch` (and so could be used to pass 403 ``method="POST"``, ``body="..."``, etc). 404 """ 405 self.http_client.fetch(self.get_url(path), self.stop, **kwargs) 406 return self.wait() 407 408 def get_httpserver_options(self): 409 """May be overridden by subclasses to return additional 410 keyword arguments for the server. 411 """ 412 return {} 413 414 def get_http_port(self): 415 """Returns the port used by the server. 416 417 A new port is chosen for each test. 418 """ 419 return self.__port 420 421 def get_protocol(self): 422 return 'http' 423 424 def get_url(self, path): 425 """Returns an absolute url for the given path on the test server.""" 426 return '%s://127.0.0.1:%s%s' % (self.get_protocol(), 427 self.get_http_port(), path) 428 429 def tearDown(self): 430 self.http_server.stop() 431 self.io_loop.run_sync(self.http_server.close_all_connections, 432 timeout=get_async_test_timeout()) 433 if (not IOLoop.initialized() or 434 self.http_client.io_loop is not IOLoop.instance()): 435 self.http_client.close() 436 super(AsyncHTTPTestCase, self).tearDown() 437 438 439class AsyncHTTPSTestCase(AsyncHTTPTestCase): 440 """A test case that starts an HTTPS server. 441 442 Interface is generally the same as `AsyncHTTPTestCase`. 443 """ 444 def get_http_client(self): 445 return AsyncHTTPClient(io_loop=self.io_loop, force_instance=True, 446 defaults=dict(validate_cert=False)) 447 448 def get_httpserver_options(self): 449 return dict(ssl_options=self.get_ssl_options()) 450 451 def get_ssl_options(self): 452 """May be overridden by subclasses to select SSL options. 453 454 By default includes a self-signed testing certificate. 455 """ 456 # Testing keys were generated with: 457 # openssl req -new -keyout tornado/test/test.key -out tornado/test/test.crt -nodes -days 3650 -x509 458 module_dir = os.path.dirname(__file__) 459 return dict( 460 certfile=os.path.join(module_dir, 'test', 'test.crt'), 461 keyfile=os.path.join(module_dir, 'test', 'test.key')) 462 463 def get_protocol(self): 464 return 'https' 465 466 467def gen_test(func=None, timeout=None): 468 """Testing equivalent of ``@gen.coroutine``, to be applied to test methods. 469 470 ``@gen.coroutine`` cannot be used on tests because the `.IOLoop` is not 471 already running. ``@gen_test`` should be applied to test methods 472 on subclasses of `AsyncTestCase`. 473 474 Example:: 475 476 class MyTest(AsyncHTTPTestCase): 477 @gen_test 478 def test_something(self): 479 response = yield gen.Task(self.fetch('/')) 480 481 By default, ``@gen_test`` times out after 5 seconds. The timeout may be 482 overridden globally with the ``ASYNC_TEST_TIMEOUT`` environment variable, 483 or for each test with the ``timeout`` keyword argument:: 484 485 class MyTest(AsyncHTTPTestCase): 486 @gen_test(timeout=10) 487 def test_something_slow(self): 488 response = yield gen.Task(self.fetch('/')) 489 490 .. versionadded:: 3.1 491 The ``timeout`` argument and ``ASYNC_TEST_TIMEOUT`` environment 492 variable. 493 494 .. versionchanged:: 4.0 495 The wrapper now passes along ``*args, **kwargs`` so it can be used 496 on functions with arguments. 497 """ 498 if timeout is None: 499 timeout = get_async_test_timeout() 500 501 def wrap(f): 502 # Stack up several decorators to allow us to access the generator 503 # object itself. In the innermost wrapper, we capture the generator 504 # and save it in an attribute of self. Next, we run the wrapped 505 # function through @gen.coroutine. Finally, the coroutine is 506 # wrapped again to make it synchronous with run_sync. 507 # 508 # This is a good case study arguing for either some sort of 509 # extensibility in the gen decorators or cancellation support. 510 @functools.wraps(f) 511 def pre_coroutine(self, *args, **kwargs): 512 result = f(self, *args, **kwargs) 513 if isinstance(result, GeneratorType) or iscoroutine(result): 514 self._test_generator = result 515 else: 516 self._test_generator = None 517 return result 518 519 if iscoroutinefunction(f): 520 coro = pre_coroutine 521 else: 522 coro = gen.coroutine(pre_coroutine) 523 524 @functools.wraps(coro) 525 def post_coroutine(self, *args, **kwargs): 526 try: 527 return self.io_loop.run_sync( 528 functools.partial(coro, self, *args, **kwargs), 529 timeout=timeout) 530 except TimeoutError as e: 531 # run_sync raises an error with an unhelpful traceback. 532 # Throw it back into the generator or coroutine so the stack 533 # trace is replaced by the point where the test is stopped. 534 self._test_generator.throw(e) 535 # In case the test contains an overly broad except clause, 536 # we may get back here. In this case re-raise the original 537 # exception, which is better than nothing. 538 raise 539 return post_coroutine 540 541 if func is not None: 542 # Used like: 543 # @gen_test 544 # def f(self): 545 # pass 546 return wrap(func) 547 else: 548 # Used like @gen_test(timeout=10) 549 return wrap 550 551 552# Without this attribute, nosetests will try to run gen_test as a test 553# anywhere it is imported. 554gen_test.__test__ = False # type: ignore 555 556 557class LogTrapTestCase(unittest.TestCase): 558 """A test case that captures and discards all logging output 559 if the test passes. 560 561 Some libraries can produce a lot of logging output even when 562 the test succeeds, so this class can be useful to minimize the noise. 563 Simply use it as a base class for your test case. It is safe to combine 564 with AsyncTestCase via multiple inheritance 565 (``class MyTestCase(AsyncHTTPTestCase, LogTrapTestCase):``) 566 567 This class assumes that only one log handler is configured and 568 that it is a `~logging.StreamHandler`. This is true for both 569 `logging.basicConfig` and the "pretty logging" configured by 570 `tornado.options`. It is not compatible with other log buffering 571 mechanisms, such as those provided by some test runners. 572 573 .. deprecated:: 4.1 574 Use the unittest module's ``--buffer`` option instead, or `.ExpectLog`. 575 """ 576 def run(self, result=None): 577 logger = logging.getLogger() 578 if not logger.handlers: 579 logging.basicConfig() 580 handler = logger.handlers[0] 581 if (len(logger.handlers) > 1 or 582 not isinstance(handler, logging.StreamHandler)): 583 # Logging has been configured in a way we don't recognize, 584 # so just leave it alone. 585 super(LogTrapTestCase, self).run(result) 586 return 587 old_stream = handler.stream 588 try: 589 handler.stream = StringIO() 590 gen_log.info("RUNNING TEST: " + str(self)) 591 old_error_count = len(result.failures) + len(result.errors) 592 super(LogTrapTestCase, self).run(result) 593 new_error_count = len(result.failures) + len(result.errors) 594 if new_error_count != old_error_count: 595 old_stream.write(handler.stream.getvalue()) 596 finally: 597 handler.stream = old_stream 598 599 600class ExpectLog(logging.Filter): 601 """Context manager to capture and suppress expected log output. 602 603 Useful to make tests of error conditions less noisy, while still 604 leaving unexpected log entries visible. *Not thread safe.* 605 606 The attribute ``logged_stack`` is set to true if any exception 607 stack trace was logged. 608 609 Usage:: 610 611 with ExpectLog('tornado.application', "Uncaught exception"): 612 error_response = self.fetch("/some_page") 613 614 .. versionchanged:: 4.3 615 Added the ``logged_stack`` attribute. 616 """ 617 def __init__(self, logger, regex, required=True): 618 """Constructs an ExpectLog context manager. 619 620 :param logger: Logger object (or name of logger) to watch. Pass 621 an empty string to watch the root logger. 622 :param regex: Regular expression to match. Any log entries on 623 the specified logger that match this regex will be suppressed. 624 :param required: If true, an exception will be raised if the end of 625 the ``with`` statement is reached without matching any log entries. 626 """ 627 if isinstance(logger, basestring_type): 628 logger = logging.getLogger(logger) 629 self.logger = logger 630 self.regex = re.compile(regex) 631 self.required = required 632 self.matched = False 633 self.logged_stack = False 634 635 def filter(self, record): 636 if record.exc_info: 637 self.logged_stack = True 638 message = record.getMessage() 639 if self.regex.match(message): 640 self.matched = True 641 return False 642 return True 643 644 def __enter__(self): 645 self.logger.addFilter(self) 646 return self 647 648 def __exit__(self, typ, value, tb): 649 self.logger.removeFilter(self) 650 if not typ and self.required and not self.matched: 651 raise Exception("did not get expected log message") 652 653 654def main(**kwargs): 655 """A simple test runner. 656 657 This test runner is essentially equivalent to `unittest.main` from 658 the standard library, but adds support for tornado-style option 659 parsing and log formatting. It is *not* necessary to use this 660 `main` function to run tests using `AsyncTestCase`; these tests 661 are self-contained and can run with any test runner. 662 663 The easiest way to run a test is via the command line:: 664 665 python -m tornado.testing tornado.test.stack_context_test 666 667 See the standard library unittest module for ways in which tests can 668 be specified. 669 670 Projects with many tests may wish to define a test script like 671 ``tornado/test/runtests.py``. This script should define a method 672 ``all()`` which returns a test suite and then call 673 `tornado.testing.main()`. Note that even when a test script is 674 used, the ``all()`` test suite may be overridden by naming a 675 single test on the command line:: 676 677 # Runs all tests 678 python -m tornado.test.runtests 679 # Runs one test 680 python -m tornado.test.runtests tornado.test.stack_context_test 681 682 Additional keyword arguments passed through to ``unittest.main()``. 683 For example, use ``tornado.testing.main(verbosity=2)`` 684 to show many test details as they are run. 685 See http://docs.python.org/library/unittest.html#unittest.main 686 for full argument list. 687 """ 688 from tornado.options import define, options, parse_command_line 689 690 define('exception_on_interrupt', type=bool, default=True, 691 help=("If true (default), ctrl-c raises a KeyboardInterrupt " 692 "exception. This prints a stack trace but cannot interrupt " 693 "certain operations. If false, the process is more reliably " 694 "killed, but does not print a stack trace.")) 695 696 # support the same options as unittest's command-line interface 697 define('verbose', type=bool) 698 define('quiet', type=bool) 699 define('failfast', type=bool) 700 define('catch', type=bool) 701 define('buffer', type=bool) 702 703 argv = [sys.argv[0]] + parse_command_line(sys.argv) 704 705 if not options.exception_on_interrupt: 706 signal.signal(signal.SIGINT, signal.SIG_DFL) 707 708 if options.verbose is not None: 709 kwargs['verbosity'] = 2 710 if options.quiet is not None: 711 kwargs['verbosity'] = 0 712 if options.failfast is not None: 713 kwargs['failfast'] = True 714 if options.catch is not None: 715 kwargs['catchbreak'] = True 716 if options.buffer is not None: 717 kwargs['buffer'] = True 718 719 if __name__ == '__main__' and len(argv) == 1: 720 print("No tests specified", file=sys.stderr) 721 sys.exit(1) 722 try: 723 # In order to be able to run tests by their fully-qualified name 724 # on the command line without importing all tests here, 725 # module must be set to None. Python 3.2's unittest.main ignores 726 # defaultTest if no module is given (it tries to do its own 727 # test discovery, which is incompatible with auto2to3), so don't 728 # set module if we're not asking for a specific test. 729 if len(argv) > 1: 730 unittest.main(module=None, argv=argv, **kwargs) 731 else: 732 unittest.main(defaultTest="all", argv=argv, **kwargs) 733 except SystemExit as e: 734 if e.code == 0: 735 gen_log.info('PASS') 736 else: 737 gen_log.error('FAIL') 738 raise 739 740 741if __name__ == '__main__': 742 main() 743