diff options
Diffstat (limited to 'lib/subunit/python/subunit/tests/test_subunit_filter.py')
| -rw-r--r-- | lib/subunit/python/subunit/tests/test_subunit_filter.py | 170 |
1 files changed, 166 insertions, 4 deletions
diff --git a/lib/subunit/python/subunit/tests/test_subunit_filter.py b/lib/subunit/python/subunit/tests/test_subunit_filter.py index 06754840eb..33b924824d 100644 --- a/lib/subunit/python/subunit/tests/test_subunit_filter.py +++ b/lib/subunit/python/subunit/tests/test_subunit_filter.py @@ -17,15 +17,18 @@ """Tests for subunit.TestResultFilter.""" from datetime import datetime +import os +import subprocess +import sys from subunit import iso8601 import unittest from testtools import TestCase -from testtools.compat import _b, BytesIO, StringIO +from testtools.compat import _b, BytesIO from testtools.testresult.doubles import ExtendedTestResult import subunit -from subunit.test_results import TestResultFilter +from subunit.test_results import make_tag_filter, TestResultFilter class TestTestResultFilter(TestCase): @@ -77,6 +80,40 @@ xfail todo filtered_result.failures]) self.assertEqual(4, filtered_result.testsRun) + def test_tag_filter(self): + tag_filter = make_tag_filter(['global'], ['local']) + result = ExtendedTestResult() + result_filter = TestResultFilter( + result, filter_success=False, filter_predicate=tag_filter) + self.run_tests(result_filter) + tests_included = [ + event[1] for event in result._events if event[0] == 'startTest'] + tests_expected = list(map( + subunit.RemotedTestCase, + ['passed', 'error', 'skipped', 'todo'])) + self.assertEquals(tests_expected, tests_included) + + def test_tags_tracked_correctly(self): + tag_filter = make_tag_filter(['a'], []) + result = ExtendedTestResult() + result_filter = TestResultFilter( + result, filter_success=False, filter_predicate=tag_filter) + input_stream = _b( + "test: foo\n" + "tags: a\n" + "successful: foo\n" + "test: bar\n" + "successful: bar\n") + self.run_tests(result_filter, input_stream) + foo = subunit.RemotedTestCase('foo') + self.assertEquals( + [('startTest', foo), + ('tags', set(['a']), set()), + ('addSuccess', foo), + ('stopTest', foo), + ], + result._events) + def test_exclude_errors(self): filtered_result = unittest.TestResult() result_filter = TestResultFilter(filtered_result, filter_error=True) @@ -151,6 +188,8 @@ xfail todo def test_filter_predicate(self): """You can filter by predicate callbacks""" + # 0.0.7 and earlier did not support the 'tags' parameter, so we need + # to test that we still support behaviour without it. filtered_result = unittest.TestResult() def filter_cb(test, outcome, err, details): return outcome == 'success' @@ -161,6 +200,18 @@ xfail todo # Only success should pass self.assertEqual(1, filtered_result.testsRun) + def test_filter_predicate_with_tags(self): + """You can filter by predicate callbacks that accept tags""" + filtered_result = unittest.TestResult() + def filter_cb(test, outcome, err, details, tags): + return outcome == 'success' + result_filter = TestResultFilter(filtered_result, + filter_predicate=filter_cb, + filter_success=False) + self.run_tests(result_filter) + # Only success should pass + self.assertEqual(1, filtered_result.testsRun) + def test_time_ordering_preserved(self): # Passing a subunit stream through TestResultFilter preserves the # relative ordering of 'time' directives and any other subunit @@ -179,14 +230,41 @@ xfail todo result_filter = TestResultFilter(result) self.run_tests(result_filter, subunit_stream) foo = subunit.RemotedTestCase('foo') - self.assertEquals( + self.maxDiff = None + self.assertEqual( [('time', date_a), - ('startTest', foo), ('time', date_b), + ('startTest', foo), ('addError', foo, {}), ('stopTest', foo), ('time', date_c)], result._events) + def test_time_passes_through_filtered_tests(self): + # Passing a subunit stream through TestResultFilter preserves 'time' + # directives even if a specific test is filtered out. + date_a = datetime(year=2000, month=1, day=1, tzinfo=iso8601.UTC) + date_b = datetime(year=2000, month=1, day=2, tzinfo=iso8601.UTC) + date_c = datetime(year=2000, month=1, day=3, tzinfo=iso8601.UTC) + subunit_stream = _b('\n'.join([ + "time: %s", + "test: foo", + "time: %s", + "success: foo", + "time: %s", + ""]) % (date_a, date_b, date_c)) + result = ExtendedTestResult() + result_filter = TestResultFilter(result) + result_filter.startTestRun() + self.run_tests(result_filter, subunit_stream) + result_filter.stopTestRun() + foo = subunit.RemotedTestCase('foo') + self.maxDiff = None + self.assertEqual( + [('startTestRun',), + ('time', date_a), + ('time', date_c), + ('stopTestRun',),], result._events) + def test_skip_preserved(self): subunit_stream = _b('\n'.join([ "test: foo", @@ -201,6 +279,90 @@ xfail todo ('addSkip', foo, {}), ('stopTest', foo), ], result._events) + if sys.version_info < (2, 7): + # These tests require Python >=2.7. + del test_fixup_expected_failures, test_fixup_expected_errors, test_fixup_unexpected_success + + +class TestFilterCommand(TestCase): + + example_subunit_stream = _b("""\ +tags: global +test passed +success passed +test failed +tags: local +failure failed +test error +error error [ +error details +] +test skipped +skip skipped +test todo +xfail todo +""") + + def run_command(self, args, stream): + root = os.path.dirname( + os.path.dirname(os.path.dirname(os.path.dirname(__file__)))) + script_path = os.path.join(root, 'filters', 'subunit-filter') + command = [sys.executable, script_path] + list(args) + ps = subprocess.Popen( + command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + out, err = ps.communicate(stream) + if ps.returncode != 0: + raise RuntimeError("%s failed: %s" % (command, err)) + return out + + def to_events(self, stream): + test = subunit.ProtocolTestCase(BytesIO(stream)) + result = ExtendedTestResult() + test.run(result) + return result._events + + def test_default(self): + output = self.run_command([], _b( + "test: foo\n" + "skip: foo\n" + )) + events = self.to_events(output) + foo = subunit.RemotedTestCase('foo') + self.assertEqual( + [('startTest', foo), + ('addSkip', foo, {}), + ('stopTest', foo)], + events) + + def test_tags(self): + output = self.run_command(['-s', '--with-tag', 'a'], _b( + "tags: a\n" + "test: foo\n" + "success: foo\n" + "tags: -a\n" + "test: bar\n" + "success: bar\n" + "test: baz\n" + "tags: a\n" + "success: baz\n" + )) + events = self.to_events(output) + foo = subunit.RemotedTestCase('foo') + baz = subunit.RemotedTestCase('baz') + self.assertEqual( + [('tags', set(['a']), set()), + ('startTest', foo), + ('addSuccess', foo), + ('stopTest', foo), + ('tags', set(), set(['a'])), + ('startTest', baz), + ('tags', set(['a']), set()), + ('addSuccess', baz), + ('stopTest', baz), + ], + events) + def test_suite(): loader = subunit.tests.TestUtil.TestLoader() |