1
# Copyright (c) 2008 testtools developers. See LICENSE for details.
3
"""Test results and related things."""
7
'ExtendedToOriginalDecorator',
10
'ThreadsafeForwardingResult',
17
from testtools.compat import all, _format_exc_info, str_is_unicode, _u
19
# From http://docs.python.org/library/datetime.html
20
_ZERO = datetime.timedelta(0)
24
class UTC(datetime.tzinfo):
27
def utcoffset(self, dt):
39
class TestResult(unittest.TestResult):
40
"""Subclass of unittest.TestResult extending the protocol for flexability.
42
This test result supports an experimental protocol for providing additional
43
data to in test outcomes. All the outcome methods take an optional dict
44
'details'. If supplied any other detail parameters like 'err' or 'reason'
45
should not be provided. The details dict is a mapping from names to
46
MIME content objects (see testtools.content). This permits attaching
47
tracebacks, log files, or even large objects like databases that were
48
part of the test fixture. Until this API is accepted into upstream
49
Python it is considered experimental: it may be replaced at any point
50
by a newer version more in line with upstream Python. Compatibility would
51
be aimed for in this case, but may not be possible.
53
:ivar skip_reasons: A dict of skip-reasons -> list of tests. See addSkip.
57
# startTestRun resets all attributes, and older clients don't know to
58
# call startTestRun, so it is called once here.
59
# Because subclasses may reasonably not expect this, we call the
60
# specific version we want to run.
61
TestResult.startTestRun(self)
63
def addExpectedFailure(self, test, err=None, details=None):
64
"""Called when a test has failed in an expected manner.
66
Like with addSuccess and addError, testStopped should still be called.
68
:param test: The test that has been skipped.
69
:param err: The exc_info of the error that was raised.
72
# This is the python 2.7 implementation
73
self.expectedFailures.append(
74
(test, self._err_details_to_string(test, err, details)))
76
def addError(self, test, err=None, details=None):
77
"""Called when an error has occurred. 'err' is a tuple of values as
78
returned by sys.exc_info().
80
:param details: Alternative way to supply details about the outcome.
81
see the class docstring for more information.
83
self.errors.append((test,
84
self._err_details_to_string(test, err, details)))
86
def addFailure(self, test, err=None, details=None):
87
"""Called when an error has occurred. 'err' is a tuple of values as
88
returned by sys.exc_info().
90
:param details: Alternative way to supply details about the outcome.
91
see the class docstring for more information.
93
self.failures.append((test,
94
self._err_details_to_string(test, err, details)))
96
def addSkip(self, test, reason=None, details=None):
97
"""Called when a test has been skipped rather than running.
99
Like with addSuccess and addError, testStopped should still be called.
101
This must be called by the TestCase. 'addError' and 'addFailure' will
102
not call addSkip, since they have no assumptions about the kind of
103
errors that a test can raise.
105
:param test: The test that has been skipped.
106
:param reason: The reason for the test being skipped. For instance,
107
u"pyGL is not available".
108
:param details: Alternative way to supply details about the outcome.
109
see the class docstring for more information.
113
reason = details.get('reason')
115
reason = 'No reason given'
117
reason = ''.join(reason.iter_text())
118
skip_list = self.skip_reasons.setdefault(reason, [])
119
skip_list.append(test)
121
def addSuccess(self, test, details=None):
122
"""Called when a test succeeded."""
124
def addUnexpectedSuccess(self, test, details=None):
125
"""Called when a test was expected to fail, but succeed."""
126
self.unexpectedSuccesses.append(test)
128
def wasSuccessful(self):
129
"""Has this result been successful so far?
131
If there have been any errors, failures or unexpected successes,
132
return False. Otherwise, return True.
134
Note: This differs from standard unittest in that we consider
135
unexpected successes to be equivalent to failures, rather than
138
return not (self.errors or self.failures or self.unexpectedSuccesses)
141
# Python 3 and IronPython strings are unicode, use parent class method
142
_exc_info_to_unicode = unittest.TestResult._exc_info_to_string
144
# For Python 2, need to decode components of traceback according to
145
# their source, so can't use traceback.format_exception
146
# Here follows a little deep magic to copy the existing method and
147
# replace the formatter with one that returns unicode instead
148
from types import FunctionType as __F, ModuleType as __M
149
__f = unittest.TestResult._exc_info_to_string.im_func
150
__g = dict(__f.func_globals)
151
__m = __M("__fake_traceback")
152
__m.format_exception = _format_exc_info
153
__g["traceback"] = __m
154
_exc_info_to_unicode = __F(__f.func_code, __g, "_exc_info_to_unicode")
155
del __F, __M, __f, __g, __m
157
def _err_details_to_string(self, test, err=None, details=None):
158
"""Convert an error in exc_info form or a contents dict to a string."""
160
return self._exc_info_to_unicode(err, test)
161
return _details_to_str(details)
164
"""Return the current 'test time'.
166
If the time() method has not been called, this is equivalent to
167
datetime.now(), otherwise its the last supplied datestamp given to the
170
if self.__now is None:
171
return datetime.datetime.now(utc)
175
def startTestRun(self):
176
"""Called before a test run starts.
178
New in Python 2.7. The testtools version resets the result to a
179
pristine condition ready for use in another test run. Note that this
180
is different from Python 2.7's startTestRun, which does nothing.
182
super(TestResult, self).__init__()
183
self.skip_reasons = {}
185
# -- Start: As per python 2.7 --
186
self.expectedFailures = []
187
self.unexpectedSuccesses = []
188
# -- End: As per python 2.7 --
190
def stopTestRun(self):
191
"""Called after a test run completes
196
def time(self, a_datetime):
197
"""Provide a timestamp to represent the current time.
199
This is useful when test activity is time delayed, or happening
200
concurrently and getting the system time between API calls will not
201
accurately represent the duration of tests (or the whole run).
203
Calling time() sets the datetime used by the TestResult object.
204
Time is permitted to go backwards when using this call.
206
:param a_datetime: A datetime.datetime object with TZ information or
207
None to reset the TestResult to gathering time from the system.
209
self.__now = a_datetime
212
"""Called when the test runner is done.
214
deprecated in favour of stopTestRun.
218
class MultiTestResult(TestResult):
219
"""A test result that dispatches to many test results."""
221
def __init__(self, *results):
222
TestResult.__init__(self)
223
self._results = list(map(ExtendedToOriginalDecorator, results))
225
def _dispatch(self, message, *args, **kwargs):
227
getattr(result, message)(*args, **kwargs)
228
for result in self._results)
230
def startTest(self, test):
231
return self._dispatch('startTest', test)
233
def stopTest(self, test):
234
return self._dispatch('stopTest', test)
236
def addError(self, test, error=None, details=None):
237
return self._dispatch('addError', test, error, details=details)
239
def addExpectedFailure(self, test, err=None, details=None):
240
return self._dispatch(
241
'addExpectedFailure', test, err, details=details)
243
def addFailure(self, test, err=None, details=None):
244
return self._dispatch('addFailure', test, err, details=details)
246
def addSkip(self, test, reason=None, details=None):
247
return self._dispatch('addSkip', test, reason, details=details)
249
def addSuccess(self, test, details=None):
250
return self._dispatch('addSuccess', test, details=details)
252
def addUnexpectedSuccess(self, test, details=None):
253
return self._dispatch('addUnexpectedSuccess', test, details=details)
255
def startTestRun(self):
256
return self._dispatch('startTestRun')
258
def stopTestRun(self):
259
return self._dispatch('stopTestRun')
261
def time(self, a_datetime):
262
return self._dispatch('time', a_datetime)
265
return self._dispatch('done')
267
def wasSuccessful(self):
268
"""Was this result successful?
270
Only returns True if every constituent result was successful.
272
return all(self._dispatch('wasSuccessful'))
275
class TextTestResult(TestResult):
276
"""A TestResult which outputs activity to a text stream."""
278
def __init__(self, stream):
279
"""Construct a TextTestResult writing to stream."""
280
super(TextTestResult, self).__init__()
282
self.sep1 = '=' * 70 + '\n'
283
self.sep2 = '-' * 70 + '\n'
285
def _delta_to_float(self, a_timedelta):
286
return (a_timedelta.days * 86400.0 + a_timedelta.seconds +
287
a_timedelta.microseconds / 1000000.0)
289
def _show_list(self, label, error_list):
290
for test, output in error_list:
291
self.stream.write(self.sep1)
292
self.stream.write("%s: %s\n" % (label, test.id()))
293
self.stream.write(self.sep2)
294
self.stream.write(output)
296
def startTestRun(self):
297
super(TextTestResult, self).startTestRun()
298
self.__start = self._now()
299
self.stream.write("Tests running...\n")
301
def stopTestRun(self):
302
if self.testsRun != 1:
307
self._show_list('ERROR', self.errors)
308
self._show_list('FAIL', self.failures)
309
for test in self.unexpectedSuccesses:
311
"%sUNEXPECTED SUCCESS: %s\n%s" % (
312
self.sep1, test.id(), self.sep2))
313
self.stream.write("Ran %d test%s in %.3fs\n\n" %
314
(self.testsRun, plural,
315
self._delta_to_float(stop - self.__start)))
316
if self.wasSuccessful():
317
self.stream.write("OK\n")
319
self.stream.write("FAILED (")
321
details.append("failures=%d" % (
323
self.failures, self.errors, self.unexpectedSuccesses)))))
324
self.stream.write(", ".join(details))
325
self.stream.write(")\n")
326
super(TextTestResult, self).stopTestRun()
329
class ThreadsafeForwardingResult(TestResult):
330
"""A TestResult which ensures the target does not receive mixed up calls.
332
This is used when receiving test results from multiple sources, and batches
333
up all the activity for a single test into a thread-safe batch where all
334
other ThreadsafeForwardingResult objects sharing the same semaphore will be
337
Typical use of ThreadsafeForwardingResult involves creating one
338
ThreadsafeForwardingResult per thread in a ConcurrentTestSuite. These
339
forward to the TestResult that the ConcurrentTestSuite run method was
342
target.done() is called once for each ThreadsafeForwardingResult that
343
forwards to the same target. If the target's done() takes special action,
344
care should be taken to accommodate this.
347
def __init__(self, target, semaphore):
348
"""Create a ThreadsafeForwardingResult forwarding to target.
350
:param target: A TestResult.
351
:param semaphore: A threading.Semaphore with limit 1.
353
TestResult.__init__(self)
354
self.result = ExtendedToOriginalDecorator(target)
355
self.semaphore = semaphore
357
def _add_result_with_semaphore(self, method, test, *args, **kwargs):
358
self.semaphore.acquire()
360
self.result.time(self._test_start)
361
self.result.startTest(test)
362
self.result.time(self._now())
364
method(test, *args, **kwargs)
366
self.result.stopTest(test)
368
self.semaphore.release()
370
def addError(self, test, err=None, details=None):
371
self._add_result_with_semaphore(self.result.addError,
372
test, err, details=details)
374
def addExpectedFailure(self, test, err=None, details=None):
375
self._add_result_with_semaphore(self.result.addExpectedFailure,
376
test, err, details=details)
378
def addFailure(self, test, err=None, details=None):
379
self._add_result_with_semaphore(self.result.addFailure,
380
test, err, details=details)
382
def addSkip(self, test, reason=None, details=None):
383
self._add_result_with_semaphore(self.result.addSkip,
384
test, reason, details=details)
386
def addSuccess(self, test, details=None):
387
self._add_result_with_semaphore(self.result.addSuccess,
388
test, details=details)
390
def addUnexpectedSuccess(self, test, details=None):
391
self._add_result_with_semaphore(self.result.addUnexpectedSuccess,
392
test, details=details)
394
def startTestRun(self):
395
self.semaphore.acquire()
397
self.result.startTestRun()
399
self.semaphore.release()
401
def stopTestRun(self):
402
self.semaphore.acquire()
404
self.result.stopTestRun()
406
self.semaphore.release()
409
self.semaphore.acquire()
413
self.semaphore.release()
415
def startTest(self, test):
416
self._test_start = self._now()
417
super(ThreadsafeForwardingResult, self).startTest(test)
419
def wasSuccessful(self):
420
return self.result.wasSuccessful()
423
class ExtendedToOriginalDecorator(object):
424
"""Permit new TestResult API code to degrade gracefully with old results.
426
This decorates an existing TestResult and converts missing outcomes
427
such as addSkip to older outcomes such as addSuccess. It also supports
428
the extended details protocol. In all cases the most recent protocol
429
is attempted first, and fallbacks only occur when the decorated result
430
does not support the newer style of calling.
433
def __init__(self, decorated):
434
self.decorated = decorated
436
def __getattr__(self, name):
437
return getattr(self.decorated, name)
439
def addError(self, test, err=None, details=None):
440
self._check_args(err, details)
441
if details is not None:
443
return self.decorated.addError(test, details=details)
446
err = self._details_to_exc_info(details)
447
return self.decorated.addError(test, err)
449
def addExpectedFailure(self, test, err=None, details=None):
450
self._check_args(err, details)
451
addExpectedFailure = getattr(
452
self.decorated, 'addExpectedFailure', None)
453
if addExpectedFailure is None:
454
return self.addSuccess(test)
455
if details is not None:
457
return addExpectedFailure(test, details=details)
460
err = self._details_to_exc_info(details)
461
return addExpectedFailure(test, err)
463
def addFailure(self, test, err=None, details=None):
464
self._check_args(err, details)
465
if details is not None:
467
return self.decorated.addFailure(test, details=details)
470
err = self._details_to_exc_info(details)
471
return self.decorated.addFailure(test, err)
473
def addSkip(self, test, reason=None, details=None):
474
self._check_args(reason, details)
475
addSkip = getattr(self.decorated, 'addSkip', None)
477
return self.decorated.addSuccess(test)
478
if details is not None:
480
return addSkip(test, details=details)
482
# extract the reason if it's available
484
reason = ''.join(details['reason'].iter_text())
486
reason = _details_to_str(details)
487
return addSkip(test, reason)
489
def addUnexpectedSuccess(self, test, details=None):
490
outcome = getattr(self.decorated, 'addUnexpectedSuccess', None)
494
except test.failureException:
495
return self.addFailure(test, sys.exc_info())
496
if details is not None:
498
return outcome(test, details=details)
503
def addSuccess(self, test, details=None):
504
if details is not None:
506
return self.decorated.addSuccess(test, details=details)
509
return self.decorated.addSuccess(test)
511
def _check_args(self, err, details):
515
if details is not None:
518
raise ValueError("Must pass only one of err '%s' and details '%s"
521
def _details_to_exc_info(self, details):
522
"""Convert a details dict to an exc_info tuple."""
523
return (_StringException,
524
_StringException(_details_to_str(details)), None)
528
return self.decorated.done()
529
except AttributeError:
532
def progress(self, offset, whence):
533
method = getattr(self.decorated, 'progress', None)
536
return method(offset, whence)
539
def shouldStop(self):
540
return self.decorated.shouldStop
542
def startTest(self, test):
543
return self.decorated.startTest(test)
545
def startTestRun(self):
547
return self.decorated.startTestRun()
548
except AttributeError:
552
return self.decorated.stop()
554
def stopTest(self, test):
555
return self.decorated.stopTest(test)
557
def stopTestRun(self):
559
return self.decorated.stopTestRun()
560
except AttributeError:
563
def tags(self, new_tags, gone_tags):
564
method = getattr(self.decorated, 'tags', None)
567
return method(new_tags, gone_tags)
569
def time(self, a_datetime):
570
method = getattr(self.decorated, 'time', None)
573
return method(a_datetime)
575
def wasSuccessful(self):
576
return self.decorated.wasSuccessful()
579
class _StringException(Exception):
580
"""An exception made from an arbitrary string."""
582
if not str_is_unicode:
583
def __init__(self, string):
584
if type(string) is not unicode:
585
raise TypeError("_StringException expects unicode, got %r" %
587
Exception.__init__(self, string)
590
return self.args[0].encode("utf-8")
592
def __unicode__(self):
594
# For 3.0 and above the default __str__ is fine, so we don't define one.
599
def __eq__(self, other):
601
return self.args == other.args
602
except AttributeError:
606
def _details_to_str(details):
607
"""Convert a details dict to a string."""
609
# sorted is for testing, may want to remove that and use a dict
610
# subclass with defined order for items instead.
611
for key, content in sorted(details.items()):
612
if content.content_type.type != 'text':
613
chars.append('Binary content: %s\n' % key)
615
chars.append('Text attachment: %s\n' % key)
616
chars.append('------------\n')
617
chars.extend(content.iter_text())
618
if not chars[-1].endswith('\n'):
620
chars.append('------------\n')
621
return _u('').join(chars)
added
removed
python-for-subunit2junitxml/testtools/testresult/real.py
