fea3b0780be3c9e1944453f30959c63cdd762983
[third_party/subunit] / 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 csv
20 import datetime
21
22 import testtools
23 from testtools.compat import all
24 from testtools.content import (
25     text_content,
26     TracebackContent,
27     )
28
29 from subunit import iso8601
30
31
32 # NOT a TestResult, because we are implementing the interface, not inheriting
33 # it.
34 class TestResultDecorator(object):
35     """General pass-through decorator.
36
37     This provides a base that other TestResults can inherit from to
38     gain basic forwarding functionality. It also takes care of
39     handling the case where the target doesn't support newer methods
40     or features by degrading them.
41     """
42
43     # XXX: Since lp:testtools r250, this is in testtools. Once it's released,
44     # we should gut this and just use that.
45
46     def __init__(self, decorated):
47         """Create a TestResultDecorator forwarding to decorated."""
48         # Make every decorator degrade gracefully.
49         self.decorated = testtools.ExtendedToOriginalDecorator(decorated)
50
51     def startTest(self, test):
52         return self.decorated.startTest(test)
53
54     def startTestRun(self):
55         return self.decorated.startTestRun()
56
57     def stopTest(self, test):
58         return self.decorated.stopTest(test)
59
60     def stopTestRun(self):
61         return self.decorated.stopTestRun()
62
63     def addError(self, test, err=None, details=None):
64         return self.decorated.addError(test, err, details=details)
65
66     def addFailure(self, test, err=None, details=None):
67         return self.decorated.addFailure(test, err, details=details)
68
69     def addSuccess(self, test, details=None):
70         return self.decorated.addSuccess(test, details=details)
71
72     def addSkip(self, test, reason=None, details=None):
73         return self.decorated.addSkip(test, reason, details=details)
74
75     def addExpectedFailure(self, test, err=None, details=None):
76         return self.decorated.addExpectedFailure(test, err, details=details)
77
78     def addUnexpectedSuccess(self, test, details=None):
79         return self.decorated.addUnexpectedSuccess(test, details=details)
80
81     def progress(self, offset, whence):
82         return self.decorated.progress(offset, whence)
83
84     def wasSuccessful(self):
85         return self.decorated.wasSuccessful()
86
87     @property
88     def shouldStop(self):
89         return self.decorated.shouldStop
90
91     def stop(self):
92         return self.decorated.stop()
93
94     @property
95     def testsRun(self):
96         return self.decorated.testsRun
97
98     def tags(self, new_tags, gone_tags):
99         return self.decorated.tags(new_tags, gone_tags)
100
101     def time(self, a_datetime):
102         return self.decorated.time(a_datetime)
103
104
105 class HookedTestResultDecorator(TestResultDecorator):
106     """A TestResult which calls a hook on every event."""
107
108     def __init__(self, decorated):
109         self.super = super(HookedTestResultDecorator, self)
110         self.super.__init__(decorated)
111
112     def startTest(self, test):
113         self._before_event()
114         return self.super.startTest(test)
115
116     def startTestRun(self):
117         self._before_event()
118         return self.super.startTestRun()
119
120     def stopTest(self, test):
121         self._before_event()
122         return self.super.stopTest(test)
123
124     def stopTestRun(self):
125         self._before_event()
126         return self.super.stopTestRun()
127
128     def addError(self, test, err=None, details=None):
129         self._before_event()
130         return self.super.addError(test, err, details=details)
131
132     def addFailure(self, test, err=None, details=None):
133         self._before_event()
134         return self.super.addFailure(test, err, details=details)
135
136     def addSuccess(self, test, details=None):
137         self._before_event()
138         return self.super.addSuccess(test, details=details)
139
140     def addSkip(self, test, reason=None, details=None):
141         self._before_event()
142         return self.super.addSkip(test, reason, details=details)
143
144     def addExpectedFailure(self, test, err=None, details=None):
145         self._before_event()
146         return self.super.addExpectedFailure(test, err, details=details)
147
148     def addUnexpectedSuccess(self, test, details=None):
149         self._before_event()
150         return self.super.addUnexpectedSuccess(test, details=details)
151
152     def progress(self, offset, whence):
153         self._before_event()
154         return self.super.progress(offset, whence)
155
156     def wasSuccessful(self):
157         self._before_event()
158         return self.super.wasSuccessful()
159
160     @property
161     def shouldStop(self):
162         self._before_event()
163         return self.super.shouldStop
164
165     def stop(self):
166         self._before_event()
167         return self.super.stop()
168
169     def time(self, a_datetime):
170         self._before_event()
171         return self.super.time(a_datetime)
172
173
174 class AutoTimingTestResultDecorator(HookedTestResultDecorator):
175     """Decorate a TestResult to add time events to a test run.
176
177     By default this will cause a time event before every test event,
178     but if explicit time data is being provided by the test run, then
179     this decorator will turn itself off to prevent causing confusion.
180     """
181
182     def __init__(self, decorated):
183         self._time = None
184         super(AutoTimingTestResultDecorator, self).__init__(decorated)
185
186     def _before_event(self):
187         time = self._time
188         if time is not None:
189             return
190         time = datetime.datetime.utcnow().replace(tzinfo=iso8601.Utc())
191         self.decorated.time(time)
192
193     def progress(self, offset, whence):
194         return self.decorated.progress(offset, whence)
195
196     @property
197     def shouldStop(self):
198         return self.decorated.shouldStop
199
200     def time(self, a_datetime):
201         """Provide a timestamp for the current test activity.
202
203         :param a_datetime: If None, automatically add timestamps before every
204             event (this is the default behaviour if time() is not called at
205             all).  If not None, pass the provided time onto the decorated
206             result object and disable automatic timestamps.
207         """
208         self._time = a_datetime
209         return self.decorated.time(a_datetime)
210
211
212 class TagCollapsingDecorator(HookedTestResultDecorator):
213     """Collapses many 'tags' calls into one where possible."""
214
215     def __init__(self, result):
216         super(TagCollapsingDecorator, self).__init__(result)
217         self._clear_tags()
218
219     def _clear_tags(self):
220         self._global_tags = set(), set()
221         self._current_test_tags = None
222
223     def _get_current_tags(self):
224         if self._current_test_tags:
225             return self._current_test_tags
226         return self._global_tags
227
228     def startTestRun(self):
229         super(TagCollapsingDecorator, self).startTestRun()
230         self._clear_tags()
231
232     def startTest(self, test):
233         """Start a test.
234
235         Not directly passed to the client, but used for handling of tags
236         correctly.
237         """
238         super(TagCollapsingDecorator, self).startTest(test)
239         self._current_test_tags = set(), set()
240
241     def stopTest(self, test):
242         super(TagCollapsingDecorator, self).stopTest(test)
243         self._current_test_tags = None
244
245     def _before_event(self):
246         new_tags, gone_tags = self._get_current_tags()
247         if new_tags or gone_tags:
248             self.decorated.tags(new_tags, gone_tags)
249         if self._current_test_tags:
250             self._current_test_tags = set(), set()
251         else:
252             self._global_tags = set(), set()
253
254     def tags(self, new_tags, gone_tags):
255         """Handle tag instructions.
256
257         Adds and removes tags as appropriate. If a test is currently running,
258         tags are not affected for subsequent tests.
259
260         :param new_tags: Tags to add,
261         :param gone_tags: Tags to remove.
262         """
263         current_new_tags, current_gone_tags = self._get_current_tags()
264         current_new_tags.update(new_tags)
265         current_new_tags.difference_update(gone_tags)
266         current_gone_tags.update(gone_tags)
267         current_gone_tags.difference_update(new_tags)
268
269
270 class TimeCollapsingDecorator(HookedTestResultDecorator):
271     """Only pass on the first and last of a consecutive sequence of times."""
272
273     def __init__(self, decorated):
274         super(TimeCollapsingDecorator, self).__init__(decorated)
275         self._last_received_time = None
276         self._last_sent_time = None
277
278     def _before_event(self):
279         if self._last_received_time is None:
280             return
281         if self._last_received_time != self._last_sent_time:
282             self.decorated.time(self._last_received_time)
283             self._last_sent_time = self._last_received_time
284         self._last_received_time = None
285
286     def time(self, a_time):
287         # Don't upcall, because we don't want to call _before_event, it's only
288         # for non-time events.
289         if self._last_received_time is None:
290             self.decorated.time(a_time)
291             self._last_sent_time = a_time
292         self._last_received_time = a_time
293
294
295 def and_predicates(predicates):
296     """Return a predicate that is true iff all predicates are true."""
297     # XXX: Should probably be in testtools to be better used by matchers. jml
298     return lambda *args, **kwargs: all(p(*args, **kwargs) for p in predicates)
299
300
301 def _make_tag_filter(with_tags, without_tags):
302     """Make a callback that checks tests against tags."""
303
304     with_tags = with_tags and set(with_tags) or None
305     without_tags = without_tags and set(without_tags) or None
306
307     def check_tags(test, outcome, err, details, tags):
308         if with_tags and not with_tags <= tags:
309             return False
310         if without_tags and bool(without_tags & tags):
311             return False
312         return True
313
314     return check_tags
315
316
317 class _PredicateFilter(TestResultDecorator):
318
319     def __init__(self, result, predicate):
320         super(_PredicateFilter, self).__init__(result)
321         self.decorated = TimeCollapsingDecorator(
322             TagCollapsingDecorator(self.decorated))
323         self._predicate = predicate
324         self._current_tags = set()
325         # The current test (for filtering tags)
326         self._current_test = None
327         # Has the current test been filtered (for outputting test tags)
328         self._current_test_filtered = None
329         # Calls to this result that we don't know whether to forward on yet.
330         self._buffered_calls = []
331
332     def filter_predicate(self, test, outcome, error, details):
333         # XXX: ExtendedToOriginalDecorator doesn't properly wrap current_tags.
334         # https://bugs.launchpad.net/testtools/+bug/978027
335         return self._predicate(
336             test, outcome, error, details, self._current_tags)
337
338     def addError(self, test, err=None, details=None):
339         if (self.filter_predicate(test, 'error', err, details)):
340             self._buffered_calls.append(
341                 ('addError', [test, err], {'details': details}))
342         else:
343             self._filtered()
344
345     def addFailure(self, test, err=None, details=None):
346         if (self.filter_predicate(test, 'failure', err, details)):
347             self._buffered_calls.append(
348                 ('addFailure', [test, err], {'details': details}))
349         else:
350             self._filtered()
351
352     def addSkip(self, test, reason=None, details=None):
353         if (self.filter_predicate(test, 'skip', reason, details)):
354             self._buffered_calls.append(
355                 ('addSkip', [test, reason], {'details': details}))
356         else:
357             self._filtered()
358
359     def addExpectedFailure(self, test, err=None, details=None):
360         if self.filter_predicate(test, 'expectedfailure', err, details):
361             self._buffered_calls.append(
362                 ('addExpectedFailure', [test, err], {'details': details}))
363         else:
364             self._filtered()
365
366     def addUnexpectedSuccess(self, test, details=None):
367         self._buffered_calls.append(
368             ('addUnexpectedSuccess', [test], {'details': details}))
369
370     def addSuccess(self, test, details=None):
371         if (self.filter_predicate(test, 'success', None, details)):
372             self._buffered_calls.append(
373                 ('addSuccess', [test], {'details': details}))
374         else:
375             self._filtered()
376
377     def _filtered(self):
378         self._current_test_filtered = True
379
380     def startTest(self, test):
381         """Start a test.
382
383         Not directly passed to the client, but used for handling of tags
384         correctly.
385         """
386         self._current_test = test
387         self._current_test_filtered = False
388         self._buffered_calls.append(('startTest', [test], {}))
389
390     def stopTest(self, test):
391         """Stop a test.
392
393         Not directly passed to the client, but used for handling of tags
394         correctly.
395         """
396         if not self._current_test_filtered:
397             for method, args, kwargs in self._buffered_calls:
398                 getattr(self.decorated, method)(*args, **kwargs)
399             self.decorated.stopTest(test)
400         self._current_test = None
401         self._current_test_filtered = None
402         self._buffered_calls = []
403
404     def tags(self, new_tags, gone_tags):
405         new_tags, gone_tags = set(new_tags), set(gone_tags)
406         self._current_tags.update(new_tags)
407         self._current_tags.difference_update(gone_tags)
408         if self._current_test is not None:
409             self._buffered_calls.append(('tags', [new_tags, gone_tags], {}))
410         else:
411             return super(_PredicateFilter, self).tags(new_tags, gone_tags)
412
413     def time(self, a_time):
414         if self._current_test is not None:
415             self._buffered_calls.append(('time', [a_time], {}))
416         else:
417             return self.decorated.time(a_time)
418
419     def id_to_orig_id(self, id):
420         if id.startswith("subunit.RemotedTestCase."):
421             return id[len("subunit.RemotedTestCase."):]
422         return id
423
424
425 class TestResultFilter(TestResultDecorator):
426     """A pyunit TestResult interface implementation which filters tests.
427
428     Tests that pass the filter are handed on to another TestResult instance
429     for further processing/reporting. To obtain the filtered results,
430     the other instance must be interrogated.
431
432     :ivar result: The result that tests are passed to after filtering.
433     :ivar filter_predicate: The callback run to decide whether to pass
434         a result.
435     """
436
437     def __init__(self, result, filter_error=False, filter_failure=False,
438         filter_success=True, filter_skip=False, filter_xfail=False,
439         filter_predicate=None, fixup_expected_failures=None):
440         """Create a FilterResult object filtering to result.
441
442         :param filter_error: Filter out errors.
443         :param filter_failure: Filter out failures.
444         :param filter_success: Filter out successful tests.
445         :param filter_skip: Filter out skipped tests.
446         :param filter_xfail: Filter out expected failure tests.
447         :param filter_predicate: A callable taking (test, outcome, err,
448             details) and returning True if the result should be passed
449             through.  err and details may be none if no error or extra
450             metadata is available. outcome is the name of the outcome such
451             as 'success' or 'failure'.
452         :param fixup_expected_failures: Set of test ids to consider known
453             failing.
454         """
455         predicates = []
456         if filter_error:
457             predicates.append(
458                 lambda t, outcome, e, d, tags: outcome != 'error')
459         if filter_failure:
460             predicates.append(
461                 lambda t, outcome, e, d, tags: outcome != 'failure')
462         if filter_success:
463             predicates.append(
464                 lambda t, outcome, e, d, tags: outcome != 'success')
465         if filter_skip:
466             predicates.append(
467                 lambda t, outcome, e, d, tags: outcome != 'skip')
468         if filter_xfail:
469             predicates.append(
470                 lambda t, outcome, e, d, tags: outcome != 'expectedfailure')
471         if filter_predicate is not None:
472             def compat(test, outcome, error, details, tags):
473                 # 0.0.7 and earlier did not support the 'tags' parameter.
474                 try:
475                     return filter_predicate(
476                         test, outcome, error, details, tags)
477                 except TypeError:
478                     return filter_predicate(test, outcome, error, details)
479             predicates.append(compat)
480         predicate = and_predicates(predicates)
481         super(TestResultFilter, self).__init__(
482             _PredicateFilter(result, predicate))
483         if fixup_expected_failures is None:
484             self._fixup_expected_failures = frozenset()
485         else:
486             self._fixup_expected_failures = fixup_expected_failures
487
488     def addError(self, test, err=None, details=None):
489         if self._failure_expected(test):
490             self.addExpectedFailure(test, err=err, details=details)
491         else:
492             super(TestResultFilter, self).addError(
493                 test, err=err, details=details)
494
495     def addFailure(self, test, err=None, details=None):
496         if self._failure_expected(test):
497             self.addExpectedFailure(test, err=err, details=details)
498         else:
499             super(TestResultFilter, self).addFailure(
500                 test, err=err, details=details)
501
502     def addSuccess(self, test, details=None):
503         if self._failure_expected(test):
504             self.addUnexpectedSuccess(test, details=details)
505         else:
506             super(TestResultFilter, self).addSuccess(test, details=details)
507
508     def _failure_expected(self, test):
509         return (test.id() in self._fixup_expected_failures)
510
511
512 class TestIdPrintingResult(testtools.TestResult):
513
514     def __init__(self, stream, show_times=False):
515         """Create a FilterResult object outputting to stream."""
516         super(TestIdPrintingResult, self).__init__()
517         self._stream = stream
518         self.failed_tests = 0
519         self.__time = None
520         self.show_times = show_times
521         self._test = None
522         self._test_duration = 0
523
524     def addError(self, test, err):
525         self.failed_tests += 1
526         self._test = test
527
528     def addFailure(self, test, err):
529         self.failed_tests += 1
530         self._test = test
531
532     def addSuccess(self, test):
533         self._test = test
534
535     def addSkip(self, test, reason=None, details=None):
536         self._test = test
537
538     def addUnexpectedSuccess(self, test, details=None):
539         self.failed_tests += 1
540         self._test = test
541
542     def addExpectedFailure(self, test, err=None, details=None):
543         self._test = test
544
545     def reportTest(self, test, duration):
546         if self.show_times:
547             seconds = duration.seconds
548             seconds += duration.days * 3600 * 24
549             seconds += duration.microseconds / 1000000.0
550             self._stream.write(test.id() + ' %0.3f\n' % seconds)
551         else:
552             self._stream.write(test.id() + '\n')
553
554     def startTest(self, test):
555         self._start_time = self._time()
556
557     def stopTest(self, test):
558         test_duration = self._time() - self._start_time
559         self.reportTest(self._test, test_duration)
560
561     def time(self, time):
562         self.__time = time
563
564     def _time(self):
565         return self.__time
566
567     def wasSuccessful(self):
568         "Tells whether or not this result was a success"
569         return self.failed_tests == 0
570
571
572 class TestByTestResult(testtools.TestResult):
573     """Call something every time a test completes."""
574
575 # XXX: Arguably belongs in testtools.
576
577     def __init__(self, on_test):
578         """Construct a ``TestByTestResult``.
579
580         :param on_test: A callable that take a test case, a status (one of
581             "success", "failure", "error", "skip", or "xfail"), a start time
582             (a ``datetime`` with timezone), a stop time, an iterable of tags,
583             and a details dict. Is called at the end of each test (i.e. on
584             ``stopTest``) with the accumulated values for that test.
585         """
586         super(TestByTestResult, self).__init__()
587         self._on_test = on_test
588
589     def startTest(self, test):
590         super(TestByTestResult, self).startTest(test)
591         self._start_time = self._now()
592         # There's no supported (i.e. tested) behaviour that relies on these
593         # being set, but it makes me more comfortable all the same. -- jml
594         self._status = None
595         self._details = None
596         self._stop_time = None
597
598     def stopTest(self, test):
599         self._stop_time = self._now()
600         super(TestByTestResult, self).stopTest(test)
601         self._on_test(
602             test=test,
603             status=self._status,
604             start_time=self._start_time,
605             stop_time=self._stop_time,
606             # current_tags is new in testtools 0.9.13.
607             tags=getattr(self, 'current_tags', None),
608             details=self._details)
609
610     def _err_to_details(self, test, err, details):
611         if details:
612             return details
613         return {'traceback': TracebackContent(err, test)}
614
615     def addSuccess(self, test, details=None):
616         super(TestByTestResult, self).addSuccess(test)
617         self._status = 'success'
618         self._details = details
619
620     def addFailure(self, test, err=None, details=None):
621         super(TestByTestResult, self).addFailure(test, err, details)
622         self._status = 'failure'
623         self._details = self._err_to_details(test, err, details)
624
625     def addError(self, test, err=None, details=None):
626         super(TestByTestResult, self).addError(test, err, details)
627         self._status = 'error'
628         self._details = self._err_to_details(test, err, details)
629
630     def addSkip(self, test, reason=None, details=None):
631         super(TestByTestResult, self).addSkip(test, reason, details)
632         self._status = 'skip'
633         if details is None:
634             details = {'reason': text_content(reason)}
635         elif reason:
636             # XXX: What if details already has 'reason' key?
637             details['reason'] = text_content(reason)
638         self._details = details
639
640     def addExpectedFailure(self, test, err=None, details=None):
641         super(TestByTestResult, self).addExpectedFailure(test, err, details)
642         self._status = 'xfail'
643         self._details = self._err_to_details(test, err, details)
644
645     def addUnexpectedSuccess(self, test, details=None):
646         super(TestByTestResult, self).addUnexpectedSuccess(test, details)
647         self._status = 'success'
648         self._details = details
649
650
651 class CsvResult(TestByTestResult):
652
653     def __init__(self, stream):
654         super(CsvResult, self).__init__(self._on_test)
655         self._write_row = csv.writer(stream).writerow
656
657     def _on_test(self, test, status, start_time, stop_time, tags, details):
658         self._write_row([test.id(), status, start_time, stop_time])
659
660     def startTestRun(self):
661         super(CsvResult, self).startTestRun()
662         self._write_row(['test', 'status', 'start_time', 'stop_time'])