From c7765fe8bf61a48cd5bcca611bfffdd55ed2f318 2013-10-15 19:16:47
From: Thomas Kluyver <takowl@gmail.com>
Date: 2013-10-15 19:16:47
Subject: [PATCH] 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.

---

diff --git a/IPython/parallel/tests/__init__.py b/IPython/parallel/tests/__init__.py
index 6e456cc..8c8c7db 100644
--- a/IPython/parallel/tests/__init__.py
+++ b/IPython/parallel/tests/__init__.py
@@ -14,7 +14,9 @@
 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 @@ 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 @@ 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 @@ 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()
diff --git a/IPython/testing/iptest.py b/IPython/testing/iptest.py
index 054574b..b85f236 100644
--- a/IPython/testing/iptest.py
+++ b/IPython/testing/iptest.py
@@ -28,10 +28,13 @@ 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 @@ 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 @@ 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 @@ 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: