Remove unused imports in IPython.utils

IPython/utils/_process_win32.py

@@ -1,188 +1,187 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
-import msvcrt
 
 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, 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.
 
     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 = DEFAULT_ENCODING
     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.
     """
     # 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
       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.bytes_to_str(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 paramter 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

IPython/utils/_process_win32_controller.py

@@ -1,574 +1,574 b''
 """Windows-specific implementation of process utilities with direct WinAPI.
 
 This file is meant to be used by process.py
 """
 
 #-----------------------------------------------------------------------------
 #  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.
 #-----------------------------------------------------------------------------
 
 from __future__ import print_function
 
 # stdlib
-import os, sys, time, threading
+import os, sys, threading
 import ctypes, msvcrt
 
 # Win32 API types needed for the API calls
 from ctypes import POINTER
 from ctypes.wintypes import HANDLE, HLOCAL, LPVOID, WORD, DWORD, BOOL, \
         ULONG, LPCWSTR
 LPDWORD = POINTER(DWORD)
 LPHANDLE = POINTER(HANDLE)
 ULONG_PTR = POINTER(ULONG)
 class SECURITY_ATTRIBUTES(ctypes.Structure):
     _fields_ = [("nLength", DWORD),
                 ("lpSecurityDescriptor", LPVOID),
                 ("bInheritHandle", BOOL)]
 LPSECURITY_ATTRIBUTES = POINTER(SECURITY_ATTRIBUTES)
 class STARTUPINFO(ctypes.Structure):
     _fields_ = [("cb", DWORD),
                 ("lpReserved", LPCWSTR),
                 ("lpDesktop", LPCWSTR),
                 ("lpTitle", LPCWSTR),
                 ("dwX", DWORD),
                 ("dwY", DWORD),
                 ("dwXSize", DWORD),
                 ("dwYSize", DWORD),
                 ("dwXCountChars", DWORD),
                 ("dwYCountChars", DWORD),
                 ("dwFillAttribute", DWORD),
                 ("dwFlags", DWORD),
                 ("wShowWindow", WORD),
                 ("cbReserved2", WORD),
                 ("lpReserved2", LPVOID),
                 ("hStdInput", HANDLE),
                 ("hStdOutput", HANDLE),
                 ("hStdError", HANDLE)]
 LPSTARTUPINFO = POINTER(STARTUPINFO)
 class PROCESS_INFORMATION(ctypes.Structure):
     _fields_ = [("hProcess", HANDLE),
                 ("hThread", HANDLE),
                 ("dwProcessId", DWORD),
                 ("dwThreadId", DWORD)]
 LPPROCESS_INFORMATION = POINTER(PROCESS_INFORMATION)
 
 # Win32 API constants needed
 ERROR_HANDLE_EOF = 38
 ERROR_BROKEN_PIPE = 109
 ERROR_NO_DATA = 232
 HANDLE_FLAG_INHERIT = 0x0001
 STARTF_USESTDHANDLES = 0x0100
 CREATE_SUSPENDED = 0x0004
 CREATE_NEW_CONSOLE = 0x0010
 CREATE_NO_WINDOW = 0x08000000
 STILL_ACTIVE = 259
 WAIT_TIMEOUT = 0x0102
 WAIT_FAILED = 0xFFFFFFFF
 INFINITE = 0xFFFFFFFF
 DUPLICATE_SAME_ACCESS = 0x00000002
 ENABLE_ECHO_INPUT = 0x0004
 ENABLE_LINE_INPUT = 0x0002
 ENABLE_PROCESSED_INPUT = 0x0001
 
 # Win32 API functions needed
 GetLastError = ctypes.windll.kernel32.GetLastError
 GetLastError.argtypes = []
 GetLastError.restype = DWORD
 
 CreateFile = ctypes.windll.kernel32.CreateFileW
 CreateFile.argtypes = [LPCWSTR, DWORD, DWORD, LPVOID, DWORD, DWORD, HANDLE]
 CreateFile.restype = HANDLE
 
 CreatePipe = ctypes.windll.kernel32.CreatePipe
 CreatePipe.argtypes = [POINTER(HANDLE), POINTER(HANDLE),
         LPSECURITY_ATTRIBUTES, DWORD]
 CreatePipe.restype = BOOL
 
 CreateProcess = ctypes.windll.kernel32.CreateProcessW
 CreateProcess.argtypes = [LPCWSTR, LPCWSTR, LPSECURITY_ATTRIBUTES,
         LPSECURITY_ATTRIBUTES, BOOL, DWORD, LPVOID, LPCWSTR, LPSTARTUPINFO,
         LPPROCESS_INFORMATION]
 CreateProcess.restype = BOOL
 
 GetExitCodeProcess = ctypes.windll.kernel32.GetExitCodeProcess
 GetExitCodeProcess.argtypes = [HANDLE, LPDWORD]
 GetExitCodeProcess.restype = BOOL
 
 GetCurrentProcess = ctypes.windll.kernel32.GetCurrentProcess
 GetCurrentProcess.argtypes = []
 GetCurrentProcess.restype = HANDLE
 
 ResumeThread = ctypes.windll.kernel32.ResumeThread
 ResumeThread.argtypes = [HANDLE]
 ResumeThread.restype = DWORD
 
 ReadFile = ctypes.windll.kernel32.ReadFile
 ReadFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, LPVOID]
 ReadFile.restype = BOOL
 
 WriteFile = ctypes.windll.kernel32.WriteFile
 WriteFile.argtypes = [HANDLE, LPVOID, DWORD, LPDWORD, LPVOID]
 WriteFile.restype = BOOL
 
 GetConsoleMode = ctypes.windll.kernel32.GetConsoleMode
 GetConsoleMode.argtypes = [HANDLE, LPDWORD]
 GetConsoleMode.restype = BOOL
 
 SetConsoleMode = ctypes.windll.kernel32.SetConsoleMode
 SetConsoleMode.argtypes = [HANDLE, DWORD]
 SetConsoleMode.restype = BOOL
 
 FlushConsoleInputBuffer = ctypes.windll.kernel32.FlushConsoleInputBuffer
 FlushConsoleInputBuffer.argtypes = [HANDLE]
 FlushConsoleInputBuffer.restype = BOOL
 
 WaitForSingleObject = ctypes.windll.kernel32.WaitForSingleObject
 WaitForSingleObject.argtypes = [HANDLE, DWORD]
 WaitForSingleObject.restype = DWORD
 
 DuplicateHandle = ctypes.windll.kernel32.DuplicateHandle
 DuplicateHandle.argtypes = [HANDLE, HANDLE, HANDLE, LPHANDLE,
         DWORD, BOOL, DWORD]
 DuplicateHandle.restype = BOOL
 
 SetHandleInformation = ctypes.windll.kernel32.SetHandleInformation
 SetHandleInformation.argtypes = [HANDLE, DWORD, DWORD]
 SetHandleInformation.restype = BOOL
 
 CloseHandle = ctypes.windll.kernel32.CloseHandle
 CloseHandle.argtypes = [HANDLE]
 CloseHandle.restype = BOOL
 
 CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW
 CommandLineToArgvW.argtypes = [LPCWSTR, POINTER(ctypes.c_int)]
 CommandLineToArgvW.restype = POINTER(LPCWSTR)
 
 LocalFree = ctypes.windll.kernel32.LocalFree
 LocalFree.argtypes = [HLOCAL]
 LocalFree.restype = HLOCAL
 
 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)
 
 
 class Win32ShellCommandController(object):
     """Runs a shell command in a 'with' context.
 
     This implementation is Win32-specific.
 
     Example:
         # Runs the command interactively with default console stdin/stdout
         with ShellCommandController('python -i') as scc:
             scc.run()
 
         # Runs the command using the provided functions for stdin/stdout
         def my_stdout_func(s):
             # print or save the string 's'
             write_to_stdout(s)
         def my_stdin_func():
             # If input is available, return it as a string.
             if input_available():
                 return get_input()
             # If no input available, return None after a short delay to
             # keep from blocking.
             else:
                 time.sleep(0.01)
                 return None
       
         with ShellCommandController('python -i') as scc:
             scc.run(my_stdout_func, my_stdin_func)
     """
 
     def __init__(self, cmd, mergeout = True):
         """Initializes the shell command controller.
 
         The cmd is the program to execute, and mergeout is
         whether to blend stdout and stderr into one output
         in stdout. Merging them together in this fashion more
         reliably keeps stdout and stderr in the correct order
         especially for interactive shell usage.
         """
         self.cmd = cmd
         self.mergeout = mergeout
 
     def __enter__(self):
         cmd = self.cmd
         mergeout = self.mergeout
 
         self.hstdout, self.hstdin, self.hstderr = None, None, None
         self.piProcInfo = None
         try:
             p_hstdout, c_hstdout, p_hstderr, \
                     c_hstderr, p_hstdin, c_hstdin = [None]*6
 
             # SECURITY_ATTRIBUTES with inherit handle set to True
             saAttr = SECURITY_ATTRIBUTES()
             saAttr.nLength = ctypes.sizeof(saAttr)
             saAttr.bInheritHandle = True
             saAttr.lpSecurityDescriptor = None
 
             def create_pipe(uninherit):
                 """Creates a Windows pipe, which consists of two handles.
 
                 The 'uninherit' parameter controls which handle is not
                 inherited by the child process.
                 """
                 handles = HANDLE(), HANDLE()
                 if not CreatePipe(ctypes.byref(handles[0]),
                             ctypes.byref(handles[1]), ctypes.byref(saAttr), 0):
                     raise ctypes.WinError()
                 if not SetHandleInformation(handles[uninherit],
                             HANDLE_FLAG_INHERIT, 0):
                     raise ctypes.WinError()
                 return handles[0].value, handles[1].value
 
             p_hstdout, c_hstdout = create_pipe(uninherit=0)
             # 'mergeout' signals that stdout and stderr should be merged.
             # We do that by using one pipe for both of them.
             if mergeout:
                 c_hstderr = HANDLE()
                 if not DuplicateHandle(GetCurrentProcess(), c_hstdout,
                                 GetCurrentProcess(), ctypes.byref(c_hstderr),
                                 0, True, DUPLICATE_SAME_ACCESS):
                     raise ctypes.WinError()
             else:
                 p_hstderr, c_hstderr = create_pipe(uninherit=0)
             c_hstdin,  p_hstdin  = create_pipe(uninherit=1)
 
             # Create the process object
             piProcInfo = PROCESS_INFORMATION()
             siStartInfo = STARTUPINFO()
             siStartInfo.cb = ctypes.sizeof(siStartInfo)
             siStartInfo.hStdInput = c_hstdin
             siStartInfo.hStdOutput = c_hstdout
             siStartInfo.hStdError = c_hstderr
             siStartInfo.dwFlags = STARTF_USESTDHANDLES
             dwCreationFlags = CREATE_SUSPENDED | CREATE_NO_WINDOW # | CREATE_NEW_CONSOLE
 
             if not CreateProcess(None,
                     u"cmd.exe /c " + cmd,
                     None, None, True, dwCreationFlags,
                     None, None, ctypes.byref(siStartInfo),
                     ctypes.byref(piProcInfo)):
                 raise ctypes.WinError()
 
             # Close this process's versions of the child handles
             CloseHandle(c_hstdin)
             c_hstdin = None
             CloseHandle(c_hstdout)
             c_hstdout = None
             if c_hstderr != None:
                 CloseHandle(c_hstderr)
                 c_hstderr = None
 
             # Transfer ownership of the parent handles to the object
             self.hstdin = p_hstdin
             p_hstdin = None
             self.hstdout = p_hstdout
             p_hstdout = None
             if not mergeout:
                 self.hstderr = p_hstderr
                 p_hstderr = None
             self.piProcInfo = piProcInfo
 
         finally:
             if p_hstdin:
                 CloseHandle(p_hstdin)
             if c_hstdin:
                 CloseHandle(c_hstdin)
             if p_hstdout:
                 CloseHandle(p_hstdout)
             if c_hstdout:
                 CloseHandle(c_hstdout)
             if p_hstderr:
                 CloseHandle(p_hstderr)
             if c_hstderr:
                 CloseHandle(c_hstderr)
 
         return self
 
     def _stdin_thread(self, handle, hprocess, func, stdout_func):
         exitCode = DWORD()
         bytesWritten = DWORD(0)
         while True:
             #print("stdin thread loop start")
             # Get the input string (may be bytes or unicode)
             data = func()
 
             # None signals to poll whether the process has exited
             if data is None:
                 #print("checking for process completion")
                 if not GetExitCodeProcess(hprocess, ctypes.byref(exitCode)):
                     raise ctypes.WinError()
                 if exitCode.value != STILL_ACTIVE:
                     return
                 # TESTING: Does zero-sized writefile help?
                 if not WriteFile(handle, "", 0,
                         ctypes.byref(bytesWritten), None):
                     raise ctypes.WinError()
                 continue
             #print("\nGot str %s\n" % repr(data), file=sys.stderr)
 
             # Encode the string to the console encoding
             if isinstance(data, unicode): #FIXME: Python3
                 data = data.encode('utf_8')
 
             # What we have now must be a string of bytes
             if not isinstance(data, str): #FIXME: Python3
                 raise RuntimeError("internal stdin function string error")
 
             # An empty string signals EOF
             if len(data) == 0:
                 return
 
             # In a windows console, sometimes the input is echoed,
             # but sometimes not. How do we determine when to do this?
             stdout_func(data)
             # WriteFile may not accept all the data at once.
             # Loop until everything is processed
             while len(data) != 0:
                 #print("Calling writefile")
                 if not WriteFile(handle, data, len(data),
                         ctypes.byref(bytesWritten), None):
                     # This occurs at exit
                     if GetLastError() == ERROR_NO_DATA:
                         return
                     raise ctypes.WinError()
                 #print("Called writefile")
                 data = data[bytesWritten.value:]
 
     def _stdout_thread(self, handle, func):
         # Allocate the output buffer
         data = ctypes.create_string_buffer(4096)
         while True:
             bytesRead = DWORD(0)
             if not ReadFile(handle, data, 4096,
                         ctypes.byref(bytesRead), None):
                 le = GetLastError()
                 if le == ERROR_BROKEN_PIPE:
                     return
                 else:
                     raise ctypes.WinError()
             # FIXME: Python3
             s = data.value[0:bytesRead.value]
             #print("\nv: %s" % repr(s), file=sys.stderr)
             func(s.decode('utf_8', 'replace'))
 
     def run(self, stdout_func = None, stdin_func = None, stderr_func = None):
         """Runs the process, using the provided functions for I/O.
 
         The function stdin_func should return strings whenever a
         character or characters become available.
         The functions stdout_func and stderr_func are called whenever
         something is printed to stdout or stderr, respectively.
         These functions are called from different threads (but not
         concurrently, because of the GIL).
         """
         if stdout_func == None and stdin_func == None and stderr_func == None:
             return self._run_stdio()
 
         if stderr_func != None and self.mergeout:
             raise RuntimeError("Shell command was initiated with "
                     "merged stdin/stdout, but a separate stderr_func "
                     "was provided to the run() method")
 
         # Create a thread for each input/output handle
         stdin_thread = None
         threads = []
         if stdin_func:
             stdin_thread = threading.Thread(target=self._stdin_thread,
                                 args=(self.hstdin, self.piProcInfo.hProcess,
                                 stdin_func, stdout_func))
         threads.append(threading.Thread(target=self._stdout_thread,
                                     args=(self.hstdout, stdout_func)))
         if not self.mergeout:
             if stderr_func == None:
                 stderr_func = stdout_func
             threads.append(threading.Thread(target=self._stdout_thread,
                                         args=(self.hstderr, stderr_func)))
         # Start the I/O threads and the process
         if ResumeThread(self.piProcInfo.hThread) == 0xFFFFFFFF:
             raise ctypes.WinError()
         if stdin_thread is not None:
             stdin_thread.start()
         for thread in threads:
             thread.start()
         # Wait for the process to complete
         if WaitForSingleObject(self.piProcInfo.hProcess, INFINITE) == \
                     WAIT_FAILED:
             raise ctypes.WinError()
         # Wait for the I/O threads to complete
         for thread in threads:
             thread.join()
 
         # Wait for the stdin thread to complete
         if stdin_thread is not None:
             stdin_thread.join()
 
     def _stdin_raw_nonblock(self):
         """Use the raw Win32 handle of sys.stdin to do non-blocking reads"""
         # WARNING: This is experimental, and produces inconsistent results.
         #          It's possible for the handle not to be appropriate for use
         #          with WaitForSingleObject, among other things.
         handle = msvcrt.get_osfhandle(sys.stdin.fileno())
         result = WaitForSingleObject(handle, 100)
         if result == WAIT_FAILED:
             raise ctypes.WinError()
         elif result == WAIT_TIMEOUT:
             print(".", end='')
             return None
         else:
             data = ctypes.create_string_buffer(256)
             bytesRead = DWORD(0)
             print('?', end='')
 
             if not ReadFile(handle, data, 256,
                         ctypes.byref(bytesRead), None):
                 raise ctypes.WinError()
             # This ensures the non-blocking works with an actual console
             # Not checking the error, so the processing will still work with
             # other handle types
             FlushConsoleInputBuffer(handle)
 
             data = data.value
             data = data.replace('\r\n', '\n')
             data = data.replace('\r', '\n')
             print(repr(data) + " ", end='')
             return data
 
     def _stdin_raw_block(self):
         """Use a blocking stdin read"""
         # The big problem with the blocking read is that it doesn't
         # exit when it's supposed to in all contexts. An extra
         # key-press may be required to trigger the exit.
         try:
             data = sys.stdin.read(1)
             data = data.replace('\r', '\n')
             return data
         except WindowsError as we:
             if we.winerror == ERROR_NO_DATA:
                 # This error occurs when the pipe is closed
                 return None
             else:
                 # Otherwise let the error propagate
                 raise we
 
     def _stdout_raw(self, s):
         """Writes the string to stdout"""
         print(s, end='', file=sys.stdout)
         sys.stdout.flush()
 
     def _stderr_raw(self, s):
         """Writes the string to stdout"""
         print(s, end='', file=sys.stderr)
         sys.stderr.flush()
 
     def _run_stdio(self):
         """Runs the process using the system standard I/O.
 
         IMPORTANT: stdin needs to be asynchronous, so the Python
                    sys.stdin object is not used. Instead,
                    msvcrt.kbhit/getwch are used asynchronously.
         """
         # Disable Line and Echo mode
         #lpMode = DWORD()
         #handle = msvcrt.get_osfhandle(sys.stdin.fileno())
         #if GetConsoleMode(handle, ctypes.byref(lpMode)):
         #    set_console_mode = True
         #    if not SetConsoleMode(handle, lpMode.value &
         #            ~(ENABLE_ECHO_INPUT | ENABLE_LINE_INPUT | ENABLE_PROCESSED_INPUT)):
         #        raise ctypes.WinError()
 
         if self.mergeout:
             return self.run(stdout_func = self._stdout_raw,
                     stdin_func = self._stdin_raw_block)
         else:
             return self.run(stdout_func = self._stdout_raw,
                     stdin_func = self._stdin_raw_block,
                     stderr_func = self._stderr_raw)
 
         # Restore the previous console mode
         #if set_console_mode:
         #    if not SetConsoleMode(handle, lpMode.value):
         #        raise ctypes.WinError()
 
     def __exit__(self, exc_type, exc_value, traceback):
         if self.hstdin:
             CloseHandle(self.hstdin)
             self.hstdin = None
         if self.hstdout:
             CloseHandle(self.hstdout)
             self.hstdout = None
         if self.hstderr:
             CloseHandle(self.hstderr)
             self.hstderr = None
         if self.piProcInfo != None:
             CloseHandle(self.piProcInfo.hProcess)
             CloseHandle(self.piProcInfo.hThread)
             self.piProcInfo = None
 
 
 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)
         with Win32ShellCommandController(cmd) as scc:
             scc.run()
 
 
 if __name__ == "__main__":
     print("Test starting!")
     #system("cmd")
     system("python -i")
     print("Test finished!")

IPython/utils/pickleutil.py

@@ -1,354 +1,353 b''
 # encoding: utf-8
 
 """Pickle related utilities. Perhaps this should be called 'can'."""
 
 __docformat__ = "restructuredtext en"
 
 #-------------------------------------------------------------------------------
 #  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 copy
 import logging
 import sys
 from types import FunctionType
 
 try:
     import cPickle as pickle
 except ImportError:
     import pickle
 
 try:
     import numpy
 except:
     numpy = None
 
-import codeutil
 import py3compat
 from importstring import import_item
 
 from IPython.config import Application
 
 if py3compat.PY3:
     buffer = memoryview
     class_type = type
 else:
     from types import ClassType
     class_type = (type, ClassType)
 
 #-------------------------------------------------------------------------------
 # Classes
 #-------------------------------------------------------------------------------
 
 
 class CannedObject(object):
     def __init__(self, obj, keys=[], hook=None):
         """can an object for safe pickling
         
         Parameters
         ==========
         
         obj:
             The object to be canned
         keys: list (optional)
             list of attribute names that will be explicitly canned / uncanned
         hook: callable (optional)
             An optional extra callable,
             which can do additional processing of the uncanned object.
         
         large data may be offloaded into the buffers list,
         used for zero-copy transfers.
         """
         self.keys = keys
         self.obj = copy.copy(obj)
         self.hook = can(hook)
         for key in keys:
             setattr(self.obj, key, can(getattr(obj, key)))
         
         self.buffers = []
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         obj = self.obj
         for key in self.keys:
             setattr(obj, key, uncan(getattr(obj, key), g))
         
         if self.hook:
             self.hook = uncan(self.hook, g)
             self.hook(obj, g)
         return self.obj
     
 
 class Reference(CannedObject):
     """object for wrapping a remote reference by name."""
     def __init__(self, name):
         if not isinstance(name, basestring):
             raise TypeError("illegal name: %r"%name)
         self.name = name
         self.buffers = []
 
     def __repr__(self):
         return "<Reference: %r>"%self.name
 
     def get_object(self, g=None):
         if g is None:
             g = {}
         
         return eval(self.name, g)
 
 
 class CannedFunction(CannedObject):
 
     def __init__(self, f):
         self._check_type(f)
         self.code = f.func_code
         if f.func_defaults:
             self.defaults = [ can(fd) for fd in f.func_defaults ]
         else:
             self.defaults = None
         self.module = f.__module__ or '__main__'
         self.__name__ = f.__name__
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, FunctionType), "Not a function type"
 
     def get_object(self, g=None):
         # try to load function back into its module:
         if not self.module.startswith('__'):
             __import__(self.module)
             g = sys.modules[self.module].__dict__
 
         if g is None:
             g = {}
         if self.defaults:
             defaults = tuple(uncan(cfd, g) for cfd in self.defaults)
         else:
             defaults = None
         newFunc = FunctionType(self.code, g, self.__name__, defaults)
         return newFunc
 
 class CannedClass(CannedObject):
 
     def __init__(self, cls):
         self._check_type(cls)
         self.name = cls.__name__
         self.old_style = not isinstance(cls, type)
         self._canned_dict = {}
         for k,v in cls.__dict__.items():
             if k not in ('__weakref__', '__dict__'):
                 self._canned_dict[k] = can(v)
         if self.old_style:
             mro = []
         else:
             mro = cls.mro()
         
         self.parents = [ can(c) for c in mro[1:] ]
         self.buffers = []
 
     def _check_type(self, obj):
         assert isinstance(obj, class_type), "Not a class type"
 
     def get_object(self, g=None):
         parents = tuple(uncan(p, g) for p in self.parents)
         return type(self.name, parents, uncan_dict(self._canned_dict, g=g))
 
 class CannedArray(CannedObject):
     def __init__(self, obj):
         self.shape = obj.shape
         self.dtype = obj.dtype.descr if obj.dtype.fields else obj.dtype.str
         if sum(obj.shape) == 0:
             # just pickle it
             self.buffers = [pickle.dumps(obj, -1)]
         else:
             # ensure contiguous
             obj = numpy.ascontiguousarray(obj, dtype=None)
             self.buffers = [buffer(obj)]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         if sum(self.shape) == 0:
             # no shape, we just pickled it
             return pickle.loads(data)
         else:
             return numpy.frombuffer(data, dtype=self.dtype).reshape(self.shape)
 
 
 class CannedBytes(CannedObject):
     wrap = bytes
     def __init__(self, obj):
         self.buffers = [obj]
     
     def get_object(self, g=None):
         data = self.buffers[0]
         return self.wrap(data)
 
 def CannedBuffer(CannedBytes):
     wrap = buffer
 
 #-------------------------------------------------------------------------------
 # Functions
 #-------------------------------------------------------------------------------
 
 def _logger():
     """get the logger for the current Application
     
     the root logger will be used if no Application is running
     """
     if Application.initialized():
         logger = Application.instance().log
     else:
         logger = logging.getLogger()
         if not logger.handlers:
             logging.basicConfig()
     
     return logger
 
 def _import_mapping(mapping, original=None):
     """import any string-keys in a type mapping
     
     """
     log = _logger()
     log.debug("Importing canning map")
     for key,value in mapping.items():
         if isinstance(key, basestring):
             try:
                 cls = import_item(key)
             except Exception:
                 if original and key not in original:
                     # only message on user-added classes
                     log.error("cannning class not importable: %r", key, exc_info=True)
                 mapping.pop(key)
             else:
                 mapping[cls] = mapping.pop(key)
 
 def istype(obj, check):
     """like isinstance(obj, check), but strict
     
     This won't catch subclasses.
     """
     if isinstance(check, tuple):
         for cls in check:
             if type(obj) is cls:
                 return True
         return False
     else:
         return type(obj) is check
 
 def can(obj):
     """prepare an object for pickling"""
     
     import_needed = False
     
     for cls,canner in can_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif istype(obj, cls):
             return canner(obj)
     
     if import_needed:
         # perform can_map imports, then try again
         # this will usually only happen once
         _import_mapping(can_map, _original_can_map)
         return can(obj)
     
     return obj
 
 def can_class(obj):
     if isinstance(obj, class_type) and obj.__module__ == '__main__':
         return CannedClass(obj)
     else:
         return obj
 
 def can_dict(obj):
     """can the *values* of a dict"""
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = can(v)
         return newobj
     else:
         return obj
 
 sequence_types = (list, tuple, set)
 
 def can_sequence(obj):
     """can the elements of a sequence"""
     if istype(obj, sequence_types):
         t = type(obj)
         return t([can(i) for i in obj])
     else:
         return obj
 
 def uncan(obj, g=None):
     """invert canning"""
     
     import_needed = False
     for cls,uncanner in uncan_map.iteritems():
         if isinstance(cls, basestring):
             import_needed = True
             break
         elif isinstance(obj, cls):
             return uncanner(obj, g)
     
     if import_needed:
         # perform uncan_map imports, then try again
         # this will usually only happen once
         _import_mapping(uncan_map, _original_uncan_map)
         return uncan(obj, g)
     
     return obj
 
 def uncan_dict(obj, g=None):
     if istype(obj, dict):
         newobj = {}
         for k, v in obj.iteritems():
             newobj[k] = uncan(v,g)
         return newobj
     else:
         return obj
 
 def uncan_sequence(obj, g=None):
     if istype(obj, sequence_types):
         t = type(obj)
         return t([uncan(i,g) for i in obj])
     else:
         return obj
 
 def _uncan_dependent_hook(dep, g=None):
     dep.check_dependency()
     
 def can_dependent(obj):
     return CannedObject(obj, keys=('f', 'df'), hook=_uncan_dependent_hook)
 
 #-------------------------------------------------------------------------------
 # API dictionaries
 #-------------------------------------------------------------------------------
 
 # These dicts can be extended for custom serialization of new objects
 
 can_map = {
     'IPython.parallel.dependent' : can_dependent,
     'numpy.ndarray' : CannedArray,
     FunctionType : CannedFunction,
     bytes : CannedBytes,
     buffer : CannedBuffer,
     class_type : can_class,
 }
 
 uncan_map = {
     CannedObject : lambda obj, g: obj.get_object(g),
 }
 
 # for use in _import_mapping:
 _original_can_map = can_map.copy()
 _original_uncan_map = uncan_map.copy()

IPython/utils/tests/test_module_paths.py

@@ -1,135 +1,125 b''
 # encoding: utf-8
-"""Tests for IPython.utils.path.py"""
+"""Tests for IPython.utils.module_paths.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
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import shutil
 import sys
 import tempfile
-import StringIO
 
 from os.path import join, abspath, split
 
-import nose.tools as nt
-
-from nose import with_setup
-
-import IPython
-from IPython.testing import decorators as dec
-from IPython.testing.decorators import skip_if_not_win32, skip_win32
 from IPython.testing.tools import make_tempfile
-from IPython.utils import path, io
-from IPython.utils import py3compat
 
 import IPython.utils.module_paths as mp
 
 env = os.environ
 TEST_FILE_PATH = split(abspath(__file__))[0]
 TMP_TEST_DIR = tempfile.mkdtemp()
 #
 # Setup/teardown functions/decorators
 #
 
 old_syspath = sys.path
 
 def make_empty_file(fname):
     f = open(fname, 'w')
     f.close()
 
 
 def setup():
     """Setup testenvironment for the module:
 
     """
     # Do not mask exceptions here.  In particular, catching WindowsError is a
     # problem because that exception is only defined on Windows...
     os.makedirs(join(TMP_TEST_DIR, "xmod"))
     os.makedirs(join(TMP_TEST_DIR, "nomod"))
     make_empty_file(join(TMP_TEST_DIR, "xmod/__init__.py"))
     make_empty_file(join(TMP_TEST_DIR, "xmod/sub.py"))
     make_empty_file(join(TMP_TEST_DIR, "pack.py"))
     make_empty_file(join(TMP_TEST_DIR, "packpyc.pyc"))
     sys.path = [TMP_TEST_DIR]
 
 def teardown():
     """Teardown testenvironment for the module:
 
             - Remove tempdir
             - restore sys.path
     """
     # Note: we remove the parent test dir, which is the root of all test
     # subdirs we may have created.  Use shutil instead of os.removedirs, so
     # that non-empty directories are all recursively removed.
     shutil.rmtree(TMP_TEST_DIR)
     sys.path = old_syspath
 
 
 def test_get_init_1():
     """See if get_init can find __init__.py in this testdir"""
     with make_tempfile(join(TMP_TEST_DIR, "__init__.py")):
         assert mp.get_init(TMP_TEST_DIR)
 
 def test_get_init_2():
     """See if get_init can find __init__.pyw in this testdir"""
     with make_tempfile(join(TMP_TEST_DIR, "__init__.pyw")):
         assert mp.get_init(TMP_TEST_DIR)
 
 def test_get_init_3():
     """get_init can't find __init__.pyc in this testdir"""
     with make_tempfile(join(TMP_TEST_DIR, "__init__.pyc")):
         assert mp.get_init(TMP_TEST_DIR) is None
 
-def test_get_init_3():
+def test_get_init_4():
     """get_init can't find __init__ in empty testdir"""
     assert mp.get_init(TMP_TEST_DIR) is None
 
 
 def test_find_mod_1():
     modpath = join(TMP_TEST_DIR, "xmod", "__init__.py")
     assert mp.find_mod("xmod") == modpath
 
 def test_find_mod_2():
     modpath = join(TMP_TEST_DIR, "xmod", "__init__.py")
     assert mp.find_mod("xmod") == modpath
 
 def test_find_mod_3():
     modpath = join(TMP_TEST_DIR, "xmod", "sub.py")
     assert mp.find_mod("xmod.sub") == modpath
 
 def test_find_mod_4():
     modpath = join(TMP_TEST_DIR, "pack.py")
     assert mp.find_mod("pack") == modpath
 
 def test_find_mod_5():
     assert mp.find_mod("packpyc") is None
 
 def test_find_module_1():
     modpath = join(TMP_TEST_DIR, "xmod")
     assert mp.find_module("xmod") == modpath
 
 def test_find_module_2():
     """Testing sys.path that is empty"""
     assert mp.find_module("xmod", []) is None
 
 def test_find_module_3():
     """Testing sys.path that is empty"""
     assert mp.find_module(None, None) is None
 
 def test_find_module_4():
     """Testing sys.path that is empty"""
     assert mp.find_module(None) is None
 
 def test_find_module_5():
     assert mp.find_module("xmod.nopack") is None

IPython/utils/tests/test_path.py

@@ -1,560 +1,559 b''
 # encoding: utf-8
 """Tests for IPython.utils.path.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
 #-----------------------------------------------------------------------------
 
 from __future__ import with_statement
 
 import os
 import shutil
 import sys
 import tempfile
-from io import StringIO
 from contextlib import contextmanager
 
 from os.path import join, abspath, split
 
 import nose.tools as nt
 
 from nose import with_setup
 
 import IPython
 from IPython.testing import decorators as dec
 from IPython.testing.decorators import skip_if_not_win32, skip_win32
 from IPython.testing.tools import make_tempfile, AssertPrints
-from IPython.utils import path, io
+from IPython.utils import path
 from IPython.utils import py3compat
 from IPython.utils.tempdir import TemporaryDirectory
 
 # Platform-dependent imports
 try:
     import _winreg as wreg
 except ImportError:
     #Fake _winreg module on none windows platforms
     import types
     wr_name = "winreg" if py3compat.PY3 else "_winreg"
     sys.modules[wr_name] = types.ModuleType(wr_name)
     import _winreg as wreg
     #Add entries that needs to be stubbed by the testing code
     (wreg.OpenKey, wreg.QueryValueEx,) = (None, None)
 
 try:
     reload
 except NameError:   # Python 3
     from imp import reload
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 env = os.environ
 TEST_FILE_PATH = split(abspath(__file__))[0]
 TMP_TEST_DIR = tempfile.mkdtemp()
 HOME_TEST_DIR = join(TMP_TEST_DIR, "home_test_dir")
 XDG_TEST_DIR = join(HOME_TEST_DIR, "xdg_test_dir")
 XDG_CACHE_DIR = join(HOME_TEST_DIR, "xdg_cache_dir")
 IP_TEST_DIR = join(HOME_TEST_DIR,'.ipython')
 #
 # Setup/teardown functions/decorators
 #
 
 def setup():
     """Setup testenvironment for the module:
 
             - Adds dummy home dir tree
     """
     # Do not mask exceptions here.  In particular, catching WindowsError is a
     # problem because that exception is only defined on Windows...
     os.makedirs(IP_TEST_DIR)
     os.makedirs(os.path.join(XDG_TEST_DIR, 'ipython'))
     os.makedirs(os.path.join(XDG_CACHE_DIR, 'ipython'))
 
 
 def teardown():
     """Teardown testenvironment for the module:
 
             - Remove dummy home dir tree
     """
     # Note: we remove the parent test dir, which is the root of all test
     # subdirs we may have created.  Use shutil instead of os.removedirs, so
     # that non-empty directories are all recursively removed.
     shutil.rmtree(TMP_TEST_DIR)
 
 
 def setup_environment():
     """Setup testenvironment for some functions that are tested
     in this module. In particular this functions stores attributes
     and other things that we need to stub in some test functions.
     This needs to be done on a function level and not module level because
     each testfunction needs a pristine environment.
     """
     global oldstuff, platformstuff
     oldstuff = (env.copy(), os.name, sys.platform, path.get_home_dir, IPython.__file__, os.getcwd())
 
     if os.name == 'nt':
         platformstuff = (wreg.OpenKey, wreg.QueryValueEx,)
 
 
 def teardown_environment():
     """Restore things that were remebered by the setup_environment function
     """
     (oldenv, os.name, sys.platform, path.get_home_dir, IPython.__file__, old_wd) = oldstuff
     os.chdir(old_wd)
     reload(path)
 
     for key in env.keys():
         if key not in oldenv:
             del env[key]
     env.update(oldenv)
     if hasattr(sys, 'frozen'):
         del sys.frozen
     if os.name == 'nt':
         (wreg.OpenKey, wreg.QueryValueEx,) = platformstuff
 
 # Build decorator that uses the setup_environment/setup_environment
 with_environment = with_setup(setup_environment, teardown_environment)
 
 @skip_if_not_win32
 @with_environment
 def test_get_home_dir_1():
     """Testcase for py2exe logic, un-compressed lib
     """
     sys.frozen = True
 
     #fake filename for IPython.__init__
     IPython.__file__ = abspath(join(HOME_TEST_DIR, "Lib/IPython/__init__.py"))
 
     home_dir = path.get_home_dir()
     nt.assert_equal(home_dir, abspath(HOME_TEST_DIR))
 
 
 @skip_if_not_win32
 @with_environment
 def test_get_home_dir_2():
     """Testcase for py2exe logic, compressed lib
     """
     sys.frozen = True
     #fake filename for IPython.__init__
     IPython.__file__ = abspath(join(HOME_TEST_DIR, "Library.zip/IPython/__init__.py")).lower()
 
     home_dir = path.get_home_dir(True)
     nt.assert_equal(home_dir, abspath(HOME_TEST_DIR).lower())
 
 
 @with_environment
 def test_get_home_dir_3():
     """get_home_dir() uses $HOME if set"""
     env["HOME"] = HOME_TEST_DIR
     home_dir = path.get_home_dir(True)
     # get_home_dir expands symlinks
     nt.assert_equal(home_dir, os.path.realpath(env["HOME"]))
 
 
 @with_environment
 def test_get_home_dir_4():
     """get_home_dir() still works if $HOME is not set"""
 
     if 'HOME' in env: del env['HOME']
     # this should still succeed, but we don't care what the answer is
     home = path.get_home_dir(False)
 
 @with_environment
 def test_get_home_dir_5():
     """raise HomeDirError if $HOME is specified, but not a writable dir"""
     env['HOME'] = abspath(HOME_TEST_DIR+'garbage')
     # set os.name = posix, to prevent My Documents fallback on Windows
     os.name = 'posix'
     nt.assert_raises(path.HomeDirError, path.get_home_dir, True)
 
 
 # Should we stub wreg fully so we can run the test on all platforms?
 @skip_if_not_win32
 @with_environment
 def test_get_home_dir_8():
     """Using registry hack for 'My Documents', os=='nt'
 
     HOMESHARE, HOMEDRIVE, HOMEPATH, USERPROFILE and others are missing.
     """
     os.name = 'nt'
     # Remove from stub environment all keys that may be set
     for key in ['HOME', 'HOMESHARE', 'HOMEDRIVE', 'HOMEPATH', 'USERPROFILE']:
         env.pop(key, None)
 
     #Stub windows registry functions
     def OpenKey(x, y):
         class key:
             def Close(self):
                 pass
         return key()
     def QueryValueEx(x, y):
         return [abspath(HOME_TEST_DIR)]
 
     wreg.OpenKey = OpenKey
     wreg.QueryValueEx = QueryValueEx
 
     home_dir = path.get_home_dir()
     nt.assert_equal(home_dir, abspath(HOME_TEST_DIR))
 
 
 @with_environment
 def test_get_ipython_dir_1():
     """test_get_ipython_dir_1, Testcase to see if we can call get_ipython_dir without Exceptions."""
     env_ipdir = os.path.join("someplace", ".ipython")
     path._writable_dir = lambda path: True
     env['IPYTHONDIR'] = env_ipdir
     ipdir = path.get_ipython_dir()
     nt.assert_equal(ipdir, env_ipdir)
 
 
 @with_environment
 def test_get_ipython_dir_2():
     """test_get_ipython_dir_2, Testcase to see if we can call get_ipython_dir without Exceptions."""
     path.get_home_dir = lambda : "someplace"
     path.get_xdg_dir = lambda : None
     path._writable_dir = lambda path: True
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     ipdir = path.get_ipython_dir()
     nt.assert_equal(ipdir, os.path.join("someplace", ".ipython"))
 
 @with_environment
 def test_get_ipython_dir_3():
     """test_get_ipython_dir_3, use XDG if defined, and .ipython doesn't exist."""
     path.get_home_dir = lambda : "someplace"
     path._writable_dir = lambda path: True
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env['XDG_CONFIG_HOME'] = XDG_TEST_DIR
     ipdir = path.get_ipython_dir()
     if sys.platform == "darwin":
         expected = os.path.join("someplace", ".ipython")
     else:
         expected = os.path.join(XDG_TEST_DIR, "ipython")
     nt.assert_equal(ipdir, expected)
 
 @with_environment
 def test_get_ipython_dir_4():
     """test_get_ipython_dir_4, use XDG if both exist."""
     path.get_home_dir = lambda : HOME_TEST_DIR
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env['XDG_CONFIG_HOME'] = XDG_TEST_DIR
     ipdir = path.get_ipython_dir()
     if sys.platform == "darwin":
         expected = os.path.join(HOME_TEST_DIR, ".ipython")
     else:
         expected = os.path.join(XDG_TEST_DIR, "ipython")
     nt.assert_equal(ipdir, expected)
 
 @with_environment
 def test_get_ipython_dir_5():
     """test_get_ipython_dir_5, use .ipython if exists and XDG defined, but doesn't exist."""
     path.get_home_dir = lambda : HOME_TEST_DIR
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env['XDG_CONFIG_HOME'] = XDG_TEST_DIR
     os.rmdir(os.path.join(XDG_TEST_DIR, 'ipython'))
     ipdir = path.get_ipython_dir()
     nt.assert_equal(ipdir, IP_TEST_DIR)
 
 @with_environment
 def test_get_ipython_dir_6():
     """test_get_ipython_dir_6, use XDG if defined and neither exist."""
     xdg = os.path.join(HOME_TEST_DIR, 'somexdg')
     os.mkdir(xdg)
     shutil.rmtree(os.path.join(HOME_TEST_DIR, '.ipython'))
     path.get_home_dir = lambda : HOME_TEST_DIR
     path.get_xdg_dir = lambda : xdg
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     xdg_ipdir = os.path.join(xdg, "ipython")
     ipdir = path.get_ipython_dir()
     nt.assert_equal(ipdir, xdg_ipdir)
 
 @with_environment
 def test_get_ipython_dir_7():
     """test_get_ipython_dir_7, test home directory expansion on IPYTHONDIR"""
     path._writable_dir = lambda path: True
     home_dir = os.path.normpath(os.path.expanduser('~'))
     env['IPYTHONDIR'] = os.path.join('~', 'somewhere')
     ipdir = path.get_ipython_dir()
     nt.assert_equal(ipdir, os.path.join(home_dir, 'somewhere'))
 
 @skip_win32
 @with_environment
 def test_get_ipython_dir_8():
     """test_get_ipython_dir_8, test / home directory"""
     old = path._writable_dir, path.get_xdg_dir
     try:
         path._writable_dir = lambda path: bool(path)
         path.get_xdg_dir = lambda: None
         env.pop('IPYTHON_DIR', None)
         env.pop('IPYTHONDIR', None)
         env['HOME'] = '/'
         nt.assert_equal(path.get_ipython_dir(), '/.ipython')
     finally:
         path._writable_dir, path.get_xdg_dir = old
 
 @with_environment
 def test_get_xdg_dir_0():
     """test_get_xdg_dir_0, check xdg_dir"""
     reload(path)
     path._writable_dir = lambda path: True
     path.get_home_dir = lambda : 'somewhere'
     os.name = "posix"
     sys.platform = "linux2"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
 
     nt.assert_equal(path.get_xdg_dir(), os.path.join('somewhere', '.config'))
 
 
 @with_environment
 def test_get_xdg_dir_1():
     """test_get_xdg_dir_1, check nonexistant xdg_dir"""
     reload(path)
     path.get_home_dir = lambda : HOME_TEST_DIR
     os.name = "posix"
     sys.platform = "linux2"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     nt.assert_equal(path.get_xdg_dir(), None)
 
 @with_environment
 def test_get_xdg_dir_2():
     """test_get_xdg_dir_2, check xdg_dir default to ~/.config"""
     reload(path)
     path.get_home_dir = lambda : HOME_TEST_DIR
     os.name = "posix"
     sys.platform = "linux2"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     cfgdir=os.path.join(path.get_home_dir(), '.config')
     if not os.path.exists(cfgdir):
         os.makedirs(cfgdir)
 
     nt.assert_equal(path.get_xdg_dir(), cfgdir)
 
 @with_environment
 def test_get_xdg_dir_3():
     """test_get_xdg_dir_3, check xdg_dir not used on OS X"""
     reload(path)
     path.get_home_dir = lambda : HOME_TEST_DIR
     os.name = "posix"
     sys.platform = "darwin"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     cfgdir=os.path.join(path.get_home_dir(), '.config')
     if not os.path.exists(cfgdir):
         os.makedirs(cfgdir)
 
     nt.assert_equal(path.get_xdg_dir(), None)
 
 def test_filefind():
     """Various tests for filefind"""
     f = tempfile.NamedTemporaryFile()
     # print 'fname:',f.name
     alt_dirs = path.get_ipython_dir()
     t = path.filefind(f.name, alt_dirs)
     # print 'found:',t
 
 @with_environment
 def test_get_ipython_cache_dir():
     os.environ["HOME"] = HOME_TEST_DIR
     if os.name == 'posix' and sys.platform != 'darwin':
         # test default
         os.makedirs(os.path.join(HOME_TEST_DIR, ".cache"))
         os.environ.pop("XDG_CACHE_HOME", None)
         ipdir = path.get_ipython_cache_dir()
         nt.assert_equal(os.path.join(HOME_TEST_DIR, ".cache", "ipython"),
                         ipdir)
         nt.assert_true(os.path.isdir(ipdir))
 
         # test env override
         os.environ["XDG_CACHE_HOME"] = XDG_CACHE_DIR
         ipdir = path.get_ipython_cache_dir()
         nt.assert_true(os.path.isdir(ipdir))
         nt.assert_equal(ipdir, os.path.join(XDG_CACHE_DIR, "ipython"))
     else:
         nt.assert_equal(path.get_ipython_cache_dir(),
                         path.get_ipython_dir())
 
 def test_get_ipython_package_dir():
     ipdir = path.get_ipython_package_dir()
     nt.assert_true(os.path.isdir(ipdir))
 
 
 def test_get_ipython_module_path():
     ipapp_path = path.get_ipython_module_path('IPython.terminal.ipapp')
     nt.assert_true(os.path.isfile(ipapp_path))
 
 
 @dec.skip_if_not_win32
 def test_get_long_path_name_win32():
     p = path.get_long_path_name('c:\\docume~1')
     nt.assert_equal(p,u'c:\\Documents and Settings')
 
 
 @dec.skip_win32
 def test_get_long_path_name():
     p = path.get_long_path_name('/usr/local')
     nt.assert_equal(p,'/usr/local')
 
 @dec.skip_win32 # can't create not-user-writable dir on win
 @with_environment
 def test_not_writable_ipdir():
     tmpdir = tempfile.mkdtemp()
     os.name = "posix"
     env.pop('IPYTHON_DIR', None)
     env.pop('IPYTHONDIR', None)
     env.pop('XDG_CONFIG_HOME', None)
     env['HOME'] = tmpdir
     ipdir = os.path.join(tmpdir, '.ipython')
     os.mkdir(ipdir)
     os.chmod(ipdir, 600)
     with AssertPrints('is not a writable location', channel='stderr'):
         ipdir = path.get_ipython_dir()
     env.pop('IPYTHON_DIR', None)
 
 def test_unquote_filename():
     for win32 in (True, False):
         nt.assert_equal(path.unquote_filename('foo.py', win32=win32), 'foo.py')
         nt.assert_equal(path.unquote_filename('foo bar.py', win32=win32), 'foo bar.py')
     nt.assert_equal(path.unquote_filename('"foo.py"', win32=True), 'foo.py')
     nt.assert_equal(path.unquote_filename('"foo bar.py"', win32=True), 'foo bar.py')
     nt.assert_equal(path.unquote_filename("'foo.py'", win32=True), 'foo.py')
     nt.assert_equal(path.unquote_filename("'foo bar.py'", win32=True), 'foo bar.py')
     nt.assert_equal(path.unquote_filename('"foo.py"', win32=False), '"foo.py"')
     nt.assert_equal(path.unquote_filename('"foo bar.py"', win32=False), '"foo bar.py"')
     nt.assert_equal(path.unquote_filename("'foo.py'", win32=False), "'foo.py'")
     nt.assert_equal(path.unquote_filename("'foo bar.py'", win32=False), "'foo bar.py'")
 
 @with_environment
 def test_get_py_filename():
     os.chdir(TMP_TEST_DIR)
     for win32 in (True, False):
         with make_tempfile('foo.py'):
             nt.assert_equal(path.get_py_filename('foo.py', force_win32=win32), 'foo.py')
             nt.assert_equal(path.get_py_filename('foo', force_win32=win32), 'foo.py')
         with make_tempfile('foo'):
             nt.assert_equal(path.get_py_filename('foo', force_win32=win32), 'foo')
             nt.assert_raises(IOError, path.get_py_filename, 'foo.py', force_win32=win32)
         nt.assert_raises(IOError, path.get_py_filename, 'foo', force_win32=win32)
         nt.assert_raises(IOError, path.get_py_filename, 'foo.py', force_win32=win32)
         true_fn = 'foo with spaces.py'
         with make_tempfile(true_fn):
             nt.assert_equal(path.get_py_filename('foo with spaces', force_win32=win32), true_fn)
             nt.assert_equal(path.get_py_filename('foo with spaces.py', force_win32=win32), true_fn)
             if win32:
                 nt.assert_equal(path.get_py_filename('"foo with spaces.py"', force_win32=True), true_fn)
                 nt.assert_equal(path.get_py_filename("'foo with spaces.py'", force_win32=True), true_fn)
             else:
                 nt.assert_raises(IOError, path.get_py_filename, '"foo with spaces.py"', force_win32=False)
                 nt.assert_raises(IOError, path.get_py_filename, "'foo with spaces.py'", force_win32=False)
                 
 def test_unicode_in_filename():
     """When a file doesn't exist, the exception raised should be safe to call
     str() on - i.e. in Python 2 it must only have ASCII characters.
     
     https://github.com/ipython/ipython/issues/875
     """
     try:
         # these calls should not throw unicode encode exceptions
         path.get_py_filename(u'fooéè.py',  force_win32=False)
     except IOError as ex:
         str(ex)
 
 
 class TestShellGlob(object):
 
     @classmethod
     def setUpClass(cls):
         cls.filenames_start_with_a = map('a{0}'.format, range(3))
         cls.filenames_end_with_b = map('{0}b'.format, range(3))
         cls.filenames = cls.filenames_start_with_a + cls.filenames_end_with_b
         cls.tempdir = TemporaryDirectory()
         td = cls.tempdir.name
 
         with cls.in_tempdir():
             # Create empty files
             for fname in cls.filenames:
                 open(os.path.join(td, fname), 'w').close()
 
     @classmethod
     def tearDownClass(cls):
         cls.tempdir.cleanup()
 
     @classmethod
     @contextmanager
     def in_tempdir(cls):
         save = os.getcwdu()
         try:
             os.chdir(cls.tempdir.name)
             yield
         finally:
             os.chdir(save)
 
     def check_match(self, patterns, matches):
         with self.in_tempdir():
             # glob returns unordered list. that's why sorted is required.
             nt.assert_equals(sorted(path.shellglob(patterns)),
                              sorted(matches))
 
     def common_cases(self):
         return [
             (['*'], self.filenames),
             (['a*'], self.filenames_start_with_a),
             (['*c'], ['*c']),
             (['*', 'a*', '*b', '*c'], self.filenames
                                       + self.filenames_start_with_a
                                       + self.filenames_end_with_b
                                       + ['*c']),
             (['a[012]'], self.filenames_start_with_a),
         ]
 
     @skip_win32
     def test_match_posix(self):
         for (patterns, matches) in self.common_cases() + [
                 ([r'\*'], ['*']),
                 ([r'a\*', 'a*'], ['a*'] + self.filenames_start_with_a),
                 ([r'a\[012]'], ['a[012]']),
                 ]:
             yield (self.check_match, patterns, matches)
 
     @skip_if_not_win32
     def test_match_windows(self):
         for (patterns, matches) in self.common_cases() + [
                 # In windows, backslash is interpreted as path
                 # separator.  Therefore, you can't escape glob
                 # using it.
                 ([r'a\*', 'a*'], [r'a\*'] + self.filenames_start_with_a),
                 ([r'a\[012]'], [r'a\[012]']),
                 ]:
             yield (self.check_match, patterns, matches)
 
 
 def test_unescape_glob():
     nt.assert_equals(path.unescape_glob(r'\*\[\!\]\?'), '*[!]?')
     nt.assert_equals(path.unescape_glob(r'\\*'), r'\*')
     nt.assert_equals(path.unescape_glob(r'\\\*'), r'\*')
     nt.assert_equals(path.unescape_glob(r'\\a'), r'\a')
     nt.assert_equals(path.unescape_glob(r'\a'), r'\a')

IPython/utils/tests/test_text.py

@@ -1,173 +1,170 b''
 # encoding: utf-8
 """Tests for IPython.utils.text"""
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 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 math
 import random
 
 import nose.tools as nt
 
-from nose import with_setup
-
-from IPython.testing import decorators as dec
 from IPython.utils import text
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 def test_columnize():
     """Basic columnize tests."""
     size = 5
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=80)
     nt.assert_equal(out, 'aaaaa  bbbbb  ccccc\n')
     out = text.columnize(items, displaywidth=12)
     nt.assert_equal(out, 'aaaaa  ccccc\nbbbbb\n')
     out = text.columnize(items, displaywidth=10)
     nt.assert_equal(out, 'aaaaa\nbbbbb\nccccc\n')
 
 def test_columnize_random():
     """Test with random input to hopfully catch edge case """
     for nitems in [random.randint(2,70) for i in range(2,20)]:
         displaywidth = random.randint(20,200)
         rand_len = [random.randint(2,displaywidth) for i in range(nitems)]
         items = ['x'*l for l in rand_len]
         out = text.columnize(items, displaywidth=displaywidth)
         longer_line = max([len(x) for x in out.split('\n')])
         longer_element = max(rand_len)
         if longer_line > displaywidth:
             print "Columnize displayed something lager than displaywidth : %s " % longer_line
             print "longer element : %s " % longer_element
             print "displaywidth : %s " % displaywidth
             print "number of element : %s " % nitems
             print "size of each element :\n %s" % rand_len
             assert False
 
 def test_columnize_medium():
     """Test with inputs than shouldn't be wider tahn 80 """
     size = 40
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=80)
     nt.assert_equal(out, '\n'.join(items+['']))
 
 def test_columnize_long():
     """Test columnize with inputs longer than the display window"""
     size = 11
     items = [l*size for l in 'abc']
     out = text.columnize(items, displaywidth=size-1)
     nt.assert_equal(out, '\n'.join(items+['']))
 
 def eval_formatter_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os, u=u"café", b="café")
     s = f.format("{n} {n//4} {stuff.split()[0]}", **ns)
     nt.assert_equal(s, "12 3 hello")
     s = f.format(' '.join(['{n//%i}'%i for i in range(1,8)]), **ns)
     nt.assert_equal(s, "12 6 4 3 2 2 1")
     s = f.format('{[n//i for i in range(1,8)]}', **ns)
     nt.assert_equal(s, "[12, 6, 4, 3, 2, 2, 1]")
     s = f.format("{stuff!s}", **ns)
     nt.assert_equal(s, ns['stuff'])
     s = f.format("{stuff!r}", **ns)
     nt.assert_equal(s, repr(ns['stuff']))
     
     # Check with unicode:
     s = f.format("{u}", **ns)
     nt.assert_equal(s, ns['u'])
     # This decodes in a platform dependent manner, but it shouldn't error out
     s = f.format("{b}", **ns)
         
     nt.assert_raises(NameError, f.format, '{dne}', **ns)
 
 def eval_formatter_slicing_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     s = f.format(" {stuff.split()[:]} ", **ns)
     nt.assert_equal(s, " ['hello', 'there'] ")
     s = f.format(" {stuff.split()[::-1]} ", **ns)
     nt.assert_equal(s, " ['there', 'hello'] ")
     s = f.format("{stuff[::2]}", **ns)
     nt.assert_equal(s, ns['stuff'][::2])
     
     nt.assert_raises(SyntaxError, f.format, "{n:x}", **ns)
 
 def eval_formatter_no_slicing_check(f):
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     
     s = f.format('{n:x} {pi**2:+f}', **ns)
     nt.assert_equal(s, "c +9.869604")
     
     s = f.format('{stuff[slice(1,4)]}', **ns)
     nt.assert_equal(s, 'ell')
     
     nt.assert_raises(SyntaxError, f.format, "{a[:]}")
 
 def test_eval_formatter():
     f = text.EvalFormatter()
     eval_formatter_check(f)
     eval_formatter_no_slicing_check(f)
 
 def test_full_eval_formatter():
     f = text.FullEvalFormatter()
     eval_formatter_check(f)
     eval_formatter_slicing_check(f)
 
 def test_dollar_formatter():
     f = text.DollarFormatter()
     eval_formatter_check(f)
     eval_formatter_slicing_check(f)
     
     ns = dict(n=12, pi=math.pi, stuff='hello there', os=os)
     s = f.format("$n", **ns)
     nt.assert_equal(s, "12")
     s = f.format("$n.real", **ns)
     nt.assert_equal(s, "12")
     s = f.format("$n/{stuff[:5]}", **ns)
     nt.assert_equal(s, "12/hello")
     s = f.format("$n $$HOME", **ns)
     nt.assert_equal(s, "12 $HOME")
     s = f.format("${foo}", foo="HOME")
     nt.assert_equal(s, "$HOME")
 
 
 def test_long_substr():
     data = ['hi']
     nt.assert_equal(text.long_substr(data), 'hi')
 
 
 def test_long_substr2():
     data = ['abc', 'abd', 'abf', 'ab']
     nt.assert_equal(text.long_substr(data), 'ab')
 
 def test_long_substr_empty():
     data = []
     nt.assert_equal(text.long_substr(data), '')
 
 def test_strip_email():
     src = """\
         >> >>> def f(x):
         >> ...   return x+1
         >> ... 
         >> >>> zz = f(2.5)"""
     cln = """\
 >>> def f(x):
 ...   return x+1
 ... 
 >>> zz = f(2.5)"""
     nt.assert_equal(text.strip_email_quotes(src), cln)
 
 
 def test_strip_email2():
     src = '> > > list()'
     cln = 'list()'
     nt.assert_equal(text.strip_email_quotes(src), cln)

IPython/utils/tests/test_wildcard.py

@@ -1,148 +1,147 b''
 """Some tests for the wildcard utilities."""
 
 #-----------------------------------------------------------------------------
 # Library imports
 #-----------------------------------------------------------------------------
 # Stdlib
-import sys
 import unittest
 
 # Our own
 from IPython.utils import wildcard
 
 #-----------------------------------------------------------------------------
 # Globals for test
 #-----------------------------------------------------------------------------
 
 class obj_t(object):
     pass
 
 root = obj_t()
 l = ["arna","abel","ABEL","active","bob","bark","abbot"]
 q = ["kate","loop","arne","vito","lucifer","koppel"]
 for x in l:
     o = obj_t()
     setattr(root,x,o)
     for y in q:
         p = obj_t()
         setattr(o,y,p)
 root._apan = obj_t()
 root._apan.a = 10
 root._apan._a = 20
 root._apan.__a = 20
 root.__anka = obj_t()
 root.__anka.a = 10
 root.__anka._a = 20
 root.__anka.__a = 20
 
 root._APAN = obj_t()
 root._APAN.a = 10
 root._APAN._a = 20
 root._APAN.__a = 20
 root.__ANKA = obj_t()
 root.__ANKA.a = 10
 root.__ANKA._a = 20
 root.__ANKA.__a = 20
 
 #-----------------------------------------------------------------------------
 # Test cases
 #-----------------------------------------------------------------------------
 
 class Tests (unittest.TestCase):
     def test_case(self):
         ns=root.__dict__
         tests=[
          ("a*",     ["abbot","abel","active","arna",]),
          ("?b*.?o*",["abbot.koppel","abbot.loop","abel.koppel","abel.loop",]),
          ("_a*",    []),
          ("_*anka", ["__anka",]),
          ("_*a*",   ["__anka",]),
         ]
         for pat,res in tests:
             res.sort()
             a=wildcard.list_namespace(ns,"all",pat,ignore_case=False,
                                       show_all=False).keys()
             a.sort()
             self.assertEqual(a,res)
 
     def test_case_showall(self):
         ns=root.__dict__
         tests=[
          ("a*",     ["abbot","abel","active","arna",]),
          ("?b*.?o*",["abbot.koppel","abbot.loop","abel.koppel","abel.loop",]),
          ("_a*",    ["_apan"]),
          ("_*anka", ["__anka",]),
          ("_*a*",   ["__anka","_apan",]),
         ]
         for pat,res in tests:
             res.sort()
             a=wildcard.list_namespace(ns,"all",pat,ignore_case=False,
                                       show_all=True).keys()
             a.sort()
             self.assertEqual(a,res)
 
 
     def test_nocase(self):
         ns=root.__dict__
         tests=[
          ("a*",     ["abbot","abel","ABEL","active","arna",]),
          ("?b*.?o*",["abbot.koppel","abbot.loop","abel.koppel","abel.loop",
                      "ABEL.koppel","ABEL.loop",]),
          ("_a*",    []),
          ("_*anka", ["__anka","__ANKA",]),
          ("_*a*",   ["__anka","__ANKA",]),
         ]
         for pat,res in tests:
             res.sort()
             a=wildcard.list_namespace(ns,"all",pat,ignore_case=True,
                                       show_all=False).keys()
             a.sort()
             self.assertEqual(a,res)
 
     def test_nocase_showall(self):
         ns=root.__dict__
         tests=[
          ("a*",     ["abbot","abel","ABEL","active","arna",]),
          ("?b*.?o*",["abbot.koppel","abbot.loop","abel.koppel","abel.loop",
                      "ABEL.koppel","ABEL.loop",]),
          ("_a*",    ["_apan","_APAN"]),
          ("_*anka", ["__anka","__ANKA",]),
          ("_*a*",   ["__anka","__ANKA","_apan","_APAN"]),
         ]
         for pat,res in tests:
             res.sort()
             a=wildcard.list_namespace(ns,"all",pat,ignore_case=True,
                                       show_all=True).keys()
             a.sort()
             self.assertEqual(a,res)
             
     def test_dict_attributes(self):
         """Dictionaries should be indexed by attributes, not by keys. This was
         causing Github issue 129."""
         ns = {"az":{"king":55}, "pq":{1:0}}
         tests = [
           ("a*", ["az"]),
           ("az.k*", ["az.keys"]),
           ("pq.k*", ["pq.keys"])
         ]
         for pat, res in tests:
             res.sort()
             a = wildcard.list_namespace(ns, "all", pat, ignore_case=False,
                                         show_all=True).keys()
             a.sort()
             self.assertEqual(a, res)
     
     def test_dict_dir(self):
         class A(object):
             def __init__(self):
                 self.a = 1
                 self.b = 2
             def __getattribute__(self, name):
                 if name=="a":
                     raise AttributeError
                 return object.__getattribute__(self, name)
                 
         a = A()
         adict = wildcard.dict_dir(a)
         assert "a" not in adict # change to assertNotIn method in >= 2.7
         self.assertEqual(adict["b"], 2)

IPython/utils/text.py

@@ -1,717 +1,713 b''
 # encoding: utf-8
 """
 Utilities for working with strings and text.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.utils.text
    :parts: 3
 """
 
 #-----------------------------------------------------------------------------
 #  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 __main__
-
 import os
 import re
-import sys
 import textwrap
 from string import Formatter
 
 from IPython.external.path import path
 from IPython.testing.skipdoctest import skip_doctest_py3, skip_doctest
 from IPython.utils import py3compat
-from IPython.utils.data import flatten
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 class LSString(str):
     """String derivative with a special access attributes.
 
     These are normal strings, but with the special attributes:
 
         .l (or .list) : value as list (split on newlines).
         .n (or .nlstr): original value (the string itself).
         .s (or .spstr): value as whitespace-separated string.
         .p (or .paths): list of path objects
 
     Any values which require transformations are computed only once and
     cached.
 
     Such strings are very useful to efficiently interact with the shell, which
     typically only understands whitespace-separated options for commands."""
 
     def get_list(self):
         try:
             return self.__list
         except AttributeError:
             self.__list = self.split('\n')
             return self.__list
 
     l = list = property(get_list)
 
     def get_spstr(self):
         try:
             return self.__spstr
         except AttributeError:
             self.__spstr = self.replace('\n',' ')
             return self.__spstr
 
     s = spstr = property(get_spstr)
 
     def get_nlstr(self):
         return self
 
     n = nlstr = property(get_nlstr)
 
     def get_paths(self):
         try:
             return self.__paths
         except AttributeError:
             self.__paths = [path(p) for p in self.split('\n') if os.path.exists(p)]
             return self.__paths
 
     p = paths = property(get_paths)
 
 # FIXME: We need to reimplement type specific displayhook and then add this
 # back as a custom printer. This should also be moved outside utils into the
 # core.
 
 # def print_lsstring(arg):
 #     """ Prettier (non-repr-like) and more informative printer for LSString """
 #     print "LSString (.p, .n, .l, .s available). Value:"
 #     print arg
 #
 #
 # print_lsstring = result_display.when_type(LSString)(print_lsstring)
 
 
 class SList(list):
     """List derivative with a special access attributes.
 
     These are normal lists, but with the special attributes:
 
         .l (or .list) : value as list (the list itself).
         .n (or .nlstr): value as a string, joined on newlines.
         .s (or .spstr): value as a string, joined on spaces.
         .p (or .paths): list of path objects
 
     Any values which require transformations are computed only once and
     cached."""
 
     def get_list(self):
         return self
 
     l = list = property(get_list)
 
     def get_spstr(self):
         try:
             return self.__spstr
         except AttributeError:
             self.__spstr = ' '.join(self)
             return self.__spstr
 
     s = spstr = property(get_spstr)
 
     def get_nlstr(self):
         try:
             return self.__nlstr
         except AttributeError:
             self.__nlstr = '\n'.join(self)
             return self.__nlstr
 
     n = nlstr = property(get_nlstr)
 
     def get_paths(self):
         try:
             return self.__paths
         except AttributeError:
             self.__paths = [path(p) for p in self if os.path.exists(p)]
             return self.__paths
 
     p = paths = property(get_paths)
 
     def grep(self, pattern, prune = False, field = None):
         """ Return all strings matching 'pattern' (a regex or callable)
 
         This is case-insensitive. If prune is true, return all items
         NOT matching the pattern.
 
         If field is specified, the match must occur in the specified
         whitespace-separated field.
 
         Examples::
 
             a.grep( lambda x: x.startswith('C') )
             a.grep('Cha.*log', prune=1)
             a.grep('chm', field=-1)
         """
 
         def match_target(s):
             if field is None:
                 return s
             parts = s.split()
             try:
                 tgt = parts[field]
                 return tgt
             except IndexError:
                 return ""
 
         if isinstance(pattern, basestring):
             pred = lambda x : re.search(pattern, x, re.IGNORECASE)
         else:
             pred = pattern
         if not prune:
             return SList([el for el in self if pred(match_target(el))])
         else:
             return SList([el for el in self if not pred(match_target(el))])
 
     def fields(self, *fields):
         """ Collect whitespace-separated fields from string list
 
         Allows quick awk-like usage of string lists.
 
         Example data (in var a, created by 'a = !ls -l')::
             -rwxrwxrwx  1 ville None      18 Dec 14  2006 ChangeLog
             drwxrwxrwx+ 6 ville None       0 Oct 24 18:05 IPython
 
         a.fields(0) is ['-rwxrwxrwx', 'drwxrwxrwx+']
         a.fields(1,0) is ['1 -rwxrwxrwx', '6 drwxrwxrwx+']
         (note the joining by space).
         a.fields(-1) is ['ChangeLog', 'IPython']
 
         IndexErrors are ignored.
 
         Without args, fields() just split()'s the strings.
         """
         if len(fields) == 0:
             return [el.split() for el in self]
 
         res = SList()
         for el in [f.split() for f in self]:
             lineparts = []
 
             for fd in fields:
                 try:
                     lineparts.append(el[fd])
                 except IndexError:
                     pass
             if lineparts:
                 res.append(" ".join(lineparts))
 
         return res
 
     def sort(self,field= None,  nums = False):
         """ sort by specified fields (see fields())
 
         Example::
             a.sort(1, nums = True)
 
         Sorts a by second field, in numerical order (so that 21 > 3)
 
         """
 
         #decorate, sort, undecorate
         if field is not None:
             dsu = [[SList([line]).fields(field),  line] for line in self]
         else:
             dsu = [[line,  line] for line in self]
         if nums:
             for i in range(len(dsu)):
                 numstr = "".join([ch for ch in dsu[i][0] if ch.isdigit()])
                 try:
                     n = int(numstr)
                 except ValueError:
                     n = 0;
                 dsu[i][0] = n
 
 
         dsu.sort()
         return SList([t[1] for t in dsu])
 
 
 # FIXME: We need to reimplement type specific displayhook and then add this
 # back as a custom printer. This should also be moved outside utils into the
 # core.
 
 # def print_slist(arg):
 #     """ Prettier (non-repr-like) and more informative printer for SList """
 #     print "SList (.p, .n, .l, .s, .grep(), .fields(), sort() available):"
 #     if hasattr(arg,  'hideonce') and arg.hideonce:
 #         arg.hideonce = False
 #         return
 #
 #     nlprint(arg)   # This was a nested list printer, now removed.
 #
 # print_slist = result_display.when_type(SList)(print_slist)
 
 
 def indent(instr,nspaces=4, ntabs=0, flatten=False):
     """Indent a string a given number of spaces or tabstops.
 
     indent(str,nspaces=4,ntabs=0) -> indent str by ntabs+nspaces.
 
     Parameters
     ----------
 
     instr : basestring
         The string to be indented.
     nspaces : int (default: 4)
         The number of spaces to be indented.
     ntabs : int (default: 0)
         The number of tabs to be indented.
     flatten : bool (default: False)
         Whether to scrub existing indentation.  If True, all lines will be
         aligned to the same indentation.  If False, existing indentation will
         be strictly increased.
 
     Returns
     -------
 
     str|unicode : string indented by ntabs and nspaces.
 
     """
     if instr is None:
         return
     ind = '\t'*ntabs+' '*nspaces
     if flatten:
         pat = re.compile(r'^\s*', re.MULTILINE)
     else:
         pat = re.compile(r'^', re.MULTILINE)
     outstr = re.sub(pat, ind, instr)
     if outstr.endswith(os.linesep+ind):
         return outstr[:-len(ind)]
     else:
         return outstr
 
 
 def list_strings(arg):
     """Always return a list of strings, given a string or list of strings
     as input.
 
     :Examples:
 
         In [7]: list_strings('A single string')
         Out[7]: ['A single string']
 
         In [8]: list_strings(['A single string in a list'])
         Out[8]: ['A single string in a list']
 
         In [9]: list_strings(['A','list','of','strings'])
         Out[9]: ['A', 'list', 'of', 'strings']
     """
 
     if isinstance(arg,basestring): return [arg]
     else: return arg
 
 
 def marquee(txt='',width=78,mark='*'):
     """Return the input string centered in a 'marquee'.
 
     :Examples:
 
         In [16]: marquee('A test',40)
         Out[16]: '**************** A test ****************'
 
         In [17]: marquee('A test',40,'-')
         Out[17]: '---------------- A test ----------------'
 
         In [18]: marquee('A test',40,' ')
         Out[18]: '                 A test                 '
 
     """
     if not txt:
         return (mark*width)[:width]
     nmark = (width-len(txt)-2)//len(mark)//2
     if nmark < 0: nmark =0
     marks = mark*nmark
     return '%s %s %s' % (marks,txt,marks)
 
 
 ini_spaces_re = re.compile(r'^(\s+)')
 
 def num_ini_spaces(strng):
     """Return the number of initial spaces in a string"""
 
     ini_spaces = ini_spaces_re.match(strng)
     if ini_spaces:
         return ini_spaces.end()
     else:
         return 0
 
 
 def format_screen(strng):
     """Format a string for screen printing.
 
     This removes some latex-type format codes."""
     # Paragraph continue
     par_re = re.compile(r'\\$',re.MULTILINE)
     strng = par_re.sub('',strng)
     return strng
 
 
 def dedent(text):
     """Equivalent of textwrap.dedent that ignores unindented first line.
 
     This means it will still dedent strings like:
     '''foo
     is a bar
     '''
 
     For use in wrap_paragraphs.
     """
 
     if text.startswith('\n'):
         # text starts with blank line, don't ignore the first line
         return textwrap.dedent(text)
 
     # split first line
     splits = text.split('\n',1)
     if len(splits) == 1:
         # only one line
         return textwrap.dedent(text)
 
     first, rest = splits
     # dedent everything but the first line
     rest = textwrap.dedent(rest)
     return '\n'.join([first, rest])
 
 
 def wrap_paragraphs(text, ncols=80):
     """Wrap multiple paragraphs to fit a specified width.
 
     This is equivalent to textwrap.wrap, but with support for multiple
     paragraphs, as separated by empty lines.
 
     Returns
     -------
 
     list of complete paragraphs, wrapped to fill `ncols` columns.
     """
     paragraph_re = re.compile(r'\n(\s*\n)+', re.MULTILINE)
     text = dedent(text).strip()
     paragraphs = paragraph_re.split(text)[::2] # every other entry is space
     out_ps = []
     indent_re = re.compile(r'\n\s+', re.MULTILINE)
     for p in paragraphs:
         # presume indentation that survives dedent is meaningful formatting,
         # so don't fill unless text is flush.
         if indent_re.search(p) is None:
             # wrap paragraph
             p = textwrap.fill(p, ncols)
         out_ps.append(p)
     return out_ps
 
 
 def long_substr(data):
     """Return the longest common substring in a list of strings.
     
     Credit: http://stackoverflow.com/questions/2892931/longest-common-substring-from-more-than-two-strings-python
     """
     substr = ''
     if len(data) > 1 and len(data[0]) > 0:
         for i in range(len(data[0])):
             for j in range(len(data[0])-i+1):
                 if j > len(substr) and all(data[0][i:i+j] in x for x in data):
                     substr = data[0][i:i+j]
     elif len(data) == 1:
         substr = data[0]
     return substr
 
 
 def strip_email_quotes(text):
     """Strip leading email quotation characters ('>').
 
     Removes any combination of leading '>' interspersed with whitespace that
     appears *identically* in all lines of the input text.
 
     Parameters
     ----------
     text : str
 
     Examples
     --------
 
     Simple uses::
 
         In [2]: strip_email_quotes('> > text')
         Out[2]: 'text'
 
         In [3]: strip_email_quotes('> > text\\n> > more')
         Out[3]: 'text\\nmore'
 
     Note how only the common prefix that appears in all lines is stripped::
 
         In [4]: strip_email_quotes('> > text\\n> > more\\n> more...')
         Out[4]: '> text\\n> more\\nmore...'
 
     So if any line has no quote marks ('>') , then none are stripped from any
     of them ::
     
         In [5]: strip_email_quotes('> > text\\n> > more\\nlast different')
         Out[5]: '> > text\\n> > more\\nlast different'
     """
     lines = text.splitlines()
     matches = set()
     for line in lines:
         prefix = re.match(r'^(\s*>[ >]*)', line)
         if prefix:
             matches.add(prefix.group(1))
         else:
             break
     else:
         prefix = long_substr(list(matches))
         if prefix:
             strip = len(prefix)
             text = '\n'.join([ ln[strip:] for ln in lines])
     return text
 
 
 class EvalFormatter(Formatter):
     """A String Formatter that allows evaluation of simple expressions.
     
     Note that this version interprets a : as specifying a format string (as per
     standard string formatting), so if slicing is required, you must explicitly
     create a slice.
     
     This is to be used in templating cases, such as the parallel batch
     script templates, where simple arithmetic on arguments is useful.
 
     Examples
     --------
     
     In  [1]: f = EvalFormatter()
     In  [2]: f.format('{n//4}', n=8)
     Out [2]: '2'
     
     In  [3]: f.format("{greeting[slice(2,4)]}", greeting="Hello")
     Out [3]: 'll'
     """
     def get_field(self, name, args, kwargs):
         v = eval(name, kwargs)
         return v, name
 
 
 @skip_doctest_py3
 class FullEvalFormatter(Formatter):
     """A String Formatter that allows evaluation of simple expressions.
     
     Any time a format key is not found in the kwargs,
     it will be tried as an expression in the kwargs namespace.
     
     Note that this version allows slicing using [1:2], so you cannot specify
     a format string. Use :class:`EvalFormatter` to permit format strings.
     
     Examples
     --------
     
     In [1]: f = FullEvalFormatter()
     In [2]: f.format('{n//4}', n=8)
     Out[2]: u'2'
     
     In [3]: f.format('{list(range(5))[2:4]}')
     Out[3]: u'[2, 3]'
 
     In [4]: f.format('{3*2}')
     Out[4]: u'6'
     """
     # copied from Formatter._vformat with minor changes to allow eval
     # and replace the format_spec code with slicing
     def _vformat(self, format_string, args, kwargs, used_args, recursion_depth):
         if recursion_depth < 0:
             raise ValueError('Max string recursion exceeded')
         result = []
         for literal_text, field_name, format_spec, conversion in \
                 self.parse(format_string):
 
             # output the literal text
             if literal_text:
                 result.append(literal_text)
 
             # if there's a field, output it
             if field_name is not None:
                 # this is some markup, find the object and do
                 # the formatting
 
                 if format_spec:
                     # override format spec, to allow slicing:
                     field_name = ':'.join([field_name, format_spec])
 
                 # eval the contents of the field for the object
                 # to be formatted
                 obj = eval(field_name, kwargs)
 
                 # do any conversion on the resulting object
                 obj = self.convert_field(obj, conversion)
 
                 # format the object and append to the result
                 result.append(self.format_field(obj, ''))
 
         return u''.join(py3compat.cast_unicode(s) for s in result)
 
 
 @skip_doctest_py3
 class DollarFormatter(FullEvalFormatter):
     """Formatter allowing Itpl style $foo replacement, for names and attribute
     access only. Standard {foo} replacement also works, and allows full
     evaluation of its arguments.
 
     Examples
     --------
     In [1]: f = DollarFormatter()
     In [2]: f.format('{n//4}', n=8)
     Out[2]: u'2'
     
     In [3]: f.format('23 * 76 is $result', result=23*76)
     Out[3]: u'23 * 76 is 1748'
     
     In [4]: f.format('$a or {b}', a=1, b=2)
     Out[4]: u'1 or 2'
     """
     _dollar_pattern = re.compile("(.*?)\$(\$?[\w\.]+)")
     def parse(self, fmt_string):
         for literal_txt, field_name, format_spec, conversion \
                     in Formatter.parse(self, fmt_string):
             
             # Find $foo patterns in the literal text.
             continue_from = 0
             txt = ""
             for m in self._dollar_pattern.finditer(literal_txt):
                 new_txt, new_field = m.group(1,2)
                 # $$foo --> $foo
                 if new_field.startswith("$"):
                     txt += new_txt + new_field
                 else:
                     yield (txt + new_txt, new_field, "", None)
                     txt = ""
                 continue_from = m.end()
             
             # Re-yield the {foo} style pattern
             yield (txt + literal_txt[continue_from:], field_name, format_spec, conversion)
 
 #-----------------------------------------------------------------------------
 # Utils to columnize a list of string
 #-----------------------------------------------------------------------------
 
 def _chunks(l, n):
     """Yield successive n-sized chunks from l."""
     for i in xrange(0, len(l), n):
         yield l[i:i+n]
 
 
 def _find_optimal(rlist , separator_size=2 , displaywidth=80):
     """Calculate optimal info to columnize a list of string"""
     for nrow in range(1, len(rlist)+1) :
         chk = map(max,_chunks(rlist, nrow))
         sumlength = sum(chk)
         ncols = len(chk)
         if sumlength+separator_size*(ncols-1) <= displaywidth :
             break;
     return {'columns_numbers' : ncols,
             'optimal_separator_width':(displaywidth - sumlength)/(ncols-1) if (ncols -1) else 0,
             'rows_numbers' : nrow,
             'columns_width' : chk
            }
 
 
 def _get_or_default(mylist, i, default=None):
     """return list item number, or default if don't exist"""
     if i >= len(mylist):
         return default
     else :
         return mylist[i]
 
 
 @skip_doctest
 def compute_item_matrix(items, empty=None, *args, **kwargs) :
     """Returns a nested list, and info to columnize items
 
     Parameters
     ----------
 
     items :
         list of strings to columize
     empty : (default None)
         default value to fill list if needed
     separator_size : int (default=2)
         How much caracters will be used as a separation between each columns.
     displaywidth : int (default=80)
         The width of the area onto wich the columns should enter
 
     Returns
     -------
 
     Returns a tuple of (strings_matrix, dict_info)
 
     strings_matrix :
 
         nested list of string, the outer most list contains as many list as
         rows, the innermost lists have each as many element as colums. If the
         total number of elements in `items` does not equal the product of
         rows*columns, the last element of some lists are filled with `None`.
 
     dict_info :
         some info to make columnize easier:
 
         columns_numbers : number of columns
         rows_numbers    : number of rows
         columns_width   : list of with of each columns
         optimal_separator_width : best separator width between columns
 
     Examples
     --------
 
     In [1]: l = ['aaa','b','cc','d','eeeee','f','g','h','i','j','k','l']
        ...: compute_item_matrix(l,displaywidth=12)
     Out[1]:
         ([['aaa', 'f', 'k'],
         ['b', 'g', 'l'],
         ['cc', 'h', None],
         ['d', 'i', None],
         ['eeeee', 'j', None]],
         {'columns_numbers': 3,
         'columns_width': [5, 1, 1],
         'optimal_separator_width': 2,
         'rows_numbers': 5})
 
     """
     info = _find_optimal(map(len, items), *args, **kwargs)
     nrow, ncol = info['rows_numbers'], info['columns_numbers']
     return ([[ _get_or_default(items, c*nrow+i, default=empty) for c in range(ncol) ] for i in range(nrow) ], info)
 
 
 def columnize(items, separator='  ', displaywidth=80):
     """ Transform a list of strings into a single string with columns.
 
     Parameters
     ----------
     items : sequence of strings
         The strings to process.
 
     separator : str, optional [default is two spaces]
         The string that separates columns.
 
     displaywidth : int, optional [default is 80]
         Width of the display in number of characters.
 
     Returns
     -------
     The formatted string.
     """
     if not items :
         return '\n'
     matrix, info = compute_item_matrix(items, separator_size=len(separator), displaywidth=displaywidth)
     fmatrix = [filter(None, x) for x in matrix]
     sjoin = lambda x : separator.join([ y.ljust(w, ' ') for y, w in zip(x, info['columns_width'])])
     return '\n'.join(map(sjoin, fmatrix))+'\n'

IPython/utils/zmqrelated.py

@@ -1,47 +1,46 b''
 """utilities for checking zmq versions"""
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING.txt, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Verify zmq version dependency >= 2.1.11
 #-----------------------------------------------------------------------------
 
-import warnings
 from IPython.utils.version import check_version
 
 
 def patch_pyzmq():
     """backport a few patches from newer pyzmq
     
     These can be removed as we bump our minimum pyzmq version
     """
     
     import zmq
     
     # fallback on stdlib json if jsonlib is selected, because jsonlib breaks things.
     # jsonlib support is removed from pyzmq >= 2.2.0
 
     from zmq.utils import jsonapi
     if jsonapi.jsonmod.__name__ == 'jsonlib':
         import json
         jsonapi.jsonmod = json
 
 
 def check_for_zmq(minimum_version, required_by='Someone'):
     try:
         import zmq
     except ImportError:
         raise ImportError("%s requires pyzmq >= %s"%(required_by, minimum_version))
     
     patch_pyzmq()
     
     pyzmq_version = zmq.__version__
     
     if not check_version(pyzmq_version, minimum_version):
         raise ImportError("%s requires pyzmq >= %s, but you have %s"%(
                         required_by, minimum_version, pyzmq_version))