Moved arg_split to _process_win32.py and _process_posix.py.

IPython/utils/_process_posix.py

@@ -1,194 +1,221 b''
 """Posix-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # Stdlib
 import subprocess as sp
 import sys
+import shlex
 
 from IPython.external import pexpect
 
 # Our own
 from .autoattr import auto_attr
 from ._process_common import getoutput
 from IPython.utils import text
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 def _find_cmd(cmd):
     """Find the full path to a command using which."""
 
     path = sp.Popen(['/usr/bin/env', 'which', cmd],
                     stdout=sp.PIPE).communicate()[0]
     return py3compat.bytes_to_str(path)
 
 
 class ProcessHandler(object):
     """Execute subprocesses under the control of pexpect.
     """
     # Timeout in seconds to wait on each reading of the subprocess' output.
     # This should not be set too low to avoid cpu overusage from our side,
     # since we read in a loop whose period is controlled by this timeout.
     read_timeout = 0.05
 
     # Timeout to give a process if we receive SIGINT, between sending the
     # SIGINT to the process and forcefully terminating it.
     terminate_timeout = 0.2
 
     # File object where stdout and stderr of the subprocess will be written
     logfile = None
 
     # Shell to call for subprocesses to execute
     sh = None
 
     @auto_attr
     def sh(self):
         sh = pexpect.which('sh')
         if sh is None:
             raise OSError('"sh" shell not found')
         return sh
 
     def __init__(self, logfile=None, read_timeout=None, terminate_timeout=None):
         """Arguments are used for pexpect calls."""
         self.read_timeout = (ProcessHandler.read_timeout if read_timeout is
                              None else read_timeout)
         self.terminate_timeout = (ProcessHandler.terminate_timeout if
                                   terminate_timeout is None else
                                   terminate_timeout)
         self.logfile = sys.stdout if logfile is None else logfile
 
     def getoutput(self, 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).
         """
         try:
             return pexpect.run(self.sh, args=['-c', cmd]).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def getoutput_pexpect(self, 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).
         """
         try:
             return pexpect.run(self.sh, args=['-c', cmd]).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def system(self, cmd):
         """Execute a command in a subshell.
 
         Parameters
         ----------
         cmd : str
           A command to be executed in the system shell.
 
         Returns
         -------
         int : child's exitstatus
         """
         # Get likely encoding for the output.
         enc = text.getdefaultencoding()
         
         # Patterns to match on the output, for pexpect.  We read input and
         # allow either a short timeout or EOF
         patterns = [pexpect.TIMEOUT, pexpect.EOF]
         # the index of the EOF pattern in the list.
         # even though we know it's 1, this call means we don't have to worry if
         # we change the above list, and forget to change this value:
         EOF_index = patterns.index(pexpect.EOF)
         # The size of the output stored so far in the process output buffer.
         # Since pexpect only appends to this buffer, each time we print we
         # record how far we've printed, so that next time we only print *new*
         # content from the buffer.
         out_size = 0
         try:
             # Since we're not really searching the buffer for text patterns, we
             # can set pexpect's search window to be tiny and it won't matter.
             # We only search for the 'patterns' timeout or EOF, which aren't in
             # the text itself.
             #child = pexpect.spawn(pcmd, searchwindowsize=1)
             if hasattr(pexpect, 'spawnb'):
                 child = pexpect.spawnb(self.sh, args=['-c', cmd]) # Pexpect-U
             else:
                 child = pexpect.spawn(self.sh, args=['-c', cmd])  # Vanilla Pexpect
             flush = sys.stdout.flush
             while True:
                 # res is the index of the pattern that caused the match, so we
                 # know whether we've finished (if we matched EOF) or not
                 res_idx = child.expect_list(patterns, self.read_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 flush()
                 if res_idx==EOF_index:
                     break
                 # Update the pointer to what we've already printed
                 out_size = len(child.before)
         except KeyboardInterrupt:
             # We need to send ^C to the process.  The ascii code for '^C' is 3
             # (the character is known as ETX for 'End of Text', see
             # curses.ascii.ETX).
             child.sendline(chr(3))
             # Read and print any more output the program might produce on its
             # way out.
             try:
                 out_size = len(child.before)
                 child.expect_list(patterns, self.terminate_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 sys.stdout.flush()
             except KeyboardInterrupt:
                 # Impatient users tend to type it multiple times
                 pass
             finally:
                 # Ensure the subprocess really is terminated
                 child.terminate(force=True)
         # add isalive check, to ensure exitstatus is set:
         child.isalive()
         return child.exitstatus
 
 
 # Make system() with a functional interface for outside use.  Note that we use
 # getoutput() from the _common utils, which is built on top of popen(). Using
 # pexpect to get subprocess output produces difficult to parse output, since
 # programs think they are talking to a tty and produce highly formatted output
 # (ls is a good example) that makes them hard.
 system = ProcessHandler().system
+
+def arg_split(s, posix=False):
+    """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."""
+
+    # 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
+    tokens = list(lex)
+    if is_unicode:
+        # Convert the tokens back to unicode.
+        tokens = [x.decode('utf-8') for x in tokens]
+    return tokens
+
+
+

IPython/utils/_process_win32.py

@@ -1,148 +1,176 b''
 """Windows-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 from __future__ import print_function
 
 # stdlib
 import os
 import sys
+import ctypes
 
+from ctypes import c_int, POINTER
+from ctypes.wintypes import LPCWSTR, HLOCAL
 from subprocess import STDOUT
 
 # our own imports
 from ._process_common import read_no_interrupt, process_handler
+from . import py3compat
 from . import text
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 class AvoidUNCPath(object):
     """A context manager to protect command execution from UNC paths.
 
     In the Win32 API, commands can't be invoked with the cwd being a UNC path.
     This context manager temporarily changes directory to the 'C:' drive on
     entering, and restores the original working directory on exit.
 
     The context manager returns the starting working directory *if* it made a
     change and None otherwise, so that users can apply the necessary adjustment
     to their system calls in the event of a change.
 
     Example
     -------
     ::
         cmd = 'dir'
         with AvoidUNCPath() as path:
             if path is not None:
                 cmd = '"pushd %s &&"%s' % (path, cmd)
             os.system(cmd)
     """
     def __enter__(self):
         self.path = os.getcwdu()
         self.is_unc_path = self.path.startswith(r"\\")
         if self.is_unc_path:
             # change to c drive (as cmd.exe cannot handle UNC addresses)
             os.chdir("C:")
             return self.path
         else:
             # We return None to signal that there was no change in the working
             # directory
             return None
 
     def __exit__(self, exc_type, exc_value, traceback):
         if self.is_unc_path:
             os.chdir(self.path)
 
 
 def _find_cmd(cmd):
     """Find the full path to a .bat or .exe using the win32api module."""
     try:
         from win32api import SearchPath
     except ImportError:
         raise ImportError('you need to have pywin32 installed for this to work')
     else:
         PATH = os.environ['PATH']
         extensions = ['.exe', '.com', '.bat', '.py']
         path = None
         for ext in extensions:
             try:
                 path = SearchPath(PATH, cmd + ext)[0]
             except:
                 pass
         if path is None:
             raise OSError("command %r not found" % cmd)
         else:
             return path
 
 
 def _system_body(p):
     """Callback for _system."""
     enc = text.getdefaultencoding()
     for line in read_no_interrupt(p.stdout).splitlines():
         line = line.decode(enc, 'replace')
         print(line, file=sys.stdout)
     for line in read_no_interrupt(p.stderr).splitlines():
         line = line.decode(enc, 'replace')
         print(line, file=sys.stderr)
 
     # Wait to finish for returncode
     return p.wait()
 
 
 def system(cmd):
     """Win32 version of os.system() that works with network shares.
 
     Note that this implementation returns None, as meant for use in IPython.
 
     Parameters
     ----------
     cmd : str
       A command to be executed in the system shell.
 
     Returns
     -------
     None : we explicitly do NOT return the subprocess status code, as this
     utility is meant to be used extensively in IPython, where any return value
     would trigger :func:`sys.displayhook` calls.
     """
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         return process_handler(cmd, _system_body)
 
 
 def getoutput(cmd):
     """Return standard output 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
     """
 
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         out = process_handler(cmd, lambda p: p.communicate()[0], STDOUT)
 
     if out is None:
         out = ''
     return out
+
+
+CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW
+CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
+CommandLineToArgvW.res_types = [POINTER(LPCWSTR)]
+LocalFree = ctypes.windll.kernel32.LocalFree
+LocalFree.res_type = HLOCAL
+LocalFree.arg_types = [HLOCAL]
+
+def arg_split(commandline, posix=False):
+    """Split a command line's arguments in a shell-like manner.
+
+    This is a special version for windows that use a ctypes call to CommandLineToArgvW
+    to do the argv splitting. The posix paramter is ignored.
+    """
+    #CommandLineToArgvW returns path to executable if called with empty string.
+    if commandline.strip() == "":
+        return []
+    argvn = c_int()
+    result_pointer = CommandLineToArgvW(py3compat.cast_unicode(commandline.lstrip()), ctypes.byref(argvn))
+    result_array_type = LPCWSTR * argvn.value
+    result = [arg for arg in result_array_type.from_address(result_pointer)]
+    retval = LocalFree(result_pointer)
+    return result

IPython/utils/process.py

@@ -1,169 +1,123 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':
-    import ctypes
-    from ctypes.wintypes import LPCWSTR, HLOCAL
-    from ctypes import c_int, POINTER
-    from ._process_win32 import _find_cmd, system, getoutput, AvoidUNCPath
-    CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW
-    CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
-    CommandLineToArgvW.res_types = [POINTER(LPCWSTR)]
-    LocalFree = ctypes.windll.kernel32.LocalFree
-    LocalFree.res_type = HLOCAL
-    LocalFree.arg_types = [HLOCAL]
+    from ._process_win32 import _find_cmd, system, getoutput, AvoidUNCPath, arg_split
 else:
-    from ._process_posix import _find_cmd, system, getoutput
+    from ._process_posix import _find_cmd, system, getoutput, arg_split
+
 
 from ._process_common import getoutputerror
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # 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`.  If
     cmd is `python` return `sys.executable`.
 
     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.frontend.terminal.ipapp'))
 
     Parameters
     ----------
     cmd : str
         The command line program to look for.
     """
     if cmd == 'python':
         return os.path.abspath(sys.executable)
     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 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:
         if sys.platform == 'win32':
             # The -u option here turns on unbuffered output, which is required
             # on Win32 to prevent wierd conflict and problems with Twisted.
             # Also, use sys.executable to make sure we are picking up the
             # right python exe.
             return [sys.executable, '-u', cmd]
         else:
             return [sys.executable, cmd]
 
-if sys.platform == 'win32':
-    def arg_split(commandline, posix=False):
-        """Split a command line's arguments in a shell-like manner.
-
-        This is a special version for windows that use a ctypes call to CommandLineToArgvW
-        to do the argv splitting. The posix paramter is ignored.
-        """
-        argvn = c_int()
-        result_pointer = CommandLineToArgvW(py3compat.str_to_unicode(commandline.lstrip()), ctypes.byref(argvn))
-        result_array_type = LPCWSTR * argvn.value
-        result = [arg for arg in result_array_type.from_address(result_pointer)]
-        retval = LocalFree(result_pointer)
-        return result
-else:
-    def arg_split(s, posix=False):
-        """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."""
-
-        # 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
-        tokens = list(lex)
-        if is_unicode:
-            # Convert the tokens back to unicode.
-            tokens = [x.decode('utf-8') for x in tokens]
-        return tokens
-
-
 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/tests/test_process.py

@@ -1,128 +1,131 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
 from unittest import TestCase
 
 import nose.tools as nt
 
 from IPython.utils.process import (find_cmd, FindCmdError, arg_split,
                                    system, getoutput, getoutputerror)
 from IPython.testing import decorators as dec
 from IPython.testing import tools as tt
 
 #-----------------------------------------------------------------------------
 # Tests
 #-----------------------------------------------------------------------------
 
 def test_find_cmd_python():
     """Make sure we find sys.exectable for python."""
     nt.assert_equals(find_cmd('python'), sys.executable)
 
     
 @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']],
-#             [u'h\N{LATIN SMALL LETTER A WITH CARON}llo', [u'h\N{LATIN SMALL LETTER A WITH CARON}llo']],
+             # \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\N{LATIN SMALL LETTER A WITH CARON}llo', [u'h\N{LATIN SMALL LETTER A WITH CARON}llo']],
+             [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('python "%s"' % self.fname)
         self.assertEquals(status, 0)
 
     def test_system_quotes(self):
         status = system('python -c "import sys"')
         self.assertEquals(status, 0)
 
     def test_getoutput(self):
         out = getoutput('python "%s"' % self.fname)
         self.assertEquals(out, 'on stdout')
 
     def test_getoutput_quoted(self):
         out = getoutput('python -c "print (1)"')
         self.assertEquals(out.strip(), '1')
 
     #Invalid quoting on windows
     @dec.skip_win32
     def test_getoutput_quoted2(self):
         out = getoutput("python -c 'print (1)'")
         self.assertEquals(out.strip(), '1')
         out = getoutput("python -c 'print (\"1\")'")
         self.assertEquals(out.strip(), '1')
 
     def test_getoutput(self):
         out, err = getoutputerror('python "%s"' % self.fname)
         self.assertEquals(out, 'on stdout')
         self.assertEquals(err, 'on stderr')