Merge updates from tag-collapsing-rigor.
authorJonathan Lange <jml@mumak.net>
Fri, 20 Apr 2012 11:34:31 +0000 (12:34 +0100)
committerJonathan Lange <jml@mumak.net>
Fri, 20 Apr 2012 11:34:31 +0000 (12:34 +0100)
1  2 
python/subunit/test_results.py

index 465b266491635f7625722efc579703e0820ddb4a,8db40e15a8d31536794b81879c7374abb1f4f648..fea3b0780be3c9e1944453f30959c63cdd762983
@@@ -20,7 -20,6 +20,7 @@@ import cs
  import datetime
  
  import testtools
 +from testtools.compat import all
  from testtools.content import (
      text_content,
      TracebackContent,
@@@ -40,9 -39,6 +40,9 @@@ class TestResultDecorator(object)
      or features by degrading them.
      """
  
 +    # XXX: Since lp:testtools r250, this is in testtools. Once it's released,
 +    # we should gut this and just use that.
 +
      def __init__(self, decorated):
          """Create a TestResultDecorator forwarding to decorated."""
          # Make every decorator degrade gracefully.
@@@ -248,6 -244,8 +248,8 @@@ class TagCollapsingDecorator(HookedTest
              self.decorated.tags(new_tags, gone_tags)
          if self._current_test_tags:
              self._current_test_tags = set(), set()
+         else:
+             self._global_tags = set(), set()
  
      def tags(self, new_tags, gone_tags):
          """Handle tag instructions.
@@@ -290,60 -288,93 +292,60 @@@ class TimeCollapsingDecorator(HookedTes
          self._last_received_time = a_time
  
  
 -def all_true(bools):
 -    """Return True if all of 'bools' are True. False otherwise."""
 -    for b in bools:
 -        if not b:
 -            return False
 -    return True
 +def and_predicates(predicates):
 +    """Return a predicate that is true iff all predicates are true."""
 +    # XXX: Should probably be in testtools to be better used by matchers. jml
 +    return lambda *args, **kwargs: all(p(*args, **kwargs) for p in predicates)
  
  
 -class TestResultFilter(TestResultDecorator):
 -    """A pyunit TestResult interface implementation which filters tests.
 +def _make_tag_filter(with_tags, without_tags):
 +    """Make a callback that checks tests against tags."""
  
 -    Tests that pass the filter are handed on to another TestResult instance
 -    for further processing/reporting. To obtain the filtered results,
 -    the other instance must be interrogated.
 +    with_tags = with_tags and set(with_tags) or None
 +    without_tags = without_tags and set(without_tags) or None
  
 -    :ivar result: The result that tests are passed to after filtering.
 -    :ivar filter_predicate: The callback run to decide whether to pass
 -        a result.
 -    """
 +    def check_tags(test, outcome, err, details, tags):
 +        if with_tags and not with_tags <= tags:
 +            return False
 +        if without_tags and bool(without_tags & tags):
 +            return False
 +        return True
  
 -    def __init__(self, result, filter_error=False, filter_failure=False,
 -        filter_success=True, filter_skip=False, filter_xfail=False,
 -        filter_predicate=None, fixup_expected_failures=None):
 -        """Create a FilterResult object filtering to result.
 +    return check_tags
  
 -        :param filter_error: Filter out errors.
 -        :param filter_failure: Filter out failures.
 -        :param filter_success: Filter out successful tests.
 -        :param filter_skip: Filter out skipped tests.
 -        :param filter_xfail: Filter out expected failure tests.
 -        :param filter_predicate: A callable taking (test, outcome, err,
 -            details) and returning True if the result should be passed
 -            through.  err and details may be none if no error or extra
 -            metadata is available. outcome is the name of the outcome such
 -            as 'success' or 'failure'.
 -        :param fixup_expected_failures: Set of test ids to consider known
 -            failing.
 -        """
 -        super(TestResultFilter, self).__init__(result)
 +
 +class _PredicateFilter(TestResultDecorator):
 +
 +    def __init__(self, result, predicate):
 +        super(_PredicateFilter, self).__init__(result)
          self.decorated = TimeCollapsingDecorator(
              TagCollapsingDecorator(self.decorated))
 -        predicates = []
 -        if filter_error:
 -            predicates.append(lambda t, outcome, e, d: outcome != 'error')
 -        if filter_failure:
 -            predicates.append(lambda t, outcome, e, d: outcome != 'failure')
 -        if filter_success:
 -            predicates.append(lambda t, outcome, e, d: outcome != 'success')
 -        if filter_skip:
 -            predicates.append(lambda t, outcome, e, d: outcome != 'skip')
 -        if filter_xfail:
 -            predicates.append(lambda t, outcome, e, d: outcome != 'expectedfailure')
 -        if filter_predicate is not None:
 -            predicates.append(filter_predicate)
 -        self.filter_predicate = (
 -            lambda test, outcome, err, details:
 -                all_true(p(test, outcome, err, details) for p in predicates))
 +        self._predicate = predicate
 +        self._current_tags = set()
          # The current test (for filtering tags)
          self._current_test = None
          # Has the current test been filtered (for outputting test tags)
          self._current_test_filtered = None
          # Calls to this result that we don't know whether to forward on yet.
          self._buffered_calls = []
 -        if fixup_expected_failures is None:
 -            self._fixup_expected_failures = frozenset()
 -        else:
 -            self._fixup_expected_failures = fixup_expected_failures
 +
 +    def filter_predicate(self, test, outcome, error, details):
 +        # XXX: ExtendedToOriginalDecorator doesn't properly wrap current_tags.
 +        # https://bugs.launchpad.net/testtools/+bug/978027
 +        return self._predicate(
 +            test, outcome, error, details, self._current_tags)
  
      def addError(self, test, err=None, details=None):
          if (self.filter_predicate(test, 'error', err, details)):
 -            if self._failure_expected(test):
 -                self._buffered_calls.append(
 -                    ('addExpectedFailure', [test, err], {'details': details}))
 -            else:
 -                self._buffered_calls.append(
 -                    ('addError', [test, err], {'details': details}))
 +            self._buffered_calls.append(
 +                ('addError', [test, err], {'details': details}))
          else:
              self._filtered()
  
      def addFailure(self, test, err=None, details=None):
          if (self.filter_predicate(test, 'failure', err, details)):
 -            if self._failure_expected(test):
 -                self._buffered_calls.append(
 -                    ('addExpectedFailure', [test, err], {'details': details}))
 -            else:
 -                self._buffered_calls.append(
 -                    ('addFailure', [test, err], {'details': details}))
 +            self._buffered_calls.append(
 +                ('addFailure', [test, err], {'details': details}))
          else:
              self._filtered()
  
          else:
              self._filtered()
  
 -    def addSuccess(self, test, details=None):
 -        if (self.filter_predicate(test, 'success', None, details)):
 -            if self._failure_expected(test):
 -                self._buffered_calls.append(
 -                    ('addUnexpectedSuccess', [test], {'details': details}))
 -            else:
 -                self._buffered_calls.append(
 -                    ('addSuccess', [test], {'details': details}))
 -        else:
 -            self._filtered()
 -
      def addExpectedFailure(self, test, err=None, details=None):
          if self.filter_predicate(test, 'expectedfailure', err, details):
              self._buffered_calls.append(
          self._buffered_calls.append(
              ('addUnexpectedSuccess', [test], {'details': details}))
  
 +    def addSuccess(self, test, details=None):
 +        if (self.filter_predicate(test, 'success', None, details)):
 +            self._buffered_calls.append(
 +                ('addSuccess', [test], {'details': details}))
 +        else:
 +            self._filtered()
 +
      def _filtered(self):
          self._current_test_filtered = True
  
 -    def _failure_expected(self, test):
 -        return (test.id() in self._fixup_expected_failures)
 -
      def startTest(self, test):
          """Start a test.
  
          correctly.
          """
          if not self._current_test_filtered:
 -            # Tags to output for this test.
              for method, args, kwargs in self._buffered_calls:
                  getattr(self.decorated, method)(*args, **kwargs)
              self.decorated.stopTest(test)
          self._current_test_filtered = None
          self._buffered_calls = []
  
 +    def tags(self, new_tags, gone_tags):
 +        new_tags, gone_tags = set(new_tags), set(gone_tags)
 +        self._current_tags.update(new_tags)
 +        self._current_tags.difference_update(gone_tags)
 +        if self._current_test is not None:
 +            self._buffered_calls.append(('tags', [new_tags, gone_tags], {}))
 +        else:
 +            return super(_PredicateFilter, self).tags(new_tags, gone_tags)
 +
      def time(self, a_time):
          if self._current_test is not None:
              self._buffered_calls.append(('time', [a_time], {}))
          return id
  
  
 +class TestResultFilter(TestResultDecorator):
 +    """A pyunit TestResult interface implementation which filters tests.
 +
 +    Tests that pass the filter are handed on to another TestResult instance
 +    for further processing/reporting. To obtain the filtered results,
 +    the other instance must be interrogated.
 +
 +    :ivar result: The result that tests are passed to after filtering.
 +    :ivar filter_predicate: The callback run to decide whether to pass
 +        a result.
 +    """
 +
 +    def __init__(self, result, filter_error=False, filter_failure=False,
 +        filter_success=True, filter_skip=False, filter_xfail=False,
 +        filter_predicate=None, fixup_expected_failures=None):
 +        """Create a FilterResult object filtering to result.
 +
 +        :param filter_error: Filter out errors.
 +        :param filter_failure: Filter out failures.
 +        :param filter_success: Filter out successful tests.
 +        :param filter_skip: Filter out skipped tests.
 +        :param filter_xfail: Filter out expected failure tests.
 +        :param filter_predicate: A callable taking (test, outcome, err,
 +            details) and returning True if the result should be passed
 +            through.  err and details may be none if no error or extra
 +            metadata is available. outcome is the name of the outcome such
 +            as 'success' or 'failure'.
 +        :param fixup_expected_failures: Set of test ids to consider known
 +            failing.
 +        """
 +        predicates = []
 +        if filter_error:
 +            predicates.append(
 +                lambda t, outcome, e, d, tags: outcome != 'error')
 +        if filter_failure:
 +            predicates.append(
 +                lambda t, outcome, e, d, tags: outcome != 'failure')
 +        if filter_success:
 +            predicates.append(
 +                lambda t, outcome, e, d, tags: outcome != 'success')
 +        if filter_skip:
 +            predicates.append(
 +                lambda t, outcome, e, d, tags: outcome != 'skip')
 +        if filter_xfail:
 +            predicates.append(
 +                lambda t, outcome, e, d, tags: outcome != 'expectedfailure')
 +        if filter_predicate is not None:
 +            def compat(test, outcome, error, details, tags):
 +                # 0.0.7 and earlier did not support the 'tags' parameter.
 +                try:
 +                    return filter_predicate(
 +                        test, outcome, error, details, tags)
 +                except TypeError:
 +                    return filter_predicate(test, outcome, error, details)
 +            predicates.append(compat)
 +        predicate = and_predicates(predicates)
 +        super(TestResultFilter, self).__init__(
 +            _PredicateFilter(result, predicate))
 +        if fixup_expected_failures is None:
 +            self._fixup_expected_failures = frozenset()
 +        else:
 +            self._fixup_expected_failures = fixup_expected_failures
 +
 +    def addError(self, test, err=None, details=None):
 +        if self._failure_expected(test):
 +            self.addExpectedFailure(test, err=err, details=details)
 +        else:
 +            super(TestResultFilter, self).addError(
 +                test, err=err, details=details)
 +
 +    def addFailure(self, test, err=None, details=None):
 +        if self._failure_expected(test):
 +            self.addExpectedFailure(test, err=err, details=details)
 +        else:
 +            super(TestResultFilter, self).addFailure(
 +                test, err=err, details=details)
 +
 +    def addSuccess(self, test, details=None):
 +        if self._failure_expected(test):
 +            self.addUnexpectedSuccess(test, details=details)
 +        else:
 +            super(TestResultFilter, self).addSuccess(test, details=details)
 +
 +    def _failure_expected(self, test):
 +        return (test.id() in self._fixup_expected_failures)
 +
 +
  class TestIdPrintingResult(testtools.TestResult):
  
      def __init__(self, stream, show_times=False):
  class TestByTestResult(testtools.TestResult):
      """Call something every time a test completes."""
  
 -    # XXX: Arguably belongs in testtools.
 +# XXX: Arguably belongs in testtools.
  
      def __init__(self, on_test):
          """Construct a ``TestByTestResult``.