subunit: Update to newer upstream version.
[samba.git] / lib / 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 datetime
20
21 import iso8601
22 import testtools
23
24 import subunit
25
26
27 # NOT a TestResult, because we are implementing the interface, not inheriting
28 # it.
29 class TestResultDecorator(object):
30     """General pass-through decorator.
31
32     This provides a base that other TestResults can inherit from to 
33     gain basic forwarding functionality. It also takes care of 
34     handling the case where the target doesn't support newer methods
35     or features by degrading them.
36     """
37
38     def __init__(self, decorated):
39         """Create a TestResultDecorator forwarding to decorated."""
40         # Make every decorator degrade gracefully.
41         self.decorated = testtools.ExtendedToOriginalDecorator(decorated)
42
43     def startTest(self, test):
44         return self.decorated.startTest(test)
45
46     def startTestRun(self):
47         return self.decorated.startTestRun()
48
49     def stopTest(self, test):
50         return self.decorated.stopTest(test)
51
52     def stopTestRun(self):
53         return self.decorated.stopTestRun()
54
55     def addError(self, test, err=None, details=None):
56         return self.decorated.addError(test, err, details=details)
57
58     def addFailure(self, test, err=None, details=None):
59         return self.decorated.addFailure(test, err, details=details)
60
61     def addSuccess(self, test, details=None):
62         return self.decorated.addSuccess(test, details=details)
63
64     def addSkip(self, test, reason=None, details=None):
65         return self.decorated.addSkip(test, reason, details=details)
66
67     def addExpectedFailure(self, test, err=None, details=None):
68         return self.decorated.addExpectedFailure(test, err, details=details)
69
70     def addUnexpectedSuccess(self, test, details=None):
71         return self.decorated.addUnexpectedSuccess(test, details=details)
72
73     def progress(self, offset, whence):
74         return self.decorated.progress(offset, whence)
75
76     def wasSuccessful(self):
77         return self.decorated.wasSuccessful()
78
79     @property
80     def shouldStop(self):
81         return self.decorated.shouldStop
82
83     def stop(self):
84         return self.decorated.stop()
85
86     def tags(self, new_tags, gone_tags):
87         return self.decorated.time(new_tags, gone_tags)
88
89     def time(self, a_datetime):
90         return self.decorated.time(a_datetime)
91
92
93 class HookedTestResultDecorator(TestResultDecorator):
94     """A TestResult which calls a hook on every event."""
95
96     def __init__(self, decorated):
97         self.super = super(HookedTestResultDecorator, self)
98         self.super.__init__(decorated)
99
100     def startTest(self, test):
101         self._before_event()
102         return self.super.startTest(test)
103
104     def startTestRun(self):
105         self._before_event()
106         return self.super.startTestRun()
107
108     def stopTest(self, test):
109         self._before_event()
110         return self.super.stopTest(test)
111
112     def stopTestRun(self):
113         self._before_event()
114         return self.super.stopTestRun()
115
116     def addError(self, test, err=None, details=None):
117         self._before_event()
118         return self.super.addError(test, err, details=details)
119
120     def addFailure(self, test, err=None, details=None):
121         self._before_event()
122         return self.super.addFailure(test, err, details=details)
123
124     def addSuccess(self, test, details=None):
125         self._before_event()
126         return self.super.addSuccess(test, details=details)
127
128     def addSkip(self, test, reason=None, details=None):
129         self._before_event()
130         return self.super.addSkip(test, reason, details=details)
131
132     def addExpectedFailure(self, test, err=None, details=None):
133         self._before_event()
134         return self.super.addExpectedFailure(test, err, details=details)
135
136     def addUnexpectedSuccess(self, test, details=None):
137         self._before_event()
138         return self.super.addUnexpectedSuccess(test, details=details)
139
140     def progress(self, offset, whence):
141         self._before_event()
142         return self.super.progress(offset, whence)
143
144     def wasSuccessful(self):
145         self._before_event()
146         return self.super.wasSuccessful()
147
148     @property
149     def shouldStop(self):
150         self._before_event()
151         return self.super.shouldStop
152
153     def stop(self):
154         self._before_event()
155         return self.super.stop()
156
157     def time(self, a_datetime):
158         self._before_event()
159         return self.super.time(a_datetime)
160
161
162 class AutoTimingTestResultDecorator(HookedTestResultDecorator):
163     """Decorate a TestResult to add time events to a test run.
164
165     By default this will cause a time event before every test event,
166     but if explicit time data is being provided by the test run, then
167     this decorator will turn itself off to prevent causing confusion.
168     """
169
170     def __init__(self, decorated):
171         self._time = None
172         super(AutoTimingTestResultDecorator, self).__init__(decorated)
173
174     def _before_event(self):
175         time = self._time
176         if time is not None:
177             return
178         time = datetime.datetime.utcnow().replace(tzinfo=iso8601.Utc())
179         self.decorated.time(time)
180
181     def progress(self, offset, whence):
182         return self.decorated.progress(offset, whence)
183
184     @property
185     def shouldStop(self):
186         return self.decorated.shouldStop
187
188     def time(self, a_datetime):
189         """Provide a timestamp for the current test activity.
190
191         :param a_datetime: If None, automatically add timestamps before every
192             event (this is the default behaviour if time() is not called at
193             all).  If not None, pass the provided time onto the decorated
194             result object and disable automatic timestamps.
195         """
196         self._time = a_datetime
197         return self.decorated.time(a_datetime)
198
199
200 class TestResultFilter(TestResultDecorator):
201     """A pyunit TestResult interface implementation which filters tests.
202
203     Tests that pass the filter are handed on to another TestResult instance
204     for further processing/reporting. To obtain the filtered results, 
205     the other instance must be interrogated.
206
207     :ivar result: The result that tests are passed to after filtering.
208     :ivar filter_predicate: The callback run to decide whether to pass 
209         a result.
210     """
211
212     def __init__(self, result, filter_error=False, filter_failure=False,
213         filter_success=True, filter_skip=False,
214         filter_predicate=None):
215         """Create a FilterResult object filtering to result.
216         
217         :param filter_error: Filter out errors.
218         :param filter_failure: Filter out failures.
219         :param filter_success: Filter out successful tests.
220         :param filter_skip: Filter out skipped tests.
221         :param filter_predicate: A callable taking (test, outcome, err,
222             details) and returning True if the result should be passed
223             through.  err and details may be none if no error or extra
224             metadata is available. outcome is the name of the outcome such
225             as 'success' or 'failure'.
226         """
227         TestResultDecorator.__init__(self, result)
228         self._filter_error = filter_error
229         self._filter_failure = filter_failure
230         self._filter_success = filter_success
231         self._filter_skip = filter_skip
232         if filter_predicate is None:
233             filter_predicate = lambda test, outcome, err, details: True
234         self.filter_predicate = filter_predicate
235         # The current test (for filtering tags)
236         self._current_test = None
237         # Has the current test been filtered (for outputting test tags)
238         self._current_test_filtered = None
239         # The (new, gone) tags for the current test.
240         self._current_test_tags = None
241         
242     def addError(self, test, err=None, details=None):
243         if (not self._filter_error and 
244             self.filter_predicate(test, 'error', err, details)):
245             self.decorated.startTest(test)
246             self.decorated.addError(test, err, details=details)
247         else:
248             self._filtered()
249
250     def addFailure(self, test, err=None, details=None):
251         if (not self._filter_failure and
252             self.filter_predicate(test, 'failure', err, details)):
253             self.decorated.startTest(test)
254             self.decorated.addFailure(test, err, details=details)
255         else:
256             self._filtered()
257
258     def addSkip(self, test, reason=None, details=None):
259         if (not self._filter_skip and
260             self.filter_predicate(test, 'skip', reason, details)):
261             self.decorated.startTest(test)
262             self.decorated.addSkip(test, reason, details=details)
263         else:
264             self._filtered()
265
266     def addSuccess(self, test, details=None):
267         if (not self._filter_success and
268             self.filter_predicate(test, 'success', None, details)):
269             self.decorated.startTest(test)
270             self.decorated.addSuccess(test, details=details)
271         else:
272             self._filtered()
273
274     def addExpectedFailure(self, test, err=None, details=None):
275         if self.filter_predicate(test, 'expectedfailure', err, details):
276             self.decorated.startTest(test)
277             return self.decorated.addExpectedFailure(test, err,
278                 details=details)
279         else:
280             self._filtered()
281
282     def addUnexpectedSuccess(self, test, details=None):
283         self.decorated.startTest(test)
284         return self.decorated.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     
299     def stopTest(self, test):
300         """Stop a test.
301         
302         Not directly passed to the client, but used for handling of tags
303         correctly.
304         """
305         if not self._current_test_filtered:
306             # Tags to output for this test.
307             if self._current_test_tags[0] or self._current_test_tags[1]:
308                 self.decorated.tags(*self._current_test_tags)
309             self.decorated.stopTest(test)
310         self._current_test = None
311         self._current_test_filtered = None
312         self._current_test_tags = None
313
314     def tags(self, new_tags, gone_tags):
315         """Handle tag instructions.
316
317         Adds and removes tags as appropriate. If a test is currently running,
318         tags are not affected for subsequent tests.
319         
320         :param new_tags: Tags to add,
321         :param gone_tags: Tags to remove.
322         """
323         if self._current_test is not None:
324             # gather the tags until the test stops.
325             self._current_test_tags[0].update(new_tags)
326             self._current_test_tags[0].difference_update(gone_tags)
327             self._current_test_tags[1].update(gone_tags)
328             self._current_test_tags[1].difference_update(new_tags)
329         return self.decorated.tags(new_tags, gone_tags)
330
331     def id_to_orig_id(self, id):
332         if id.startswith("subunit.RemotedTestCase."):
333             return id[len("subunit.RemotedTestCase."):]
334         return id