diff options
Diffstat (limited to 'lib/subunit/python/testtools/testresult/real.py')
| -rw-r--r-- | lib/subunit/python/testtools/testresult/real.py | 540 |
1 files changed, 0 insertions, 540 deletions
diff --git a/lib/subunit/python/testtools/testresult/real.py b/lib/subunit/python/testtools/testresult/real.py deleted file mode 100644 index 8c8a3edd6e..0000000000 --- a/lib/subunit/python/testtools/testresult/real.py +++ /dev/null @@ -1,540 +0,0 @@ -# Copyright (c) 2008 Jonathan M. Lange. See LICENSE for details. - -"""Test results and related things.""" - -__metaclass__ = type -__all__ = [ - 'ExtendedToOriginalDecorator', - 'MultiTestResult', - 'TestResult', - 'ThreadsafeForwardingResult', - ] - -import datetime -import unittest - - -class TestResult(unittest.TestResult): - """Subclass of unittest.TestResult extending the protocol for flexability. - - This test result supports an experimental protocol for providing additional - data to in test outcomes. All the outcome methods take an optional dict - 'details'. If supplied any other detail parameters like 'err' or 'reason' - should not be provided. The details dict is a mapping from names to - MIME content objects (see testtools.content). This permits attaching - tracebacks, log files, or even large objects like databases that were - part of the test fixture. Until this API is accepted into upstream - Python it is considered experimental: it may be replaced at any point - by a newer version more in line with upstream Python. Compatibility would - be aimed for in this case, but may not be possible. - - :ivar skip_reasons: A dict of skip-reasons -> list of tests. See addSkip. - """ - - def __init__(self): - super(TestResult, self).__init__() - self.skip_reasons = {} - self.__now = None - # -- Start: As per python 2.7 -- - self.expectedFailures = [] - self.unexpectedSuccesses = [] - # -- End: As per python 2.7 -- - - def addExpectedFailure(self, test, err=None, details=None): - """Called when a test has failed in an expected manner. - - Like with addSuccess and addError, testStopped should still be called. - - :param test: The test that has been skipped. - :param err: The exc_info of the error that was raised. - :return: None - """ - # This is the python 2.7 implementation - self.expectedFailures.append( - (test, self._err_details_to_string(test, err, details))) - - def addError(self, test, err=None, details=None): - """Called when an error has occurred. 'err' is a tuple of values as - returned by sys.exc_info(). - - :param details: Alternative way to supply details about the outcome. - see the class docstring for more information. - """ - self.errors.append((test, - self._err_details_to_string(test, err, details))) - - def addFailure(self, test, err=None, details=None): - """Called when an error has occurred. 'err' is a tuple of values as - returned by sys.exc_info(). - - :param details: Alternative way to supply details about the outcome. - see the class docstring for more information. - """ - self.failures.append((test, - self._err_details_to_string(test, err, details))) - - def addSkip(self, test, reason=None, details=None): - """Called when a test has been skipped rather than running. - - Like with addSuccess and addError, testStopped should still be called. - - This must be called by the TestCase. 'addError' and 'addFailure' will - not call addSkip, since they have no assumptions about the kind of - errors that a test can raise. - - :param test: The test that has been skipped. - :param reason: The reason for the test being skipped. For instance, - u"pyGL is not available". - :param details: Alternative way to supply details about the outcome. - see the class docstring for more information. - :return: None - """ - if reason is None: - reason = details.get('reason') - if reason is None: - reason = 'No reason given' - else: - reason = ''.join(reason.iter_text()) - skip_list = self.skip_reasons.setdefault(reason, []) - skip_list.append(test) - - def addSuccess(self, test, details=None): - """Called when a test succeeded.""" - - def addUnexpectedSuccess(self, test, details=None): - """Called when a test was expected to fail, but succeed.""" - self.unexpectedSuccesses.append(test) - - def _err_details_to_string(self, test, err=None, details=None): - """Convert an error in exc_info form or a contents dict to a string.""" - if err is not None: - return self._exc_info_to_string(err, test) - return _details_to_str(details) - - def _now(self): - """Return the current 'test time'. - - If the time() method has not been called, this is equivalent to - datetime.now(), otherwise its the last supplied datestamp given to the - time() method. - """ - if self.__now is None: - return datetime.datetime.now() - else: - return self.__now - - def startTestRun(self): - """Called before a test run starts. - - New in python 2.7 - """ - - def stopTestRun(self): - """Called after a test run completes - - New in python 2.7 - """ - - def time(self, a_datetime): - """Provide a timestamp to represent the current time. - - This is useful when test activity is time delayed, or happening - concurrently and getting the system time between API calls will not - accurately represent the duration of tests (or the whole run). - - Calling time() sets the datetime used by the TestResult object. - Time is permitted to go backwards when using this call. - - :param a_datetime: A datetime.datetime object with TZ information or - None to reset the TestResult to gathering time from the system. - """ - self.__now = a_datetime - - def done(self): - """Called when the test runner is done. - - deprecated in favour of stopTestRun. - """ - - -class MultiTestResult(TestResult): - """A test result that dispatches to many test results.""" - - def __init__(self, *results): - TestResult.__init__(self) - self._results = map(ExtendedToOriginalDecorator, results) - - def _dispatch(self, message, *args, **kwargs): - for result in self._results: - getattr(result, message)(*args, **kwargs) - - def startTest(self, test): - self._dispatch('startTest', test) - - def stopTest(self, test): - self._dispatch('stopTest', test) - - def addError(self, test, error=None, details=None): - self._dispatch('addError', test, error, details=details) - - def addExpectedFailure(self, test, err=None, details=None): - self._dispatch('addExpectedFailure', test, err, details=details) - - def addFailure(self, test, err=None, details=None): - self._dispatch('addFailure', test, err, details=details) - - def addSkip(self, test, reason=None, details=None): - self._dispatch('addSkip', test, reason, details=details) - - def addSuccess(self, test, details=None): - self._dispatch('addSuccess', test, details=details) - - def addUnexpectedSuccess(self, test, details=None): - self._dispatch('addUnexpectedSuccess', test, details=details) - - def startTestRun(self): - self._dispatch('startTestRun') - - def stopTestRun(self): - self._dispatch('stopTestRun') - - def done(self): - self._dispatch('done') - - -class TextTestResult(TestResult): - """A TestResult which outputs activity to a text stream.""" - - def __init__(self, stream): - """Construct a TextTestResult writing to stream.""" - super(TextTestResult, self).__init__() - self.stream = stream - self.sep1 = '=' * 70 + '\n' - self.sep2 = '-' * 70 + '\n' - - def _delta_to_float(self, a_timedelta): - return (a_timedelta.days * 86400.0 + a_timedelta.seconds + - a_timedelta.microseconds / 1000000.0) - - def _show_list(self, label, error_list): - for test, output in error_list: - self.stream.write(self.sep1) - self.stream.write("%s: %s\n" % (label, test.id())) - self.stream.write(self.sep2) - self.stream.write(output) - - def startTestRun(self): - super(TextTestResult, self).startTestRun() - self.__start = self._now() - self.stream.write("Tests running...\n") - - def stopTestRun(self): - if self.testsRun != 1: - plural = 's' - else: - plural = '' - stop = self._now() - self._show_list('ERROR', self.errors) - self._show_list('FAIL', self.failures) - self.stream.write("Ran %d test%s in %.3fs\n\n" % - (self.testsRun, plural, - self._delta_to_float(stop - self.__start))) - if self.wasSuccessful(): - self.stream.write("OK\n") - else: - self.stream.write("FAILED (") - details = [] - details.append("failures=%d" % ( - len(self.failures) + len(self.errors))) - self.stream.write(", ".join(details)) - self.stream.write(")\n") - super(TextTestResult, self).stopTestRun() - - -class ThreadsafeForwardingResult(TestResult): - """A TestResult which ensures the target does not receive mixed up calls. - - This is used when receiving test results from multiple sources, and batches - up all the activity for a single test into a thread-safe batch where all - other ThreadsafeForwardingResult objects sharing the same semaphore will be - locked out. - - Typical use of ThreadsafeForwardingResult involves creating one - ThreadsafeForwardingResult per thread in a ConcurrentTestSuite. These - forward to the TestResult that the ConcurrentTestSuite run method was - called with. - - target.done() is called once for each ThreadsafeForwardingResult that - forwards to the same target. If the target's done() takes special action, - care should be taken to accommodate this. - """ - - def __init__(self, target, semaphore): - """Create a ThreadsafeForwardingResult forwarding to target. - - :param target: A TestResult. - :param semaphore: A threading.Semaphore with limit 1. - """ - TestResult.__init__(self) - self.result = ExtendedToOriginalDecorator(target) - self.semaphore = semaphore - - def addError(self, test, err=None, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addError(test, err, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def addExpectedFailure(self, test, err=None, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addExpectedFailure(test, err, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def addFailure(self, test, err=None, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addFailure(test, err, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def addSkip(self, test, reason=None, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addSkip(test, reason, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def addSuccess(self, test, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addSuccess(test, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def addUnexpectedSuccess(self, test, details=None): - self.semaphore.acquire() - try: - self.result.startTest(test) - self.result.addUnexpectedSuccess(test, details=details) - self.result.stopTest(test) - finally: - self.semaphore.release() - - def startTestRun(self): - self.semaphore.acquire() - try: - self.result.startTestRun() - finally: - self.semaphore.release() - - def stopTestRun(self): - self.semaphore.acquire() - try: - self.result.stopTestRun() - finally: - self.semaphore.release() - - def done(self): - self.semaphore.acquire() - try: - self.result.done() - finally: - self.semaphore.release() - - -class ExtendedToOriginalDecorator(object): - """Permit new TestResult API code to degrade gracefully with old results. - - This decorates an existing TestResult and converts missing outcomes - such as addSkip to older outcomes such as addSuccess. It also supports - the extended details protocol. In all cases the most recent protocol - is attempted first, and fallbacks only occur when the decorated result - does not support the newer style of calling. - """ - - def __init__(self, decorated): - self.decorated = decorated - - def __getattr__(self, name): - return getattr(self.decorated, name) - - def addError(self, test, err=None, details=None): - self._check_args(err, details) - if details is not None: - try: - return self.decorated.addError(test, details=details) - except TypeError: - # have to convert - err = self._details_to_exc_info(details) - return self.decorated.addError(test, err) - - def addExpectedFailure(self, test, err=None, details=None): - self._check_args(err, details) - addExpectedFailure = getattr( - self.decorated, 'addExpectedFailure', None) - if addExpectedFailure is None: - return self.addSuccess(test) - if details is not None: - try: - return addExpectedFailure(test, details=details) - except TypeError: - # have to convert - err = self._details_to_exc_info(details) - return addExpectedFailure(test, err) - - def addFailure(self, test, err=None, details=None): - self._check_args(err, details) - if details is not None: - try: - return self.decorated.addFailure(test, details=details) - except TypeError: - # have to convert - err = self._details_to_exc_info(details) - return self.decorated.addFailure(test, err) - - def addSkip(self, test, reason=None, details=None): - self._check_args(reason, details) - addSkip = getattr(self.decorated, 'addSkip', None) - if addSkip is None: - return self.decorated.addSuccess(test) - if details is not None: - try: - return addSkip(test, details=details) - except TypeError: - # have to convert - reason = _details_to_str(details) - return addSkip(test, reason) - - def addUnexpectedSuccess(self, test, details=None): - outcome = getattr(self.decorated, 'addUnexpectedSuccess', None) - if outcome is None: - return self.decorated.addSuccess(test) - if details is not None: - try: - return outcome(test, details=details) - except TypeError: - pass - return outcome(test) - - def addSuccess(self, test, details=None): - if details is not None: - try: - return self.decorated.addSuccess(test, details=details) - except TypeError: - pass - return self.decorated.addSuccess(test) - - def _check_args(self, err, details): - param_count = 0 - if err is not None: - param_count += 1 - if details is not None: - param_count += 1 - if param_count != 1: - raise ValueError("Must pass only one of err '%s' and details '%s" - % (err, details)) - - def _details_to_exc_info(self, details): - """Convert a details dict to an exc_info tuple.""" - return (_StringException, - _StringException(_details_to_str(details)), None) - - def done(self): - try: - return self.decorated.done() - except AttributeError: - return - - def progress(self, offset, whence): - method = getattr(self.decorated, 'progress', None) - if method is None: - return - return method(offset, whence) - - @property - def shouldStop(self): - return self.decorated.shouldStop - - def startTest(self, test): - return self.decorated.startTest(test) - - def startTestRun(self): - try: - return self.decorated.startTestRun() - except AttributeError: - return - - def stop(self): - return self.decorated.stop() - - def stopTest(self, test): - return self.decorated.stopTest(test) - - def stopTestRun(self): - try: - return self.decorated.stopTestRun() - except AttributeError: - return - - def tags(self, new_tags, gone_tags): - method = getattr(self.decorated, 'tags', None) - if method is None: - return - return method(new_tags, gone_tags) - - def time(self, a_datetime): - method = getattr(self.decorated, 'time', None) - if method is None: - return - return method(a_datetime) - - def wasSuccessful(self): - return self.decorated.wasSuccessful() - - -class _StringException(Exception): - """An exception made from an arbitrary string.""" - - def __hash__(self): - return id(self) - - def __str__(self): - """Stringify better than 2.x's default behaviour of ascii encoding.""" - return self.args[0] - - def __eq__(self, other): - try: - return self.args == other.args - except AttributeError: - return False - - -def _details_to_str(details): - """Convert a details dict to a string.""" - chars = [] - # sorted is for testing, may want to remove that and use a dict - # subclass with defined order for items instead. - for key, content in sorted(details.items()): - if content.content_type.type != 'text': - chars.append('Binary content: %s\n' % key) - continue - chars.append('Text attachment: %s\n' % key) - chars.append('------------\n') - chars.extend(content.iter_text()) - if not chars[-1].endswith('\n'): - chars.append('\n') - chars.append('------------\n') - return ''.join(chars) |