* ``stdin``, ``stdout``, and ``stderr`` may have the value
``tornado.process.Subprocess.STREAM``, which will make the corresponding
- attribute of the resulting Subprocess a `.PipeIOStream`.
+ attribute of the resulting Subprocess a `.PipeIOStream`. If this option
+ is used, the caller is responsible for closing the streams when done
+ with them.
The ``Subprocess.STREAM`` option and the ``set_exit_callback`` and
``wait_for_exit`` methods do not work on Windows. There is
stdin=Subprocess.STREAM,
stdout=Subprocess.STREAM, stderr=subprocess.STDOUT)
self.addCleanup(lambda: (subproc.proc.terminate(), subproc.proc.wait()))
+ self.addCleanup(subproc.stdout.close)
+ self.addCleanup(subproc.stdin.close)
subproc.stdout.read_until(b'>>> ', self.stop)
self.wait()
subproc.stdin.write(b"print('hello')\n")
subproc.stderr.read_until(b'\n', self.stop)
data = self.wait()
self.assertEqual(data, b'hello\n')
+ # More mysterious EBADF: This fails if done with self.addCleanup instead of here.
+ subproc.stderr.close()
def test_sigchild(self):
# Twisted's SIGCHLD handler and Subprocess's conflict with each other.
subproc = Subprocess([sys.executable, '-c',
'import time; time.sleep(30)'],
stdout=Subprocess.STREAM)
+ self.addCleanup(subproc.stdout.close)
subproc.set_exit_callback(self.stop)
os.kill(subproc.pid, signal.SIGTERM)
try:
from __future__ import absolute_import, division, print_function
import gc
+import io
import locale # system locale module, not tornado.locale
import logging
import operator
return unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULES)
-class TornadoTextTestRunner(unittest.TextTestRunner):
- def run(self, test):
- result = super(TornadoTextTestRunner, self).run(test)
- if result.skipped:
- skip_reasons = set(reason for (test, reason) in result.skipped)
- self.stream.write(textwrap.fill(
- "Some tests were skipped because: %s" %
- ", ".join(sorted(skip_reasons))))
- self.stream.write("\n")
- return result
+def test_runner_factory(stderr):
+ class TornadoTextTestRunner(unittest.TextTestRunner):
+ def __init__(self, *args, **kwargs):
+ super(TornadoTextTestRunner, self).__init__(*args, stream=stderr, **kwargs)
+
+ def run(self, test):
+ result = super(TornadoTextTestRunner, self).run(test)
+ if result.skipped:
+ skip_reasons = set(reason for (test, reason) in result.skipped)
+ self.stream.write(textwrap.fill(
+ "Some tests were skipped because: %s" %
+ ", ".join(sorted(skip_reasons))))
+ self.stream.write("\n")
+ return result
+ return TornadoTextTestRunner
class LogCounter(logging.Filter):
return True
+class CountingStderr(io.IOBase):
+ def __init__(self, real):
+ self.real = real
+ self.byte_count = 0
+
+ def write(self, data):
+ self.byte_count += len(data)
+ return self.real.write(data)
+
+ def flush(self):
+ return self.real.flush()
+
+
def main():
# The -W command-line option does not work in a virtualenv with
# python 3 (as of virtualenv 1.7), so configure warnings
add_parse_callback(
lambda: logging.getLogger().handlers[0].addFilter(log_counter))
+ # Certain errors (especially "unclosed resource" errors raised in
+ # destructors) go directly to stderr instead of logging. Count
+ # anything written by anything but the test runner as an error.
+ orig_stderr = sys.stderr
+ sys.stderr = CountingStderr(orig_stderr)
+
import tornado.testing
kwargs = {}
if sys.version_info >= (3, 2):
# suppresses this behavior, although this looks like an implementation
# detail. http://bugs.python.org/issue15626
kwargs['warnings'] = False
- kwargs['testRunner'] = TornadoTextTestRunner
+ kwargs['testRunner'] = test_runner_factory(orig_stderr)
try:
tornado.testing.main(**kwargs)
finally:
# The tests should run clean; consider it a failure if they
- # logged anything at info level or above (except for the one
- # allowed info message "PASS")
- if (log_counter.info_count > 1 or
+ # logged anything at info level or above.
+ if (log_counter.info_count > 0 or
log_counter.warning_count > 0 or
- log_counter.error_count > 0):
- logging.error("logged %d infos, %d warnings, and %d errors",
+ log_counter.error_count > 0 or
+ sys.stderr.byte_count > 0):
+ logging.error("logged %d infos, %d warnings, %d errors, and %d bytes to stderr",
log_counter.info_count, log_counter.warning_count,
- log_counter.error_count)
+ log_counter.error_count, sys.stderr.byte_count)
sys.exit(1)
from tornado.web import Application
import contextlib
import os
+import platform
import traceback
import warnings
self.assertIn("should be decorated", result.errors[0][1])
@skipBefore35
+ @unittest.skipIf(platform.python_implementation() == 'PyPy',
+ 'pypy destructor warnings cannot be silenced')
def test_undecorated_coroutine(self):
namespace = exec_test(globals(), locals(), """
class Test(AsyncTestCase):
netutil = None # type: ignore
SimpleAsyncHTTPClient = None # type: ignore
Subprocess = None # type: ignore
-from tornado.log import gen_log, app_log
+from tornado.log import app_log
from tornado.stack_context import ExceptionStackContext
from tornado.util import raise_exc_info, basestring_type, PY3
import functools
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
if __name__ == '__main__' and len(argv) == 1:
print("No tests specified", file=sys.stderr)
sys.exit(1)
- try:
- # 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)
- except SystemExit as e:
- if e.code == 0:
- gen_log.info('PASS')
- else:
- gen_log.error('FAIL')
- raise
+ # 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__':
projects / thirdparty / tornado.git / commitdiff
