Add a CSV filter.

1 files changed, 99 insertions, 0 deletions

diff --git a/python/subunit/test_results.py b/python/subunit/test_results.py
index 1c91daa..4d6a77d 100644
--- a/python/subunit/test_results.py
+++ b/python/subunit/test_results.py
@@ -16,10 +16,15 @@
 
 """TestResult helper classes used to by subunit."""
 
+import csv
 import datetime
 
 import iso8601
 import testtools
+from testtools.content import (
+    text_content,
+    TracebackContent,
+    )
 
 
 # NOT a TestResult, because we are implementing the interface, not inheriting
@@ -380,3 +385,97 @@ 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: No tests. Naughty, naughty.
+
+    # 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._current_test = test
+        self._status = None
+        self._details = None
+        self._start_time = self._now()
+        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,
+            # XXX: This is unset when I run it.  Is current_tags a new part of
+            # the testtools API?
+            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
+
+
+def csv_result(stream):
+    writer = csv.writer(stream)
+    w = writer.writerow
+    # XXX: Not great that we write this out immediately. Probably better to
+    # wait for startTestRun.
+    w(['test', 'status', 'start_time', 'stop_time'])
+    def on_test(test, status, start_time, stop_time, tags, details):
+        # XXX: Don't really know how to serialize tags or details into csv in
+        # a useful way.
+        w([test.id(), status, start_time, stop_time])
+    return TestByTestResult(on_test)