Merge pull request #4393 from takluyver/tests-subproc-stream-capture

Capture output from subprocs during test, and display on failure

IPython/kernel/tests/utils.py

@@ -14,9 +14,10 b''
 import atexit
 
 from contextlib import contextmanager
-from subprocess import PIPE
+from subprocess import PIPE, STDOUT
 from Queue import Empty
 
+import nose
 import nose.tools as nt
 
 from IPython.kernel import KernelManager
@@ -39,10 +40,12 b' KC = None'
 def start_new_kernel(argv=None):
     """start a new kernel, and return its Manager and Client"""
     km = KernelManager()
-    kwargs = dict(stdout=PIPE, stderr=PIPE)
+    kwargs = dict(stdout=PIPE, stderr=STDOUT)
     if argv:
         kwargs['extra_arguments'] = argv
     km.start_kernel(**kwargs)
+    nose.ipy_stream_capturer.add_stream(km.kernel.stdout.fileno())
+    nose.ipy_stream_capturer.ensure_started()
     kc = km.client()
     kc.start_channels()
     

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,12 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 +42,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
@@ -354,6 +357,94 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
+        captured = self.stream_capturer.get_buffer().decode('utf-8', 'replace')
+        if captured.strip():
+            ev = safe_str(ev)
+            out = [ev, '>> begin captured subprocess output <<',
+                    captured,
+                    '>> end captured subprocess output <<']
+            return ec, '\n'.join(out), tb
+
+        return err
+    
+    formatError = formatFailure
+    
+    def finalize(self, result):
+        if self.stream_capturer.started:
+            self.stream_capturer.stop.set()
+            self.stream_capturer.join()
+
+
 def run_iptest():
     """Run the IPython test suite using nose.
 
@@ -408,7 +499,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: