git.samba.org - third_party/subunit/blob - python/subunit/test_results.py

   1 #
   2 #  subunit: extensions to Python unittest to get test results from subprocesses.
   3 #  Copyright (C) 2009  Robert Collins <robertc@robertcollins.net>
   4 #
   5 #  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
   6 #  license at the users choice. A copy of both licenses are available in the
   7 #  project source as Apache-2.0 and BSD. You may not use this file except in
   8 #  compliance with one of these two licences.
   9 #
  10 #  Unless required by applicable law or agreed to in writing, software
  11 #  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
  12 #  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
  13 #  license you chose for the specific language governing permissions and
  14 #  limitations under that license.
  15 #
  16
  17 """TestResult helper classes used to by subunit."""
  18
  19 import datetime
  20
  21 import iso8601
  22 import testtools
  23
  24
  25 # NOT a TestResult, because we are implementing the interface, not inheriting
  26 # it.
  27 class TestResultDecorator(object):
  28     """General pass-through decorator.
  29
  30     This provides a base that other TestResults can inherit from to
  31     gain basic forwarding functionality. It also takes care of
  32     handling the case where the target doesn't support newer methods
  33     or features by degrading them.
  34     """
  35
  36     def __init__(self, decorated):
  37         """Create a TestResultDecorator forwarding to decorated."""
  38         # Make every decorator degrade gracefully.
  39         self.decorated = testtools.ExtendedToOriginalDecorator(decorated)
  40
  41     def startTest(self, test):
  42         return self.decorated.startTest(test)
  43
  44     def startTestRun(self):
  45         return self.decorated.startTestRun()
  46
  47     def stopTest(self, test):
  48         return self.decorated.stopTest(test)
  49
  50     def stopTestRun(self):
  51         return self.decorated.stopTestRun()
  52
  53     def addError(self, test, err=None, details=None):
  54         return self.decorated.addError(test, err, details=details)
  55
  56     def addFailure(self, test, err=None, details=None):
  57         return self.decorated.addFailure(test, err, details=details)
  58
  59     def addSuccess(self, test, details=None):
  60         return self.decorated.addSuccess(test, details=details)
  61
  62     def addSkip(self, test, reason=None, details=None):
  63         return self.decorated.addSkip(test, reason, details=details)
  64
  65     def addExpectedFailure(self, test, err=None, details=None):
  66         return self.decorated.addExpectedFailure(test, err, details=details)
  67
  68     def addUnexpectedSuccess(self, test, details=None):
  69         return self.decorated.addUnexpectedSuccess(test, details=details)
  70
  71     def progress(self, offset, whence):
  72         return self.decorated.progress(offset, whence)
  73
  74     def wasSuccessful(self):
  75         return self.decorated.wasSuccessful()
  76
  77     @property
  78     def shouldStop(self):
  79         return self.decorated.shouldStop
  80
  81     def stop(self):
  82         return self.decorated.stop()
  83
  84     def tags(self, new_tags, gone_tags):
  85         return self.decorated.time(new_tags, gone_tags)
  86
  87     def time(self, a_datetime):
  88         return self.decorated.time(a_datetime)
  89
  90
  91 class HookedTestResultDecorator(TestResultDecorator):
  92     """A TestResult which calls a hook on every event."""
  93
  94     def __init__(self, decorated):
  95         self.super = super(HookedTestResultDecorator, self)
  96         self.super.__init__(decorated)
  97
  98     def startTest(self, test):
  99         self._before_event()
 100         return self.super.startTest(test)
 101
 102     def startTestRun(self):
 103         self._before_event()
 104         return self.super.startTestRun()
 105
 106     def stopTest(self, test):
 107         self._before_event()
 108         return self.super.stopTest(test)
 109
 110     def stopTestRun(self):
 111         self._before_event()
 112         return self.super.stopTestRun()
 113
 114     def addError(self, test, err=None, details=None):
 115         self._before_event()
 116         return self.super.addError(test, err, details=details)
 117
 118     def addFailure(self, test, err=None, details=None):
 119         self._before_event()
 120         return self.super.addFailure(test, err, details=details)
 121
 122     def addSuccess(self, test, details=None):
 123         self._before_event()
 124         return self.super.addSuccess(test, details=details)
 125
 126     def addSkip(self, test, reason=None, details=None):
 127         self._before_event()
 128         return self.super.addSkip(test, reason, details=details)
 129
 130     def addExpectedFailure(self, test, err=None, details=None):
 131         self._before_event()
 132         return self.super.addExpectedFailure(test, err, details=details)
 133
 134     def addUnexpectedSuccess(self, test, details=None):
 135         self._before_event()
 136         return self.super.addUnexpectedSuccess(test, details=details)
 137
 138     def progress(self, offset, whence):
 139         self._before_event()
 140         return self.super.progress(offset, whence)
 141
 142     def wasSuccessful(self):
 143         self._before_event()
 144         return self.super.wasSuccessful()
 145
 146     @property
 147     def shouldStop(self):
 148         self._before_event()
 149         return self.super.shouldStop
 150
 151     def stop(self):
 152         self._before_event()
 153         return self.super.stop()
 154
 155     def time(self, a_datetime):
 156         self._before_event()
 157         return self.super.time(a_datetime)
 158
 159
 160 class AutoTimingTestResultDecorator(HookedTestResultDecorator):
 161     """Decorate a TestResult to add time events to a test run.
 162
 163     By default this will cause a time event before every test event,
 164     but if explicit time data is being provided by the test run, then
 165     this decorator will turn itself off to prevent causing confusion.
 166     """
 167
 168     def __init__(self, decorated):
 169         self._time = None
 170         super(AutoTimingTestResultDecorator, self).__init__(decorated)
 171
 172     def _before_event(self):
 173         time = self._time
 174         if time is not None:
 175             return
 176         time = datetime.datetime.utcnow().replace(tzinfo=iso8601.Utc())
 177         self.decorated.time(time)
 178
 179     def progress(self, offset, whence):
 180         return self.decorated.progress(offset, whence)
 181
 182     @property
 183     def shouldStop(self):
 184         return self.decorated.shouldStop
 185
 186     def time(self, a_datetime):
 187         """Provide a timestamp for the current test activity.
 188
 189         :param a_datetime: If None, automatically add timestamps before every
 190             event (this is the default behaviour if time() is not called at
 191             all).  If not None, pass the provided time onto the decorated
 192             result object and disable automatic timestamps.
 193         """
 194         self._time = a_datetime
 195         return self.decorated.time(a_datetime)
 196
 197
 198 class TestResultFilter(TestResultDecorator):
 199     """A pyunit TestResult interface implementation which filters tests.
 200
 201     Tests that pass the filter are handed on to another TestResult instance
 202     for further processing/reporting. To obtain the filtered results,
 203     the other instance must be interrogated.
 204
 205     :ivar result: The result that tests are passed to after filtering.
 206     :ivar filter_predicate: The callback run to decide whether to pass
 207         a result.
 208     """
 209
 210     def __init__(self, result, filter_error=False, filter_failure=False,
 211         filter_success=True, filter_skip=False,
 212         filter_predicate=None):
 213         """Create a FilterResult object filtering to result.
 214
 215         :param filter_error: Filter out errors.
 216         :param filter_failure: Filter out failures.
 217         :param filter_success: Filter out successful tests.
 218         :param filter_skip: Filter out skipped tests.
 219         :param filter_predicate: A callable taking (test, outcome, err,
 220             details) and returning True if the result should be passed
 221             through.  err and details may be none if no error or extra
 222             metadata is available. outcome is the name of the outcome such
 223             as 'success' or 'failure'.
 224         """
 225         TestResultDecorator.__init__(self, result)
 226         self._filter_error = filter_error
 227         self._filter_failure = filter_failure
 228         self._filter_success = filter_success
 229         self._filter_skip = filter_skip
 230         if filter_predicate is None:
 231             filter_predicate = lambda test, outcome, err, details: True
 232         self.filter_predicate = filter_predicate
 233         # The current test (for filtering tags)
 234         self._current_test = None
 235         # Has the current test been filtered (for outputting test tags)
 236         self._current_test_filtered = None
 237         # The (new, gone) tags for the current test.
 238         self._current_test_tags = None
 239
 240         # Calls to this result that we don't know whether to forward on yet.
 241         self._buffered_calls = []
 242
 243     def addError(self, test, err=None, details=None):
 244         if (not self._filter_error and
 245             self.filter_predicate(test, 'error', err, details)):
 246             self._buffered_calls.append(
 247                 ('addError', [test, err], {'details': details}))
 248         else:
 249             self._filtered()
 250
 251     def addFailure(self, test, err=None, details=None):
 252         if (not self._filter_failure and
 253             self.filter_predicate(test, 'failure', err, details)):
 254             self._buffered_calls.append(
 255                 ('addFailure', [test, err], {'details': details}))
 256         else:
 257             self._filtered()
 258
 259     def addSkip(self, test, reason=None, details=None):
 260         if (not self._filter_skip and
 261             self.filter_predicate(test, 'skip', reason, details)):
 262             self._buffered_calls.append(
 263                 ('addSkip', [reason], {'details': details}))
 264         else:
 265             self._filtered()
 266
 267     def addSuccess(self, test, details=None):
 268         if (not self._filter_success and
 269             self.filter_predicate(test, 'success', None, details)):
 270             self._buffered_calls.append(
 271                 ('addSuccess', [test], {'details': details}))
 272         else:
 273             self._filtered()
 274
 275     def addExpectedFailure(self, test, err=None, details=None):
 276         if self.filter_predicate(test, 'expectedfailure', err, details):
 277             self._buffered_calls.append(
 278                 ('addExpectedFailure', [test, err], {'details': details}))
 279         else:
 280             self._filtered()
 281
 282     def addUnexpectedSuccess(self, test, details=None):
 283         self._buffered_calls.append(
 284             ('addUnexpectedSuccess', [test], {'details': details}))
 285
 286     def _filtered(self):
 287         self._current_test_filtered = True
 288
 289     def startTest(self, test):
 290         """Start a test.
 291
 292         Not directly passed to the client, but used for handling of tags
 293         correctly.
 294         """
 295         self._current_test = test
 296         self._current_test_filtered = False
 297         self._current_test_tags = set(), set()
 298         self._buffered_calls.append(('startTest', [test], {}))
 299
 300     def stopTest(self, test):
 301         """Stop a test.
 302
 303         Not directly passed to the client, but used for handling of tags
 304         correctly.
 305         """
 306         if not self._current_test_filtered:
 307             # Tags to output for this test.
 308             for method, args, kwargs in self._buffered_calls:
 309                 getattr(self.decorated, method)(*args, **kwargs)
 310             if self._current_test_tags[0] or self._current_test_tags[1]:
 311                 self.decorated.tags(*self._current_test_tags)
 312             self.decorated.stopTest(test)
 313         self._current_test = None
 314         self._current_test_filtered = None
 315         self._current_test_tags = None
 316         self._buffered_calls = []
 317
 318     def tags(self, new_tags, gone_tags):
 319         """Handle tag instructions.
 320
 321         Adds and removes tags as appropriate. If a test is currently running,
 322         tags are not affected for subsequent tests.
 323
 324         :param new_tags: Tags to add,
 325         :param gone_tags: Tags to remove.
 326         """
 327         if self._current_test is not None:
 328             # gather the tags until the test stops.
 329             self._current_test_tags[0].update(new_tags)
 330             self._current_test_tags[0].difference_update(gone_tags)
 331             self._current_test_tags[1].update(gone_tags)
 332             self._current_test_tags[1].difference_update(new_tags)
 333         return self.decorated.tags(new_tags, gone_tags)
 334
 335     def time(self, a_time):
 336         if self._current_test is not None:
 337             self._buffered_calls.append(('time', [a_time], {}))
 338         else:
 339             return self.decorated.time(a_time)
 340
 341     def id_to_orig_id(self, id):
 342         if id.startswith("subunit.RemotedTestCase."):
 343             return id[len("subunit.RemotedTestCase."):]
 344         return id
 345
 346
 347 class TestIdPrintingResult(testtools.TestResult):
 348
 349     def __init__(self, stream, show_times=False):
 350         """Create a FilterResult object outputting to stream."""
 351         testtools.TestResult.__init__(self)
 352         self._stream = stream
 353         self.failed_tests = 0
 354         self.__time = 0
 355         self.show_times = show_times
 356         self._test = None
 357         self._test_duration = 0
 358
 359     def addError(self, test, err):
 360         self.failed_tests += 1
 361         self._test = test
 362
 363     def addFailure(self, test, err):
 364         self.failed_tests += 1
 365         self._test = test
 366
 367     def addSuccess(self, test):
 368         self._test = test
 369
 370     def reportTest(self, test, duration):
 371         if self.show_times:
 372             seconds = duration.seconds
 373             seconds += duration.days * 3600 * 24
 374             seconds += duration.microseconds / 1000000.0
 375             self._stream.write(test.id() + ' %0.3f\n' % seconds)
 376         else:
 377             self._stream.write(test.id() + '\n')
 378
 379     def startTest(self, test):
 380         self._start_time = self._time()
 381
 382     def stopTest(self, test):
 383         test_duration = self._time() - self._start_time
 384         self.reportTest(self._test, test_duration)
 385
 386     def time(self, time):
 387         self.__time = time
 388
 389     def _time(self):
 390         return self.__time
 391
 392     def wasSuccessful(self):
 393         "Tells whether or not this result was a success"
 394         return self.failed_tests == 0