1 # Copyright (c) 2008 Jonathan M. Lange. See LICENSE for details.
3 """Test results and related things."""
7 'ExtendedToOriginalDecorator',
10 'ThreadsafeForwardingResult',
16 from testtools.compat import _format_exc_info, str_is_unicode, _u
19 class TestResult(unittest.TestResult):
20 """Subclass of unittest.TestResult extending the protocol for flexability.
22 This test result supports an experimental protocol for providing additional
23 data to in test outcomes. All the outcome methods take an optional dict
24 'details'. If supplied any other detail parameters like 'err' or 'reason'
25 should not be provided. The details dict is a mapping from names to
26 MIME content objects (see testtools.content). This permits attaching
27 tracebacks, log files, or even large objects like databases that were
28 part of the test fixture. Until this API is accepted into upstream
29 Python it is considered experimental: it may be replaced at any point
30 by a newer version more in line with upstream Python. Compatibility would
31 be aimed for in this case, but may not be possible.
33 :ivar skip_reasons: A dict of skip-reasons -> list of tests. See addSkip.
37 super(TestResult, self).__init__()
38 self.skip_reasons = {}
40 # -- Start: As per python 2.7 --
41 self.expectedFailures = []
42 self.unexpectedSuccesses = []
43 # -- End: As per python 2.7 --
45 def addExpectedFailure(self, test, err=None, details=None):
46 """Called when a test has failed in an expected manner.
48 Like with addSuccess and addError, testStopped should still be called.
50 :param test: The test that has been skipped.
51 :param err: The exc_info of the error that was raised.
54 # This is the python 2.7 implementation
55 self.expectedFailures.append(
56 (test, self._err_details_to_string(test, err, details)))
58 def addError(self, test, err=None, details=None):
59 """Called when an error has occurred. 'err' is a tuple of values as
60 returned by sys.exc_info().
62 :param details: Alternative way to supply details about the outcome.
63 see the class docstring for more information.
65 self.errors.append((test,
66 self._err_details_to_string(test, err, details)))
68 def addFailure(self, test, err=None, details=None):
69 """Called when an error has occurred. 'err' is a tuple of values as
70 returned by sys.exc_info().
72 :param details: Alternative way to supply details about the outcome.
73 see the class docstring for more information.
75 self.failures.append((test,
76 self._err_details_to_string(test, err, details)))
78 def addSkip(self, test, reason=None, details=None):
79 """Called when a test has been skipped rather than running.
81 Like with addSuccess and addError, testStopped should still be called.
83 This must be called by the TestCase. 'addError' and 'addFailure' will
84 not call addSkip, since they have no assumptions about the kind of
85 errors that a test can raise.
87 :param test: The test that has been skipped.
88 :param reason: The reason for the test being skipped. For instance,
89 u"pyGL is not available".
90 :param details: Alternative way to supply details about the outcome.
91 see the class docstring for more information.
95 reason = details.get('reason')
97 reason = 'No reason given'
99 reason = ''.join(reason.iter_text())
100 skip_list = self.skip_reasons.setdefault(reason, [])
101 skip_list.append(test)
103 def addSuccess(self, test, details=None):
104 """Called when a test succeeded."""
106 def addUnexpectedSuccess(self, test, details=None):
107 """Called when a test was expected to fail, but succeed."""
108 self.unexpectedSuccesses.append(test)
111 # Python 3 and IronPython strings are unicode, use parent class method
112 _exc_info_to_unicode = unittest.TestResult._exc_info_to_string
114 # For Python 2, need to decode components of traceback according to
115 # their source, so can't use traceback.format_exception
116 # Here follows a little deep magic to copy the existing method and
117 # replace the formatter with one that returns unicode instead
118 from types import FunctionType as __F, ModuleType as __M
119 __f = unittest.TestResult._exc_info_to_string.im_func
120 __g = dict(__f.func_globals)
121 __m = __M("__fake_traceback")
122 __m.format_exception = _format_exc_info
123 __g["traceback"] = __m
124 _exc_info_to_unicode = __F(__f.func_code, __g, "_exc_info_to_unicode")
125 del __F, __M, __f, __g, __m
127 def _err_details_to_string(self, test, err=None, details=None):
128 """Convert an error in exc_info form or a contents dict to a string."""
130 return self._exc_info_to_unicode(err, test)
131 return _details_to_str(details)
134 """Return the current 'test time'.
136 If the time() method has not been called, this is equivalent to
137 datetime.now(), otherwise its the last supplied datestamp given to the
140 if self.__now is None:
141 return datetime.datetime.now()
145 def startTestRun(self):
146 """Called before a test run starts.
151 def stopTestRun(self):
152 """Called after a test run completes
157 def time(self, a_datetime):
158 """Provide a timestamp to represent the current time.
160 This is useful when test activity is time delayed, or happening
161 concurrently and getting the system time between API calls will not
162 accurately represent the duration of tests (or the whole run).
164 Calling time() sets the datetime used by the TestResult object.
165 Time is permitted to go backwards when using this call.
167 :param a_datetime: A datetime.datetime object with TZ information or
168 None to reset the TestResult to gathering time from the system.
170 self.__now = a_datetime
173 """Called when the test runner is done.
175 deprecated in favour of stopTestRun.
179 class MultiTestResult(TestResult):
180 """A test result that dispatches to many test results."""
182 def __init__(self, *results):
183 TestResult.__init__(self)
184 self._results = map(ExtendedToOriginalDecorator, results)
186 def _dispatch(self, message, *args, **kwargs):
188 getattr(result, message)(*args, **kwargs)
189 for result in self._results)
191 def startTest(self, test):
192 return self._dispatch('startTest', test)
194 def stopTest(self, test):
195 return self._dispatch('stopTest', test)
197 def addError(self, test, error=None, details=None):
198 return self._dispatch('addError', test, error, details=details)
200 def addExpectedFailure(self, test, err=None, details=None):
201 return self._dispatch(
202 'addExpectedFailure', test, err, details=details)
204 def addFailure(self, test, err=None, details=None):
205 return self._dispatch('addFailure', test, err, details=details)
207 def addSkip(self, test, reason=None, details=None):
208 return self._dispatch('addSkip', test, reason, details=details)
210 def addSuccess(self, test, details=None):
211 return self._dispatch('addSuccess', test, details=details)
213 def addUnexpectedSuccess(self, test, details=None):
214 return self._dispatch('addUnexpectedSuccess', test, details=details)
216 def startTestRun(self):
217 return self._dispatch('startTestRun')
219 def stopTestRun(self):
220 return self._dispatch('stopTestRun')
223 return self._dispatch('done')
226 class TextTestResult(TestResult):
227 """A TestResult which outputs activity to a text stream."""
229 def __init__(self, stream):
230 """Construct a TextTestResult writing to stream."""
231 super(TextTestResult, self).__init__()
233 self.sep1 = '=' * 70 + '\n'
234 self.sep2 = '-' * 70 + '\n'
236 def _delta_to_float(self, a_timedelta):
237 return (a_timedelta.days * 86400.0 + a_timedelta.seconds +
238 a_timedelta.microseconds / 1000000.0)
240 def _show_list(self, label, error_list):
241 for test, output in error_list:
242 self.stream.write(self.sep1)
243 self.stream.write("%s: %s\n" % (label, test.id()))
244 self.stream.write(self.sep2)
245 self.stream.write(output)
247 def startTestRun(self):
248 super(TextTestResult, self).startTestRun()
249 self.__start = self._now()
250 self.stream.write("Tests running...\n")
252 def stopTestRun(self):
253 if self.testsRun != 1:
258 self._show_list('ERROR', self.errors)
259 self._show_list('FAIL', self.failures)
260 self.stream.write("Ran %d test%s in %.3fs\n\n" %
261 (self.testsRun, plural,
262 self._delta_to_float(stop - self.__start)))
263 if self.wasSuccessful():
264 self.stream.write("OK\n")
266 self.stream.write("FAILED (")
268 details.append("failures=%d" % (
269 len(self.failures) + len(self.errors)))
270 self.stream.write(", ".join(details))
271 self.stream.write(")\n")
272 super(TextTestResult, self).stopTestRun()
275 class ThreadsafeForwardingResult(TestResult):
276 """A TestResult which ensures the target does not receive mixed up calls.
278 This is used when receiving test results from multiple sources, and batches
279 up all the activity for a single test into a thread-safe batch where all
280 other ThreadsafeForwardingResult objects sharing the same semaphore will be
283 Typical use of ThreadsafeForwardingResult involves creating one
284 ThreadsafeForwardingResult per thread in a ConcurrentTestSuite. These
285 forward to the TestResult that the ConcurrentTestSuite run method was
288 target.done() is called once for each ThreadsafeForwardingResult that
289 forwards to the same target. If the target's done() takes special action,
290 care should be taken to accommodate this.
293 def __init__(self, target, semaphore):
294 """Create a ThreadsafeForwardingResult forwarding to target.
296 :param target: A TestResult.
297 :param semaphore: A threading.Semaphore with limit 1.
299 TestResult.__init__(self)
300 self.result = ExtendedToOriginalDecorator(target)
301 self.semaphore = semaphore
303 def addError(self, test, err=None, details=None):
304 self.semaphore.acquire()
306 self.result.startTest(test)
307 self.result.addError(test, err, details=details)
308 self.result.stopTest(test)
310 self.semaphore.release()
312 def addExpectedFailure(self, test, err=None, details=None):
313 self.semaphore.acquire()
315 self.result.startTest(test)
316 self.result.addExpectedFailure(test, err, details=details)
317 self.result.stopTest(test)
319 self.semaphore.release()
321 def addFailure(self, test, err=None, details=None):
322 self.semaphore.acquire()
324 self.result.startTest(test)
325 self.result.addFailure(test, err, details=details)
326 self.result.stopTest(test)
328 self.semaphore.release()
330 def addSkip(self, test, reason=None, details=None):
331 self.semaphore.acquire()
333 self.result.startTest(test)
334 self.result.addSkip(test, reason, details=details)
335 self.result.stopTest(test)
337 self.semaphore.release()
339 def addSuccess(self, test, details=None):
340 self.semaphore.acquire()
342 self.result.startTest(test)
343 self.result.addSuccess(test, details=details)
344 self.result.stopTest(test)
346 self.semaphore.release()
348 def addUnexpectedSuccess(self, test, details=None):
349 self.semaphore.acquire()
351 self.result.startTest(test)
352 self.result.addUnexpectedSuccess(test, details=details)
353 self.result.stopTest(test)
355 self.semaphore.release()
357 def startTestRun(self):
358 self.semaphore.acquire()
360 self.result.startTestRun()
362 self.semaphore.release()
364 def stopTestRun(self):
365 self.semaphore.acquire()
367 self.result.stopTestRun()
369 self.semaphore.release()
372 self.semaphore.acquire()
376 self.semaphore.release()
379 class ExtendedToOriginalDecorator(object):
380 """Permit new TestResult API code to degrade gracefully with old results.
382 This decorates an existing TestResult and converts missing outcomes
383 such as addSkip to older outcomes such as addSuccess. It also supports
384 the extended details protocol. In all cases the most recent protocol
385 is attempted first, and fallbacks only occur when the decorated result
386 does not support the newer style of calling.
389 def __init__(self, decorated):
390 self.decorated = decorated
392 def __getattr__(self, name):
393 return getattr(self.decorated, name)
395 def addError(self, test, err=None, details=None):
396 self._check_args(err, details)
397 if details is not None:
399 return self.decorated.addError(test, details=details)
402 err = self._details_to_exc_info(details)
403 return self.decorated.addError(test, err)
405 def addExpectedFailure(self, test, err=None, details=None):
406 self._check_args(err, details)
407 addExpectedFailure = getattr(
408 self.decorated, 'addExpectedFailure', None)
409 if addExpectedFailure is None:
410 return self.addSuccess(test)
411 if details is not None:
413 return addExpectedFailure(test, details=details)
416 err = self._details_to_exc_info(details)
417 return addExpectedFailure(test, err)
419 def addFailure(self, test, err=None, details=None):
420 self._check_args(err, details)
421 if details is not None:
423 return self.decorated.addFailure(test, details=details)
426 err = self._details_to_exc_info(details)
427 return self.decorated.addFailure(test, err)
429 def addSkip(self, test, reason=None, details=None):
430 self._check_args(reason, details)
431 addSkip = getattr(self.decorated, 'addSkip', None)
433 return self.decorated.addSuccess(test)
434 if details is not None:
436 return addSkip(test, details=details)
439 reason = _details_to_str(details)
440 return addSkip(test, reason)
442 def addUnexpectedSuccess(self, test, details=None):
443 outcome = getattr(self.decorated, 'addUnexpectedSuccess', None)
445 return self.decorated.addSuccess(test)
446 if details is not None:
448 return outcome(test, details=details)
453 def addSuccess(self, test, details=None):
454 if details is not None:
456 return self.decorated.addSuccess(test, details=details)
459 return self.decorated.addSuccess(test)
461 def _check_args(self, err, details):
465 if details is not None:
468 raise ValueError("Must pass only one of err '%s' and details '%s"
471 def _details_to_exc_info(self, details):
472 """Convert a details dict to an exc_info tuple."""
473 return (_StringException,
474 _StringException(_details_to_str(details)), None)
478 return self.decorated.done()
479 except AttributeError:
482 def progress(self, offset, whence):
483 method = getattr(self.decorated, 'progress', None)
486 return method(offset, whence)
489 def shouldStop(self):
490 return self.decorated.shouldStop
492 def startTest(self, test):
493 return self.decorated.startTest(test)
495 def startTestRun(self):
497 return self.decorated.startTestRun()
498 except AttributeError:
502 return self.decorated.stop()
504 def stopTest(self, test):
505 return self.decorated.stopTest(test)
507 def stopTestRun(self):
509 return self.decorated.stopTestRun()
510 except AttributeError:
513 def tags(self, new_tags, gone_tags):
514 method = getattr(self.decorated, 'tags', None)
517 return method(new_tags, gone_tags)
519 def time(self, a_datetime):
520 method = getattr(self.decorated, 'time', None)
523 return method(a_datetime)
525 def wasSuccessful(self):
526 return self.decorated.wasSuccessful()
529 class _StringException(Exception):
530 """An exception made from an arbitrary string."""
532 if not str_is_unicode:
533 def __init__(self, string):
534 if type(string) is not unicode:
535 raise TypeError("_StringException expects unicode, got %r" %
537 Exception.__init__(self, string)
540 return self.args[0].encode("utf-8")
542 def __unicode__(self):
544 # For 3.0 and above the default __str__ is fine, so we don't define one.
549 def __eq__(self, other):
551 return self.args == other.args
552 except AttributeError:
556 def _details_to_str(details):
557 """Convert a details dict to a string."""
559 # sorted is for testing, may want to remove that and use a dict
560 # subclass with defined order for items instead.
561 for key, content in sorted(details.items()):
562 if content.content_type.type != 'text':
563 chars.append('Binary content: %s\n' % key)
565 chars.append('Text attachment: %s\n' % key)
566 chars.append('------------\n')
567 chars.extend(content.iter_text())
568 if not chars[-1].endswith('\n'):
570 chars.append('------------\n')
571 return _u('').join(chars)