[Zope3-checkins]
SVN: zope.testing/branches/output-refactoring/src/zope/testing/testrunner.py
Extract (about half of) output operations into a single class.
Marius Gedminas
marius at pov.lt
Fri Jul 13 06:06:09 EDT 2007
Log message for revision 77826:
Extract (about half of) output operations into a single class.
The goal is to have the output logic in a single place so it's easier to modify
(e.g. add colours if the user asks for them), or replace it with alternative
output formats (e.g., HTML or GUI).
Changed:
U zope.testing/branches/output-refactoring/src/zope/testing/testrunner.py
-=-
Modified: zope.testing/branches/output-refactoring/src/zope/testing/testrunner.py
===================================================================
--- zope.testing/branches/output-refactoring/src/zope/testing/testrunner.py 2007-07-13 10:04:33 UTC (rev 77825)
+++ zope.testing/branches/output-refactoring/src/zope/testing/testrunner.py 2007-07-13 10:06:09 UTC (rev 77826)
@@ -246,6 +246,83 @@
return "__init__.pyo" in fnamelist
return False
+
+class OutputFormatter(object):
+ """Test runner output formatter."""
+
+ # Implementation note: be careful about printing stuff to sys.stderr.
+ # It is used for interprocess communication between the parent and the
+ # child test runner, when you run some test layers in a subprocess.
+ # resume_layer() reasigns sys.stderr for this reason, but be careful
+ # and don't store the original one in __init__ or something.
+
+ def info(self, message):
+ """Print an informative message."""
+ print message
+
+ def error(self, message):
+ """Report an error."""
+ print message
+
+ def profiler_stats(self, stats):
+ """Report profiler stats."""
+ stats.print_stats(50)
+
+ def import_errors(self, import_errors):
+ """Report test-module import errors (if any)."""
+ if import_errors:
+ print "Test-module import failures:"
+ for error in import_errors:
+ print_traceback("Module: %s\n" % error.module, error.exc_info),
+ print
+
+ def tests_with_errors(self, errors):
+ """Report names of tests with errors (if any)."""
+ if errors:
+ print
+ print "Tests with errors:"
+ for test, exc_info in errors:
+ print " ", test
+
+ def tests_with_failures(self, failures):
+ """Report names of tests with failures (if any)."""
+ if failures:
+ print
+ print "Tests with failures:"
+ for test, exc_info in failures:
+ print " ", test
+
+ def modules_with_import_problems(self, import_errors):
+ """Report names of modules with import problems (if any)."""
+ if import_errors:
+ print
+ print "Test-modules with import problems:"
+ for test in import_errors:
+ print " " + test.module
+
+ def totals(self, n_tests, n_failures, n_errors):
+ """Report totals (number of tests, failures, and errors)."""
+ print "Total: %s tests, %s failures, %s errors" % (
+ n_tests, n_failures, n_errors)
+
+ def summary(self, n_tests, n_failures, n_errors, n_seconds):
+ """Summarize the results."""
+ print (" Ran %s tests with %s failures and %s errors in %.3f seconds."
+ % (n_tests, n_failures, n_errors, n_seconds))
+
+ def list_of_tests(self, tests, layer_name):
+ """Report a list of test names."""
+ print "Listing %s tests:" % layer_name
+ for test in tests:
+ print ' ', test
+
+ def garbage(self, garbage):
+ """Report garbage generated by tests."""
+ if garbage:
+ print "Tests generated new (%d) garbage:" % len(garbage)
+ print garbage
+
+
def run(defaults=None, args=None):
if args is None:
args = sys.argv
@@ -277,6 +354,8 @@
if options.fail:
return True
+ output = options.output
+
options.testrunner_defaults = defaults
options.resume_layer = resume_layer
options.resume_number = resume_number
@@ -288,9 +367,9 @@
if (options.profile
and sys.version_info[:3] <= (2,4,1)
and __debug__):
- print ('Because of a bug in Python < 2.4.1, profiling '
- 'during tests requires the -O option be passed to '
- 'Python (not the test runner).')
+ output.error('Because of a bug in Python < 2.4.1, profiling '
+ 'during tests requires the -O option be passed to '
+ 'Python (not the test runner).')
sys.exit()
if options.coverage:
@@ -334,7 +413,7 @@
if options.profile and not options.resume_layer:
stats = profiler.loadStats(prof_glob)
stats.sort_stats('cumulative', 'calls')
- stats.print_stats(50)
+ output.profiler_stats(stats)
if tracer:
coverdir = os.path.join(os.getcwd(), options.coverage)
@@ -362,26 +441,29 @@
global _layer_name_cache
_layer_name_cache = {} # Reset to enforce test isolation
+ output = options.output
+
if options.resume_layer:
original_stderr = sys.stderr
sys.stderr = sys.stdout
elif options.verbose:
if options.all:
- print "Running tests at all levels"
+ msg = "Running tests at all levels"
else:
- print "Running tests at level %d" % options.at_level
+ msg = "Running tests at level %d" % options.at_level
+ output.info(msg)
old_threshold = gc.get_threshold()
if options.gc:
if len(options.gc) > 3:
- print "Too many --gc options"
+ output.error("Too many --gc options")
sys.exit(1)
if options.gc[0]:
- print ("Cyclic garbage collection threshold set to: %s" %
- `tuple(options.gc)`)
+ output.info("Cyclic garbage collection threshold set to: %s" %
+ repr(tuple(options.gc)))
else:
- print "Cyclic garbage collection is disabled."
+ output.info("Cyclic garbage collection is disabled.")
gc.set_threshold(*options.gc)
@@ -398,12 +480,12 @@
reporting_flags = doctest.REPORT_NDIFF
if options.udiff:
if reporting_flags:
- print "Can only give one of --ndiff, --udiff, or --cdiff"
+ output.error("Can only give one of --ndiff, --udiff, or --cdiff")
sys.exit(1)
reporting_flags = doctest.REPORT_UDIFF
if options.cdiff:
if reporting_flags:
- print "Can only give one of --ndiff, --udiff, or --cdiff"
+ output.error("Can only give one of --ndiff, --udiff, or --cdiff")
sys.exit(1)
reporting_flags = doctest.REPORT_CDIFF
if options.report_only_first_failure:
@@ -430,11 +512,7 @@
nlayers = 0
import_errors = tests_by_layer_name.pop(None, None)
- if import_errors:
- print "Test-module import failures:"
- for error in import_errors:
- print_traceback("Module: %s\n" % error.module, error.exc_info),
- print
+ output.import_errors(import_errors)
if 'unit' in tests_by_layer_name:
tests = tests_by_layer_name.pop('unit')
@@ -452,7 +530,7 @@
if options.list_tests:
list_tests(options, tests, 'unit')
else:
- print "Running unit tests:"
+ output.info("Running unit tests:")
nlayers += 1
ran += run_tests(options, tests, 'unit', failures, errors)
@@ -492,11 +570,12 @@
if setup_layers:
if options.resume_layer == None:
- print "Tearing down left over layers:"
+ output.info("Tearing down left over layers:")
tear_down_unneeded((), setup_layers, True)
if options.resume_layer:
sys.stdout.close()
+ # Communicate with the parent. The protocol is obvious:
print >> original_stderr, ran, len(failures), len(errors)
for test, exc_info in failures:
print >> original_stderr, ' '.join(str(test).strip().split('\n'))
@@ -505,27 +584,13 @@
else:
if options.verbose > 1:
- if errors:
- print
- print "Tests with errors:"
- for test, exc_info in errors:
- print " ", test
+ output.tests_with_errors(errors)
+ output.tests_with_failures(failures)
- if failures:
- print
- print "Tests with failures:"
- for test, exc_info in failures:
- print " ", test
-
if nlayers != 1:
- print "Total: %s tests, %s failures, %s errors" % (
- ran, len(failures), len(errors))
+ output.totals(ran, len(failures), len(errors))
- if import_errors:
- print
- print "Test-modules with import problems:"
- for test in import_errors:
- print " " + test.module
+ output.modules_with_import_problems(import_errors)
doctest.set_unittest_reportflags(old_reporting_flags)
@@ -538,15 +603,16 @@
return not bool(import_errors or failures or errors)
def list_tests(options, tests, layer_name):
- print "Listing %s tests:" % layer_name
- for test in tests:
- print ' ', test
+ # TODO: inline this method
+ options.output.list_of_tests(tests, layer_name)
def run_tests(options, tests, name, failures, errors):
repeat = options.repeat or 1
repeat_range = iter(range(repeat))
ran = 0
+ output = options.output
+
gc.collect()
lgarbage = len(gc.garbage)
@@ -558,10 +624,10 @@
for i in repeat_range:
if repeat > 1:
- print "Iteration", i+1
+ output.info("Iteration %d" % (i+1))
if options.verbose > 0 or options.progress:
- print ' Running:'
+ output.info(' Running:')
result = TestResult(options, tests, layer_name=name)
t = time.time()
@@ -603,20 +669,16 @@
t = time.time() - t
if options.verbose == 1 or options.progress:
result.stopTests()
+ # XXX: figure out how to move this print into the OutputFormatter
print
failures.extend(result.failures)
errors.extend(result.errors)
- print (
- " Ran %s tests with %s failures and %s errors in %.3f seconds." %
- (result.testsRun, len(result.failures), len(result.errors), t)
- )
+ output.summary(result.testsRun, len(result.failures), len(result.errors), t)
ran = result.testsRun
gc.collect()
if len(gc.garbage) > lgarbage:
- print ("Tests generated new (%d) garbage:"
- % (len(gc.garbage)-lgarbage))
- print gc.garbage[lgarbage:]
+ output.garbage(gc.garbage[lgarbage:])
lgarbage = len(gc.garbage)
if options.report_refcounts:
@@ -630,6 +692,7 @@
prev = rc
rc = sys.gettotalrefcount()
+ # TODO: move the output into OutputFormatter
if options.verbose:
track.update()
if i:
@@ -650,20 +713,22 @@
def run_layer(options, layer_name, layer, tests, setup_layers,
failures, errors):
+ output = options.output
gathered = []
gather_layers(layer, gathered)
needed = dict([(l, 1) for l in gathered])
if options.resume_number != 0:
- print "Running %s tests:" % layer_name
+ output.info("Running %s tests:" % layer_name)
tear_down_unneeded(needed, setup_layers)
if options.resume_layer != None:
- print " Running in a subprocess."
+ output.info( " Running in a subprocess.")
setup_layer(layer, setup_layers)
return run_tests(options, tests, layer_name, failures, errors)
def resume_tests(options, layer_name, layers, failures, errors):
+ output = options.output
layers = [l for (l, _, _) in layers]
layers = layers[layers.index(layer_name):]
rantotal = 0
@@ -695,7 +760,7 @@
for l in subout:
sys.stdout.write(l)
except IOError:
- print "Error reading subprocess output for", layer_name
+ output.error("Error reading subprocess output for %s" % layer_name)
line = suberr.readline()
try:
@@ -731,6 +796,7 @@
unneeded = order_by_bases(unneeded)
unneeded.reverse()
for l in unneeded:
+ # TODO: figure out how to move this to OutputFormatter
print " Tear down %s" % name_from_layer(l),
t = time.time()
try:
@@ -750,6 +816,7 @@
for base in layer.__bases__:
if base is not object:
setup_layer(base, setup_layers)
+ # TODO: figure out how to move this to OutputFormatter
print " Set up %s" % name_from_layer(layer),
t = time.time()
if hasattr(layer, 'setUp'):
@@ -848,6 +915,7 @@
options = self.options
self.test_width = 0
+ # TODO: figure out how to move this to OutputFormatter
if options.progress:
if self.last_width:
sys.stdout.write('\r' + (' ' * self.last_width) + '\r')
@@ -881,6 +949,7 @@
self._start_time = time.time()
def addSuccess(self, test):
+ # TODO: figure out how to move this to OutputFormatter
if self.options.verbose > 2:
t = max(time.time() - self._start_time, 0.0)
s = " (%.3f s)" % t
@@ -888,6 +957,7 @@
self.test_width += len(s) + 1
def addError(self, test, exc_info):
+ # TODO: figure out how to move this to OutputFormatter
if self.options.verbose > 2:
print " (%.3f s)" % (time.time() - self._start_time)
@@ -909,6 +979,7 @@
self.test_width = self.last_width = 0
def addFailure(self, test, exc_info):
+ # TODO: figure out how to move this to OutputFormatter
if self.options.verbose > 2:
@@ -925,6 +996,7 @@
def stopTests(self):
+ # TODO: figure out how to move this to OutputFormatter
if self.options.progress and self.last_width:
sys.stdout.write('\r' + (' ' * self.last_width) + '\r')
@@ -935,6 +1007,7 @@
elif self.options.verbose > 1:
print
+ # TODO: figure out how to move this to OutputFormatter
if gc.garbage:
print "The following test left garbage:"
print test
@@ -956,6 +1029,7 @@
def _print_traceback(self, msg, exc_info):
+ # TODO: figure out how to move this to OutputFormatter
print_traceback(msg, exc_info)
doctest_template = """
@@ -971,6 +1045,7 @@
class FakeInputContinueGenerator:
def readline(self):
+ # TODO: figure out how to move this to OutputFormatter
print 'c\n'
print '*'*70
print ("Can't use pdb.set_trace when running a layer"
@@ -981,6 +1056,7 @@
def print_traceback(msg, exc_info):
+ # TODO: figure out how to move this to OutputFormatter
print
print msg
@@ -1304,7 +1380,8 @@
for file in files:
if file[-4:] in compiled_sufixes and file[:-1] not in files:
fullname = os.path.join(dirname, file)
- print "Removing stale bytecode file", fullname
+ options.output.info("Removing stale bytecode file %s"
+ % fullname)
os.unlink(fullname)
@@ -1422,6 +1499,7 @@
def output(self):
+ # TODO: figure out how to move this to OutputFormatter
printed = False
s1 = s2 = 0
for t, delta1, delta2 in self.delta:
@@ -1868,6 +1946,8 @@
merge_options(options, defaults)
options.original_testrunner_args = original_testrunner_args
+ options.output = OutputFormatter()
+
options.fail = False
if positional:
More information about the Zope3-Checkins
mailing list