subunit-stats filter added.

3 files changed, 149 insertions, 0 deletions

diff --git a/python/subunit/__init__.py b/python/subunit/__init__.py
index ff2ad66..1322248 100644
--- a/python/subunit/__init__.py
+++ b/python/subunit/__init__.py
@@ -617,3 +617,66 @@ def tag_stream(original, filtered, tags):
         else:
             filtered.write(line)
     return 0
+
+
+class ProtocolTestCase(object):
+    """A test case which reports a subunit stream."""
+
+    def __init__(self, stream):
+        self._stream = stream
+
+    def __call__(self, result=None):
+        return self.run(result)
+
+    def run(self, result=None):
+        if result is None: result = self.defaultTestResult()
+        protocol = TestProtocolServer(result)
+        for line in self._stream:
+            protocol.lineReceived(line)
+        protocol.lostConnection()
+
+
+class TestResultStats(unittest.TestResult):
+    """A pyunit TestResult interface implementation for making statistics.
+    
+    :ivar total_tests: The total tests seen.
+    :ivar passed_tests: The tests that passed.
+    :ivar failed_tests: The tests that failed.
+    :ivar tags: The tags seen across all tests.
+    """
+
+    def __init__(self, stream):
+        """Create a TestResultStats which outputs to stream."""
+        unittest.TestResult.__init__(self)
+        self._stream = stream
+        self.failed_tests = 0
+        self.tags = set()
+
+    @property
+    def total_tests(self):
+        return self.testsRun
+
+    def addError(self, test, err):
+        self.failed_tests += 1
+
+    def addFailure(self, test, err):
+        self.failed_tests += 1
+
+    def formatStats(self):
+        self._stream.write("Total tests:  %5d\n" % self.total_tests)
+        self._stream.write("Passed tests: %5d\n" % self.passed_tests)
+        self._stream.write("Failed tests: %5d\n" % self.failed_tests)
+        tags = sorted(self.tags)
+        self._stream.write("Tags: %s\n" % (", ".join(tags)))
+
+    @property
+    def passed_tests(self):
+        return self.total_tests - self.failed_tests
+
+    def stopTest(self, test):
+        unittest.TestResult.stopTest(self, test)
+        self.tags.update(test.tags)
+
+    def wasSuccessful(self):
+        "Tells whether or not this result was a success"
+        return self.failed_tests == 0
diff --git a/python/subunit/tests/__init__.py b/python/subunit/tests/__init__.py
index 6703514..a506281 100644
--- a/python/subunit/tests/__init__.py
+++ b/python/subunit/tests/__init__.py
@@ -19,6 +19,7 @@
 
 from subunit.tests import (
     TestUtil,
+    test_subunit_stats,
     test_subunit_tags,
     test_tap2subunit,
     test_test_protocol,
@@ -29,4 +30,5 @@ def test_suite():
     result.addTest(test_test_protocol.test_suite())
     result.addTest(test_tap2subunit.test_suite())
     result.addTest(test_subunit_tags.test_suite())
+    result.addTest(test_subunit_stats.test_suite())
     return result
diff --git a/python/subunit/tests/test_subunit_stats.py b/python/subunit/tests/test_subunit_stats.py
new file mode 100644
index 0000000..b789089
--- /dev/null
+++ b/python/subunit/tests/test_subunit_stats.py
@@ -0,0 +1,84 @@
+#
+#  subunit: extensions to python unittest to get test results from subprocesses.
+#  Copyright (C) 2005  Robert Collins <robertc@robertcollins.net>
+#
+#  This program is free software; you can redistribute it and/or modify
+#  it under the terms of the GNU General Public License as published by
+#  the Free Software Foundation; either version 2 of the License, or
+#  (at your option) any later version.
+#
+#  This program is distributed in the hope that it will be useful,
+#  but WITHOUT ANY WARRANTY; without even the implied warranty of
+#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#  GNU General Public License for more details.
+#
+#  You should have received a copy of the GNU General Public License
+#  along with this program; if not, write to the Free Software
+#  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA
+#
+
+"""Tests for subunit.TestResultStats."""
+
+import unittest
+from StringIO import StringIO
+
+import subunit
+
+
+class TestTestResultStats(unittest.TestCase):
+    """Test for TestResultStats, a TestResult object that generates stats."""
+
+    def setUp(self):
+        self.output = StringIO()
+        self.result = subunit.TestResultStats(self.output)
+        self.input_stream = StringIO()
+        self.test = subunit.ProtocolTestCase(self.input_stream)
+
+    def test_stats_empty(self):
+        self.test.run(self.result)
+        self.assertEqual(0, self.result.total_tests)
+        self.assertEqual(0, self.result.passed_tests)
+        self.assertEqual(0, self.result.failed_tests)
+        self.assertEqual(set(), self.result.tags)
+
+    def setUpUsedStream(self):
+        self.input_stream.write("""tags: global
+test passed
+success passed
+test failed
+tags: local
+failure failed
+test error
+error error
+test skipped
+skip skipped
+test todo
+xfail todo
+""")
+        self.input_stream.seek(0)
+        self.test.run(self.result)
+    
+    def test_stats_smoke_everything(self):
+        # Statistics are calculated usefully.
+        self.setUpUsedStream()
+        self.assertEqual(5, self.result.total_tests)
+        self.assertEqual(3, self.result.passed_tests)
+        self.assertEqual(2, self.result.failed_tests)
+        self.assertEqual(set(["global", "local"]), self.result.tags)
+
+    def test_stat_formatting(self):
+        expected = ("""
+Total tests:      5
+Passed tests:     3
+Failed tests:     2
+Tags: global, local
+""")[1:]
+        self.setUpUsedStream()
+        self.result.formatStats()
+        self.assertEqual(expected, self.output.getvalue())
+
+
+def test_suite():
+    loader = subunit.tests.TestUtil.TestLoader()
+    result = loader.loadTestsFromName(__name__)
+    return result