s4-python: Install external included packages only if they're not present on the...
[nivanova/samba-autobuild/.git] / lib / testtools / testtools / testresult / real.py
1 # Copyright (c) 2008 Jonathan M. Lange. See LICENSE for details.
2
3 """Test results and related things."""
4
5 __metaclass__ = type
6 __all__ = [
7     'ExtendedToOriginalDecorator',
8     'MultiTestResult',
9     'TestResult',
10     'ThreadsafeForwardingResult',
11     ]
12
13 import datetime
14 import unittest
15
16
17 class TestResult(unittest.TestResult):
18     """Subclass of unittest.TestResult extending the protocol for flexability.
19
20     This test result supports an experimental protocol for providing additional
21     data to in test outcomes. All the outcome methods take an optional dict
22     'details'. If supplied any other detail parameters like 'err' or 'reason'
23     should not be provided. The details dict is a mapping from names to
24     MIME content objects (see testtools.content). This permits attaching
25     tracebacks, log files, or even large objects like databases that were
26     part of the test fixture. Until this API is accepted into upstream
27     Python it is considered experimental: it may be replaced at any point
28     by a newer version more in line with upstream Python. Compatibility would
29     be aimed for in this case, but may not be possible.
30
31     :ivar skip_reasons: A dict of skip-reasons -> list of tests. See addSkip.
32     """
33
34     def __init__(self):
35         super(TestResult, self).__init__()
36         self.skip_reasons = {}
37         self.__now = None
38         # -- Start: As per python 2.7 --
39         self.expectedFailures = []
40         self.unexpectedSuccesses = []
41         # -- End:   As per python 2.7 --
42
43     def addExpectedFailure(self, test, err=None, details=None):
44         """Called when a test has failed in an expected manner.
45
46         Like with addSuccess and addError, testStopped should still be called.
47
48         :param test: The test that has been skipped.
49         :param err: The exc_info of the error that was raised.
50         :return: None
51         """
52         # This is the python 2.7 implementation
53         self.expectedFailures.append(
54             (test, self._err_details_to_string(test, err, details)))
55
56     def addError(self, test, err=None, details=None):
57         """Called when an error has occurred. 'err' is a tuple of values as
58         returned by sys.exc_info().
59
60         :param details: Alternative way to supply details about the outcome.
61             see the class docstring for more information.
62         """
63         self.errors.append((test,
64             self._err_details_to_string(test, err, details)))
65
66     def addFailure(self, test, err=None, details=None):
67         """Called when an error has occurred. 'err' is a tuple of values as
68         returned by sys.exc_info().
69
70         :param details: Alternative way to supply details about the outcome.
71             see the class docstring for more information.
72         """
73         self.failures.append((test,
74             self._err_details_to_string(test, err, details)))
75
76     def addSkip(self, test, reason=None, details=None):
77         """Called when a test has been skipped rather than running.
78
79         Like with addSuccess and addError, testStopped should still be called.
80
81         This must be called by the TestCase. 'addError' and 'addFailure' will
82         not call addSkip, since they have no assumptions about the kind of
83         errors that a test can raise.
84
85         :param test: The test that has been skipped.
86         :param reason: The reason for the test being skipped. For instance,
87             u"pyGL is not available".
88         :param details: Alternative way to supply details about the outcome.
89             see the class docstring for more information.
90         :return: None
91         """
92         if reason is None:
93             reason = details.get('reason')
94             if reason is None:
95                 reason = 'No reason given'
96             else:
97                 reason = ''.join(reason.iter_text())
98         skip_list = self.skip_reasons.setdefault(reason, [])
99         skip_list.append(test)
100
101     def addSuccess(self, test, details=None):
102         """Called when a test succeeded."""
103
104     def addUnexpectedSuccess(self, test, details=None):
105         """Called when a test was expected to fail, but succeed."""
106         self.unexpectedSuccesses.append(test)
107
108     def _err_details_to_string(self, test, err=None, details=None):
109         """Convert an error in exc_info form or a contents dict to a string."""
110         if err is not None:
111             return self._exc_info_to_string(err, test)
112         return _details_to_str(details)
113
114     def _now(self):
115         """Return the current 'test time'.
116
117         If the time() method has not been called, this is equivalent to
118         datetime.now(), otherwise its the last supplied datestamp given to the
119         time() method.
120         """
121         if self.__now is None:
122             return datetime.datetime.now()
123         else:
124             return self.__now
125
126     def startTestRun(self):
127         """Called before a test run starts.
128
129         New in python 2.7
130         """
131
132     def stopTestRun(self):
133         """Called after a test run completes
134
135         New in python 2.7
136         """
137
138     def time(self, a_datetime):
139         """Provide a timestamp to represent the current time.
140
141         This is useful when test activity is time delayed, or happening
142         concurrently and getting the system time between API calls will not
143         accurately represent the duration of tests (or the whole run).
144
145         Calling time() sets the datetime used by the TestResult object.
146         Time is permitted to go backwards when using this call.
147
148         :param a_datetime: A datetime.datetime object with TZ information or
149             None to reset the TestResult to gathering time from the system.
150         """
151         self.__now = a_datetime
152
153     def done(self):
154         """Called when the test runner is done.
155
156         deprecated in favour of stopTestRun.
157         """
158
159
160 class MultiTestResult(TestResult):
161     """A test result that dispatches to many test results."""
162
163     def __init__(self, *results):
164         TestResult.__init__(self)
165         self._results = map(ExtendedToOriginalDecorator, results)
166
167     def _dispatch(self, message, *args, **kwargs):
168         for result in self._results:
169             getattr(result, message)(*args, **kwargs)
170
171     def startTest(self, test):
172         self._dispatch('startTest', test)
173
174     def stopTest(self, test):
175         self._dispatch('stopTest', test)
176
177     def addError(self, test, error=None, details=None):
178         self._dispatch('addError', test, error, details=details)
179
180     def addExpectedFailure(self, test, err=None, details=None):
181         self._dispatch('addExpectedFailure', test, err, details=details)
182
183     def addFailure(self, test, err=None, details=None):
184         self._dispatch('addFailure', test, err, details=details)
185
186     def addSkip(self, test, reason=None, details=None):
187         self._dispatch('addSkip', test, reason, details=details)
188
189     def addSuccess(self, test, details=None):
190         self._dispatch('addSuccess', test, details=details)
191
192     def addUnexpectedSuccess(self, test, details=None):
193         self._dispatch('addUnexpectedSuccess', test, details=details)
194
195     def startTestRun(self):
196         self._dispatch('startTestRun')
197
198     def stopTestRun(self):
199         self._dispatch('stopTestRun')
200
201     def done(self):
202         self._dispatch('done')
203
204
205 class TextTestResult(TestResult):
206     """A TestResult which outputs activity to a text stream."""
207
208     def __init__(self, stream):
209         """Construct a TextTestResult writing to stream."""
210         super(TextTestResult, self).__init__()
211         self.stream = stream
212         self.sep1 = '=' * 70 + '\n'
213         self.sep2 = '-' * 70 + '\n'
214
215     def _delta_to_float(self, a_timedelta):
216         return (a_timedelta.days * 86400.0 + a_timedelta.seconds +
217             a_timedelta.microseconds / 1000000.0)
218
219     def _show_list(self, label, error_list):
220         for test, output in error_list:
221             self.stream.write(self.sep1)
222             self.stream.write("%s: %s\n" % (label, test.id()))
223             self.stream.write(self.sep2)
224             self.stream.write(output)
225
226     def startTestRun(self):
227         super(TextTestResult, self).startTestRun()
228         self.__start = self._now()
229         self.stream.write("Tests running...\n")
230
231     def stopTestRun(self):
232         if self.testsRun != 1:
233             plural = 's'
234         else:
235             plural = ''
236         stop = self._now()
237         self._show_list('ERROR', self.errors)
238         self._show_list('FAIL', self.failures)
239         self.stream.write("Ran %d test%s in %.3fs\n\n" %
240             (self.testsRun, plural,
241              self._delta_to_float(stop - self.__start)))
242         if self.wasSuccessful():
243             self.stream.write("OK\n")
244         else:
245             self.stream.write("FAILED (")
246             details = []
247             details.append("failures=%d" % (
248                 len(self.failures) + len(self.errors)))
249             self.stream.write(", ".join(details))
250             self.stream.write(")\n")
251         super(TextTestResult, self).stopTestRun()
252
253
254 class ThreadsafeForwardingResult(TestResult):
255     """A TestResult which ensures the target does not receive mixed up calls.
256
257     This is used when receiving test results from multiple sources, and batches
258     up all the activity for a single test into a thread-safe batch where all
259     other ThreadsafeForwardingResult objects sharing the same semaphore will be
260     locked out.
261
262     Typical use of ThreadsafeForwardingResult involves creating one
263     ThreadsafeForwardingResult per thread in a ConcurrentTestSuite. These
264     forward to the TestResult that the ConcurrentTestSuite run method was
265     called with.
266
267     target.done() is called once for each ThreadsafeForwardingResult that
268     forwards to the same target. If the target's done() takes special action,
269     care should be taken to accommodate this.
270     """
271
272     def __init__(self, target, semaphore):
273         """Create a ThreadsafeForwardingResult forwarding to target.
274
275         :param target: A TestResult.
276         :param semaphore: A threading.Semaphore with limit 1.
277         """
278         TestResult.__init__(self)
279         self.result = ExtendedToOriginalDecorator(target)
280         self.semaphore = semaphore
281
282     def addError(self, test, err=None, details=None):
283         self.semaphore.acquire()
284         try:
285             self.result.startTest(test)
286             self.result.addError(test, err, details=details)
287             self.result.stopTest(test)
288         finally:
289             self.semaphore.release()
290
291     def addExpectedFailure(self, test, err=None, details=None):
292         self.semaphore.acquire()
293         try:
294             self.result.startTest(test)
295             self.result.addExpectedFailure(test, err, details=details)
296             self.result.stopTest(test)
297         finally:
298             self.semaphore.release()
299
300     def addFailure(self, test, err=None, details=None):
301         self.semaphore.acquire()
302         try:
303             self.result.startTest(test)
304             self.result.addFailure(test, err, details=details)
305             self.result.stopTest(test)
306         finally:
307             self.semaphore.release()
308
309     def addSkip(self, test, reason=None, details=None):
310         self.semaphore.acquire()
311         try:
312             self.result.startTest(test)
313             self.result.addSkip(test, reason, details=details)
314             self.result.stopTest(test)
315         finally:
316             self.semaphore.release()
317
318     def addSuccess(self, test, details=None):
319         self.semaphore.acquire()
320         try:
321             self.result.startTest(test)
322             self.result.addSuccess(test, details=details)
323             self.result.stopTest(test)
324         finally:
325             self.semaphore.release()
326
327     def addUnexpectedSuccess(self, test, details=None):
328         self.semaphore.acquire()
329         try:
330             self.result.startTest(test)
331             self.result.addUnexpectedSuccess(test, details=details)
332             self.result.stopTest(test)
333         finally:
334             self.semaphore.release()
335
336     def startTestRun(self):
337         self.semaphore.acquire()
338         try:
339             self.result.startTestRun()
340         finally:
341             self.semaphore.release()
342
343     def stopTestRun(self):
344         self.semaphore.acquire()
345         try:
346             self.result.stopTestRun()
347         finally:
348             self.semaphore.release()
349
350     def done(self):
351         self.semaphore.acquire()
352         try:
353             self.result.done()
354         finally:
355             self.semaphore.release()
356
357
358 class ExtendedToOriginalDecorator(object):
359     """Permit new TestResult API code to degrade gracefully with old results.
360
361     This decorates an existing TestResult and converts missing outcomes
362     such as addSkip to older outcomes such as addSuccess. It also supports
363     the extended details protocol. In all cases the most recent protocol
364     is attempted first, and fallbacks only occur when the decorated result
365     does not support the newer style of calling.
366     """
367
368     def __init__(self, decorated):
369         self.decorated = decorated
370
371     def __getattr__(self, name):
372         return getattr(self.decorated, name)
373
374     def addError(self, test, err=None, details=None):
375         self._check_args(err, details)
376         if details is not None:
377             try:
378                 return self.decorated.addError(test, details=details)
379             except TypeError:
380                 # have to convert
381                 err = self._details_to_exc_info(details)
382         return self.decorated.addError(test, err)
383
384     def addExpectedFailure(self, test, err=None, details=None):
385         self._check_args(err, details)
386         addExpectedFailure = getattr(
387             self.decorated, 'addExpectedFailure', None)
388         if addExpectedFailure is None:
389             return self.addSuccess(test)
390         if details is not None:
391             try:
392                 return addExpectedFailure(test, details=details)
393             except TypeError:
394                 # have to convert
395                 err = self._details_to_exc_info(details)
396         return addExpectedFailure(test, err)
397
398     def addFailure(self, test, err=None, details=None):
399         self._check_args(err, details)
400         if details is not None:
401             try:
402                 return self.decorated.addFailure(test, details=details)
403             except TypeError:
404                 # have to convert
405                 err = self._details_to_exc_info(details)
406         return self.decorated.addFailure(test, err)
407
408     def addSkip(self, test, reason=None, details=None):
409         self._check_args(reason, details)
410         addSkip = getattr(self.decorated, 'addSkip', None)
411         if addSkip is None:
412             return self.decorated.addSuccess(test)
413         if details is not None:
414             try:
415                 return addSkip(test, details=details)
416             except TypeError:
417                 # have to convert
418                 reason = _details_to_str(details)
419         return addSkip(test, reason)
420
421     def addUnexpectedSuccess(self, test, details=None):
422         outcome = getattr(self.decorated, 'addUnexpectedSuccess', None)
423         if outcome is None:
424             return self.decorated.addSuccess(test)
425         if details is not None:
426             try:
427                 return outcome(test, details=details)
428             except TypeError:
429                 pass
430         return outcome(test)
431
432     def addSuccess(self, test, details=None):
433         if details is not None:
434             try:
435                 return self.decorated.addSuccess(test, details=details)
436             except TypeError:
437                 pass
438         return self.decorated.addSuccess(test)
439
440     def _check_args(self, err, details):
441         param_count = 0
442         if err is not None:
443             param_count += 1
444         if details is not None:
445             param_count += 1
446         if param_count != 1:
447             raise ValueError("Must pass only one of err '%s' and details '%s"
448                 % (err, details))
449
450     def _details_to_exc_info(self, details):
451         """Convert a details dict to an exc_info tuple."""
452         return (_StringException,
453             _StringException(_details_to_str(details)), None)
454
455     def done(self):
456         try:
457             return self.decorated.done()
458         except AttributeError:
459             return
460
461     def progress(self, offset, whence):
462         method = getattr(self.decorated, 'progress', None)
463         if method is None:
464             return
465         return method(offset, whence)
466
467     @property
468     def shouldStop(self):
469         return self.decorated.shouldStop
470
471     def startTest(self, test):
472         return self.decorated.startTest(test)
473
474     def startTestRun(self):
475         try:
476             return self.decorated.startTestRun()
477         except AttributeError:
478             return
479
480     def stop(self):
481         return self.decorated.stop()
482
483     def stopTest(self, test):
484         return self.decorated.stopTest(test)
485
486     def stopTestRun(self):
487         try:
488             return self.decorated.stopTestRun()
489         except AttributeError:
490             return
491
492     def tags(self, new_tags, gone_tags):
493         method = getattr(self.decorated, 'tags', None)
494         if method is None:
495             return
496         return method(new_tags, gone_tags)
497
498     def time(self, a_datetime):
499         method = getattr(self.decorated, 'time', None)
500         if method is None:
501             return
502         return method(a_datetime)
503
504     def wasSuccessful(self):
505         return self.decorated.wasSuccessful()
506
507
508 class _StringException(Exception):
509     """An exception made from an arbitrary string."""
510
511     def __hash__(self):
512         return id(self)
513
514     def __str__(self):
515         """Stringify better than 2.x's default behaviour of ascii encoding."""
516         return self.args[0]
517
518     def __eq__(self, other):
519         try:
520             return self.args == other.args
521         except AttributeError:
522             return False
523
524
525 def _details_to_str(details):
526     """Convert a details dict to a string."""
527     chars = []
528     # sorted is for testing, may want to remove that and use a dict
529     # subclass with defined order for items instead.
530     for key, content in sorted(details.items()):
531         if content.content_type.type != 'text':
532             chars.append('Binary content: %s\n' % key)
533             continue
534         chars.append('Text attachment: %s\n' % key)
535         chars.append('------------\n')
536         chars.extend(content.iter_text())
537         if not chars[-1].endswith('\n'):
538             chars.append('\n')
539         chars.append('------------\n')
540     return ''.join(chars)