Merge pull request #13242 from Kojoley/remove-unused-_find_cmd

Remove unused _find_cmd

IPython/utils/_process_cli.py

@@ -1,78 +1,70 b''
 """cli-specific implementation of process utilities.
 
 cli - Common Language Infrastructure for IronPython. Code
       can run on any operating system. Check os.name for os-
       specific settings.
 
 This file is only meant to be imported by process.py, not by end-users.
 
 This file is largely untested. To become a full drop-in process
 interface for IronPython will probably require you to help fill
 in the details. 
 """
 
 # Import cli libraries:
 import clr
 import System
 
 # Import Python libraries:
 import os
 
 # Import IPython libraries:
 from IPython.utils import py3compat
 from ._process_common import arg_split
 
-def _find_cmd(cmd):
-    """Find the full path to a command using which."""
-    paths = System.Environment.GetEnvironmentVariable("PATH").Split(os.pathsep)
-    for path in paths:
-        filename = os.path.join(path, cmd)
-        if System.IO.File.Exists(filename):
-            return py3compat.decode(filename)
-    raise OSError("command %r not found" % cmd)
 
 def system(cmd):
     """
     system(cmd) should work in a cli environment on Mac OSX, Linux,
     and Windows
     """
     psi = System.Diagnostics.ProcessStartInfo(cmd)
     psi.RedirectStandardOutput = True
     psi.RedirectStandardError = True
     psi.WindowStyle = System.Diagnostics.ProcessWindowStyle.Normal
     psi.UseShellExecute = False
     # Start up process:
     reg = System.Diagnostics.Process.Start(psi)
 
 def getoutput(cmd):
     """
     getoutput(cmd) should work in a cli environment on Mac OSX, Linux,
     and Windows
     """
     psi = System.Diagnostics.ProcessStartInfo(cmd)
     psi.RedirectStandardOutput = True
     psi.RedirectStandardError = True
     psi.WindowStyle = System.Diagnostics.ProcessWindowStyle.Normal
     psi.UseShellExecute = False
     # Start up process:
     reg = System.Diagnostics.Process.Start(psi)
     myOutput = reg.StandardOutput
     output = myOutput.ReadToEnd()
     myError = reg.StandardError
     error = myError.ReadToEnd()
     return output
 
 def check_pid(pid):
     """
     Check if a process with the given PID (pid) exists
     """
     try:
         System.Diagnostics.Process.GetProcessById(pid)
         # process with given pid is running
         return True
     except System.InvalidOperationException:
         # process wasn't started by this object (but is running)
         return True
     except System.ArgumentException:
         # process with given pid isn't running
         return False 

IPython/utils/_process_posix.py

@@ -1,225 +1,217 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
 #-----------------------------------------------------------------------------
 
 # Stdlib
 import errno
 import os
 import subprocess as sp
 import sys
 
 import pexpect
 
 # Our own
 from ._process_common import getoutput, arg_split
 from IPython.utils import py3compat
 from IPython.utils.encoding import DEFAULT_ENCODING
 
 #-----------------------------------------------------------------------------
 # 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, stderr=sp.PIPE).communicate()[0]
-    return py3compat.decode(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
 
     @property
     def sh(self):
         if self._sh is None:
             shell_name = os.environ.get("SHELL", "sh")
             self._sh = pexpect.which(shell_name)
             if self._sh is None:
                 raise OSError('"{}" shell not found'.format(shell_name))
 
         return self._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 = DEFAULT_ENCODING
         
         # 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()
 
         # We follow the subprocess pattern, returning either the exit status
         # as a positive number, or the terminating signal as a negative
         # number.
         # on Linux, sh returns 128+n for signals terminating child processes on Linux
         # on BSD (OS X), the signal code is set instead
         if child.exitstatus is None:
             # on WIFSIGNALED, pexpect sets signalstatus, leaving exitstatus=None
             if child.signalstatus is None:
                 # this condition may never occur,
                 # but let's be certain we always return an integer.
                 return 0
             return -child.signalstatus
         if child.exitstatus > 128:
             return -(child.exitstatus - 128)
         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 check_pid(pid):
     try:
         os.kill(pid, 0)
     except OSError as err:
         if err.errno == errno.ESRCH:
             return False
         elif err.errno == errno.EPERM:
             # Don't have permission to signal the process - probably means it exists
             return True
         raise
     else:
         return True

IPython/utils/_process_win32.py

@@ -1,205 +1,184 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
 #-----------------------------------------------------------------------------
 
 # stdlib
 import os
 import sys
 import ctypes
 import time
 
 from ctypes import c_int, POINTER
 from ctypes.wintypes import LPCWSTR, HLOCAL
 from subprocess import STDOUT, TimeoutExpired
 from threading import Thread
 
 # our own imports
 from ._process_common import read_no_interrupt, process_handler, arg_split as py_arg_split
 from . import py3compat
 from .encoding import DEFAULT_ENCODING
 
 #-----------------------------------------------------------------------------
 # 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.
 
     Examples
     --------
     ::
         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.getcwd()
         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 as e:
-        raise ImportError('you need to have pywin32 installed for this to work') from e
-    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 = DEFAULT_ENCODING
 
     def stdout_read():
         for line in read_no_interrupt(p.stdout).splitlines():
             line = line.decode(enc, 'replace')
             print(line, file=sys.stdout)
 
     def stderr_read():
         for line in read_no_interrupt(p.stderr).splitlines():
             line = line.decode(enc, 'replace')
             print(line, file=sys.stderr)
 
     Thread(target=stdout_read).start()
     Thread(target=stderr_read).start()
 
     # Wait to finish for returncode. Unfortunately, Python has a bug where
     # wait() isn't interruptible (https://bugs.python.org/issue28168) so poll in
     # a loop instead of just doing `return p.wait()`.
     while True:
         result = p.poll()
         if result is None:
             time.sleep(0.01)
         else:
             return result
 
 
 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 or list
         A command to be executed in the system shell.
 
     Returns
     -------
     int : child process' exit code.
     """
     # The controller provides interactivity with both
     # stdin and stdout
     #import _process_win32_controller
     #_process_win32_controller.system(cmd)
 
     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 or list
         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 = b''
     return py3compat.decode(out)
 
 try:
     CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW
     CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
     CommandLineToArgvW.restype = POINTER(LPCWSTR)
     LocalFree = ctypes.windll.kernel32.LocalFree
     LocalFree.res_type = HLOCAL
     LocalFree.arg_types = [HLOCAL]
     
     def arg_split(commandline, posix=False, strict=True):
         """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 parameter is ignored.
 
         If strict=False, process_common.arg_split(...strict=False) is used instead.
         """
         #CommandLineToArgvW returns path to executable if called with empty string.
         if commandline.strip() == "":
             return []
         if not strict:
             # not really a cl-arg, fallback on _process_common
             return py_arg_split(commandline, posix=posix, strict=strict)
         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(ctypes.addressof(result_pointer.contents))]
         retval = LocalFree(result_pointer)
         return result
 except AttributeError:
     arg_split = py_arg_split
 
 def check_pid(pid):
     # OpenProcess returns 0 if no such process (of ours) exists
     # positive int otherwise
     return bool(ctypes.windll.kernel32.OpenProcess(1,0,pid))