Merge lp:~jml/subunit/to-csv -- Add subunit2csv script to filter subunit output

3 files changed, 421 insertions, 0 deletions

diff --git a/python/subunit/filters.py b/python/subunit/filters.py
new file mode 100644
index 0000000..dc3fd8a
--- /dev/null
+++ b/python/subunit/filters.py
@@ -0,0 +1,125 @@
+#  subunit: extensions to python unittest to get test results from subprocesses.
+#  Copyright (C) 2009  Robert Collins <robertc@robertcollins.net>
+#
+#  Licensed under either the Apache License, Version 2.0 or the BSD 3-clause
+#  license at the users choice. A copy of both licenses are available in the
+#  project source as Apache-2.0 and BSD. You may not use this file except in
+#  compliance with one of these two licences.
+#  
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under these licenses is distributed on an "AS IS" BASIS, WITHOUT
+#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+#  license you chose for the specific language governing permissions and
+#  limitations under that license.
+#
+
+
+from optparse import OptionParser
+import sys
+
+from subunit import DiscardStream, ProtocolTestCase
+
+
+def make_options(description):
+    parser = OptionParser(description=description)
+    parser.add_option(
+        "--no-passthrough", action="store_true",
+        help="Hide all non subunit input.", default=False,
+        dest="no_passthrough")
+    parser.add_option(
+        "-o", "--output-to",
+        help="Send the output to this path rather than stdout.")
+    parser.add_option(
+        "-f", "--forward", action="store_true", default=False,
+        help="Forward subunit stream on stdout.")
+    return parser
+
+
+def run_tests_from_stream(input_stream, result, passthrough_stream=None,
+                          forward_stream=None):
+    """Run tests from a subunit input stream through 'result'.
+
+    :param input_stream: A stream containing subunit input.
+    :param result: A TestResult that will receive the test events.
+    :param passthrough_stream: All non-subunit input received will be
+        sent to this stream.  If not provided, uses the ``TestProtocolServer``
+        default, which is ``sys.stdout``.
+    :param forward_stream: All subunit input received will be forwarded
+        to this stream.  If not provided, uses the ``TestProtocolServer``
+        default, which is to not forward any input.
+    """
+    test = ProtocolTestCase(
+        input_stream, passthrough=passthrough_stream,
+        forward=forward_stream)
+    result.startTestRun()
+    test.run(result)
+    result.stopTestRun()
+
+
+def filter_by_result(result_factory, output_path, passthrough, forward,
+                     input_stream=sys.stdin):
+    """Filter an input stream using a test result.
+
+    :param result_factory: A callable that when passed an output stream
+        returns a TestResult.  It is expected that this result will output
+        to the given stream.
+    :param output_path: A path send output to.  If None, output will be go
+        to ``sys.stdout``.
+    :param passthrough: If True, all non-subunit input will be sent to
+        ``sys.stdout``.  If False, that input will be discarded.
+    :param forward: If True, all subunit input will be forwarded directly to
+        ``sys.stdout`` as well as to the ``TestResult``.
+    :param input_stream: The source of subunit input.  Defaults to
+        ``sys.stdin``.
+    :return: A test result with the resultts of the run.
+    """
+    if passthrough:
+        passthrough_stream = sys.stdout
+    else:
+        passthrough_stream = DiscardStream()
+
+    if forward:
+        forward_stream = sys.stdout
+    else:
+        forward_stream = DiscardStream()
+
+    if output_path is None:
+        output_to = sys.stdout
+    else:
+        output_to = file(output_path, 'wb')
+
+    try:
+        result = result_factory(output_to)
+        run_tests_from_stream(
+            input_stream, result, passthrough_stream, forward_stream)
+    finally:
+        if output_path:
+            output_to.close()
+    return result
+
+
+def run_filter_script(result_factory, description, post_run_hook=None):
+    """Main function for simple subunit filter scripts.
+
+    Many subunit filter scripts take a stream of subunit input and use a
+    TestResult to handle the events generated by that stream.  This function
+    wraps a lot of the boiler-plate around that by making a script with
+    options for handling passthrough information and stream forwarding, and
+    that will exit with a successful return code (i.e. 0) if the input stream
+    represents a successful test run.
+
+    :param result_factory: A callable that takes an output stream and returns
+        a test result that outputs to that stream.
+    :param description: A description of the filter script.
+    """
+    parser = make_options(description)
+    (options, args) = parser.parse_args()
+    result = filter_by_result(
+        result_factory, options.output_to, not options.no_passthrough,
+        options.forward)
+    if post_run_hook:
+        post_run_hook(result)
+    if result.wasSuccessful():
+        sys.exit(0)
+    else:
+        sys.exit(1)
diff --git a/python/subunit/test_results.py b/python/subunit/test_results.py
index 33fb50e..fb7affd 100644
--- a/python/subunit/test_results.py
+++ b/python/subunit/test_results.py
@@ -16,9 +16,14 @@
 
 """TestResult helper classes used to by subunit."""
 
+import csv
 import datetime
 
 import testtools
+from testtools.content import (
+    text_content,
+    TracebackContent,
+    )
 
 from subunit import iso8601
 
@@ -493,3 +498,96 @@ class TestIdPrintingResult(testtools.TestResult):
     def wasSuccessful(self):
         "Tells whether or not this result was a success"
         return self.failed_tests == 0
+
+
+class TestByTestResult(testtools.TestResult):
+    """Call something every time a test completes."""
+
+    # XXX: Arguably belongs in testtools.
+
+    def __init__(self, on_test):
+        """Construct a ``TestByTestResult``.
+
+        :param on_test: A callable that take a test case, a status (one of
+            "success", "failure", "error", "skip", or "xfail"), a start time
+            (a ``datetime`` with timezone), a stop time, an iterable of tags,
+            and a details dict. Is called at the end of each test (i.e. on
+            ``stopTest``) with the accumulated values for that test.
+        """
+        super(TestByTestResult, self).__init__()
+        self._on_test = on_test
+
+    def startTest(self, test):
+        super(TestByTestResult, self).startTest(test)
+        self._start_time = self._now()
+        # There's no supported (i.e. tested) behaviour that relies on these
+        # being set, but it makes me more comfortable all the same. -- jml
+        self._status = None
+        self._details = None
+        self._stop_time = None
+
+    def stopTest(self, test):
+        self._stop_time = self._now()
+        super(TestByTestResult, self).stopTest(test)
+        self._on_test(
+            test=test,
+            status=self._status,
+            start_time=self._start_time,
+            stop_time=self._stop_time,
+            # current_tags is new in testtools 0.9.13.
+            tags=getattr(self, 'current_tags', None),
+            details=self._details)
+
+    def _err_to_details(self, test, err, details):
+        if details:
+            return details
+        return {'traceback': TracebackContent(err, test)}
+
+    def addSuccess(self, test, details=None):
+        super(TestByTestResult, self).addSuccess(test)
+        self._status = 'success'
+        self._details = details
+
+    def addFailure(self, test, err=None, details=None):
+        super(TestByTestResult, self).addFailure(test, err, details)
+        self._status = 'failure'
+        self._details = self._err_to_details(test, err, details)
+
+    def addError(self, test, err=None, details=None):
+        super(TestByTestResult, self).addError(test, err, details)
+        self._status = 'error'
+        self._details = self._err_to_details(test, err, details)
+
+    def addSkip(self, test, reason=None, details=None):
+        super(TestByTestResult, self).addSkip(test, reason, details)
+        self._status = 'skip'
+        if details is None:
+            details = {'reason': text_content(reason)}
+        elif reason:
+            # XXX: What if details already has 'reason' key?
+            details['reason'] = text_content(reason)
+        self._details = details
+
+    def addExpectedFailure(self, test, err=None, details=None):
+        super(TestByTestResult, self).addExpectedFailure(test, err, details)
+        self._status = 'xfail'
+        self._details = self._err_to_details(test, err, details)
+
+    def addUnexpectedSuccess(self, test, details=None):
+        super(TestByTestResult, self).addUnexpectedSuccess(test, details)
+        self._status = 'success'
+        self._details = details
+
+
+class CsvResult(TestByTestResult):
+
+    def __init__(self, stream):
+        super(CsvResult, self).__init__(self._on_test)
+        self._write_row = csv.writer(stream).writerow
+
+    def _on_test(self, test, status, start_time, stop_time, tags, details):
+        self._write_row([test.id(), status, start_time, stop_time])
+
+    def startTestRun(self):
+        super(CsvResult, self).startTestRun()
+        self._write_row(['test', 'status', 'start_time', 'stop_time'])
diff --git a/python/subunit/tests/test_test_results.py b/python/subunit/tests/test_test_results.py
index 94d2274..6beb57a 100644
--- a/python/subunit/tests/test_test_results.py
+++ b/python/subunit/tests/test_test_results.py
@@ -14,16 +14,25 @@
 #  limitations under that license.
 #
 
+import csv
 import datetime
+import sys
 import unittest
 
 from testtools import TestCase
+from testtools.compat import StringIO
+from testtools.content import (
+    text_content,
+    TracebackContent,
+    )
 from testtools.testresult.doubles import ExtendedTestResult
 
 import subunit
 import subunit.iso8601 as iso8601
 import subunit.test_results
 
+import testtools
+
 
 class LoggingDecorator(subunit.test_results.HookedTestResultDecorator):
 
@@ -294,6 +303,195 @@ class TestTimeCollapsingDecorator(TestCase):
              ('stopTest', foo)], result._events)
 
 
+class TestByTestResultTests(testtools.TestCase):
+
+    def setUp(self):
+        super(TestByTestResultTests, self).setUp()
+        self.log = []
+        self.result = subunit.test_results.TestByTestResult(self.on_test)
+        self.result._now = iter(range(5)).next
+
+    def assertCalled(self, **kwargs):
+        defaults = {
+            'test': self,
+            'tags': set(),
+            'details': None,
+            'start_time': 0,
+            'stop_time': 1,
+            }
+        defaults.update(kwargs)
+        self.assertEqual([defaults], self.log)
+
+    def on_test(self, **kwargs):
+        self.log.append(kwargs)
+
+    def test_no_tests_nothing_reported(self):
+        self.result.startTestRun()
+        self.result.stopTestRun()
+        self.assertEqual([], self.log)
+
+    def test_add_success(self):
+        self.result.startTest(self)
+        self.result.addSuccess(self)
+        self.result.stopTest(self)
+        self.assertCalled(status='success')
+
+    def test_add_success_details(self):
+        self.result.startTest(self)
+        details = {'foo': 'bar'}
+        self.result.addSuccess(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='success', details=details)
+
+    def test_tags(self):
+        if not getattr(self.result, 'tags', None):
+            self.skipTest("No tags in testtools")
+        self.result.tags(['foo'], [])
+        self.result.startTest(self)
+        self.result.addSuccess(self)
+        self.result.stopTest(self)
+        self.assertCalled(status='success', tags=set(['foo']))
+
+    def test_add_error(self):
+        self.result.startTest(self)
+        try:
+            1/0
+        except ZeroDivisionError:
+            error = sys.exc_info()
+        self.result.addError(self, error)
+        self.result.stopTest(self)
+        self.assertCalled(
+            status='error',
+            details={'traceback': TracebackContent(error, self)})
+
+    def test_add_error_details(self):
+        self.result.startTest(self)
+        details = {"foo": text_content("bar")}
+        self.result.addError(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='error', details=details)
+
+    def test_add_failure(self):
+        self.result.startTest(self)
+        try:
+            self.fail("intentional failure")
+        except self.failureException:
+            failure = sys.exc_info()
+        self.result.addFailure(self, failure)
+        self.result.stopTest(self)
+        self.assertCalled(
+            status='failure',
+            details={'traceback': TracebackContent(failure, self)})
+
+    def test_add_failure_details(self):
+        self.result.startTest(self)
+        details = {"foo": text_content("bar")}
+        self.result.addFailure(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='failure', details=details)
+
+    def test_add_xfail(self):
+        self.result.startTest(self)
+        try:
+            1/0
+        except ZeroDivisionError:
+            error = sys.exc_info()
+        self.result.addExpectedFailure(self, error)
+        self.result.stopTest(self)
+        self.assertCalled(
+            status='xfail',
+            details={'traceback': TracebackContent(error, self)})
+
+    def test_add_xfail_details(self):
+        self.result.startTest(self)
+        details = {"foo": text_content("bar")}
+        self.result.addExpectedFailure(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='xfail', details=details)
+
+    def test_add_unexpected_success(self):
+        self.result.startTest(self)
+        details = {'foo': 'bar'}
+        self.result.addUnexpectedSuccess(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='success', details=details)
+
+    def test_add_skip_reason(self):
+        self.result.startTest(self)
+        reason = self.getUniqueString()
+        self.result.addSkip(self, reason)
+        self.result.stopTest(self)
+        self.assertCalled(
+            status='skip', details={'reason': text_content(reason)})
+
+    def test_add_skip_details(self):
+        self.result.startTest(self)
+        details = {'foo': 'bar'}
+        self.result.addSkip(self, details=details)
+        self.result.stopTest(self)
+        self.assertCalled(status='skip', details=details)
+
+    def test_twice(self):
+        self.result.startTest(self)
+        self.result.addSuccess(self, details={'foo': 'bar'})
+        self.result.stopTest(self)
+        self.result.startTest(self)
+        self.result.addSuccess(self)
+        self.result.stopTest(self)
+        self.assertEqual(
+            [{'test': self,
+              'status': 'success',
+              'start_time': 0,
+              'stop_time': 1,
+              'tags': set(),
+              'details': {'foo': 'bar'}},
+             {'test': self,
+              'status': 'success',
+              'start_time': 2,
+              'stop_time': 3,
+              'tags': set(),
+              'details': None},
+             ],
+            self.log)
+
+
+class TestCsvResult(testtools.TestCase):
+
+    def parse_stream(self, stream):
+        stream.seek(0)
+        reader = csv.reader(stream)
+        return list(reader)
+
+    def test_csv_output(self):
+        stream = StringIO()
+        result = subunit.test_results.CsvResult(stream)
+        result._now = iter(range(5)).next
+        result.startTestRun()
+        result.startTest(self)
+        result.addSuccess(self)
+        result.stopTest(self)
+        result.stopTestRun()
+        self.assertEqual(
+            [['test', 'status', 'start_time', 'stop_time'],
+             [self.id(), 'success', '0', '1'],
+             ],
+            self.parse_stream(stream))
+
+    def test_just_header_when_no_tests(self):
+        stream = StringIO()
+        result = subunit.test_results.CsvResult(stream)
+        result.startTestRun()
+        result.stopTestRun()
+        self.assertEqual(
+            [['test', 'status', 'start_time', 'stop_time']],
+            self.parse_stream(stream))
+
+    def test_no_output_before_events(self):
+        stream = StringIO()
+        subunit.test_results.CsvResult(stream)
+        self.assertEqual([], self.parse_stream(stream))
+
+
 def test_suite():
     loader = subunit.tests.TestUtil.TestLoader()
     result = loader.loadTestsFromName(__name__)