Capture output from subprocs during test, and display on failure

This probably needs some more thought on synchronisation so we don't miss
critical bits of output as the test ends.

IPython/parallel/tests/__init__.py

@@ -14,7 +14,9 b''
 import os
 import tempfile
 import time
-from subprocess import Popen
+from subprocess import Popen, PIPE, STDOUT
+
+import nose
 
 from IPython.utils.path import get_ipython_dir
 from IPython.parallel import Client
@@ -35,12 +37,16 b' class TestProcessLauncher(LocalProcessLauncher):'
     def start(self):
         if self.state == 'before':
             self.process = Popen(self.args,
-                stdout=blackhole, stderr=blackhole,
+                stdout=PIPE, stderr=STDOUT,
                 env=os.environ,
                 cwd=self.work_dir
             )
             self.notify_start(self.process.pid)
             self.poll = self.process.poll
+            # Store stdout & stderr to show with failing tests.
+            # This is defined in IPython.testing.iptest
+            nose.ipy_stream_capturer.add_stream(self.process.stdout.fileno())
+            nose.ipy_stream_capturer.ensure_started()
         else:
             s = 'The process was already started and has state: %r' % self.state
             raise ProcessStateError(s)
@@ -57,7 +63,7 b' def setup():'
     
     cp = TestProcessLauncher()
     cp.cmd_and_args = ipcontroller_cmd_argv + \
-                ['--profile=iptest', '--log-level=50', '--ping=250', '--dictdb']
+                ['--profile=iptest', '--log-level=20', '--ping=250', '--dictdb']
     cp.start()
     launchers.append(cp)
     tic = time.time()
@@ -107,6 +113,7 b' def teardown():'
     time.sleep(1)
     while launchers:
         p = launchers.pop()
+        nose.ipy_stream_capturer.remove_stream(p.process.stdout.fileno())
         if p.poll() is None:
             try:
                 p.stop()

IPython/testing/iptest.py

@@ -28,10 +28,13 b' from __future__ import print_function'
 
 # Stdlib
 import glob
+from io import BytesIO
 import os
 import os.path as path
 import re
+from select import select
 import sys
+from threading import Thread, Lock, Event
 import warnings
 
 # Now, proceed to import nose itself
@@ -40,6 +43,7 b' from nose.plugins.xunit import Xunit'
 from nose import SkipTest
 from nose.core import TestProgram
 from nose.plugins import Plugin
+from nose.util import safe_str
 
 # Our own imports
 from IPython.utils.importstring import import_item
@@ -353,6 +357,89 b' class ExclusionPlugin(Plugin):'
         return None
 
 
+class StreamCapturer(Thread):
+    started = False
+    def __init__(self):
+        super(StreamCapturer, self).__init__()
+        self.streams = []
+        self.buffer = BytesIO()
+        self.streams_lock = Lock()
+        self.buffer_lock = Lock()
+        self.stream_added = Event()
+        self.stop = Event()
+
+    def run(self):
+        self.started = True
+        while not self.stop.is_set():
+            with self.streams_lock:
+                streams = self.streams
+
+            if not streams:
+                self.stream_added.wait(timeout=1)
+                self.stream_added.clear()
+                continue
+
+            ready = select(streams, [], [], 0.5)[0]
+            with self.buffer_lock:
+                for fd in ready:
+                    self.buffer.write(os.read(fd, 1024))
+    
+    def add_stream(self, fd):
+        with self.streams_lock:
+            self.streams.append(fd)
+        self.stream_added.set()
+    
+    def remove_stream(self, fd):
+        with self.streams_lock:
+            self.streams.remove(fd)
+        
+    def reset_buffer(self):
+        with self.buffer_lock:
+            self.buffer.truncate(0)
+            self.buffer.seek(0)
+    
+    def get_buffer(self):
+        with self.buffer_lock:
+            return self.buffer.getvalue()
+    
+    def ensure_started(self):
+        if not self.started:
+            self.start()
+
+class SubprocessStreamCapturePlugin(Plugin):
+    name='subprocstreams'
+    def __init__(self):
+        Plugin.__init__(self)
+        self.stream_capturer = StreamCapturer()
+        # This is ugly, but distant parts of the test machinery need to be able
+        # to add streams, so we make the object globally accessible.
+        nose.ipy_stream_capturer = self.stream_capturer
+    
+    def configure(self, options, config):
+        Plugin.configure(self, options, config)
+        # Override nose trying to disable plugin.
+        self.enabled = True
+    
+    def startTest(self, test):
+        # Reset log capture
+        self.stream_capturer.reset_buffer()
+    
+    def formatFailure(self, test, err):
+        # Show output
+        ec, ev, tb = err
+        ev = safe_str(ev)
+        out = [ev, '>> begin captured subprocess output <<',
+                self.stream_capturer.get_buffer().decode('utf-8', 'replace'),
+                '>> end captured subprocess output <<']
+        return ec, '\n'.join(out), tb
+    
+    formatError = formatFailure
+    
+    def finalize(self, result):
+        self.stream_capturer.stop.set()
+        self.stream_capturer.join()
+
+
 def run_iptest():
     """Run the IPython test suite using nose.
 
@@ -407,7 +494,8 b' def run_iptest():'
 
     # use our plugin for doctesting.  It will remove the standard doctest plugin
     # if it finds it enabled
-    plugins = [ExclusionPlugin(section.excludes), IPythonDoctest(), KnownFailure()]
+    plugins = [ExclusionPlugin(section.excludes), IPythonDoctest(), KnownFailure(),
+               SubprocessStreamCapturePlugin() ]
     
     # Use working directory set by parent process (see iptestcontroller)
     if 'IPTEST_WORKING_DIR' in os.environ: