From: Jonathan Lange Date: Fri, 20 Apr 2012 11:34:31 +0000 (+0100) Subject: Merge updates from tag-collapsing-rigor. X-Git-Url: http://git.samba.org/samba.git/?p=third_party%2Fsubunit;a=commitdiff_plain;h=89f529754fbb52f5015e487261529f2b40ab2df1;hp=-c Merge updates from tag-collapsing-rigor. --- 89f529754fbb52f5015e487261529f2b40ab2df1 diff --combined python/subunit/test_results.py index 465b266,8db40e1..fea3b07 --- a/python/subunit/test_results.py +++ b/python/subunit/test_results.py @@@ -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() @@@ -354,6 -385,17 +356,6 @@@ 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( @@@ -365,16 -407,12 +367,16 @@@ 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. @@@ -392,6 -430,7 +394,6 @@@ 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) @@@ -399,15 -438,6 +401,15 @@@ 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], {})) @@@ -420,93 -450,6 +422,93 @@@ 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): @@@ -570,7 -513,7 +572,7 @@@ 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``.