Merge pull request #3834 from ivanov/nbconvert-better-tests

This PR fixes a few issues with nbconvert tests

The code for testing 'ipython nbconvert' prior to this PR did not work as
intended, and simply swallowed errors when pandoc wasn't installed, for example.

This PR adds a new get_output_error_code utility for easier checking of error
(looking at return code as opposed to the contents of stdout for the word
'error').

This new machinery is leveraged when calling nbconvert during tests.

IPython/nbconvert/tests/base.py

@@ -1,177 +1,138 b''
 """
 Contains base test class for nbconvert
 """
 #-----------------------------------------------------------------------------
 #Copyright (c) 2013, the IPython Development Team.
 #
 #Distributed under the terms of the Modified BSD License.
 #
 #The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
-import subprocess
 import os
 import glob
 import shutil
-import sys
 
 import IPython
-from IPython.utils.tempdir import TemporaryDirectory
-from IPython.utils import py3compat
+from IPython.utils.tempdir import TemporaryWorkingDirectory
+from IPython.utils.process import get_output_error_code
+from IPython.testing.tools import get_ipython_cmd
+
+# a trailing space allows for simpler concatenation with the other arguments
+ipy_cmd = get_ipython_cmd(as_string=True) + " "
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
-class TemporaryWorkingDirectory(TemporaryDirectory):
-    """
-    Creates a temporary directory and sets the cwd to that directory.
-    Automatically reverts to previous cwd upon cleanup.
-    Usage example:
-
-        with TemporaryWorakingDirectory() as tmpdir:
-            ...
-    """
-
-    def __init__(self, **kw):
-        """
-        Constructor
-        """
-        super(TemporaryWorkingDirectory, self).__init__(**kw)
-
-        #Change cwd to new temp dir.  Remember old cwd.
-        self.old_wd = os.getcwd()
-        os.chdir(self.name)
-
-
-    def cleanup(self):
-        """
-        Destructor
-        """
-
-        #Revert to old cwd.
-        os.chdir(self.old_wd)
-
-        #Cleanup
-        super(TemporaryWorkingDirectory, self).cleanup()
-
 
 class TestsBase(object):
-    """Base tests class.  Contains usefull fuzzy comparison and nbconvert
+    """Base tests class.  Contains useful fuzzy comparison and nbconvert
     functions."""
 
 
     def fuzzy_compare(self, a, b, newlines_are_spaces=True, tabs_are_spaces=True, 
                       fuzzy_spacing=True, ignore_spaces=False, 
                       ignore_newlines=False, case_sensitive=False):
         """
         Performs a fuzzy comparison of two strings.  A fuzzy comparison is a
         comparison that ignores insignificant differences in the two comparands.
         The significance of certain differences can be specified via the keyword
         parameters of this method.
         """
 
         if newlines_are_spaces:
             a = a.replace('\n', ' ')
             b = b.replace('\n', ' ')
 
         if tabs_are_spaces:
             a = a.replace('\t', ' ')
             b = b.replace('\t', ' ')
 
         if ignore_spaces:
             a = a.replace(' ', '')
             b = b.replace(' ', '')
 
         if fuzzy_spacing:
             a = self.recursive_replace(a, '  ', ' ')
             b = self.recursive_replace(b, '  ', ' ')
 
         if ignore_newlines:
             a = a.replace('\n', '')
             b = b.replace('\n', '')
 
         if not case_sensitive:
             a = a.lower()
             b = b.lower()
 
         return a == b
 
 
     def recursive_replace(self, text, search, replacement):
         """
         Performs a recursive replacement operation.  Replaces all instances
         of a search string in a text string with a replacement string until
         the search string no longer exists.  Recursion is needed because the
         replacement string may generate additional search strings.
 
         For example:
            Replace "ii" with "i" in the string "Hiiii" yields "Hii"
            Another replacement yields "Hi" (the desired output)
 
         Parameters:
         -----------
         text : string
             Text to replace in.
         search : string
             String to search for within "text"
         replacement : string
             String to replace "search" with
         """
         while search in text:
             text = text.replace(search, replacement)
         return text
 
-            
     def create_temp_cwd(self, copy_filenames=None):
         temp_dir = TemporaryWorkingDirectory()
 
         #Copy the files if requested.
-        if not copy_filenames is None:
+        if copy_filenames is not None:
             self.copy_files_to(copy_filenames)
 
         #Return directory handler
         return temp_dir
 
 
-    def copy_files_to(self, copy_filenames=None, destination=None):
-        
-        #Copy test files into the destination directory.
-        if copy_filenames:
-            for pattern in copy_filenames:
-                for match in glob.glob(os.path.join(self._get_files_path(), pattern)):
-                    if destination is None:
-                        shutil.copyfile(match, os.path.basename(match))
-                    else:
-                        if not os.path.isdir(destination):
-                            os.makedirs(destination)
-                        shutil.copyfile(match, os.path.join(destination, os.path.basename(match)))
+    def copy_files_to(self, copy_filenames, dest='.'):
+        "Copy test files into the destination directory"
+        if not os.path.isdir(dest):
+            os.makedirs(dest)
+        files_path = self._get_files_path()
+        for pattern in copy_filenames:
+            for match in glob.glob(os.path.join(files_path, pattern)):
+                shutil.copyfile(match, os.path.join(dest, os.path.basename(match)))
 
 
     def _get_files_path(self):
 
         #Get the relative path to this module in the IPython directory.
         names = self.__module__.split('.')[1:-1]
         names.append('files')
         
         #Build a path using the IPython directory and the relative path we just
         #found.
         path = IPython.__path__[0]
         for name in names:
             path = os.path.join(path, name)
         return path
 
 
-    def call(self, parameters):
-        output = subprocess.Popen(parameters, stdout=subprocess.PIPE).communicate()[0]
-        
-        #Convert the output to a string if running Python3
-        if py3compat.PY3:
-            return output.decode('utf-8')
-        else:
-            return output
-     
  No newline at end of file
+    def call(self, parameters, raise_on_error=True):
+        stdout, stderr, retcode = get_output_error_code(ipy_cmd + parameters)
+        if retcode != 0 and raise_on_error:
+            raise OSError(stderr)
+        return stdout, stderr

IPython/nbconvert/tests/test_nbconvertapp.py

@@ -1,149 +1,145 b''
 """
 Contains tests for the nbconvertapp
 """
 #-----------------------------------------------------------------------------
 #Copyright (c) 2013, the IPython Development Team.
 #
 #Distributed under the terms of the Modified BSD License.
 #
 #The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
 from .base import TestsBase
 
-from IPython.utils import py3compat
 from IPython.testing import decorators as dec
 
     
 #-----------------------------------------------------------------------------
 # Constants
 #-----------------------------------------------------------------------------
 
-# Define ipython commandline name
-if py3compat.PY3:
-    IPYTHON = 'ipython3'
-else:
-    IPYTHON = 'ipython'
-
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 class TestNbConvertApp(TestsBase):
     """Collection of NbConvertApp tests"""
 
 
     def test_notebook_help(self):
         """
         Will help show if no notebooks are specified?
         """
         with self.create_temp_cwd():
-            assert "see '--help-all'" in self.call([IPYTHON, 'nbconvert'])
+            out, err = self.call('nbconvert', raise_on_error=False)
+            assert "see '--help-all'" in out
 
 
     def test_glob(self):
         """
         Do search patterns work for notebook names?
         """
         with self.create_temp_cwd(['notebook*.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', 
-                '--to="python"', '--notebooks=["*.ipynb"]']).lower()
+            self.call('nbconvert --to="python" --notebooks=\'["*.ipynb"]\'')
             assert os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')
 
 
     def test_glob_subdir(self):
         """
         Do search patterns work for subdirectory notebook names?
         """
-        with self.create_temp_cwd() as cwd:
+        with self.create_temp_cwd():
             self.copy_files_to(['notebook*.ipynb'], 'subdir/')
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to="python"', 
-                '--notebooks=["%s"]' % os.path.join('subdir', '*.ipynb')]).lower()
+            self.call('nbconvert --to="python" --notebooks='
+                      '\'["%s"]\'' % os.path.join('subdir', '*.ipynb'))
             assert os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')
 
 
     def test_explicit(self):
         """
         Do explicit notebook names work?
         """
         with self.create_temp_cwd(['notebook*.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to="python"', 
-                '--notebooks=["notebook2.ipynb"]']).lower()
+            self.call('nbconvert --to="python" --notebooks='
+                      '\'["notebook2.ipynb"]\'')
             assert not os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')
 
 
     @dec.onlyif_cmds_exist('pdflatex')
+    @dec.onlyif_cmds_exist('pandoc')
     def test_post_processor(self):
         """
         Do post processors work?
         """
         with self.create_temp_cwd(['notebook1.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to="latex"', 
-                'notebook1', '--post="PDF"', '--PDFPostProcessor.verbose=True']).lower()
+            self.call('nbconvert --to="latex" notebook1'
+                      ' --post="PDF" --PDFPostProcessor.verbose=True')
             assert os.path.isfile('notebook1.tex')
             print("\n\n\t" + "\n\t".join([f for f in os.listdir('.') if os.path.isfile(f)]) + "\n\n")
             assert os.path.isfile('notebook1.pdf')
 
 
+    @dec.onlyif_cmds_exist('pandoc')
     def test_template(self):
         """
         Do export templates work?
         """
         with self.create_temp_cwd(['notebook2.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to=slides', 
-                '--notebooks=["notebook2.ipynb"]', '--template=reveal']).lower()
+            self.call('nbconvert --to=slides --notebooks='
+                      '\'["notebook2.ipynb"]\' --template=reveal')
             assert os.path.isfile('notebook2.slides.html')
             with open('notebook2.slides.html') as f:
                 assert '/reveal.css' in f.read()
 
 
     def test_glob_explicit(self):
         """
         Can a search pattern be used along with matching explicit notebook names?
         """
         with self.create_temp_cwd(['notebook*.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to="python"', 
-                '--notebooks=["*.ipynb", "notebook1.ipynb", "notebook2.ipynb"]']).lower()
+            self.call('nbconvert --to="python" --notebooks='
+                      '\'["*.ipynb","notebook1.ipynb","notebook2.ipynb"]\'')
             assert os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')
 
 
     def test_explicit_glob(self):
         """
         Can explicit notebook names be used and then a matching search pattern?
         """
         with self.create_temp_cwd(['notebook*.ipynb']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--to="python"', 
-                '--notebooks=["notebook1.ipynb", "notebook2.ipynb", "*.ipynb"]']).lower()
+            self.call('nbconvert --to="python" --notebooks='
+                      '\'["notebook1.ipynb","notebook2.ipynb","*.ipynb"]\'')
             assert os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')
 
 
     def test_default_config(self):
         """
         Does the default config work?
         """
         with self.create_temp_cwd(['notebook*.ipynb', 'ipython_nbconvert_config.py']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert']).lower()
+            self.call('nbconvert')
             assert os.path.isfile('notebook1.py')
             assert not os.path.isfile('notebook2.py')
 
 
     def test_override_config(self):
         """
         Can the default config be overriden?
         """
-        with self.create_temp_cwd(['notebook*.ipynb', 'ipython_nbconvert_config.py', 
+        with self.create_temp_cwd(['notebook*.ipynb',
+                                   'ipython_nbconvert_config.py',
                                    'override.py']):
-            assert not 'error' in self.call([IPYTHON, 'nbconvert', '--config="override.py"']).lower()
+            self.call('nbconvert --config="override.py"')
             assert not os.path.isfile('notebook1.py')
             assert os.path.isfile('notebook2.py')

IPython/testing/tools.py

@@ -1,399 +1,414 b''
 """Generic testing tools.
 
 Authors
 -------
 - Fernando Perez <[e-mail]>
 """
 
 from __future__ import absolute_import
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2009-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import os
-import pipes
 import re
 import sys
 import tempfile
 
 from contextlib import contextmanager
 from io import StringIO
 from subprocess import Popen, PIPE
 
 try:
     # These tools are used by parts of the runtime, so we make the nose
     # dependency optional at this point.  Nose is a hard dependency to run the
     # test suite, but NOT to use ipython itself.
     import nose.tools as nt
     has_nose = True
 except ImportError:
     has_nose = False
 
 from IPython.config.loader import Config
-from IPython.utils.process import getoutputerror
 from IPython.utils.text import list_strings
 from IPython.utils.io import temp_pyfile, Tee
 from IPython.utils import py3compat
 from IPython.utils.encoding import DEFAULT_ENCODING
 
 from . import decorators as dec
 from . import skipdoctest
 
 #-----------------------------------------------------------------------------
 # Functions and classes
 #-----------------------------------------------------------------------------
 
 # The docstring for full_path doctests differently on win32 (different path
 # separator) so just skip the doctest there.  The example remains informative.
 doctest_deco = skipdoctest.skip_doctest if sys.platform == 'win32' else dec.null_deco
 
 @doctest_deco
 def full_path(startPath,files):
     """Make full paths for all the listed files, based on startPath.
 
     Only the base part of startPath is kept, since this routine is typically
     used with a script's __file__ variable as startPath.  The base of startPath
     is then prepended to all the listed files, forming the output list.
 
     Parameters
     ----------
       startPath : string
         Initial path to use as the base for the results.  This path is split
       using os.path.split() and only its first component is kept.
 
       files : string or list
         One or more files.
 
     Examples
     --------
 
     >>> full_path('/foo/bar.py',['a.txt','b.txt'])
     ['/foo/a.txt', '/foo/b.txt']
 
     >>> full_path('/foo',['a.txt','b.txt'])
     ['/a.txt', '/b.txt']
 
     If a single file is given, the output is still a list:
     >>> full_path('/foo','a.txt')
     ['/a.txt']
     """
 
     files = list_strings(files)
     base = os.path.split(startPath)[0]
     return [ os.path.join(base,f) for f in files ]
 
 
 def parse_test_output(txt):
     """Parse the output of a test run and return errors, failures.
 
     Parameters
     ----------
     txt : str
       Text output of a test run, assumed to contain a line of one of the
       following forms::
       
         'FAILED (errors=1)'
         'FAILED (failures=1)'
         'FAILED (errors=1, failures=1)'
 
     Returns
     -------
     nerr, nfail: number of errors and failures.
     """
 
     err_m = re.search(r'^FAILED \(errors=(\d+)\)', txt, re.MULTILINE)
     if err_m:
         nerr = int(err_m.group(1))
         nfail = 0
         return  nerr, nfail
 
     fail_m = re.search(r'^FAILED \(failures=(\d+)\)', txt, re.MULTILINE)
     if fail_m:
         nerr = 0
         nfail = int(fail_m.group(1))
         return  nerr, nfail
 
     both_m = re.search(r'^FAILED \(errors=(\d+), failures=(\d+)\)', txt,
                        re.MULTILINE)
     if both_m:
         nerr = int(both_m.group(1))
         nfail = int(both_m.group(2))
         return  nerr, nfail
 
     # If the input didn't match any of these forms, assume no error/failures
     return 0, 0
 
 
 # So nose doesn't think this is a test
 parse_test_output.__test__ = False
 
 
 def default_argv():
     """Return a valid default argv for creating testing instances of ipython"""
 
     return ['--quick', # so no config file is loaded
             # Other defaults to minimize side effects on stdout
             '--colors=NoColor', '--no-term-title','--no-banner',
             '--autocall=0']
 
 
 def default_config():
     """Return a config object with good defaults for testing."""
     config = Config()
     config.TerminalInteractiveShell.colors = 'NoColor'
     config.TerminalTerminalInteractiveShell.term_title = False,
     config.TerminalInteractiveShell.autocall = 0
     config.HistoryManager.hist_file = tempfile.mktemp(u'test_hist.sqlite')
     config.HistoryManager.db_cache_size = 10000
     return config
 
 
+def get_ipython_cmd(as_string=False):
+    """
+    Return appropriate IPython command line name. By default, this will return
+    a list that can be used with subprocess.Popen, for example, but passing
+    `as_string=True` allows for returning the IPython command as a string.
+
+    Parameters
+    ----------
+    as_string: bool
+        Flag to allow to return the command as a string.
+    """
+    # FIXME: remove workaround for 2.6 support
+    if sys.version_info[:2] > (2,6):
+        ipython_cmd = [sys.executable, "-m", "IPython"]
+    else:
+        ipython_cmd = ["ipython"]
+
+    if as_string:
+        ipython_cmd = " ".join(ipython_cmd)
+
+    return ipython_cmd
+
 def ipexec(fname, options=None):
     """Utility to call 'ipython filename'.
 
     Starts IPython with a minimal and safe configuration to make startup as fast
     as possible.
 
     Note that this starts IPython in a subprocess!
 
     Parameters
     ----------
     fname : str
       Name of file to be executed (should have .py or .ipy extension).
 
     options : optional, list
       Extra command-line flags to be passed to IPython.
 
     Returns
     -------
     (stdout, stderr) of ipython subprocess.
     """
     if options is None: options = []
 
     # For these subprocess calls, eliminate all prompt printing so we only see
     # output from script execution
     prompt_opts = [ '--PromptManager.in_template=""',
                     '--PromptManager.in2_template=""',
                     '--PromptManager.out_template=""'
     ]
     cmdargs = default_argv() + prompt_opts + options
 
-    _ip = get_ipython()
     test_dir = os.path.dirname(__file__)
-    
-    # FIXME: remove workaround for 2.6 support
-    if sys.version_info[:2] > (2,6):
-        ipython_cmd = [sys.executable, "-m", "IPython"]
-    else:
-        ipython_cmd = ["ipython"]
+
+    ipython_cmd = get_ipython_cmd()
     # Absolute path for filename
     full_fname = os.path.join(test_dir, fname)
     full_cmd = ipython_cmd + cmdargs + [full_fname]
     p = Popen(full_cmd, stdout=PIPE, stderr=PIPE)
     out, err = p.communicate()
     out, err = py3compat.bytes_to_str(out), py3compat.bytes_to_str(err)
     # `import readline` causes 'ESC[?1034h' to be output sometimes,
     # so strip that out before doing comparisons
     if out:
         out = re.sub(r'\x1b\[[^h]+h', '', out)
     return out, err
 
 
 def ipexec_validate(fname, expected_out, expected_err='',
                     options=None):
     """Utility to call 'ipython filename' and validate output/error.
 
     This function raises an AssertionError if the validation fails.
 
     Note that this starts IPython in a subprocess!
 
     Parameters
     ----------
     fname : str
       Name of the file to be executed (should have .py or .ipy extension).
 
     expected_out : str
       Expected stdout of the process.
 
     expected_err : optional, str
       Expected stderr of the process.
 
     options : optional, list
       Extra command-line flags to be passed to IPython.
 
     Returns
     -------
     None
     """
 
     import nose.tools as nt
 
     out, err = ipexec(fname, options)
     #print 'OUT', out  # dbg
     #print 'ERR', err  # dbg
     # If there are any errors, we must check those befor stdout, as they may be
     # more informative than simply having an empty stdout.
     if err:
         if expected_err:
             nt.assert_equal("\n".join(err.strip().splitlines()), "\n".join(expected_err.strip().splitlines()))
         else:
             raise ValueError('Running file %r produced error: %r' %
                              (fname, err))
     # If no errors or output on stderr was expected, match stdout
     nt.assert_equal("\n".join(out.strip().splitlines()), "\n".join(expected_out.strip().splitlines()))
 
 
 class TempFileMixin(object):
     """Utility class to create temporary Python/IPython files.
 
     Meant as a mixin class for test cases."""
 
     def mktmp(self, src, ext='.py'):
         """Make a valid python temp file."""
         fname, f = temp_pyfile(src, ext)
         self.tmpfile = f
         self.fname = fname
 
     def tearDown(self):
         if hasattr(self, 'tmpfile'):
             # If the tmpfile wasn't made because of skipped tests, like in
             # win32, there's nothing to cleanup.
             self.tmpfile.close()
             try:
                 os.unlink(self.fname)
             except:
                 # On Windows, even though we close the file, we still can't
                 # delete it.  I have no clue why
                 pass
 
 pair_fail_msg = ("Testing {0}\n\n"
                 "In:\n"
                 "  {1!r}\n"
                 "Expected:\n"
                 "  {2!r}\n"
                 "Got:\n"
                 "  {3!r}\n")
 def check_pairs(func, pairs):
     """Utility function for the common case of checking a function with a
     sequence of input/output pairs.
 
     Parameters
     ----------
     func : callable
       The function to be tested. Should accept a single argument.
     pairs : iterable
       A list of (input, expected_output) tuples.
 
     Returns
     -------
     None. Raises an AssertionError if any output does not match the expected
     value.
     """
     name = getattr(func, "func_name", getattr(func, "__name__", "<unknown>"))
     for inp, expected in pairs:
         out = func(inp)
         assert out == expected, pair_fail_msg.format(name, inp, expected, out)
 
 
 if py3compat.PY3:
     MyStringIO = StringIO
 else:
     # In Python 2, stdout/stderr can have either bytes or unicode written to them,
     # so we need a class that can handle both.
     class MyStringIO(StringIO):
         def write(self, s):
             s = py3compat.cast_unicode(s, encoding=DEFAULT_ENCODING)
             super(MyStringIO, self).write(s)
 
 notprinted_msg = """Did not find {0!r} in printed output (on {1}):
 -------
 {2!s}
 -------
 """
 
 class AssertPrints(object):
     """Context manager for testing that code prints certain text.
     
     Examples
     --------
     >>> with AssertPrints("abc", suppress=False):
     ...     print "abcd"
     ...     print "def"
     ... 
     abcd
     def
     """
     def __init__(self, s, channel='stdout', suppress=True):
         self.s = s
         self.channel = channel
         self.suppress = suppress
     
     def __enter__(self):
         self.orig_stream = getattr(sys, self.channel)
         self.buffer = MyStringIO()
         self.tee = Tee(self.buffer, channel=self.channel)
         setattr(sys, self.channel, self.buffer if self.suppress else self.tee)
     
     def __exit__(self, etype, value, traceback):
         self.tee.flush()
         setattr(sys, self.channel, self.orig_stream)
         printed = self.buffer.getvalue()
         assert self.s in printed, notprinted_msg.format(self.s, self.channel, printed)
         return False
 
 printed_msg = """Found {0!r} in printed output (on {1}):
 -------
 {2!s}
 -------
 """
 
 class AssertNotPrints(AssertPrints):
     """Context manager for checking that certain output *isn't* produced.
     
     Counterpart of AssertPrints"""
     def __exit__(self, etype, value, traceback):
         self.tee.flush()
         setattr(sys, self.channel, self.orig_stream)
         printed = self.buffer.getvalue()
         assert self.s not in printed, printed_msg.format(self.s, self.channel, printed)
         return False
 
 @contextmanager
 def mute_warn():
     from IPython.utils import warn
     save_warn = warn.warn
     warn.warn = lambda *a, **kw: None
     try:
         yield
     finally:
         warn.warn = save_warn
 
 @contextmanager
 def make_tempfile(name):
     """ Create an empty, named, temporary file for the duration of the context.
     """
     f = open(name, 'w')
     f.close()
     try:
         yield
     finally:
         os.unlink(name)
 
 
 @contextmanager
 def monkeypatch(obj, name, attr):
     """
     Context manager to replace attribute named `name` in `obj` with `attr`.
     """
     orig = getattr(obj, name)
     setattr(obj, name, attr)
     yield
     setattr(obj, name, orig)

IPython/utils/_process_common.py

@@ -1,196 +1,214 b''
 """Common utilities for the various process_* implementations.
 
 This file is only meant to be imported by the platform-specific implementations
 of subprocess utilities, and it contains tools that are common to all of them.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 import subprocess
 import shlex
 import sys
 
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 def read_no_interrupt(p):
     """Read from a pipe ignoring EINTR errors.
 
     This is necessary because when reading from pipes with GUI event loops
     running in the background, often interrupts are raised that stop the
     command from completing."""
     import errno
 
     try:
         return p.read()
     except IOError as err:
         if err.errno != errno.EINTR:
             raise
 
 
 def process_handler(cmd, callback, stderr=subprocess.PIPE):
     """Open a command in a shell subprocess and execute a callback.
 
     This function provides common scaffolding for creating subprocess.Popen()
     calls.  It creates a Popen object and then calls the callback with it.
 
     Parameters
     ----------
     cmd : str
       A string to be executed with the underlying system shell (by calling
     :func:`Popen` with ``shell=True``.
 
     callback : callable
       A one-argument function that will be called with the Popen object.
 
     stderr : file descriptor number, optional
       By default this is set to ``subprocess.PIPE``, but you can also pass the
       value ``subprocess.STDOUT`` to force the subprocess' stderr to go into
       the same file descriptor as its stdout.  This is useful to read stdout
       and stderr combined in the order they are generated.
 
     Returns
     -------
     The return value of the provided callback is returned.
     """
     sys.stdout.flush()
     sys.stderr.flush()
     # On win32, close_fds can't be true when using pipes for stdin/out/err
     close_fds = sys.platform != 'win32'
     p = subprocess.Popen(cmd, shell=True,
                          stdin=subprocess.PIPE,
                          stdout=subprocess.PIPE,
                          stderr=stderr,
                          close_fds=close_fds)
 
     try:
         out = callback(p)
     except KeyboardInterrupt:
         print('^C')
         sys.stdout.flush()
         sys.stderr.flush()
         out = None
     finally:
         # Make really sure that we don't leave processes behind, in case the
         # call above raises an exception
         # We start by assuming the subprocess finished (to avoid NameErrors
         # later depending on the path taken)
         if p.returncode is None:
             try:
                 p.terminate()
                 p.poll()
             except OSError:
                 pass
         # One last try on our way out
         if p.returncode is None:
             try:
                 p.kill()
             except OSError:
                 pass
 
     return out
 
 
 def getoutput(cmd):
     """Run a command and return its stdout/stderr as a string.
 
     Parameters
     ----------
     cmd : str
       A command to be executed in the system shell.
 
     Returns
     -------
     output : str
       A string containing the combination of stdout and stderr from the
     subprocess, in whatever order the subprocess originally wrote to its
     file descriptors (so the order of the information in this string is the
     correct order as would be seen if running the command in a terminal).
     """
     out = process_handler(cmd, lambda p: p.communicate()[0], subprocess.STDOUT)
     if out is None:
         return ''
     return py3compat.bytes_to_str(out)
 
 
 def getoutputerror(cmd):
     """Return (standard output, standard error) of executing cmd in a shell.
 
     Accepts the same arguments as os.system().
 
     Parameters
     ----------
     cmd : str
       A command to be executed in the system shell.
 
     Returns
     -------
     stdout : str
     stderr : str
     """
+    return get_output_error_code(cmd)[:2]
 
-    out_err = process_handler(cmd, lambda p: p.communicate())
+def get_output_error_code(cmd):
+    """Return (standard output, standard error, return code) of executing cmd
+    in a shell.
+
+    Accepts the same arguments as os.system().
+
+    Parameters
+    ----------
+    cmd : str
+      A command to be executed in the system shell.
+
+    Returns
+    -------
+    stdout : str
+    stderr : str
+    returncode: int
+    """
+
+    out_err, p = process_handler(cmd, lambda p: (p.communicate(), p))
     if out_err is None:
-        return '', ''
+        return '', '', p.returncode
     out, err = out_err
-    return py3compat.bytes_to_str(out), py3compat.bytes_to_str(err)
-
+    return py3compat.bytes_to_str(out), py3compat.bytes_to_str(err), p.returncode
 
 def arg_split(s, posix=False, strict=True):
     """Split a command line's arguments in a shell-like manner.
 
     This is a modified version of the standard library's shlex.split()
     function, but with a default of posix=False for splitting, so that quotes
     in inputs are respected.
 
     if strict=False, then any errors shlex.split would raise will result in the
     unparsed remainder being the last element of the list, rather than raising.
     This is because we sometimes use arg_split to parse things other than
     command-line args.
     """
 
     # Unfortunately, python's shlex module is buggy with unicode input:
     # http://bugs.python.org/issue1170
     # At least encoding the input when it's unicode seems to help, but there
     # may be more problems lurking.  Apparently this is fixed in python3.
     is_unicode = False
     if (not py3compat.PY3) and isinstance(s, unicode):
         is_unicode = True
         s = s.encode('utf-8')
     lex = shlex.shlex(s, posix=posix)
     lex.whitespace_split = True
     # Extract tokens, ensuring that things like leaving open quotes
     # does not cause this to raise.  This is important, because we
     # sometimes pass Python source through this (e.g. %timeit f(" ")),
     # and it shouldn't raise an exception.
     # It may be a bad idea to parse things that are not command-line args
     # through this function, but we do, so let's be safe about it.
     lex.commenters='' #fix for GH-1269
     tokens = []
     while True:
         try:
             tokens.append(next(lex))
         except StopIteration:
             break
         except ValueError:
             if strict:
                 raise
             # couldn't parse, get remaining blob as last token
             tokens.append(lex.token)
             break
     
     if is_unicode:
         # Convert the tokens back to unicode.
         tokens = [x.decode('utf-8') for x in tokens]
     return tokens

IPython/utils/process.py

@@ -1,122 +1,122 b''
 # encoding: utf-8
 """
 Utilities for working with external processes.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import os
 import sys
 import shlex
 
 # Our own
 if sys.platform == 'win32':
     from ._process_win32 import _find_cmd, system, getoutput, AvoidUNCPath, arg_split
 else:
     from ._process_posix import _find_cmd, system, getoutput, arg_split
 
 
-from ._process_common import getoutputerror
+from ._process_common import getoutputerror, get_output_error_code
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 class FindCmdError(Exception):
     pass
 
 
 def find_cmd(cmd):
     """Find absolute path to executable cmd in a cross platform manner.
 
     This function tries to determine the full path to a command line program
     using `which` on Unix/Linux/OS X and `win32api` on Windows.  Most of the
     time it will use the version that is first on the users `PATH`.
 
     Warning, don't use this to find IPython command line programs as there
     is a risk you will find the wrong one.  Instead find those using the
     following code and looking for the application itself::
 
         from IPython.utils.path import get_ipython_module_path
         from IPython.utils.process import pycmd2argv
         argv = pycmd2argv(get_ipython_module_path('IPython.terminal.ipapp'))
 
     Parameters
     ----------
     cmd : str
         The command line program to look for.
     """
     try:
         path = _find_cmd(cmd).rstrip()
     except OSError:
         raise FindCmdError('command could not be found: %s' % cmd)
     # which returns empty if not found
     if path == '':
         raise FindCmdError('command could not be found: %s' % cmd)
     return os.path.abspath(path)
 
 
 def is_cmd_found(cmd):
     """Check whether executable `cmd` exists or not and return a bool."""
     try:
         find_cmd(cmd)
         return True
     except FindCmdError:
         return False
 
 
 def pycmd2argv(cmd):
     r"""Take the path of a python command and return a list (argv-style).
 
     This only works on Python based command line programs and will find the
     location of the ``python`` executable using ``sys.executable`` to make
     sure the right version is used.
 
     For a given path ``cmd``, this returns [cmd] if cmd's extension is .exe,
     .com or .bat, and [, cmd] otherwise.
 
     Parameters
     ----------
     cmd : string
       The path of the command.
 
     Returns
     -------
     argv-style list.
     """
     ext = os.path.splitext(cmd)[1]
     if ext in ['.exe', '.com', '.bat']:
         return [cmd]
     else:
         return [sys.executable, cmd]
 
 
 def abbrev_cwd():
     """ Return abbreviated version of cwd, e.g. d:mydir """
     cwd = os.getcwdu().replace('\\','/')
     drivepart = ''
     tail = cwd
     if sys.platform == 'win32':
         if len(cwd) < 4:
             return cwd
         drivepart,tail = os.path.splitdrive(cwd)
 
 
     parts = tail.split('/')
     if len(parts) > 2:
         tail = '/'.join(parts[-2:])
 
     return (drivepart + (
         cwd == '/' and '/' or tail))

IPython/utils/tempdir.py

@@ -1,106 +1,132 b''
 """TemporaryDirectory class, copied from Python 3.2.
 
 This is copied from the stdlib and will be standard in Python 3.2 and onwards.
 """
 
 import os as _os
 
 # This code should only be used in Python versions < 3.2, since after that we
 # can rely on the stdlib itself.
 try:
     from tempfile import TemporaryDirectory
 
 except ImportError:
     from tempfile import mkdtemp, template
 
     class TemporaryDirectory(object):
         """Create and return a temporary directory.  This has the same
         behavior as mkdtemp but can be used as a context manager.  For
         example:
 
             with TemporaryDirectory() as tmpdir:
                 ...
 
         Upon exiting the context, the directory and everthing contained
         in it are removed.
         """
 
         def __init__(self, suffix="", prefix=template, dir=None):
             self.name = mkdtemp(suffix, prefix, dir)
             self._closed = False
 
         def __enter__(self):
             return self.name
 
         def cleanup(self):
             if not self._closed:
                 self._rmtree(self.name)
                 self._closed = True
 
         def __exit__(self, exc, value, tb):
             self.cleanup()
 
         __del__ = cleanup
 
 
         # XXX (ncoghlan): The following code attempts to make
         # this class tolerant of the module nulling out process
         # that happens during CPython interpreter shutdown
         # Alas, it doesn't actually manage it. See issue #10188
         _listdir = staticmethod(_os.listdir)
         _path_join = staticmethod(_os.path.join)
         _isdir = staticmethod(_os.path.isdir)
         _remove = staticmethod(_os.remove)
         _rmdir = staticmethod(_os.rmdir)
         _os_error = _os.error
 
         def _rmtree(self, path):
             # Essentially a stripped down version of shutil.rmtree.  We can't
             # use globals because they may be None'ed out at shutdown.
             for name in self._listdir(path):
                 fullname = self._path_join(path, name)
                 try:
                     isdir = self._isdir(fullname)
                 except self._os_error:
                     isdir = False
                 if isdir:
                     self._rmtree(fullname)
                 else:
                     try:
                         self._remove(fullname)
                     except self._os_error:
                         pass
             try:
                 self._rmdir(path)
             except self._os_error:
                 pass
 
 
 class NamedFileInTemporaryDirectory(object):
 
     def __init__(self, filename, mode='w+b', bufsize=-1, **kwds):
         """
         Open a file named `filename` in a temporary directory.
 
         This context manager is preferred over `NamedTemporaryFile` in
         stdlib `tempfile` when one needs to reopen the file.
 
         Arguments `mode` and `bufsize` are passed to `open`.
         Rest of the arguments are passed to `TemporaryDirectory`.
 
         """
         self._tmpdir = TemporaryDirectory(**kwds)
         path = _os.path.join(self._tmpdir.name, filename)
         self.file = open(path, mode, bufsize)
 
     def cleanup(self):
         self.file.close()
         self._tmpdir.cleanup()
 
     __del__ = cleanup
 
     def __enter__(self):
         return self.file
 
     def __exit__(self, type, value, traceback):
         self.cleanup()
+
+
+class TemporaryWorkingDirectory(TemporaryDirectory):
+    """
+    Creates a temporary directory and sets the cwd to that directory.
+    Automatically reverts to previous cwd upon cleanup.
+    Usage example:
+
+        with TemporaryWorakingDirectory() as tmpdir:
+            ...
+    """
+
+    def __init__(self, **kw):
+        super(TemporaryWorkingDirectory, self).__init__(**kw)
+
+        #Change cwd to new temp dir.  Remember old cwd.
+        self.old_wd = _os.getcwd()
+        _os.chdir(self.name)
+
+
+    def cleanup(self):
+        #Revert to old cwd.
+        _os.chdir(self.old_wd)
+
+        #Cleanup
+        super(TemporaryWorkingDirectory, self).cleanup()

IPython/utils/tests/test_process.py

@@ -1,134 +1,146 b''
 # encoding: utf-8
 """
 Tests for platutils.py
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import sys
 import os
 from unittest import TestCase
 
 import nose.tools as nt
 
 from IPython.utils.process import (find_cmd, FindCmdError, arg_split,
-                                   system, getoutput, getoutputerror)
+                                   system, getoutput, getoutputerror,
+                                   get_output_error_code)
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 
 python = os.path.basename(sys.executable)
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 
 @dec.skip_win32
 def test_find_cmd_ls():
     """Make sure we can find the full path to ls."""
     path = find_cmd('ls')
     nt.assert_true(path.endswith('ls'))
 
     
 def has_pywin32():
     try:
         import win32api
     except ImportError:
         return False
     return True
 
 
 @dec.onlyif(has_pywin32, "This test requires win32api to run")
 def test_find_cmd_pythonw():
     """Try to find pythonw on Windows."""
     path = find_cmd('pythonw')
     nt.assert_true(path.endswith('pythonw.exe'))
 
 
 @dec.onlyif(lambda : sys.platform != 'win32' or has_pywin32(),
             "This test runs on posix or in win32 with win32api installed")
 def test_find_cmd_fail():
     """Make sure that FindCmdError is raised if we can't find the cmd."""
     nt.assert_raises(FindCmdError,find_cmd,'asdfasdf')
 
     
 @dec.skip_win32
 def test_arg_split():
     """Ensure that argument lines are correctly split like in a shell."""
     tests = [['hi', ['hi']],
              [u'hi', [u'hi']],
              ['hello there', ['hello', 'there']],
              # \u01ce == \N{LATIN SMALL LETTER A WITH CARON}
              # Do not use \N because the tests crash with syntax error in
              # some cases, for example windows python2.6.
              [u'h\u01cello', [u'h\u01cello']],
              ['something "with quotes"', ['something', '"with quotes"']],
              ]
     for argstr, argv in tests:
         nt.assert_equal(arg_split(argstr), argv)
     
 @dec.skip_if_not_win32
 def test_arg_split_win32():
     """Ensure that argument lines are correctly split like in a shell."""
     tests = [['hi', ['hi']],
              [u'hi', [u'hi']],
              ['hello there', ['hello', 'there']],
              [u'h\u01cello', [u'h\u01cello']],
              ['something "with quotes"', ['something', 'with quotes']],
              ]
     for argstr, argv in tests:
         nt.assert_equal(arg_split(argstr), argv)
 
 
 class SubProcessTestCase(TestCase, tt.TempFileMixin):
     def setUp(self):
         """Make a valid python temp file."""
         lines = ["from __future__ import print_function",
                  "import sys",
                  "print('on stdout', end='', file=sys.stdout)",
                  "print('on stderr', end='', file=sys.stderr)",
                  "sys.stdout.flush()",
                  "sys.stderr.flush()"]
         self.mktmp('\n'.join(lines))
 
     def test_system(self):
         status = system('%s "%s"' % (python, self.fname))
         self.assertEqual(status, 0)
 
     def test_system_quotes(self):
         status = system('%s -c "import sys"' % python)
         self.assertEqual(status, 0)
 
     def test_getoutput(self):
         out = getoutput('%s "%s"' % (python, self.fname))
         # we can't rely on the order the line buffered streams are flushed
         try:
             self.assertEqual(out, 'on stderron stdout')
         except AssertionError:
             self.assertEqual(out, 'on stdouton stderr')
 
     def test_getoutput_quoted(self):
         out = getoutput('%s -c "print (1)"' % python)
         self.assertEqual(out.strip(), '1')
 
     #Invalid quoting on windows
     @dec.skip_win32
     def test_getoutput_quoted2(self):
         out = getoutput("%s -c 'print (1)'" % python)
         self.assertEqual(out.strip(), '1')
         out = getoutput("%s -c 'print (\"1\")'" % python)
         self.assertEqual(out.strip(), '1')
 
     def test_getoutput_error(self):
         out, err = getoutputerror('%s "%s"' % (python, self.fname))
         self.assertEqual(out, 'on stdout')
         self.assertEqual(err, 'on stderr')
+    
+    def test_get_output_error_code(self):
+        quiet_exit = '%s -c "import sys; sys.exit(1)"' % python
+        out, err, code = get_output_error_code(quiet_exit)
+        self.assertEqual(out, '')
+        self.assertEqual(err, '')
+        self.assertEqual(code, 1)
+        out, err, code = get_output_error_code('%s "%s"' % (python, self.fname))
+        self.assertEqual(out, 'on stdout')
+        self.assertEqual(err, 'on stderr')
+        self.assertEqual(code, 0)

IPython/utils/tests/test_tempdir.py

@@ -1,20 +1,28 b''
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2012-  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 import os
 
 from IPython.utils.tempdir import NamedFileInTemporaryDirectory
+from IPython.utils.tempdir import TemporaryWorkingDirectory
 
 
 def test_named_file_in_temporary_directory():
     with NamedFileInTemporaryDirectory('filename') as file:
         name = file.name
         assert not file.closed
         assert os.path.exists(name)
         file.write(b'test')
     assert file.closed
     assert not os.path.exists(name)
+
+def test_temporary_working_directory():
+    with TemporaryWorkingDirectory() as dir:
+        assert os.path.exists(dir)
+        assert os.path.abspath(os.curdir) == dir
+    assert not os.path.exists(dir)
+    assert os.path.abspath(os.curdir) != dir