Always forward time events from TestResultFilter.
[third_party/subunit] / python / subunit / tests / test_subunit_filter.py
1 #
2 #  subunit: extensions to python unittest to get test results from subprocesses.
3 #  Copyright (C) 2005  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 """Tests for subunit.TestResultFilter."""
18
19 from datetime import datetime
20 from subunit import iso8601
21 import unittest
22
23 from testtools import TestCase
24 from testtools.compat import _b, BytesIO, StringIO
25 from testtools.testresult.doubles import ExtendedTestResult
26
27 import subunit
28 from subunit.test_results import TestResultFilter
29
30
31 class TestTestResultFilter(TestCase):
32     """Test for TestResultFilter, a TestResult object which filters tests."""
33
34     # While TestResultFilter works on python objects, using a subunit stream
35     # is an easy pithy way of getting a series of test objects to call into
36     # the TestResult, and as TestResultFilter is intended for use with subunit
37     # also has the benefit of detecting any interface skew issues.
38     example_subunit_stream = _b("""\
39 tags: global
40 test passed
41 success passed
42 test failed
43 tags: local
44 failure failed
45 test error
46 error error [
47 error details
48 ]
49 test skipped
50 skip skipped
51 test todo
52 xfail todo
53 """)
54
55     def run_tests(self, result_filter, input_stream=None):
56         """Run tests through the given filter.
57
58         :param result_filter: A filtering TestResult object.
59         :param input_stream: Bytes of subunit stream data. If not provided,
60             uses TestTestResultFilter.example_subunit_stream.
61         """
62         if input_stream is None:
63             input_stream = self.example_subunit_stream
64         test = subunit.ProtocolTestCase(BytesIO(input_stream))
65         test.run(result_filter)
66
67     def test_default(self):
68         """The default is to exclude success and include everything else."""
69         filtered_result = unittest.TestResult()
70         result_filter = TestResultFilter(filtered_result)
71         self.run_tests(result_filter)
72         # skips are seen as success by default python TestResult.
73         self.assertEqual(['error'],
74             [error[0].id() for error in filtered_result.errors])
75         self.assertEqual(['failed'],
76             [failure[0].id() for failure in
77             filtered_result.failures])
78         self.assertEqual(4, filtered_result.testsRun)
79
80     def test_exclude_errors(self):
81         filtered_result = unittest.TestResult()
82         result_filter = TestResultFilter(filtered_result, filter_error=True)
83         self.run_tests(result_filter)
84         # skips are seen as errors by default python TestResult.
85         self.assertEqual([], filtered_result.errors)
86         self.assertEqual(['failed'],
87             [failure[0].id() for failure in
88             filtered_result.failures])
89         self.assertEqual(3, filtered_result.testsRun)
90
91     def test_fixup_expected_failures(self):
92         filtered_result = unittest.TestResult()
93         result_filter = TestResultFilter(filtered_result,
94             fixup_expected_failures=set(["failed"]))
95         self.run_tests(result_filter)
96         self.assertEqual(['failed', 'todo'],
97             [failure[0].id() for failure in filtered_result.expectedFailures])
98         self.assertEqual([], filtered_result.failures)
99         self.assertEqual(4, filtered_result.testsRun)
100
101     def test_fixup_expected_errors(self):
102         filtered_result = unittest.TestResult()
103         result_filter = TestResultFilter(filtered_result,
104             fixup_expected_failures=set(["error"]))
105         self.run_tests(result_filter)
106         self.assertEqual(['error', 'todo'],
107             [failure[0].id() for failure in filtered_result.expectedFailures])
108         self.assertEqual([], filtered_result.errors)
109         self.assertEqual(4, filtered_result.testsRun)
110
111     def test_fixup_unexpected_success(self):
112         filtered_result = unittest.TestResult()
113         result_filter = TestResultFilter(filtered_result, filter_success=False,
114             fixup_expected_failures=set(["passed"]))
115         self.run_tests(result_filter)
116         self.assertEqual(['passed'],
117             [passed.id() for passed in filtered_result.unexpectedSuccesses])
118         self.assertEqual(5, filtered_result.testsRun)
119
120     def test_exclude_failure(self):
121         filtered_result = unittest.TestResult()
122         result_filter = TestResultFilter(filtered_result, filter_failure=True)
123         self.run_tests(result_filter)
124         self.assertEqual(['error'],
125             [error[0].id() for error in filtered_result.errors])
126         self.assertEqual([],
127             [failure[0].id() for failure in
128             filtered_result.failures])
129         self.assertEqual(3, filtered_result.testsRun)
130
131     def test_exclude_skips(self):
132         filtered_result = subunit.TestResultStats(None)
133         result_filter = TestResultFilter(filtered_result, filter_skip=True)
134         self.run_tests(result_filter)
135         self.assertEqual(0, filtered_result.skipped_tests)
136         self.assertEqual(2, filtered_result.failed_tests)
137         self.assertEqual(3, filtered_result.testsRun)
138
139     def test_include_success(self):
140         """Successes can be included if requested."""
141         filtered_result = unittest.TestResult()
142         result_filter = TestResultFilter(filtered_result,
143             filter_success=False)
144         self.run_tests(result_filter)
145         self.assertEqual(['error'],
146             [error[0].id() for error in filtered_result.errors])
147         self.assertEqual(['failed'],
148             [failure[0].id() for failure in
149             filtered_result.failures])
150         self.assertEqual(5, filtered_result.testsRun)
151
152     def test_filter_predicate(self):
153         """You can filter by predicate callbacks"""
154         filtered_result = unittest.TestResult()
155         def filter_cb(test, outcome, err, details):
156             return outcome == 'success'
157         result_filter = TestResultFilter(filtered_result,
158             filter_predicate=filter_cb,
159             filter_success=False)
160         self.run_tests(result_filter)
161         # Only success should pass
162         self.assertEqual(1, filtered_result.testsRun)
163
164     def test_time_ordering_preserved(self):
165         # Passing a subunit stream through TestResultFilter preserves the
166         # relative ordering of 'time' directives and any other subunit
167         # directives that are still included.
168         date_a = datetime(year=2000, month=1, day=1, tzinfo=iso8601.UTC)
169         date_b = datetime(year=2000, month=1, day=2, tzinfo=iso8601.UTC)
170         date_c = datetime(year=2000, month=1, day=3, tzinfo=iso8601.UTC)
171         subunit_stream = _b('\n'.join([
172             "time: %s",
173             "test: foo",
174             "time: %s",
175             "error: foo",
176             "time: %s",
177             ""]) % (date_a, date_b, date_c))
178         result = ExtendedTestResult()
179         result_filter = TestResultFilter(result)
180         self.run_tests(result_filter, subunit_stream)
181         foo = subunit.RemotedTestCase('foo')
182         self.maxDiff = None
183         self.assertSequenceEqual(
184             [('time', date_a),
185              ('time', date_b),
186              ('startTest', foo),
187              ('addError', foo, {}),
188              ('stopTest', foo),
189              ('time', date_c)], result._events)
190
191     def test_skip_preserved(self):
192         subunit_stream = _b('\n'.join([
193             "test: foo",
194             "skip: foo",
195             ""]))
196         result = ExtendedTestResult()
197         result_filter = TestResultFilter(result)
198         self.run_tests(result_filter, subunit_stream)
199         foo = subunit.RemotedTestCase('foo')
200         self.assertEquals(
201             [('startTest', foo),
202              ('addSkip', foo, {}),
203              ('stopTest', foo), ], result._events)
204
205
206 def test_suite():
207     loader = subunit.tests.TestUtil.TestLoader()
208     result = loader.loadTestsFromName(__name__)
209     return result