Exit non-zero when listing includes import failures.

[third_party/subunit] / python / subunit / run.py

diff --git a/python/subunit/run.py b/python/subunit/run.py
index ca5fe5c17e0e4ac1506de39a3bae2a1fffe9af88..7e4d783bdedf2bfe8bcfa9428c22145163c2a24f 100755 (executable)
--- a/python/subunit/run.py
+++ b/python/subunit/run.py
@@ -20,30 +20,74 @@
   $ python -m subunit.run mylib.tests.test_suite
 """
 
+import io
+import os
 import sys
 
-from subunit import TestProtocolClient, get_default_formatter
+from testtools import ExtendedToStreamDecorator
+from testtools.testsuite import iterate_tests
+
+from subunit import StreamResultToBytes, get_default_formatter
 from subunit.test_results import AutoTimingTestResultDecorator
 from testtools.run import (
     BUFFEROUTPUT,
     CATCHBREAK,
     FAILFAST,
+    list_test,
     TestProgram,
     USAGE_AS_MAIN,
     )
 
 
 class SubunitTestRunner(object):
-    def __init__(self, stream=sys.stdout):
-        self.stream = stream
+    def __init__(self, verbosity=None, failfast=None, buffer=None, stream=None):
+        """Create a TestToolsTestRunner.
+
+        :param verbosity: Ignored.
+        :param failfast: Stop running tests at the first failure.
+        :param buffer: Ignored.
+        """
+        self.failfast = failfast
+        self.stream = stream or sys.stdout
 
     def run(self, test):
         "Run the given test case or test suite."
-        result = TestProtocolClient(self.stream)
+        result, _ = self._list(test)
+        result = ExtendedToStreamDecorator(result)
         result = AutoTimingTestResultDecorator(result)
-        test(result)
+        if self.failfast is not None:
+            result.failfast = self.failfast
+        result.startTestRun()
+        try:
+            test(result)
+        finally:
+            result.stopTestRun()
         return result
 
+    def list(self, test):
+        "List the test."
+        result, errors = self._list(test)
+        if errors:
+            failed_descr = '\n'.join(errors).encode('utf8')
+            result.status(file_name="import errors", runnable=False,
+                file_bytes=failed_descr, mime_type="text/plain;charset=utf8")
+            sys.exit(2)
+
+    def _list(self, test):
+        test_ids, errors = list_test(test)
+        try:
+            fileno = self.stream.fileno()
+        except:
+            fileno = None
+        if fileno is not None:
+            stream = os.fdopen(fileno, 'wb', 0)
+        else:
+            stream = self.stream
+        result = StreamResultToBytes(stream)
+        for test_id in test_ids:
+            result.status(test_id=test_id, test_status='exists')
+        return result, errors
+
 
 class SubunitTestProgram(TestProgram):
 
@@ -68,8 +112,20 @@ class SubunitTestProgram(TestProgram):
         sys.exit(2)
 
 
-if __name__ == '__main__':
+def main():
+    # Disable the default buffering, for Python 2.x where pdb doesn't do it
+    # on non-ttys.
     stream = get_default_formatter()
-    runner = SubunitTestRunner(stream)
+    runner = SubunitTestRunner
+    # Patch stdout to be unbuffered, so that pdb works well on 2.6/2.7.
+    binstdout = io.open(sys.stdout.fileno(), 'wb', 0)
+    if sys.version_info[0] > 2:
+        sys.stdout = io.TextIOWrapper(binstdout, encoding=sys.stdout.encoding)
+    else:
+        sys.stdout = binstdout
     SubunitTestProgram(module=None, argv=sys.argv, testRunner=runner,
         stdout=sys.stdout)
+
+
+if __name__ == '__main__':
+    main()