2 # subunit: extensions to Python unittest to get test results from subprocesses.
3 # Copyright (C) 2009 Robert Collins <robertc@robertcollins.net>
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.
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.
22 from testtools import TestCase
23 from testtools.compat import StringIO
24 from testtools.content import (
28 from testtools.testresult.doubles import ExtendedTestResult
31 import subunit.iso8601 as iso8601
32 import subunit.test_results
37 class LoggingDecorator(subunit.test_results.HookedTestResultDecorator):
39 def __init__(self, decorated):
41 super(LoggingDecorator, self).__init__(decorated)
43 def _before_event(self):
47 class AssertBeforeTestResult(LoggingDecorator):
48 """A TestResult for checking preconditions."""
50 def __init__(self, decorated, test):
52 super(AssertBeforeTestResult, self).__init__(decorated)
54 def _before_event(self):
55 self.test.assertEqual(1, self.earlier._calls)
56 super(AssertBeforeTestResult, self)._before_event()
59 class TimeCapturingResult(unittest.TestResult):
62 super(TimeCapturingResult, self).__init__()
66 def time(self, a_datetime):
67 self._calls.append(a_datetime)
70 class TestHookedTestResultDecorator(unittest.TestCase):
74 terminal = unittest.TestResult()
75 # Asserts that the call was made to self.result before asserter was
77 asserter = AssertBeforeTestResult(terminal, self)
78 # The result object we call, which much increase its call count.
79 self.result = LoggingDecorator(asserter)
80 asserter.earlier = self.result
81 self.decorated = asserter
84 # The hook in self.result must have been called
85 self.assertEqual(1, self.result._calls)
86 # The hook in asserter must have been called too, otherwise the
87 # assertion about ordering won't have completed.
88 self.assertEqual(1, self.decorated._calls)
90 def test_startTest(self):
91 self.result.startTest(self)
93 def test_startTestRun(self):
94 self.result.startTestRun()
96 def test_stopTest(self):
97 self.result.stopTest(self)
99 def test_stopTestRun(self):
100 self.result.stopTestRun()
102 def test_addError(self):
103 self.result.addError(self, subunit.RemoteError())
105 def test_addError_details(self):
106 self.result.addError(self, details={})
108 def test_addFailure(self):
109 self.result.addFailure(self, subunit.RemoteError())
111 def test_addFailure_details(self):
112 self.result.addFailure(self, details={})
114 def test_addSuccess(self):
115 self.result.addSuccess(self)
117 def test_addSuccess_details(self):
118 self.result.addSuccess(self, details={})
120 def test_addSkip(self):
121 self.result.addSkip(self, "foo")
123 def test_addSkip_details(self):
124 self.result.addSkip(self, details={})
126 def test_addExpectedFailure(self):
127 self.result.addExpectedFailure(self, subunit.RemoteError())
129 def test_addExpectedFailure_details(self):
130 self.result.addExpectedFailure(self, details={})
132 def test_addUnexpectedSuccess(self):
133 self.result.addUnexpectedSuccess(self)
135 def test_addUnexpectedSuccess_details(self):
136 self.result.addUnexpectedSuccess(self, details={})
138 def test_progress(self):
139 self.result.progress(1, subunit.PROGRESS_SET)
141 def test_wasSuccessful(self):
142 self.result.wasSuccessful()
144 def test_shouldStop(self):
145 self.result.shouldStop
151 self.result.time(None)
154 class TestAutoTimingTestResultDecorator(unittest.TestCase):
157 # And end to the chain which captures time events.
158 terminal = TimeCapturingResult()
159 # The result object under test.
160 self.result = subunit.test_results.AutoTimingTestResultDecorator(
162 self.decorated = terminal
164 def test_without_time_calls_time_is_called_and_not_None(self):
165 self.result.startTest(self)
166 self.assertEqual(1, len(self.decorated._calls))
167 self.assertNotEqual(None, self.decorated._calls[0])
169 def test_no_time_from_progress(self):
170 self.result.progress(1, subunit.PROGRESS_CUR)
171 self.assertEqual(0, len(self.decorated._calls))
173 def test_no_time_from_shouldStop(self):
174 self.decorated.stop()
175 self.result.shouldStop
176 self.assertEqual(0, len(self.decorated._calls))
178 def test_calling_time_inhibits_automatic_time(self):
179 # Calling time() outputs a time signal immediately and prevents
180 # automatically adding one when other methods are called.
181 time = datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc())
182 self.result.time(time)
183 self.result.startTest(self)
184 self.result.stopTest(self)
185 self.assertEqual(1, len(self.decorated._calls))
186 self.assertEqual(time, self.decorated._calls[0])
188 def test_calling_time_None_enables_automatic_time(self):
189 time = datetime.datetime(2009,10,11,12,13,14,15, iso8601.Utc())
190 self.result.time(time)
191 self.assertEqual(1, len(self.decorated._calls))
192 self.assertEqual(time, self.decorated._calls[0])
193 # Calling None passes the None through, in case other results care.
194 self.result.time(None)
195 self.assertEqual(2, len(self.decorated._calls))
196 self.assertEqual(None, self.decorated._calls[1])
197 # Calling other methods doesn't generate an automatic time event.
198 self.result.startTest(self)
199 self.assertEqual(3, len(self.decorated._calls))
200 self.assertNotEqual(None, self.decorated._calls[2])
202 def test_set_failfast_True(self):
203 self.assertFalse(self.decorated.failfast)
204 self.result.failfast = True
205 self.assertTrue(self.decorated.failfast)
208 class TestTagCollapsingDecorator(TestCase):
210 def test_tags_collapsed_outside_of_tests(self):
211 result = ExtendedTestResult()
212 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
213 tag_collapser.tags(set(['a']), set())
214 tag_collapser.tags(set(['b']), set())
215 tag_collapser.startTest(self)
217 [('tags', set(['a', 'b']), set([])),
221 def test_tags_collapsed_outside_of_tests_are_flushed(self):
222 result = ExtendedTestResult()
223 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
224 tag_collapser.startTestRun()
225 tag_collapser.tags(set(['a']), set())
226 tag_collapser.tags(set(['b']), set())
227 tag_collapser.startTest(self)
228 tag_collapser.addSuccess(self)
229 tag_collapser.stopTest(self)
230 tag_collapser.stopTestRun()
233 ('tags', set(['a', 'b']), set([])),
235 ('addSuccess', self),
240 def test_tags_forwarded_after_tests(self):
241 test = subunit.RemotedTestCase('foo')
242 result = ExtendedTestResult()
243 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
244 tag_collapser.startTestRun()
245 tag_collapser.startTest(test)
246 tag_collapser.addSuccess(test)
247 tag_collapser.stopTest(test)
248 tag_collapser.tags(set(['a']), set(['b']))
249 tag_collapser.stopTestRun()
253 ('addSuccess', test),
255 ('tags', set(['a']), set(['b'])),
260 def test_tags_collapsed_inside_of_tests(self):
261 result = ExtendedTestResult()
262 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
263 test = subunit.RemotedTestCase('foo')
264 tag_collapser.startTest(test)
265 tag_collapser.tags(set(['a']), set())
266 tag_collapser.tags(set(['b']), set(['a']))
267 tag_collapser.tags(set(['c']), set())
268 tag_collapser.stopTest(test)
270 [('startTest', test),
271 ('tags', set(['b', 'c']), set(['a'])),
275 def test_tags_collapsed_inside_of_tests_different_ordering(self):
276 result = ExtendedTestResult()
277 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
278 test = subunit.RemotedTestCase('foo')
279 tag_collapser.startTest(test)
280 tag_collapser.tags(set(), set(['a']))
281 tag_collapser.tags(set(['a', 'b']), set())
282 tag_collapser.tags(set(['c']), set())
283 tag_collapser.stopTest(test)
285 [('startTest', test),
286 ('tags', set(['a', 'b', 'c']), set()),
290 def test_tags_sent_before_result(self):
291 # Because addSuccess and friends tend to send subunit output
292 # immediately, and because 'tags:' before a result line means
293 # something different to 'tags:' after a result line, we need to be
294 # sure that tags are emitted before 'addSuccess' (or whatever).
295 result = ExtendedTestResult()
296 tag_collapser = subunit.test_results.TagCollapsingDecorator(result)
297 test = subunit.RemotedTestCase('foo')
298 tag_collapser.startTest(test)
299 tag_collapser.tags(set(['a']), set())
300 tag_collapser.addSuccess(test)
301 tag_collapser.stopTest(test)
303 [('startTest', test),
304 ('tags', set(['a']), set()),
305 ('addSuccess', test),
310 class TestTimeCollapsingDecorator(TestCase):
314 return datetime.datetime(
315 2000, 1, self.getUniqueInteger(), tzinfo=iso8601.UTC)
317 def test_initial_time_forwarded(self):
318 # We always forward the first time event we see.
319 result = ExtendedTestResult()
320 tag_collapser = subunit.test_results.TimeCollapsingDecorator(result)
321 a_time = self.make_time()
322 tag_collapser.time(a_time)
323 self.assertEquals([('time', a_time)], result._events)
325 def test_time_collapsed_to_first_and_last(self):
326 # If there are many consecutive time events, only the first and last
328 result = ExtendedTestResult()
329 tag_collapser = subunit.test_results.TimeCollapsingDecorator(result)
330 times = [self.make_time() for i in range(5)]
332 tag_collapser.time(a_time)
333 tag_collapser.startTest(subunit.RemotedTestCase('foo'))
335 [('time', times[0]), ('time', times[-1])], result._events[:-1])
337 def test_only_one_time_sent(self):
338 # If we receive a single time event followed by a non-time event, we
339 # send exactly one time event.
340 result = ExtendedTestResult()
341 tag_collapser = subunit.test_results.TimeCollapsingDecorator(result)
342 a_time = self.make_time()
343 tag_collapser.time(a_time)
344 tag_collapser.startTest(subunit.RemotedTestCase('foo'))
345 self.assertEquals([('time', a_time)], result._events[:-1])
347 def test_duplicate_times_not_sent(self):
348 # Many time events with the exact same time are collapsed into one
350 result = ExtendedTestResult()
351 tag_collapser = subunit.test_results.TimeCollapsingDecorator(result)
352 a_time = self.make_time()
354 tag_collapser.time(a_time)
355 tag_collapser.startTest(subunit.RemotedTestCase('foo'))
356 self.assertEquals([('time', a_time)], result._events[:-1])
358 def test_no_times_inserted(self):
359 result = ExtendedTestResult()
360 tag_collapser = subunit.test_results.TimeCollapsingDecorator(result)
361 a_time = self.make_time()
362 tag_collapser.time(a_time)
363 foo = subunit.RemotedTestCase('foo')
364 tag_collapser.startTest(foo)
365 tag_collapser.addSuccess(foo)
366 tag_collapser.stopTest(foo)
371 ('stopTest', foo)], result._events)
374 class TestByTestResultTests(testtools.TestCase):
377 super(TestByTestResultTests, self).setUp()
379 self.result = subunit.test_results.TestByTestResult(self.on_test)
380 if sys.version_info >= (3, 0):
381 self.result._now = iter(range(5)).__next__
383 self.result._now = iter(range(5)).next
385 def assertCalled(self, **kwargs):
393 defaults.update(kwargs)
394 self.assertEqual([defaults], self.log)
396 def on_test(self, **kwargs):
397 self.log.append(kwargs)
399 def test_no_tests_nothing_reported(self):
400 self.result.startTestRun()
401 self.result.stopTestRun()
402 self.assertEqual([], self.log)
404 def test_add_success(self):
405 self.result.startTest(self)
406 self.result.addSuccess(self)
407 self.result.stopTest(self)
408 self.assertCalled(status='success')
410 def test_add_success_details(self):
411 self.result.startTest(self)
412 details = {'foo': 'bar'}
413 self.result.addSuccess(self, details=details)
414 self.result.stopTest(self)
415 self.assertCalled(status='success', details=details)
418 if not getattr(self.result, 'tags', None):
419 self.skipTest("No tags in testtools")
420 self.result.tags(['foo'], [])
421 self.result.startTest(self)
422 self.result.addSuccess(self)
423 self.result.stopTest(self)
424 self.assertCalled(status='success', tags=set(['foo']))
426 def test_add_error(self):
427 self.result.startTest(self)
430 except ZeroDivisionError:
431 error = sys.exc_info()
432 self.result.addError(self, error)
433 self.result.stopTest(self)
436 details={'traceback': TracebackContent(error, self)})
438 def test_add_error_details(self):
439 self.result.startTest(self)
440 details = {"foo": text_content("bar")}
441 self.result.addError(self, details=details)
442 self.result.stopTest(self)
443 self.assertCalled(status='error', details=details)
445 def test_add_failure(self):
446 self.result.startTest(self)
448 self.fail("intentional failure")
449 except self.failureException:
450 failure = sys.exc_info()
451 self.result.addFailure(self, failure)
452 self.result.stopTest(self)
455 details={'traceback': TracebackContent(failure, self)})
457 def test_add_failure_details(self):
458 self.result.startTest(self)
459 details = {"foo": text_content("bar")}
460 self.result.addFailure(self, details=details)
461 self.result.stopTest(self)
462 self.assertCalled(status='failure', details=details)
464 def test_add_xfail(self):
465 self.result.startTest(self)
468 except ZeroDivisionError:
469 error = sys.exc_info()
470 self.result.addExpectedFailure(self, error)
471 self.result.stopTest(self)
474 details={'traceback': TracebackContent(error, self)})
476 def test_add_xfail_details(self):
477 self.result.startTest(self)
478 details = {"foo": text_content("bar")}
479 self.result.addExpectedFailure(self, details=details)
480 self.result.stopTest(self)
481 self.assertCalled(status='xfail', details=details)
483 def test_add_unexpected_success(self):
484 self.result.startTest(self)
485 details = {'foo': 'bar'}
486 self.result.addUnexpectedSuccess(self, details=details)
487 self.result.stopTest(self)
488 self.assertCalled(status='success', details=details)
490 def test_add_skip_reason(self):
491 self.result.startTest(self)
492 reason = self.getUniqueString()
493 self.result.addSkip(self, reason)
494 self.result.stopTest(self)
496 status='skip', details={'reason': text_content(reason)})
498 def test_add_skip_details(self):
499 self.result.startTest(self)
500 details = {'foo': 'bar'}
501 self.result.addSkip(self, details=details)
502 self.result.stopTest(self)
503 self.assertCalled(status='skip', details=details)
505 def test_twice(self):
506 self.result.startTest(self)
507 self.result.addSuccess(self, details={'foo': 'bar'})
508 self.result.stopTest(self)
509 self.result.startTest(self)
510 self.result.addSuccess(self)
511 self.result.stopTest(self)
518 'details': {'foo': 'bar'}},
529 class TestCsvResult(testtools.TestCase):
531 def parse_stream(self, stream):
533 reader = csv.reader(stream)
536 def test_csv_output(self):
538 result = subunit.test_results.CsvResult(stream)
539 if sys.version_info >= (3, 0):
540 result._now = iter(range(5)).__next__
542 result._now = iter(range(5)).next
543 result.startTestRun()
544 result.startTest(self)
545 result.addSuccess(self)
546 result.stopTest(self)
549 [['test', 'status', 'start_time', 'stop_time'],
550 [self.id(), 'success', '0', '1'],
552 self.parse_stream(stream))
554 def test_just_header_when_no_tests(self):
556 result = subunit.test_results.CsvResult(stream)
557 result.startTestRun()
560 [['test', 'status', 'start_time', 'stop_time']],
561 self.parse_stream(stream))
563 def test_no_output_before_events(self):
565 subunit.test_results.CsvResult(stream)
566 self.assertEqual([], self.parse_stream(stream))
570 loader = subunit.tests.TestUtil.TestLoader()
571 result = loader.loadTestsFromName(__name__)