Put testtools directly under lib/ to make it easier to install from Samba 4.

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)