Merge pull request #12893 from Carreau/doc-reformat

reformat docstring in IPython utils

IPython/utils/_process_common.py

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

IPython/utils/_process_posix.py

@@ -1,225 +1,225 b''
 """Posix-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Stdlib
 import errno
 import os
 import subprocess as sp
 import sys
 
 import pexpect
 
 # Our own
 from ._process_common import getoutput, arg_split
 from IPython.utils import py3compat
 from IPython.utils.encoding import DEFAULT_ENCODING
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 def _find_cmd(cmd):
     """Find the full path to a command using which."""
 
     path = sp.Popen(['/usr/bin/env', 'which', cmd],
                     stdout=sp.PIPE, stderr=sp.PIPE).communicate()[0]
     return py3compat.decode(path)
 
 
 class ProcessHandler(object):
     """Execute subprocesses under the control of pexpect.
     """
     # Timeout in seconds to wait on each reading of the subprocess' output.
     # This should not be set too low to avoid cpu overusage from our side,
     # since we read in a loop whose period is controlled by this timeout.
     read_timeout = 0.05
 
     # Timeout to give a process if we receive SIGINT, between sending the
     # SIGINT to the process and forcefully terminating it.
     terminate_timeout = 0.2
 
     # File object where stdout and stderr of the subprocess will be written
     logfile = None
 
     # Shell to call for subprocesses to execute
     _sh = None
 
     @property
     def sh(self):
         if self._sh is None:
             shell_name = os.environ.get("SHELL", "sh")
             self._sh = pexpect.which(shell_name)
             if self._sh is None:
                 raise OSError('"{}" shell not found'.format(shell_name))
 
         return self._sh
 
     def __init__(self, logfile=None, read_timeout=None, terminate_timeout=None):
         """Arguments are used for pexpect calls."""
         self.read_timeout = (ProcessHandler.read_timeout if read_timeout is
                              None else read_timeout)
         self.terminate_timeout = (ProcessHandler.terminate_timeout if
                                   terminate_timeout is None else
                                   terminate_timeout)
         self.logfile = sys.stdout if logfile is None else logfile
 
     def getoutput(self, cmd):
         """Run a command and return its stdout/stderr as a string.
 
         Parameters
         ----------
         cmd : str
-          A command to be executed in the system shell.
+            A command to be executed in the system shell.
 
         Returns
         -------
         output : str
-          A string containing the combination of stdout and stderr from the
+            A string containing the combination of stdout and stderr from the
         subprocess, in whatever order the subprocess originally wrote to its
         file descriptors (so the order of the information in this string is the
         correct order as would be seen if running the command in a terminal).
         """
         try:
             return pexpect.run(self.sh, args=['-c', cmd]).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def getoutput_pexpect(self, cmd):
         """Run a command and return its stdout/stderr as a string.
 
         Parameters
         ----------
         cmd : str
-          A command to be executed in the system shell.
+            A command to be executed in the system shell.
 
         Returns
         -------
         output : str
-          A string containing the combination of stdout and stderr from the
+            A string containing the combination of stdout and stderr from the
         subprocess, in whatever order the subprocess originally wrote to its
         file descriptors (so the order of the information in this string is the
         correct order as would be seen if running the command in a terminal).
         """
         try:
             return pexpect.run(self.sh, args=['-c', cmd]).replace('\r\n', '\n')
         except KeyboardInterrupt:
             print('^C', file=sys.stderr, end='')
 
     def system(self, cmd):
         """Execute a command in a subshell.
 
         Parameters
         ----------
         cmd : str
-          A command to be executed in the system shell.
+            A command to be executed in the system shell.
 
         Returns
         -------
         int : child's exitstatus
         """
         # Get likely encoding for the output.
         enc = DEFAULT_ENCODING
         
         # Patterns to match on the output, for pexpect.  We read input and
         # allow either a short timeout or EOF
         patterns = [pexpect.TIMEOUT, pexpect.EOF]
         # the index of the EOF pattern in the list.
         # even though we know it's 1, this call means we don't have to worry if
         # we change the above list, and forget to change this value:
         EOF_index = patterns.index(pexpect.EOF)
         # The size of the output stored so far in the process output buffer.
         # Since pexpect only appends to this buffer, each time we print we
         # record how far we've printed, so that next time we only print *new*
         # content from the buffer.
         out_size = 0
         try:
             # Since we're not really searching the buffer for text patterns, we
             # can set pexpect's search window to be tiny and it won't matter.
             # We only search for the 'patterns' timeout or EOF, which aren't in
             # the text itself.
             #child = pexpect.spawn(pcmd, searchwindowsize=1)
             if hasattr(pexpect, 'spawnb'):
                 child = pexpect.spawnb(self.sh, args=['-c', cmd]) # Pexpect-U
             else:
                 child = pexpect.spawn(self.sh, args=['-c', cmd])  # Vanilla Pexpect
             flush = sys.stdout.flush
             while True:
                 # res is the index of the pattern that caused the match, so we
                 # know whether we've finished (if we matched EOF) or not
                 res_idx = child.expect_list(patterns, self.read_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 flush()
                 if res_idx==EOF_index:
                     break
                 # Update the pointer to what we've already printed
                 out_size = len(child.before)
         except KeyboardInterrupt:
             # We need to send ^C to the process.  The ascii code for '^C' is 3
             # (the character is known as ETX for 'End of Text', see
             # curses.ascii.ETX).
             child.sendline(chr(3))
             # Read and print any more output the program might produce on its
             # way out.
             try:
                 out_size = len(child.before)
                 child.expect_list(patterns, self.terminate_timeout)
                 print(child.before[out_size:].decode(enc, 'replace'), end='')
                 sys.stdout.flush()
             except KeyboardInterrupt:
                 # Impatient users tend to type it multiple times
                 pass
             finally:
                 # Ensure the subprocess really is terminated
                 child.terminate(force=True)
         # add isalive check, to ensure exitstatus is set:
         child.isalive()
 
         # We follow the subprocess pattern, returning either the exit status
         # as a positive number, or the terminating signal as a negative
         # number.
         # on Linux, sh returns 128+n for signals terminating child processes on Linux
         # on BSD (OS X), the signal code is set instead
         if child.exitstatus is None:
             # on WIFSIGNALED, pexpect sets signalstatus, leaving exitstatus=None
             if child.signalstatus is None:
                 # this condition may never occur,
                 # but let's be certain we always return an integer.
                 return 0
             return -child.signalstatus
         if child.exitstatus > 128:
             return -(child.exitstatus - 128)
         return child.exitstatus
 
 
 # Make system() with a functional interface for outside use.  Note that we use
 # getoutput() from the _common utils, which is built on top of popen(). Using
 # pexpect to get subprocess output produces difficult to parse output, since
 # programs think they are talking to a tty and produce highly formatted output
 # (ls is a good example) that makes them hard.
 system = ProcessHandler().system
 
 def check_pid(pid):
     try:
         os.kill(pid, 0)
     except OSError as err:
         if err.errno == errno.ESRCH:
             return False
         elif err.errno == errno.EPERM:
             # Don't have permission to signal the process - probably means it exists
             return True
         raise
     else:
         return True

IPython/utils/_process_win32.py

@@ -1,205 +1,205 b''
 """Windows-specific implementation of process utilities.
 
 This file is only meant to be imported by process.py, not by end-users.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2010-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # stdlib
 import os
 import sys
 import ctypes
 import time
 
 from ctypes import c_int, POINTER
 from ctypes.wintypes import LPCWSTR, HLOCAL
 from subprocess import STDOUT, TimeoutExpired
 from threading import Thread
 
 # our own imports
 from ._process_common import read_no_interrupt, process_handler, arg_split as py_arg_split
 from . import py3compat
 from .encoding import DEFAULT_ENCODING
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 class AvoidUNCPath(object):
     """A context manager to protect command execution from UNC paths.
 
     In the Win32 API, commands can't be invoked with the cwd being a UNC path.
     This context manager temporarily changes directory to the 'C:' drive on
     entering, and restores the original working directory on exit.
 
     The context manager returns the starting working directory *if* it made a
     change and None otherwise, so that users can apply the necessary adjustment
     to their system calls in the event of a change.
 
     Examples
     --------
     ::
         cmd = 'dir'
         with AvoidUNCPath() as path:
             if path is not None:
                 cmd = '"pushd %s &&"%s' % (path, cmd)
             os.system(cmd)
     """
     def __enter__(self):
         self.path = os.getcwd()
         self.is_unc_path = self.path.startswith(r"\\")
         if self.is_unc_path:
             # change to c drive (as cmd.exe cannot handle UNC addresses)
             os.chdir("C:")
             return self.path
         else:
             # We return None to signal that there was no change in the working
             # directory
             return None
 
     def __exit__(self, exc_type, exc_value, traceback):
         if self.is_unc_path:
             os.chdir(self.path)
 
 
 def _find_cmd(cmd):
     """Find the full path to a .bat or .exe using the win32api module."""
     try:
         from win32api import SearchPath
     except ImportError as e:
         raise ImportError('you need to have pywin32 installed for this to work') from e
     else:
         PATH = os.environ['PATH']
         extensions = ['.exe', '.com', '.bat', '.py']
         path = None
         for ext in extensions:
             try:
                 path = SearchPath(PATH, cmd, ext)[0]
             except:
                 pass
         if path is None:
             raise OSError("command %r not found" % cmd)
         else:
             return path
 
 
 def _system_body(p):
     """Callback for _system."""
     enc = DEFAULT_ENCODING
 
     def stdout_read():
         for line in read_no_interrupt(p.stdout).splitlines():
             line = line.decode(enc, 'replace')
             print(line, file=sys.stdout)
 
     def stderr_read():
         for line in read_no_interrupt(p.stderr).splitlines():
             line = line.decode(enc, 'replace')
             print(line, file=sys.stderr)
 
     Thread(target=stdout_read).start()
     Thread(target=stderr_read).start()
 
     # Wait to finish for returncode. Unfortunately, Python has a bug where
     # wait() isn't interruptible (https://bugs.python.org/issue28168) so poll in
     # a loop instead of just doing `return p.wait()`.
     while True:
         result = p.poll()
         if result is None:
             time.sleep(0.01)
         else:
             return result
 
 
 def system(cmd):
     """Win32 version of os.system() that works with network shares.
 
     Note that this implementation returns None, as meant for use in IPython.
 
     Parameters
     ----------
     cmd : str or list
-      A command to be executed in the system shell.
+        A command to be executed in the system shell.
 
     Returns
     -------
     int : child process' exit code.
     """
     # The controller provides interactivity with both
     # stdin and stdout
     #import _process_win32_controller
     #_process_win32_controller.system(cmd)
 
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         return process_handler(cmd, _system_body)
 
 def getoutput(cmd):
     """Return standard output of executing cmd in a shell.
 
     Accepts the same arguments as os.system().
 
     Parameters
     ----------
     cmd : str or list
-      A command to be executed in the system shell.
+        A command to be executed in the system shell.
 
     Returns
     -------
     stdout : str
     """
 
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         out = process_handler(cmd, lambda p: p.communicate()[0], STDOUT)
 
     if out is None:
         out = b''
     return py3compat.decode(out)
 
 try:
     CommandLineToArgvW = ctypes.windll.shell32.CommandLineToArgvW
     CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
     CommandLineToArgvW.restype = POINTER(LPCWSTR)
     LocalFree = ctypes.windll.kernel32.LocalFree
     LocalFree.res_type = HLOCAL
     LocalFree.arg_types = [HLOCAL]
     
     def arg_split(commandline, posix=False, strict=True):
         """Split a command line's arguments in a shell-like manner.
 
         This is a special version for windows that use a ctypes call to CommandLineToArgvW
         to do the argv splitting. The posix parameter is ignored.
-        
+
         If strict=False, process_common.arg_split(...strict=False) is used instead.
         """
         #CommandLineToArgvW returns path to executable if called with empty string.
         if commandline.strip() == "":
             return []
         if not strict:
             # not really a cl-arg, fallback on _process_common
             return py_arg_split(commandline, posix=posix, strict=strict)
         argvn = c_int()
         result_pointer = CommandLineToArgvW(py3compat.cast_unicode(commandline.lstrip()), ctypes.byref(argvn))
         result_array_type = LPCWSTR * argvn.value
         result = [arg for arg in result_array_type.from_address(ctypes.addressof(result_pointer.contents))]
         retval = LocalFree(result_pointer)
         return result
 except AttributeError:
     arg_split = py_arg_split
 
 def check_pid(pid):
     # OpenProcess returns 0 if no such process (of ours) exists
     # positive int otherwise
     return bool(ctypes.windll.kernel32.OpenProcess(1,0,pid))

IPython/utils/_process_win32_controller.py

@@ -1,573 +1,573 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.
 #-----------------------------------------------------------------------------
 
 
 # stdlib
 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.
 
     Examples
     --------
     ::
         cmd = 'dir'
         with AvoidUNCPath() as path:
             if path is not None:
                 cmd = '"pushd %s &&"%s' % (path, cmd)
             os.system(cmd)
     """
     def __enter__(self):
         self.path = os.getcwd()
         self.is_unc_path = self.path.startswith(r"\\")
         if self.is_unc_path:
             # change to c drive (as cmd.exe cannot handle UNC addresses)
             os.chdir("C:")
             return self.path
         else:
             # We return None to signal that there was no change in the working
             # directory
             return None
 
     def __exit__(self, exc_type, exc_value, traceback):
         if self.is_unc_path:
             os.chdir(self.path)
 
 
 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 is not 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 is None and stdin_func is None and stderr_func is None:
             return self._run_stdio()
 
         if stderr_func is not 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 is 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 is not 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.
+        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.
+    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/decorators.py

@@ -1,58 +1,58 b''
 # encoding: utf-8
 """Decorators that don't go anywhere else.
 
 This module contains misc. decorators that don't really go with another module
 in :mod:`IPython.utils`. Beore putting something here please see if it should
 go into another topical module in :mod:`IPython.utils`.
 """
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def flag_calls(func):
     """Wrap a function to detect and flag when it gets called.
 
     This is a decorator which takes a function and wraps it in a function with
     a 'called' attribute. wrapper.called is initialized to False.
 
     The wrapper.called attribute is set to False right before each call to the
     wrapped function, so if the call fails it remains False.  After the call
     completes, wrapper.called is set to True and the output is returned.
 
     Testing for truth in wrapper.called allows you to determine if a call to
     func() was attempted and succeeded."""
     
     # don't wrap twice
     if hasattr(func, 'called'):
         return func
 
     def wrapper(*args,**kw):
         wrapper.called = False
         out = func(*args,**kw)
         wrapper.called = True
         return out
 
     wrapper.called = False
     wrapper.__doc__ = func.__doc__
     return wrapper
 
 def undoc(func):
     """Mark a function or class as undocumented.
-    
+
     This is found by inspecting the AST, so for now it must be used directly
     as @undoc, not as e.g. @decorators.undoc
     """
     return func
 

IPython/utils/encoding.py

@@ -1,71 +1,71 b''
 # coding: utf-8
 """
 Utilities for dealing with text encodings
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2012  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 import sys
 import locale
 import warnings
 
 # to deal with the possibility of sys.std* not being a stream at all
 def get_stream_enc(stream, default=None):
     """Return the given stream's encoding or a default.
 
     There are cases where ``sys.std*`` might not actually be a stream, so
     check for the encoding attribute prior to returning it, and return
     a default if it doesn't exist or evaluates as False. ``default``
     is None if not provided.
     """
     if not hasattr(stream, 'encoding') or not stream.encoding:
         return default
     else:
         return stream.encoding
 
 # Less conservative replacement for sys.getdefaultencoding, that will try
 # to match the environment.
 # Defined here as central function, so if we find better choices, we
 # won't need to make changes all over IPython.
 def getdefaultencoding(prefer_stream=True):
     """Return IPython's guess for the default encoding for bytes as text.
-    
+
     If prefer_stream is True (default), asks for stdin.encoding first,
     to match the calling Terminal, but that is often None for subprocesses.
-    
+
     Then fall back on locale.getpreferredencoding(),
     which should be a sensible platform default (that respects LANG environment),
     and finally to sys.getdefaultencoding() which is the most conservative option,
     and usually UTF8 as of Python 3.
     """
     enc = None
     if prefer_stream:
         enc = get_stream_enc(sys.stdin)
     if not enc or enc=='ascii':
         try:
             # There are reports of getpreferredencoding raising errors
             # in some cases, which may well be fixed, but let's be conservative here.
             enc = locale.getpreferredencoding()
         except Exception:
             pass
     enc = enc or sys.getdefaultencoding()
     # On windows `cp0` can be returned to indicate that there is no code page.
     # Since cp0 is an invalid encoding return instead cp1252 which is the
     # Western European default.
     if enc == 'cp0':
         warnings.warn(
             "Invalid code page cp0 detected - using cp1252 instead."
             "If cp1252 is incorrect please ensure a valid code page "
             "is defined for the process.", RuntimeWarning)
         return 'cp1252'
     return enc
 
 DEFAULT_ENCODING = getdefaultencoding()

IPython/utils/frame.py

@@ -1,94 +1,92 b''
 # encoding: utf-8
 """
 Utilities for working with stack frames.
 """
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2008-2011  The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 import sys
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def extract_vars(*names,**kw):
     """Extract a set of variables by name from another frame.
 
     Parameters
     ----------
     *names : str
         One or more variable names which will be extracted from the caller's
         frame.
-
-    depth : integer, optional
+    **kw : integer, optional
         How many frames in the stack to walk when looking for your variables.
         The default is 0, which will use the frame where the call was made.
 
-
     Examples
     --------
     ::
 
         In [2]: def func(x):
            ...:     y = 1
            ...:     print(sorted(extract_vars('x','y').items()))
            ...:
 
         In [3]: func('hello')
         [('x', 'hello'), ('y', 1)]
     """
 
     depth = kw.get('depth',0)
 
     callerNS = sys._getframe(depth+1).f_locals
     return dict((k,callerNS[k]) for k in names)
 
 
 def extract_vars_above(*names):
     """Extract a set of variables by name from another frame.
 
     Similar to extractVars(), but with a specified depth of 1, so that names
     are extracted exactly from above the caller.
 
     This is simply a convenience function so that the very common case (for us)
     of skipping exactly 1 frame doesn't have to construct a special dict for
     keyword passing."""
 
     callerNS = sys._getframe(2).f_locals
     return dict((k,callerNS[k]) for k in names)
 
 
 def debugx(expr,pre_msg=''):
     """Print the value of an expression from the caller's frame.
 
     Takes an expression, evaluates it in the caller's frame and prints both
     the given expression and the resulting value (as well as a debug mark
     indicating the name of the calling function.  The input must be of a form
     suitable for eval().
 
     An optional message can be passed, which will be prepended to the printed
     expr->value pair."""
 
     cf = sys._getframe(1)
     print('[DBG:%s] %s%s -> %r' % (cf.f_code.co_name,pre_msg,expr,
                                    eval(expr,cf.f_globals,cf.f_locals)))
 
 
 # deactivate it by uncommenting the following line, which makes it a no-op
 #def debugx(expr,pre_msg=''): pass
 
 def extract_module_locals(depth=0):
     """Returns (module, locals) of the function `depth` frames away from the caller"""
     f = sys._getframe(depth + 1)
     global_ns = f.f_globals
     module = sys.modules[global_ns['__name__']]
     return (module, f.f_locals)

IPython/utils/generics.py

@@ -1,30 +1,29 b''
 # encoding: utf-8
 """Generic functions for extending IPython.
 """
 
 from IPython.core.error import TryNext
 from functools import singledispatch
 
 
 @singledispatch
 def inspect_object(obj):
     """Called when you do obj?"""
     raise TryNext
 
 
 @singledispatch
 def complete_object(obj, prev_completions):
     """Custom completer dispatching for python objects.
 
     Parameters
     ----------
     obj : object
         The object to complete.
     prev_completions : list
         List of attributes discovered so far.
-
     This should return the list of attributes in obj. If you only wish to
     add to the attributes already discovered normally, return
     own_attrs + prev_completions.
     """
     raise TryNext

IPython/utils/importstring.py

@@ -1,39 +1,39 b''
 # encoding: utf-8
 """
 A simple utility to import something by its string name.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 
 def import_item(name):
     """Import and return ``bar`` given the string ``foo.bar``.
 
     Calling ``bar = import_item("foo.bar")`` is the functional equivalent of
     executing the code ``from foo import bar``.
 
     Parameters
     ----------
     name : string
-      The fully qualified name of the module/package being imported.
+        The fully qualified name of the module/package being imported.
 
     Returns
     -------
     mod : module object
-       The module that was imported.
+        The module that was imported.
     """
     
     parts = name.rsplit('.', 1)
     if len(parts) == 2:
         # called with 'foo.bar....'
         package, obj = parts
         module = __import__(package, fromlist=[obj])
         try:
             pak = getattr(module, obj)
         except AttributeError as e:
             raise ImportError('No module named %s' % obj) from e
         return pak
     else:
         # called with un-dotted string
         return __import__(parts[0])

IPython/utils/io.py

@@ -1,249 +1,246 b''
 # encoding: utf-8
 """
 IO related utilities.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 
 
 import atexit
 import os
 import sys
 import tempfile
 import warnings
 from pathlib import Path
 from warnings import warn
 
 from IPython.utils.decorators import undoc
 from .capture import CapturedIO, capture_output
 
 @undoc
 class IOStream:
 
     def __init__(self, stream, fallback=None):
         warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
              DeprecationWarning, stacklevel=2)
         if not hasattr(stream,'write') or not hasattr(stream,'flush'):
             if fallback is not None:
                 stream = fallback
             else:
                 raise ValueError("fallback required, but not specified")
         self.stream = stream
         self._swrite = stream.write
 
         # clone all methods not overridden:
         def clone(meth):
             return not hasattr(self, meth) and not meth.startswith('_')
         for meth in filter(clone, dir(stream)):
             try:
                 val = getattr(stream, meth)
             except AttributeError:
                 pass
             else:
                 setattr(self, meth, val)
 
     def __repr__(self):
         cls = self.__class__
         tpl = '{mod}.{cls}({args})'
         return tpl.format(mod=cls.__module__, cls=cls.__name__, args=self.stream)
 
     def write(self,data):
         warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
              DeprecationWarning, stacklevel=2)
         try:
             self._swrite(data)
         except:
             try:
                 # print handles some unicode issues which may trip a plain
                 # write() call.  Emulate write() by using an empty end
                 # argument.
                 print(data, end='', file=self.stream)
             except:
                 # if we get here, something is seriously broken.
                 print('ERROR - failed to write data to stream:', self.stream,
                       file=sys.stderr)
 
     def writelines(self, lines):
         warn('IOStream is deprecated since IPython 5.0, use sys.{stdin,stdout,stderr} instead',
              DeprecationWarning, stacklevel=2)
         if isinstance(lines, str):
             lines = [lines]
         for line in lines:
             self.write(line)
 
     # This class used to have a writeln method, but regular files and streams
     # in Python don't have this method. We need to keep this completely
     # compatible so we removed it.
 
     @property
     def closed(self):
         return self.stream.closed
 
     def close(self):
         pass
 
 # setup stdin/stdout/stderr to sys.stdin/sys.stdout/sys.stderr
 devnull = open(os.devnull, 'w')
 atexit.register(devnull.close)
 
 # io.std* are deprecated, but don't show our own deprecation warnings
 # during initialization of the deprecated API.
 with warnings.catch_warnings():
     warnings.simplefilter('ignore', DeprecationWarning)
     stdin = IOStream(sys.stdin, fallback=devnull)
     stdout = IOStream(sys.stdout, fallback=devnull)
     stderr = IOStream(sys.stderr, fallback=devnull)
 
 class Tee(object):
     """A class to duplicate an output stream to stdout/err.
 
     This works in a manner very similar to the Unix 'tee' command.
 
     When the object is closed or deleted, it closes the original file given to
     it for duplication.
     """
     # Inspired by:
     # http://mail.python.org/pipermail/python-list/2007-May/442737.html
 
     def __init__(self, file_or_name, mode="w", channel='stdout'):
         """Construct a new Tee object.
 
         Parameters
         ----------
         file_or_name : filename or open filehandle (writable)
-          File that will be duplicated
-
+            File that will be duplicated
         mode : optional, valid mode for open().
-          If a filename was give, open with this mode.
-
+            If a filename was give, open with this mode.
         channel : str, one of ['stdout', 'stderr']
         """
         if channel not in ['stdout', 'stderr']:
             raise ValueError('Invalid channel spec %s' % channel)
 
         if hasattr(file_or_name, 'write') and hasattr(file_or_name, 'seek'):
             self.file = file_or_name
         else:
             self.file = open(file_or_name, mode)
         self.channel = channel
         self.ostream = getattr(sys, channel)
         setattr(sys, channel, self)
         self._closed = False
 
     def close(self):
         """Close the file and restore the channel."""
         self.flush()
         setattr(sys, self.channel, self.ostream)
         self.file.close()
         self._closed = True
 
     def write(self, data):
         """Write data to both channels."""
         self.file.write(data)
         self.ostream.write(data)
         self.ostream.flush()
 
     def flush(self):
         """Flush both channels."""
         self.file.flush()
         self.ostream.flush()
 
     def __del__(self):
         if not self._closed:
             self.close()
 
 
 def ask_yes_no(prompt, default=None, interrupt=None):
     """Asks a question and returns a boolean (y/n) answer.
 
     If default is given (one of 'y','n'), it is used if the user input is
     empty. If interrupt is given (one of 'y','n'), it is used if the user
     presses Ctrl-C. Otherwise the question is repeated until an answer is
     given.
 
     An EOF is treated as the default answer.  If there is no default, an
     exception is raised to prevent infinite loops.
 
     Valid answers are: y/yes/n/no (match is not case sensitive)."""
 
     answers = {'y':True,'n':False,'yes':True,'no':False}
     ans = None
     while ans not in answers.keys():
         try:
             ans = input(prompt+' ').lower()
             if not ans:  # response was an empty string
                 ans = default
         except KeyboardInterrupt:
             if interrupt:
                 ans = interrupt
             print("\r")
         except EOFError:
             if default in answers.keys():
                 ans = default
                 print()
             else:
                 raise
 
     return answers[ans]
 
 
 def temp_pyfile(src, ext='.py'):
     """Make a temporary python file, return filename and filehandle.
 
     Parameters
     ----------
     src : string or list of strings (no need for ending newlines if list)
-      Source code to be written to the file.
-
+        Source code to be written to the file.
     ext : optional, string
-      Extension for the generated file.
+        Extension for the generated file.
 
     Returns
     -------
     (filename, open filehandle)
-      It is the caller's responsibility to close the open file and unlink it.
+        It is the caller's responsibility to close the open file and unlink it.
     """
     fname = tempfile.mkstemp(ext)[1]
     with open(Path(fname), "w") as f:
         f.write(src)
         f.flush()
     return fname
 
 @undoc
 def atomic_writing(*args, **kwargs):
     """DEPRECATED: moved to notebook.services.contents.fileio"""
     warn("IPython.utils.io.atomic_writing has moved to notebook.services.contents.fileio since IPython 4.0", DeprecationWarning, stacklevel=2)
     from notebook.services.contents.fileio import atomic_writing
     return atomic_writing(*args, **kwargs)
 
 @undoc
 def raw_print(*args, **kw):
     """DEPRECATED: Raw print to sys.__stdout__, otherwise identical interface to print()."""
     warn("IPython.utils.io.raw_print has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2)
 
     print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
           file=sys.__stdout__)
     sys.__stdout__.flush()
 
 @undoc
 def raw_print_err(*args, **kw):
     """DEPRECATED: Raw print to sys.__stderr__, otherwise identical interface to print()."""
     warn("IPython.utils.io.raw_print_err has been deprecated since IPython 7.0", DeprecationWarning, stacklevel=2)
 
     print(*args, sep=kw.get('sep', ' '), end=kw.get('end', '\n'),
           file=sys.__stderr__)
     sys.__stderr__.flush()
 
 # used by IPykernel <- 4.9. Removed during IPython 7-dev period and re-added 
 # Keep for a version or two then should remove
 rprint = raw_print
 rprinte = raw_print_err
 
 @undoc
 def unicode_std_stream(stream='stdout'):
     """DEPRECATED, moved to nbconvert.utils.io"""
     warn("IPython.utils.io.unicode_std_stream has moved to nbconvert.utils.io since IPython 4.0", DeprecationWarning, stacklevel=2)
     from nbconvert.utils.io import unicode_std_stream
     return unicode_std_stream(stream)

IPython/utils/ipstruct.py

@@ -1,391 +1,379 b''
 # encoding: utf-8
 """A dict subclass that supports attribute style access.
 
 Authors:
 
 * Fernando Perez (original)
 * Brian Granger (refactoring to a dict subclass)
 """
 
 #-----------------------------------------------------------------------------
 #  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
 #-----------------------------------------------------------------------------
 
 __all__ = ['Struct']
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 
 class Struct(dict):
     """A dict subclass with attribute style access.
 
     This dict subclass has a a few extra features:
 
     * Attribute style access.
     * Protection of class members (like keys, items) when using attribute
       style access.
     * The ability to restrict assignment to only existing keys.
     * Intelligent merging.
     * Overloaded operators.
     """
     _allownew = True
     def __init__(self, *args, **kw):
         """Initialize with a dictionary, another Struct, or data.
 
         Parameters
         ----------
-        args : dict, Struct
+        *args : dict, Struct
             Initialize with one dict or Struct
-        kw : dict
+        **kw : dict
             Initialize with key, value pairs.
 
         Examples
         --------
-
         >>> s = Struct(a=10,b=30)
         >>> s.a
         10
         >>> s.b
         30
         >>> s2 = Struct(s,c=30)
         >>> sorted(s2.keys())
         ['a', 'b', 'c']
         """
         object.__setattr__(self, '_allownew', True)
         dict.__init__(self, *args, **kw)
 
     def __setitem__(self, key, value):
         """Set an item with check for allownew.
 
         Examples
         --------
-
         >>> s = Struct()
         >>> s['a'] = 10
         >>> s.allow_new_attr(False)
         >>> s['a'] = 10
         >>> s['a']
         10
         >>> try:
         ...     s['b'] = 20
         ... except KeyError:
         ...     print('this is not allowed')
         ...
         this is not allowed
         """
         if not self._allownew and key not in self:
             raise KeyError(
                 "can't create new attribute %s when allow_new_attr(False)" % key)
         dict.__setitem__(self, key, value)
 
     def __setattr__(self, key, value):
         """Set an attr with protection of class members.
 
         This calls :meth:`self.__setitem__` but convert :exc:`KeyError` to
         :exc:`AttributeError`.
 
         Examples
         --------
-
         >>> s = Struct()
         >>> s.a = 10
         >>> s.a
         10
         >>> try:
         ...     s.get = 10
         ... except AttributeError:
         ...     print("you can't set a class member")
         ...
         you can't set a class member
         """
         # If key is an str it might be a class member or instance var
         if isinstance(key, str):
             # I can't simply call hasattr here because it calls getattr, which
             # calls self.__getattr__, which returns True for keys in
             # self._data.  But I only want keys in the class and in
             # self.__dict__
             if key in self.__dict__ or hasattr(Struct, key):
                 raise AttributeError(
                     'attr %s is a protected member of class Struct.' % key
                 )
         try:
             self.__setitem__(key, value)
         except KeyError as e:
             raise AttributeError(e) from e
 
     def __getattr__(self, key):
         """Get an attr by calling :meth:`dict.__getitem__`.
 
         Like :meth:`__setattr__`, this method converts :exc:`KeyError` to
         :exc:`AttributeError`.
 
         Examples
         --------
-
         >>> s = Struct(a=10)
         >>> s.a
         10
         >>> type(s.get)
         <... 'builtin_function_or_method'>
         >>> try:
         ...     s.b
         ... except AttributeError:
         ...     print("I don't have that key")
         ...
         I don't have that key
         """
         try:
             result = self[key]
         except KeyError as e:
             raise AttributeError(key) from e
         else:
             return result
 
     def __iadd__(self, other):
         """s += s2 is a shorthand for s.merge(s2).
 
         Examples
         --------
-
         >>> s = Struct(a=10,b=30)
         >>> s2 = Struct(a=20,c=40)
         >>> s += s2
         >>> sorted(s.keys())
         ['a', 'b', 'c']
         """
         self.merge(other)
         return self
 
     def __add__(self,other):
         """s + s2 -> New Struct made from s.merge(s2).
 
         Examples
         --------
-
         >>> s1 = Struct(a=10,b=30)
         >>> s2 = Struct(a=20,c=40)
         >>> s = s1 + s2
         >>> sorted(s.keys())
         ['a', 'b', 'c']
         """
         sout = self.copy()
         sout.merge(other)
         return sout
 
     def __sub__(self,other):
         """s1 - s2 -> remove keys in s2 from s1.
 
         Examples
         --------
-
         >>> s1 = Struct(a=10,b=30)
         >>> s2 = Struct(a=40)
         >>> s = s1 - s2
         >>> s
         {'b': 30}
         """
         sout = self.copy()
         sout -= other
         return sout
 
     def __isub__(self,other):
         """Inplace remove keys from self that are in other.
 
         Examples
         --------
-
         >>> s1 = Struct(a=10,b=30)
         >>> s2 = Struct(a=40)
         >>> s1 -= s2
         >>> s1
         {'b': 30}
         """
         for k in other.keys():
             if k in self:
                 del self[k]
         return self
 
     def __dict_invert(self, data):
         """Helper function for merge.
 
         Takes a dictionary whose values are lists and returns a dict with
         the elements of each list as keys and the original keys as values.
         """
         outdict = {}
         for k,lst in data.items():
             if isinstance(lst, str):
                 lst = lst.split()
             for entry in lst:
                 outdict[entry] = k
         return outdict
 
     def dict(self):
         return self
 
     def copy(self):
         """Return a copy as a Struct.
 
         Examples
         --------
-
         >>> s = Struct(a=10,b=30)
         >>> s2 = s.copy()
         >>> type(s2) is Struct
         True
         """
         return Struct(dict.copy(self))
 
     def hasattr(self, key):
         """hasattr function available as a method.
 
         Implemented like has_key.
 
         Examples
         --------
-
         >>> s = Struct(a=10)
         >>> s.hasattr('a')
         True
         >>> s.hasattr('b')
         False
         >>> s.hasattr('get')
         False
         """
         return key in self
 
     def allow_new_attr(self, allow = True):
         """Set whether new attributes can be created in this Struct.
 
         This can be used to catch typos by verifying that the attribute user
         tries to change already exists in this Struct.
         """
         object.__setattr__(self, '_allownew', allow)
 
     def merge(self, __loc_data__=None, __conflict_solve=None, **kw):
         """Merge two Structs with customizable conflict resolution.
 
         This is similar to :meth:`update`, but much more flexible. First, a
         dict is made from data+key=value pairs. When merging this dict with
         the Struct S, the optional dictionary 'conflict' is used to decide
         what to do.
 
         If conflict is not given, the default behavior is to preserve any keys
         with their current value (the opposite of the :meth:`update` method's
         behavior).
 
         Parameters
         ----------
-        __loc_data : dict, Struct
+        __loc_data__ : dict, Struct
             The data to merge into self
         __conflict_solve : dict
             The conflict policy dict.  The keys are binary functions used to
             resolve the conflict and the values are lists of strings naming
             the keys the conflict resolution function applies to.  Instead of
             a list of strings a space separated string can be used, like
             'a b c'.
-        kw : dict
+        **kw : dict
             Additional key, value pairs to merge in
 
         Notes
         -----
-
         The `__conflict_solve` dict is a dictionary of binary functions which will be used to
         solve key conflicts.  Here is an example::
 
             __conflict_solve = dict(
                 func1=['a','b','c'],
                 func2=['d','e']
             )
 
         In this case, the function :func:`func1` will be used to resolve
         keys 'a', 'b' and 'c' and the function :func:`func2` will be used for
         keys 'd' and 'e'.  This could also be written as::
 
             __conflict_solve = dict(func1='a b c',func2='d e')
 
         These functions will be called for each key they apply to with the
         form::
 
             func1(self['a'], other['a'])
 
         The return value is used as the final merged value.
 
         As a convenience, merge() provides five (the most commonly needed)
         pre-defined policies: preserve, update, add, add_flip and add_s. The
         easiest explanation is their implementation::
 
             preserve = lambda old,new: old
             update   = lambda old,new: new
             add      = lambda old,new: old + new
             add_flip = lambda old,new: new + old  # note change of order!
             add_s    = lambda old,new: old + ' ' + new  # only for str!
 
         You can use those four words (as strings) as keys instead
         of defining them as functions, and the merge method will substitute
         the appropriate functions for you.
 
         For more complicated conflict resolution policies, you still need to
         construct your own functions.
 
         Examples
         --------
-
         This show the default policy:
 
         >>> s = Struct(a=10,b=30)
         >>> s2 = Struct(a=20,c=40)
         >>> s.merge(s2)
         >>> sorted(s.items())
         [('a', 10), ('b', 30), ('c', 40)]
 
         Now, show how to specify a conflict dict:
 
         >>> s = Struct(a=10,b=30)
         >>> s2 = Struct(a=20,b=40)
         >>> conflict = {'update':'a','add':'b'}
         >>> s.merge(s2,conflict)
         >>> sorted(s.items())
         [('a', 20), ('b', 70)]
         """
 
         data_dict = dict(__loc_data__,**kw)
 
         # policies for conflict resolution: two argument functions which return
         # the value that will go in the new struct
         preserve = lambda old,new: old
         update   = lambda old,new: new
         add      = lambda old,new: old + new
         add_flip = lambda old,new: new + old  # note change of order!
         add_s    = lambda old,new: old + ' ' + new
 
         # default policy is to keep current keys when there's a conflict
         conflict_solve = dict.fromkeys(self, preserve)
 
         # the conflict_solve dictionary is given by the user 'inverted': we
         # need a name-function mapping, it comes as a function -> names
         # dict. Make a local copy (b/c we'll make changes), replace user
         # strings for the three builtin policies and invert it.
         if __conflict_solve:
             inv_conflict_solve_user = __conflict_solve.copy()
             for name, func in [('preserve',preserve), ('update',update),
                                ('add',add), ('add_flip',add_flip),
                                ('add_s',add_s)]:
                 if name in inv_conflict_solve_user.keys():
                     inv_conflict_solve_user[func] = inv_conflict_solve_user[name]
                     del inv_conflict_solve_user[name]
             conflict_solve.update(self.__dict_invert(inv_conflict_solve_user))
         for key in data_dict:
             if key not in self:
                 self[key] = data_dict[key]
             else:
                 self[key] = conflict_solve[key](self[key],data_dict[key])
 

IPython/utils/module_paths.py

@@ -1,73 +1,73 b''
 """Utility functions for finding modules
 
 Utility functions for finding modules on sys.path.
 
 `find_module` returns a path to module or None, given certain conditions.
 
 """
 #-----------------------------------------------------------------------------
 # Copyright (c) 2011, the IPython Development Team.
 #
 # Distributed under the terms of the Modified BSD License.
 #
 # The full license is in the file COPYING.txt, distributed with this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 # Stdlib imports
 import importlib
 import os
 import sys
 
 # Third-party imports
 
 # Our own imports
 
 
 #-----------------------------------------------------------------------------
 # Globals and constants
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Local utilities
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Classes and functions
 #-----------------------------------------------------------------------------
 
 def find_mod(module_name):
     """
     Find module `module_name` on sys.path, and return the path to module `module_name`.
 
-      - If `module_name` refers to a module directory, then return path to __init__ file. 
+      - If `module_name` refers to a module directory, then return path to __init__ file.
         - If `module_name` is a directory without an __init__file, return None.
       - If module is missing or does not have a `.py` or `.pyw` extension, return None.
         - Note that we are not interested in running bytecode.
       - Otherwise, return the fill path of the module.
 
     Parameters
     ----------
     module_name : str
-    
+
     Returns
     -------
     module_path : str
         Path to module `module_name`, its __init__.py, or None,
         depending on above conditions.
     """
     loader = importlib.util.find_spec(module_name)
     module_path = loader.origin
     if module_path is None:
         if loader.loader in sys.meta_path:
             return loader.loader
         return None
     else:
         split_path = module_path.split(".")
         if split_path[-1] in ["py", "pyw"]:
             return module_path
         else:
             return None

IPython/utils/openpy.py

@@ -1,105 +1,105 b''
 """
 Tools to open .py files as Unicode, using the encoding specified within the file,
 as per PEP 263.
 
 Much of the code is taken from the tokenize module in Python 3.2.
 """
 
 import io
 from io import TextIOWrapper, BytesIO
 from pathlib import Path
 import re
 from tokenize import open, detect_encoding
 
 cookie_re = re.compile(r"coding[:=]\s*([-\w.]+)", re.UNICODE)
 cookie_comment_re = re.compile(r"^\s*#.*coding[:=]\s*([-\w.]+)", re.UNICODE)
 
 def source_to_unicode(txt, errors='replace', skip_encoding_cookie=True):
     """Converts a bytes string with python source code to unicode.
 
     Unicode strings are passed through unchanged. Byte strings are checked
     for the python source file encoding cookie to determine encoding.
     txt can be either a bytes buffer or a string containing the source
     code.
     """
     if isinstance(txt, str):
         return txt
     if isinstance(txt, bytes):
         buffer = BytesIO(txt)
     else:
         buffer = txt
     try:
         encoding, _ = detect_encoding(buffer.readline)
     except SyntaxError:
         encoding = "ascii"
     buffer.seek(0)
     with TextIOWrapper(buffer, encoding, errors=errors, line_buffering=True) as text:
         text.mode = 'r'
         if skip_encoding_cookie:
             return u"".join(strip_encoding_cookie(text))
         else:
             return text.read()
 
 def strip_encoding_cookie(filelike):
     """Generator to pull lines from a text-mode file, skipping the encoding
     cookie if it is found in the first two lines.
     """
     it = iter(filelike)
     try:
         first = next(it)
         if not cookie_comment_re.match(first):
             yield first
         second = next(it)
         if not cookie_comment_re.match(second):
             yield second
     except StopIteration:
         return
     
     for line in it:
         yield line
 
 def read_py_file(filename, skip_encoding_cookie=True):
     """Read a Python file, using the encoding declared inside the file.
-    
+
     Parameters
     ----------
     filename : str
-      The path to the file to read.
+        The path to the file to read.
     skip_encoding_cookie : bool
-      If True (the default), and the encoding declaration is found in the first
-      two lines, that line will be excluded from the output.
-    
+        If True (the default), and the encoding declaration is found in the first
+        two lines, that line will be excluded from the output.
+
     Returns
     -------
     A unicode string containing the contents of the file.
     """
     filepath = Path(filename)
     with open(filepath) as f:  # the open function defined in this module.
         if skip_encoding_cookie:
             return "".join(strip_encoding_cookie(f))
         else:
             return f.read()
 
 def read_py_url(url, errors='replace', skip_encoding_cookie=True):
     """Read a Python file from a URL, using the encoding declared inside the file.
-    
+
     Parameters
     ----------
     url : str
-      The URL from which to fetch the file.
+        The URL from which to fetch the file.
     errors : str
-      How to handle decoding errors in the file. Options are the same as for
-      bytes.decode(), but here 'replace' is the default.
+        How to handle decoding errors in the file. Options are the same as for
+        bytes.decode(), but here 'replace' is the default.
     skip_encoding_cookie : bool
-      If True (the default), and the encoding declaration is found in the first
-      two lines, that line will be excluded from the output.
-    
+        If True (the default), and the encoding declaration is found in the first
+        two lines, that line will be excluded from the output.
+
     Returns
     -------
     A unicode string containing the contents of the file.
     """
     # Deferred import for faster start
     from urllib.request import urlopen 
     response = urlopen(url)
     buffer = io.BytesIO(response.read())
     return source_to_unicode(buffer, errors, skip_encoding_cookie)

IPython/utils/path.py

@@ -1,436 +1,440 b''
 # encoding: utf-8
 """
 Utilities for path handling.
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import os
 import sys
 import errno
 import shutil
 import random
 import glob
 from warnings import warn
 
 from IPython.utils.process import system
 from IPython.utils.decorators import undoc
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 fs_encoding = sys.getfilesystemencoding()
 
 def _writable_dir(path):
     """Whether `path` is a directory, to which the user has write access."""
     return os.path.isdir(path) and os.access(path, os.W_OK)
 
 if sys.platform == 'win32':
     def _get_long_path_name(path):
         """Get a long path name (expand ~) on Windows using ctypes.
 
         Examples
         --------
 
         >>> get_long_path_name('c:\\docume~1')
         'c:\\\\Documents and Settings'
 
         """
         try:
             import ctypes
         except ImportError as e: 
             raise ImportError('you need to have ctypes installed for this to work') from e
         _GetLongPathName = ctypes.windll.kernel32.GetLongPathNameW
         _GetLongPathName.argtypes = [ctypes.c_wchar_p, ctypes.c_wchar_p,
             ctypes.c_uint ]
 
         buf = ctypes.create_unicode_buffer(260)
         rv = _GetLongPathName(path, buf, 260)
         if rv == 0 or rv > 260:
             return path
         else:
             return buf.value
 else:
     def _get_long_path_name(path):
         """Dummy no-op."""
         return path
 
 
 
 def get_long_path_name(path):
     """Expand a path into its long form.
 
     On Windows this expands any ~ in the paths. On other platforms, it is
     a null operation.
     """
     return _get_long_path_name(path)
 
 
 def unquote_filename(name, win32=(sys.platform=='win32')):
     """ On Windows, remove leading and trailing quotes from filenames.
 
     This function has been deprecated and should not be used any more:
     unquoting is now taken care of by :func:`IPython.utils.process.arg_split`.
     """
     warn("'unquote_filename' is deprecated since IPython 5.0 and should not "
          "be used anymore", DeprecationWarning, stacklevel=2)
     if win32:
         if name.startswith(("'", '"')) and name.endswith(("'", '"')):
             name = name[1:-1]
     return name
 
 
 def compress_user(path):
     """Reverse of :func:`os.path.expanduser`
     """
     home = os.path.expanduser('~')
     if path.startswith(home):
         path =  "~" + path[len(home):]
     return path
 
 def get_py_filename(name, force_win32=None):
     """Return a valid python filename in the current directory.
 
     If the given name is not a file, it adds '.py' and searches again.
     Raises IOError with an informative message if the file isn't found.
     """
 
     name = os.path.expanduser(name)
     if force_win32 is not None:
         warn("The 'force_win32' argument to 'get_py_filename' is deprecated "
              "since IPython 5.0 and should not be used anymore",
             DeprecationWarning, stacklevel=2)
     if not os.path.isfile(name) and not name.endswith('.py'):
         name += '.py'
     if os.path.isfile(name):
         return name
     else:
         raise IOError('File `%r` not found.' % name)
 
 
-def filefind(filename, path_dirs=None):
+def filefind(filename: str, path_dirs=None) -> str:
     """Find a file by looking through a sequence of paths.
 
     This iterates through a sequence of paths looking for a file and returns
     the full, absolute path of the first occurrence of the file.  If no set of
     path dirs is given, the filename is tested as is, after running through
     :func:`expandvars` and :func:`expanduser`.  Thus a simple call::
 
         filefind('myfile.txt')
 
     will find the file in the current working dir, but::
 
         filefind('~/myfile.txt')
 
     Will find the file in the users home directory.  This function does not
     automatically try any paths, such as the cwd or the user's home directory.
 
     Parameters
     ----------
     filename : str
         The filename to look for.
     path_dirs : str, None or sequence of str
         The sequence of paths to look for the file in.  If None, the filename
         need to be absolute or be in the cwd.  If a string, the string is
         put into a sequence and the searched.  If a sequence, walk through
         each element and join with ``filename``, calling :func:`expandvars`
         and :func:`expanduser` before testing for existence.
 
     Returns
     -------
-    Raises :exc:`IOError` or returns absolute path to file.
+    path : str
+        returns absolute path to file.
+
+    Raises
+    ------
+    IOError
     """
 
     # If paths are quoted, abspath gets confused, strip them...
     filename = filename.strip('"').strip("'")
     # If the input is an absolute path, just check it exists
     if os.path.isabs(filename) and os.path.isfile(filename):
         return filename
 
     if path_dirs is None:
         path_dirs = ("",)
     elif isinstance(path_dirs, str):
         path_dirs = (path_dirs,)
 
     for path in path_dirs:
         if path == '.': path = os.getcwd()
         testname = expand_path(os.path.join(path, filename))
         if os.path.isfile(testname):
             return os.path.abspath(testname)
 
     raise IOError("File %r does not exist in any of the search paths: %r" %
                   (filename, path_dirs) )
 
 
 class HomeDirError(Exception):
     pass
 
 
 def get_home_dir(require_writable=False) -> str:
     """Return the 'home' directory, as a unicode string.
 
     Uses os.path.expanduser('~'), and checks for writability.
 
     See stdlib docs for how this is determined.
     For Python <3.8, $HOME is first priority on *ALL* platforms.
     For Python >=3.8 on Windows, %HOME% is no longer considered.
 
     Parameters
     ----------
-
     require_writable : bool [default: False]
         if True:
             guarantees the return value is a writable directory, otherwise
             raises HomeDirError
         if False:
             The path is resolved, but it is not guaranteed to exist or be writable.
     """
 
     homedir = os.path.expanduser('~')
     # Next line will make things work even when /home/ is a symlink to
     # /usr/home as it is on FreeBSD, for example
     homedir = os.path.realpath(homedir)
 
     if not _writable_dir(homedir) and os.name == 'nt':
         # expanduser failed, use the registry to get the 'My Documents' folder.
         try:
             import winreg as wreg
             with wreg.OpenKey(
                 wreg.HKEY_CURRENT_USER,
                 r"Software\Microsoft\Windows\CurrentVersion\Explorer\Shell Folders"
             ) as key:
                 homedir = wreg.QueryValueEx(key,'Personal')[0]
         except:
             pass
 
     if (not require_writable) or _writable_dir(homedir):
         assert isinstance(homedir, str), "Homedir shoudl be unicode not bytes"
         return homedir
     else:
         raise HomeDirError('%s is not a writable dir, '
                 'set $HOME environment variable to override' % homedir)
 
 def get_xdg_dir():
     """Return the XDG_CONFIG_HOME, if it is defined and exists, else None.
 
     This is only for non-OS X posix (Linux,Unix,etc.) systems.
     """
 
     env = os.environ
 
     if os.name == 'posix' and sys.platform != 'darwin':
         # Linux, Unix, AIX, etc.
         # use ~/.config if empty OR not set
         xdg = env.get("XDG_CONFIG_HOME", None) or os.path.join(get_home_dir(), '.config')
         if xdg and _writable_dir(xdg):
             assert isinstance(xdg, str)
             return xdg
 
     return None
 
 
 def get_xdg_cache_dir():
     """Return the XDG_CACHE_HOME, if it is defined and exists, else None.
 
     This is only for non-OS X posix (Linux,Unix,etc.) systems.
     """
 
     env = os.environ
 
     if os.name == 'posix' and sys.platform != 'darwin':
         # Linux, Unix, AIX, etc.
         # use ~/.cache if empty OR not set
         xdg = env.get("XDG_CACHE_HOME", None) or os.path.join(get_home_dir(), '.cache')
         if xdg and _writable_dir(xdg):
             assert isinstance(xdg, str)
             return xdg
 
     return None
 
 
 @undoc
 def get_ipython_dir():
     warn("get_ipython_dir has moved to the IPython.paths module since IPython 4.0.", DeprecationWarning, stacklevel=2)
     from IPython.paths import get_ipython_dir
     return get_ipython_dir()
 
 @undoc
 def get_ipython_cache_dir():
     warn("get_ipython_cache_dir has moved to the IPython.paths module since IPython 4.0.", DeprecationWarning, stacklevel=2)
     from IPython.paths import get_ipython_cache_dir
     return get_ipython_cache_dir()
 
 @undoc
 def get_ipython_package_dir():
     warn("get_ipython_package_dir has moved to the IPython.paths module since IPython 4.0.", DeprecationWarning, stacklevel=2)
     from IPython.paths import get_ipython_package_dir
     return get_ipython_package_dir()
 
 @undoc
 def get_ipython_module_path(module_str):
     warn("get_ipython_module_path has moved to the IPython.paths module since IPython 4.0.", DeprecationWarning, stacklevel=2)
     from IPython.paths import get_ipython_module_path
     return get_ipython_module_path(module_str)
 
 @undoc
 def locate_profile(profile='default'):
     warn("locate_profile has moved to the IPython.paths module since IPython 4.0.", DeprecationWarning, stacklevel=2)
     from IPython.paths import locate_profile
     return locate_profile(profile=profile)
 
 def expand_path(s):
     """Expand $VARS and ~names in a string, like a shell
 
     :Examples:
 
        In [2]: os.environ['FOO']='test'
 
        In [3]: expand_path('variable FOO is $FOO')
        Out[3]: 'variable FOO is test'
     """
     # This is a pretty subtle hack. When expand user is given a UNC path
     # on Windows (\\server\share$\%username%), os.path.expandvars, removes
     # the $ to get (\\server\share\%username%). I think it considered $
     # alone an empty var. But, we need the $ to remains there (it indicates
     # a hidden share).
     if os.name=='nt':
         s = s.replace('$\\', 'IPYTHON_TEMP')
     s = os.path.expandvars(os.path.expanduser(s))
     if os.name=='nt':
         s = s.replace('IPYTHON_TEMP', '$\\')
     return s
 
 
 def unescape_glob(string):
     """Unescape glob pattern in `string`."""
     def unescape(s):
         for pattern in '*[]!?':
             s = s.replace(r'\{0}'.format(pattern), pattern)
         return s
     return '\\'.join(map(unescape, string.split('\\\\')))
 
 
 def shellglob(args):
     """
     Do glob expansion for each element in `args` and return a flattened list.
 
     Unmatched glob pattern will remain as-is in the returned list.
 
     """
     expanded = []
     # Do not unescape backslash in Windows as it is interpreted as
     # path separator:
     unescape = unescape_glob if sys.platform != 'win32' else lambda x: x
     for a in args:
         expanded.extend(glob.glob(a) or [unescape(a)])
     return expanded
 
 
 def target_outdated(target,deps):
     """Determine whether a target is out of date.
 
     target_outdated(target,deps) -> 1/0
 
     deps: list of filenames which MUST exist.
     target: single filename which may or may not exist.
 
     If target doesn't exist or is older than any file listed in deps, return
     true, otherwise return false.
     """
     try:
         target_time = os.path.getmtime(target)
     except os.error:
         return 1
     for dep in deps:
         dep_time = os.path.getmtime(dep)
         if dep_time > target_time:
             #print "For target",target,"Dep failed:",dep # dbg
             #print "times (dep,tar):",dep_time,target_time # dbg
             return 1
     return 0
 
 
 def target_update(target,deps,cmd):
     """Update a target with a given command given a list of dependencies.
 
     target_update(target,deps,cmd) -> runs cmd if target is outdated.
 
     This is just a wrapper around target_outdated() which calls the given
     command if target is outdated."""
 
     if target_outdated(target,deps):
         system(cmd)
 
 
 ENOLINK = 1998
 
 def link(src, dst):
     """Hard links ``src`` to ``dst``, returning 0 or errno.
 
     Note that the special errno ``ENOLINK`` will be returned if ``os.link`` isn't
     supported by the operating system.
     """
 
     if not hasattr(os, "link"):
         return ENOLINK
     link_errno = 0
     try:
         os.link(src, dst)
     except OSError as e:
         link_errno = e.errno
     return link_errno
 
 
 def link_or_copy(src, dst):
     """Attempts to hardlink ``src`` to ``dst``, copying if the link fails.
 
     Attempts to maintain the semantics of ``shutil.copy``.
 
     Because ``os.link`` does not overwrite files, a unique temporary file
     will be used if the target already exists, then that file will be moved
     into place.
     """
 
     if os.path.isdir(dst):
         dst = os.path.join(dst, os.path.basename(src))
 
     link_errno = link(src, dst)
     if link_errno == errno.EEXIST:
         if os.stat(src).st_ino == os.stat(dst).st_ino:
             # dst is already a hard link to the correct file, so we don't need
             # to do anything else. If we try to link and rename the file
             # anyway, we get duplicate files - see http://bugs.python.org/issue21876
             return
 
         new_dst = dst + "-temp-%04X" %(random.randint(1, 16**4), )
         try:
             link_or_copy(src, new_dst)
         except:
             try:
                 os.remove(new_dst)
             except OSError:
                 pass
             raise
         os.rename(new_dst, dst)
     elif link_errno != 0:
         # Either link isn't supported, or the filesystem doesn't support
         # linking, or 'src' and 'dst' are on different filesystems.
         shutil.copy(src, dst)
 
 def ensure_dir_exists(path, mode=0o755):
     """ensure that a directory exists
 
     If it doesn't exist, try to create it and protect against a race condition
     if another process is doing the same.
 
     The default permissions are 755, which differ from os.makedirs default of 777.
     """
     if not os.path.exists(path):
         try:
             os.makedirs(path, mode=mode)
         except OSError as e:
             if e.errno != errno.EEXIST:
                 raise
     elif not os.path.isdir(path):
         raise IOError("%r exists but is not a directory" % path)

IPython/utils/sysinfo.py

@@ -1,166 +1,166 b''
 # encoding: utf-8
 """
 Utilities for getting information about IPython and the system it's running in.
 """
 
 #-----------------------------------------------------------------------------
 #  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 os
 import platform
 import pprint
 import sys
 import subprocess
 
 from IPython.core import release
 from IPython.utils import _sysinfo, encoding
 
 #-----------------------------------------------------------------------------
 # Code
 #-----------------------------------------------------------------------------
 
 def pkg_commit_hash(pkg_path):
     """Get short form of commit hash given directory `pkg_path`
 
     We get the commit hash from (in order of preference):
 
     * IPython.utils._sysinfo.commit
     * git output, if we are in a git repository
 
     If these fail, we return a not-found placeholder tuple
 
     Parameters
     ----------
     pkg_path : str
-       directory containing package
-       only used for getting commit from active repo
+        directory containing package
+        only used for getting commit from active repo
 
     Returns
     -------
     hash_from : str
-       Where we got the hash from - description
+        Where we got the hash from - description
     hash_str : str
-       short form of hash
+        short form of hash
     """
     # Try and get commit from written commit text file
     if _sysinfo.commit:
         return "installation", _sysinfo.commit
 
     # maybe we are in a repository
     proc = subprocess.Popen('git rev-parse --short HEAD'.split(' '),
                             stdout=subprocess.PIPE,
                             stderr=subprocess.PIPE,
                             cwd=pkg_path)
     repo_commit, _ = proc.communicate()
     if repo_commit:
         return 'repository', repo_commit.strip().decode('ascii')
     return '(none found)', '<not found>'
 
 
 def pkg_info(pkg_path):
     """Return dict describing the context of this package
 
     Parameters
     ----------
     pkg_path : str
-       path containing __init__.py for package
+        path containing __init__.py for package
 
     Returns
     -------
     context : dict
-       with named parameters of interest
+        with named parameters of interest
     """
     src, hsh = pkg_commit_hash(pkg_path)
     return dict(
         ipython_version=release.version,
         ipython_path=pkg_path,
         commit_source=src,
         commit_hash=hsh,
         sys_version=sys.version,
         sys_executable=sys.executable,
         sys_platform=sys.platform,
         platform=platform.platform(),
         os_name=os.name,
         default_encoding=encoding.DEFAULT_ENCODING,
         )
 
 def get_sys_info():
     """Return useful information about IPython and the system, as a dict."""
     p = os.path
     path = p.realpath(p.dirname(p.abspath(p.join(__file__, '..'))))
     return pkg_info(path)
 
 def sys_info():
     """Return useful information about IPython and the system, as a string.
 
     Examples
     --------
     ::
     
         In [2]: print(sys_info())
         {'commit_hash': '144fdae',      # random
          'commit_source': 'repository',
          'ipython_path': '/home/fperez/usr/lib/python2.6/site-packages/IPython',
          'ipython_version': '0.11.dev',
          'os_name': 'posix',
          'platform': 'Linux-2.6.35-22-generic-i686-with-Ubuntu-10.10-maverick',
          'sys_executable': '/usr/bin/python',
          'sys_platform': 'linux2',
          'sys_version': '2.6.6 (r266:84292, Sep 15 2010, 15:52:39) \\n[GCC 4.4.5]'}
     """
     return pprint.pformat(get_sys_info())
 
 def _num_cpus_unix():
     """Return the number of active CPUs on a Unix system."""
     return os.sysconf("SC_NPROCESSORS_ONLN")
 
 
 def _num_cpus_darwin():
     """Return the number of active CPUs on a Darwin system."""
     p = subprocess.Popen(['sysctl','-n','hw.ncpu'],stdout=subprocess.PIPE)
     return p.stdout.read()
 
 
 def _num_cpus_windows():
     """Return the number of active CPUs on a Windows system."""
     return os.environ.get("NUMBER_OF_PROCESSORS")
 
 
 def num_cpus():
    """Return the effective number of CPUs in the system as an integer.
 
    This cross-platform function makes an attempt at finding the total number of
    available CPUs in the system, as returned by various underlying system and
    python calls.
 
    If it can't find a sensible answer, it returns 1 (though an error *may* make
    it return a large positive number that's actually incorrect).
    """
 
    # Many thanks to the Parallel Python project (http://www.parallelpython.com)
    # for the names of the keys we needed to look up for this function.  This
    # code was inspired by their equivalent function.
 
    ncpufuncs = {'Linux':_num_cpus_unix,
                 'Darwin':_num_cpus_darwin,
                 'Windows':_num_cpus_windows
                 }
 
    ncpufunc = ncpufuncs.get(platform.system(),
                             # default to unix version (Solaris, AIX, etc)
                             _num_cpus_unix)
 
    try:
        ncpus = max(1,int(ncpufunc()))
    except:
        ncpus = 1
    return ncpus
 

IPython/utils/terminal.py

@@ -1,129 +1,129 b''
 # encoding: utf-8
 """
 Utilities for working with terminals.
 
 Authors:
 
 * Brian E. Granger
 * Fernando Perez
 * Alexander Belchenko (e-mail: bialix AT ukr.net)
 """
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import os
 import sys
 import warnings
 from shutil import get_terminal_size as _get_terminal_size
 
 # This variable is part of the expected API of the module:
 ignore_termtitle = True
 
 
 
 if os.name == 'posix':
     def _term_clear():
         os.system('clear')
 elif sys.platform == 'win32':
     def _term_clear():
         os.system('cls')
 else:
     def _term_clear():
         pass
 
 
 
 def toggle_set_term_title(val):
     """Control whether set_term_title is active or not.
 
     set_term_title() allows writing to the console titlebar.  In embedded
     widgets this can cause problems, so this call can be used to toggle it on
     or off as needed.
 
     The default state of the module is for the function to be disabled.
 
     Parameters
     ----------
-      val : bool
+    val : bool
         If True, set_term_title() actually writes to the terminal (using the
         appropriate platform-specific module).  If False, it is a no-op.
     """
     global ignore_termtitle
     ignore_termtitle = not(val)
 
 
 def _set_term_title(*args,**kw):
     """Dummy no-op."""
     pass
 
 
 def _restore_term_title():
     pass
 
 
 def _set_term_title_xterm(title):
     """ Change virtual terminal title in xterm-workalikes """
     # save the current title to the xterm "stack"
     sys.stdout.write('\033[22;0t') 
     sys.stdout.write('\033]0;%s\007' % title)
 
 
 def _restore_term_title_xterm():
     sys.stdout.write('\033[23;0t') 
 
 
 if os.name == 'posix':
     TERM = os.environ.get('TERM','')
     if TERM.startswith('xterm'):
         _set_term_title = _set_term_title_xterm
         _restore_term_title = _restore_term_title_xterm
 elif sys.platform == 'win32':
     try:
         import ctypes
 
         SetConsoleTitleW = ctypes.windll.kernel32.SetConsoleTitleW
         SetConsoleTitleW.argtypes = [ctypes.c_wchar_p]
     
         def _set_term_title(title):
             """Set terminal title using ctypes to access the Win32 APIs."""
             SetConsoleTitleW(title)
     except ImportError:
         def _set_term_title(title):
             """Set terminal title using the 'title' command."""
             global ignore_termtitle
 
             try:
                 # Cannot be on network share when issuing system commands
                 curr = os.getcwd()
                 os.chdir("C:")
                 ret = os.system("title " + title)
             finally:
                 os.chdir(curr)
             if ret:
                 # non-zero return code signals error, don't try again
                 ignore_termtitle = True
 
 
 def set_term_title(title):
     """Set terminal title using the necessary platform-dependent calls."""
     if ignore_termtitle:
         return
     _set_term_title(title)
 
 
 def restore_term_title():
     """Restore, if possible, terminal title to the original state"""
     if ignore_termtitle:
         return
     _restore_term_title()
 
 
 def freeze_term_title():
     warnings.warn("This function is deprecated, use toggle_set_term_title()")
     global ignore_termtitle
     ignore_termtitle = True
 
 
 def get_terminal_size(defaultx=80, defaulty=25):
     return _get_terminal_size((defaultx, defaulty))

IPython/utils/tests/test_capture.py

@@ -1,172 +1,172 b''
 # encoding: utf-8
 """Tests for IPython.utils.capture"""
 
 #-----------------------------------------------------------------------------
 #  Copyright (C) 2013 The IPython Development Team
 #
 #  Distributed under the terms of the BSD License.  The full license is in
 #  the file COPYING, distributed as part of this software.
 #-----------------------------------------------------------------------------
 
 #-----------------------------------------------------------------------------
 # Imports
 #-----------------------------------------------------------------------------
 
 
 import sys
 
 import nose.tools as nt
 import pytest
 
 from IPython.testing.decorators import skip_iptest_but_not_pytest
 
 from IPython.utils import capture
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 
 _mime_map = dict(
     _repr_png_="image/png",
     _repr_jpeg_="image/jpeg",
     _repr_svg_="image/svg+xml",
     _repr_html_="text/html",
     _repr_json_="application/json",
     _repr_javascript_="application/javascript",
 )
 
 basic_data = {
     'image/png' : b'binarydata',
     'text/html' : "<b>bold</b>",
 }
 basic_metadata = {
     'image/png' : {
         'width' : 10,
         'height' : 20,
     },
 }
 
 full_data = {
     'image/png' : b'binarydata',
     'image/jpeg' : b'binarydata',
     'image/svg+xml' : "<svg>",
     'text/html' : "<b>bold</b>",
     'application/javascript' : "alert();",
     'application/json' : "{}",
 }
 full_metadata = {
     'image/png' : {"png" : "exists"},
     'image/jpeg' : {"jpeg" : "exists"},
     'image/svg+xml' : {"svg" : "exists"},
     'text/html' : {"html" : "exists"},
     'application/javascript' : {"js" : "exists"},
     'application/json' : {"json" : "exists"},
 }
 
 hello_stdout = "hello, stdout"
 hello_stderr = "hello, stderr"
 
 #-----------------------------------------------------------------------------
 # Test Functions
 #-----------------------------------------------------------------------------
 @pytest.mark.parametrize("method_mime", _mime_map.items())
 @skip_iptest_but_not_pytest
 def test_rich_output_empty(method_mime):
     """RichOutput with no args"""
     rich = capture.RichOutput()
     method, mime = method_mime
     nt.assert_equal(getattr(rich, method)(), None)
     
 def test_rich_output():
     """test RichOutput basics"""
     data = basic_data
     metadata = basic_metadata
     rich = capture.RichOutput(data=data, metadata=metadata)
     nt.assert_equal(rich._repr_html_(), data["text/html"])
     nt.assert_equal(rich._repr_png_(), (data["image/png"], metadata["image/png"]))
     nt.assert_equal(rich._repr_latex_(), None)
     nt.assert_equal(rich._repr_javascript_(), None)
     nt.assert_equal(rich._repr_svg_(), None)
 
 
 @skip_iptest_but_not_pytest
 @pytest.mark.parametrize("method_mime", _mime_map.items())
 def test_rich_output_no_metadata(method_mime):
     """test RichOutput with no metadata"""
     data = full_data
     rich = capture.RichOutput(data=data)
     method, mime = method_mime
     nt.assert_equal(getattr(rich, method)(), data[mime])
 
 
 @skip_iptest_but_not_pytest
 @pytest.mark.parametrize("method_mime", _mime_map.items())
 def test_rich_output_metadata(method_mime):
     """test RichOutput with metadata"""
     data = full_data
     metadata = full_metadata
     rich = capture.RichOutput(data=data, metadata=metadata)
     method, mime = method_mime
     nt.assert_equal(getattr(rich, method)(), (data[mime], metadata[mime]))
 
 def test_rich_output_display():
     """test RichOutput.display
-    
+
     This is a bit circular, because we are actually using the capture code we are testing
     to test itself.
     """
     data = full_data
     rich = capture.RichOutput(data=data)
     with capture.capture_output() as cap:
         rich.display()
     nt.assert_equal(len(cap.outputs), 1)
     rich2 = cap.outputs[0]
     nt.assert_equal(rich2.data, rich.data)
     nt.assert_equal(rich2.metadata, rich.metadata)
 
 def test_capture_output():
     """capture_output works"""
     rich = capture.RichOutput(data=full_data)
     with capture.capture_output() as cap:
         print(hello_stdout, end="")
         print(hello_stderr, end="", file=sys.stderr)
         rich.display()
     nt.assert_equal(hello_stdout, cap.stdout)
     nt.assert_equal(hello_stderr, cap.stderr)
 
 
 def test_capture_output_no_stdout():
     """test capture_output(stdout=False)"""
     rich = capture.RichOutput(data=full_data)
     with capture.capture_output(stdout=False) as cap:
         print(hello_stdout, end="")
         print(hello_stderr, end="", file=sys.stderr)
         rich.display()
     nt.assert_equal("", cap.stdout)
     nt.assert_equal(hello_stderr, cap.stderr)
     nt.assert_equal(len(cap.outputs), 1)
 
 
 def test_capture_output_no_stderr():
     """test capture_output(stderr=False)"""
     rich = capture.RichOutput(data=full_data)
     # add nested capture_output so stderr doesn't make it to nose output
     with capture.capture_output(), capture.capture_output(stderr=False) as cap:
         print(hello_stdout, end="")
         print(hello_stderr, end="", file=sys.stderr)
         rich.display()
     nt.assert_equal(hello_stdout, cap.stdout)
     nt.assert_equal("", cap.stderr)
     nt.assert_equal(len(cap.outputs), 1)
 
 
 def test_capture_output_no_display():
     """test capture_output(display=False)"""
     rich = capture.RichOutput(data=full_data)
     with capture.capture_output(display=False) as cap:
         print(hello_stdout, end="")
         print(hello_stderr, end="", file=sys.stderr)
         rich.display()
     nt.assert_equal(hello_stdout, cap.stdout)
     nt.assert_equal(hello_stderr, cap.stderr)
     nt.assert_equal(cap.outputs, [])

IPython/utils/tests/test_module_paths.py

@@ -1,111 +1,111 b''
 # encoding: utf-8
 """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
 #-----------------------------------------------------------------------------
 
 import shutil
 import sys
 import tempfile
 
 from pathlib import Path
 
 from IPython.testing.tools import make_tempfile
 
 import IPython.utils.module_paths as mp
 
 import nose.tools as nt
 
 TEST_FILE_PATH = Path(__file__).resolve().parent
 
 TMP_TEST_DIR = Path(tempfile.mkdtemp(suffix="with.dot"))
 #
 # Setup/teardown functions/decorators
 #
 
 old_syspath = sys.path
 
 def make_empty_file(fname):
     open(fname, 'w').close()
 
 
 def setup_module():
     """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...
     Path(TMP_TEST_DIR / "xmod").mkdir(parents=True)
     Path(TMP_TEST_DIR / "nomod").mkdir(parents=True)
     make_empty_file(TMP_TEST_DIR / "xmod/__init__.py")
     make_empty_file(TMP_TEST_DIR / "xmod/sub.py")
     make_empty_file(TMP_TEST_DIR / "pack.py")
     make_empty_file(TMP_TEST_DIR / "packpyc.pyc")
     sys.path = [str(TMP_TEST_DIR)]
 
 def teardown_module():
     """Teardown testenvironment for the module:
 
-            - Remove tempdir
-            - restore sys.path
+    - 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_tempdir():
     """
     Ensure the test are done with a temporary file that have a dot somewhere.
     """
     nt.assert_in(".", str(TMP_TEST_DIR))
 
 
 def test_find_mod_1():
     """
     Search for a directory's file path.
     Expected output: a path to that directory's __init__.py file.
     """
     modpath = TMP_TEST_DIR / "xmod" / "__init__.py"
     nt.assert_equal(Path(mp.find_mod("xmod")), modpath)
 
 def test_find_mod_2():
     """
     Search for a directory's file path.
     Expected output: a path to that directory's __init__.py file.
     TODO: Confirm why this is a duplicate test.
     """
     modpath = TMP_TEST_DIR / "xmod" / "__init__.py"
     nt.assert_equal(Path(mp.find_mod("xmod")), modpath)
 
 def test_find_mod_3():
     """
     Search for a directory + a filename without its .py extension
     Expected output: full path with .py extension.
     """
     modpath = TMP_TEST_DIR / "xmod" / "sub.py"
     nt.assert_equal(Path(mp.find_mod("xmod.sub")), modpath)
 
 def test_find_mod_4():
     """
     Search for a filename without its .py extension
     Expected output: full path with .py extension
     """
     modpath = TMP_TEST_DIR / "pack.py"
     nt.assert_equal(Path(mp.find_mod("pack")), modpath)
 
 def test_find_mod_5():
     """
     Search for a filename with a .pyc extension
     Expected output: TODO: do we exclude or include .pyc files?
     """
     nt.assert_equal(mp.find_mod("packpyc"), None)

IPython/utils/tests/test_path.py

@@ -1,492 +1,492 b''
 # encoding: utf-8
 """Tests for IPython.utils.path.py"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import os
 import shutil
 import sys
 import tempfile
 import unittest
 from contextlib import contextmanager
 from unittest.mock import patch
 from os.path import join, abspath
 from imp import reload
 
 from nose import SkipTest, with_setup
 import nose.tools as nt
 
 import IPython
 from IPython import paths
 from IPython.testing import decorators as dec
 from IPython.testing.decorators import (skip_if_not_win32, skip_win32,
                                         onlyif_unicode_paths,
                                         skip_win32_py38,)
 from IPython.testing.tools import make_tempfile
 from IPython.utils import path
 from IPython.utils.tempdir import TemporaryDirectory
 
 
 # Platform-dependent imports
 try:
     import winreg as wreg
 except ImportError:
     #Fake _winreg module on non-windows platforms
     import types
     wr_name = "winreg"
     sys.modules[wr_name] = types.ModuleType(wr_name)
     try:
         import winreg as wreg
     except ImportError:
         import _winreg as wreg
         #Add entries that needs to be stubbed by the testing code
         (wreg.OpenKey, wreg.QueryValueEx,) = (None, None)
 
 #-----------------------------------------------------------------------------
 # Globals
 #-----------------------------------------------------------------------------
 env = os.environ
 TMP_TEST_DIR = tempfile.mkdtemp()
 HOME_TEST_DIR = join(TMP_TEST_DIR, "home_test_dir")
 #
 # Setup/teardown functions/decorators
 #
 
 def setup_module():
     """Setup testenvironment for the module:
 
-            - Adds dummy home dir tree
+    - 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(os.path.join(HOME_TEST_DIR, 'ipython'))
 
 
 def teardown_module():
     """Teardown testenvironment for the module:
 
-            - Remove dummy home dir tree
+    - 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())
 
 def teardown_environment():
     """Restore things that were remembered 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 list(env):
         if key not in oldenv:
             del env[key]
     env.update(oldenv)
     if hasattr(sys, 'frozen'):
         del sys.frozen
 
 # 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
     """
     unfrozen = path.get_home_dir()
     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, unfrozen)
 
 
 @skip_if_not_win32
 @with_environment
 def test_get_home_dir_2():
     """Testcase for py2exe logic, compressed lib
     """
     unfrozen = path.get_home_dir()
     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, unfrozen)
 
 
 @skip_win32_py38
 @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)
 
 @skip_win32_py38
 @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)
 
     class key:
         def __enter__(self):
             pass
         def Close(self):
             pass
         def __exit__(*args, **kwargs):
             pass
 
     with patch.object(wreg, 'OpenKey', return_value=key()), \
          patch.object(wreg, 'QueryValueEx', return_value=[abspath(HOME_TEST_DIR)]):
         home_dir = path.get_home_dir()
     nt.assert_equal(home_dir, abspath(HOME_TEST_DIR))
 
 @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 nonexistent 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 = paths.get_ipython_dir()
     t = path.filefind(f.name, alt_dirs)
     # print 'found:',t
 
 
 @dec.skip_if_not_win32
 def test_get_long_path_name_win32():
     with TemporaryDirectory() as tmpdir:
 
         # Make a long path. Expands the path of tmpdir prematurely as it may already have a long
         # path component, so ensure we include the long form of it
         long_path = os.path.join(path.get_long_path_name(tmpdir), 'this is my long path name')
         os.makedirs(long_path)
 
         # Test to see if the short path evaluates correctly.
         short_path = os.path.join(tmpdir, 'THISIS~1')
         evaluated_path = path.get_long_path_name(short_path)
         nt.assert_equal(evaluated_path.lower(), long_path.lower())
 
 
 @dec.skip_win32
 def test_get_long_path_name():
     p = path.get_long_path_name('/usr/local')
     nt.assert_equal(p,'/usr/local')
 
 
 class TestRaiseDeprecation(unittest.TestCase):
 
     @dec.skip_win32 # can't create not-user-writable dir on win
     @with_environment
     def test_not_writable_ipdir(self):
         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, 0o555)
         try:
             open(os.path.join(ipdir, "_foo_"), 'w').close()
         except IOError:
             pass
         else:
             # I can still write to an unwritable dir,
             # assume I'm root and skip the test
             raise SkipTest("I can't create directories that I can't write to")
         with self.assertWarnsRegex(UserWarning, 'is not a writable location'):
             ipdir = paths.get_ipython_dir()
         env.pop('IPYTHON_DIR', None)
 
 @with_environment
 def test_get_py_filename():
     os.chdir(TMP_TEST_DIR)
     with make_tempfile('foo.py'):
         nt.assert_equal(path.get_py_filename('foo.py'), 'foo.py')
         nt.assert_equal(path.get_py_filename('foo'), 'foo.py')
     with make_tempfile('foo'):
         nt.assert_equal(path.get_py_filename('foo'), 'foo')
         nt.assert_raises(IOError, path.get_py_filename, 'foo.py')
     nt.assert_raises(IOError, path.get_py_filename, 'foo')
     nt.assert_raises(IOError, path.get_py_filename, 'foo.py')
     true_fn = 'foo with spaces.py'
     with make_tempfile(true_fn):
         nt.assert_equal(path.get_py_filename('foo with spaces'), true_fn)
         nt.assert_equal(path.get_py_filename('foo with spaces.py'), true_fn)
         nt.assert_raises(IOError, path.get_py_filename, '"foo with spaces.py"')
         nt.assert_raises(IOError, path.get_py_filename, "'foo with spaces.py'")
 
 @onlyif_unicode_paths
 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('fooéè.py')
     except IOError as ex:
         str(ex)
 
 
 class TestShellGlob(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls):
         cls.filenames_start_with_a = ['a0', 'a1', 'a2']
         cls.filenames_end_with_b = ['0b', '1b', '2b']
         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.getcwd()
         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_equal(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_equal(path.unescape_glob(r'\*\[\!\]\?'), '*[!]?')
     nt.assert_equal(path.unescape_glob(r'\\*'), r'\*')
     nt.assert_equal(path.unescape_glob(r'\\\*'), r'\*')
     nt.assert_equal(path.unescape_glob(r'\\a'), r'\a')
     nt.assert_equal(path.unescape_glob(r'\a'), r'\a')
 
 
 @onlyif_unicode_paths
 def test_ensure_dir_exists():
     with TemporaryDirectory() as td:
         d = os.path.join(td, '∂ir')
         path.ensure_dir_exists(d) # create it
         assert os.path.isdir(d)
         path.ensure_dir_exists(d) # no-op
         f = os.path.join(td, 'ƒile')
         open(f, 'w').close() # touch
         with nt.assert_raises(IOError):
             path.ensure_dir_exists(f)
 
 class TestLinkOrCopy(unittest.TestCase):
     def setUp(self):
         self.tempdir = TemporaryDirectory()
         self.src = self.dst("src")
         with open(self.src, "w") as f:
             f.write("Hello, world!")
 
     def tearDown(self):
         self.tempdir.cleanup()
 
     def dst(self, *args):
         return os.path.join(self.tempdir.name, *args)
 
     def assert_inode_not_equal(self, a, b):
         nt.assert_not_equal(os.stat(a).st_ino, os.stat(b).st_ino,
                             "%r and %r do reference the same indoes" %(a, b))
 
     def assert_inode_equal(self, a, b):
         nt.assert_equal(os.stat(a).st_ino, os.stat(b).st_ino,
                         "%r and %r do not reference the same indoes" %(a, b))
 
     def assert_content_equal(self, a, b):
         with open(a) as a_f:
             with open(b) as b_f:
                 nt.assert_equal(a_f.read(), b_f.read())
 
     @skip_win32
     def test_link_successful(self):
         dst = self.dst("target")
         path.link_or_copy(self.src, dst)
         self.assert_inode_equal(self.src, dst)
 
     @skip_win32
     def test_link_into_dir(self):
         dst = self.dst("some_dir")
         os.mkdir(dst)
         path.link_or_copy(self.src, dst)
         expected_dst = self.dst("some_dir", os.path.basename(self.src))
         self.assert_inode_equal(self.src, expected_dst)
 
     @skip_win32
     def test_target_exists(self):
         dst = self.dst("target")
         open(dst, "w").close()
         path.link_or_copy(self.src, dst)
         self.assert_inode_equal(self.src, dst)
 
     @skip_win32
     def test_no_link(self):
         real_link = os.link
         try:
             del os.link
             dst = self.dst("target")
             path.link_or_copy(self.src, dst)
             self.assert_content_equal(self.src, dst)
             self.assert_inode_not_equal(self.src, dst)
         finally:
             os.link = real_link
 
     @skip_if_not_win32
     def test_windows(self):
         dst = self.dst("target")
         path.link_or_copy(self.src, dst)
         self.assert_content_equal(self.src, dst)
 
     def test_link_twice(self):
         # Linking the same file twice shouldn't leave duplicates around.
         # See https://github.com/ipython/ipython/issues/6450
         dst = self.dst('target')
         path.link_or_copy(self.src, dst)
         path.link_or_copy(self.src, dst)
         self.assert_inode_equal(self.src, dst)
         nt.assert_equal(sorted(os.listdir(self.tempdir.name)), ['src', 'target'])

IPython/utils/tests/test_sysinfo.py

@@ -1,17 +1,17 b''
 # coding: utf-8
 """Test suite for our sysinfo utilities."""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 import json
 import nose.tools as nt
 
 from IPython.utils import sysinfo
 
 
 def test_json_getsysinfo():
     """
-    test that it is easily jsonable and don't return bytes somewhere. 
+    test that it is easily jsonable and don't return bytes somewhere.
     """
     json.dumps(sysinfo.get_sys_info())

IPython/utils/text.py

@@ -1,771 +1,761 b''
 # encoding: utf-8
 """
 Utilities for working with strings and text.
 
 Inheritance diagram:
 
 .. inheritance-diagram:: IPython.utils.text
    :parts: 3
 """
 
 import os
 import re
 import sys
 import textwrap
 from string import Formatter
 from pathlib import Path
 
 
 # datetime.strftime date format for ipython
 if sys.platform == 'win32':
     date_format = "%B %d, %Y"
 else:
     date_format = "%B %-d, %Y"
 
 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 (requires path.py package)
 
     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.register(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 (requires path.py package)
 
     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, str):
             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.register(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, str):
         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
 
 def strip_ansi(source):
     """
     Remove ansi escape codes from text.
-    
+
     Parameters
     ----------
     source : str
         Source to remove the ansi from
     """
     return re.sub(r'\033\[(\d|;)+?m', '', source)
 
 
 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
 
 #XXX: As of Python 3.4, the format string parsing no longer splits on a colon
 # inside [], so EvalFormatter can handle slicing. Once we only support 3.4 and
 # above, it should be possible to remove FullEvalFormatter.
 
 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]: '2'
 
         In [3]: f.format('{list(range(5))[2:4]}')
         Out[3]: '[2, 3]'
 
         In [4]: f.format('{3*2}')
         Out[4]: '6'
     """
     # copied from Formatter._vformat with minor changes to allow eval
     # and replace the format_spec code with slicing
     def vformat(self, format_string:str, args, kwargs)->str:
         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 ''.join(result)
 
 
 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]: '2'
 
         In [3]: f.format('23 * 76 is $result', result=23*76)
         Out[3]: '23 * 76 is 1748'
 
         In [4]: f.format('$a or {b}', a=1, b=2)
         Out[4]: '1 or 2'
     """
     _dollar_pattern_ignore_single_quote = re.compile(r"(.*?)\$(\$?[\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_ignore_single_quote.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 _col_chunks(l, max_rows, row_first=False):
     """Yield successive max_rows-sized column chunks from l."""
     if row_first:
         ncols = (len(l) // max_rows) + (len(l) % max_rows > 0)
         for i in range(ncols):
             yield [l[j] for j in range(i, len(l), ncols)]
     else:
         for i in range(0, len(l), max_rows):
             yield l[i:(i + max_rows)]
 
 
 def _find_optimal(rlist, row_first=False, separator_size=2, displaywidth=80):
     """Calculate optimal info to columnize a list of string"""
     for max_rows in range(1, len(rlist) + 1):
         col_widths = list(map(max, _col_chunks(rlist, max_rows, row_first)))
         sumlength = sum(col_widths)
         ncols = len(col_widths)
         if sumlength + separator_size * (ncols - 1) <= displaywidth:
             break
     return {'num_columns': ncols,
             'optimal_separator_width': (displaywidth - sumlength) // (ncols - 1) if (ncols - 1) else 0,
             'max_rows': max_rows,
             'column_widths': col_widths
             }
 
 
 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]
 
 
 def compute_item_matrix(items, row_first=False, empty=None, *args, **kwargs) :
     """Returns a nested list, and info to columnize items
 
     Parameters
     ----------
-
     items
         list of strings to columize
     row_first : (default False)
         Whether to compute columns for a row-first matrix instead of
         column-first (default).
     empty : (default None)
         default value to fill list if needed
     separator_size : int (default=2)
         How much characters will be used as a separation between each columns.
     displaywidth : int (default=80)
         The width of the area onto which the columns should enter
 
     Returns
     -------
-
     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 columns. 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:
 
         num_columns
           number of columns
         max_rows
           maximum number of rows (final number may be less)
         column_widths
           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']
         In [2]: list, info = compute_item_matrix(l, displaywidth=12)
         In [3]: list
         Out[3]: [['aaa', 'f', 'k'], ['b', 'g', 'l'], ['cc', 'h', None], ['d', 'i', None], ['eeeee', 'j', None]]
         In [4]: ideal = {'num_columns': 3, 'column_widths': [5, 1, 1], 'optimal_separator_width': 2, 'max_rows': 5}
         In [5]: all((info[k] == ideal[k] for k in ideal.keys()))
         Out[5]: True
     """
     info = _find_optimal(list(map(len, items)), row_first, *args, **kwargs)
     nrow, ncol = info['max_rows'], info['num_columns']
     if row_first:
         return ([[_get_or_default(items, r * ncol + c, default=empty) for c in range(ncol)] for r in range(nrow)], info)
     else:
         return ([[_get_or_default(items, c * nrow + r, default=empty) for c in range(ncol)] for r in range(nrow)], info)
 
 
 def columnize(items, row_first=False, separator='  ', displaywidth=80, spread=False):
     """ Transform a list of strings into a single string with columns.
 
     Parameters
     ----------
     items : sequence of strings
         The strings to process.
-
     row_first : (default False)
         Whether to compute columns for a row-first matrix instead of
         column-first (default).
-
     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, row_first=row_first, separator_size=len(separator), displaywidth=displaywidth)
     if spread:
         separator = separator.ljust(int(info['optimal_separator_width']))
     fmatrix = [filter(None, x) for x in matrix]
     sjoin = lambda x : separator.join([ y.ljust(w, ' ') for y, w in zip(x, info['column_widths'])])
     return '\n'.join(map(sjoin, fmatrix))+'\n'
 
 
 def get_text_list(list_, last_sep=' and ', sep=", ", wrap_item_with=""):
     """
     Return a string with a natural enumeration of items
 
     >>> get_text_list(['a', 'b', 'c', 'd'])
     'a, b, c and d'
     >>> get_text_list(['a', 'b', 'c'], ' or ')
     'a, b or c'
     >>> get_text_list(['a', 'b', 'c'], ', ')
     'a, b, c'
     >>> get_text_list(['a', 'b'], ' or ')
     'a or b'
     >>> get_text_list(['a'])
     'a'
     >>> get_text_list([])
     ''
     >>> get_text_list(['a', 'b'], wrap_item_with="`")
     '`a` and `b`'
     >>> get_text_list(['a', 'b', 'c', 'd'], " = ", sep=" + ")
     'a + b + c = d'
     """
     if len(list_) == 0:
         return ''
     if wrap_item_with:
         list_ = ['%s%s%s' % (wrap_item_with, item, wrap_item_with) for
                  item in list_]
     if len(list_) == 1:
         return list_[0]
     return '%s%s%s' % (
         sep.join(i for i in list_[:-1]),
         last_sep, list_[-1])

IPython/utils/tokenutil.py

@@ -1,130 +1,127 b''
 """Token-related utilities"""
 
 # Copyright (c) IPython Development Team.
 # Distributed under the terms of the Modified BSD License.
 
 from collections import namedtuple
 from io import StringIO
 from keyword import iskeyword
 
 import tokenize
 
 
 Token = namedtuple('Token', ['token', 'text', 'start', 'end', 'line'])
 
 def generate_tokens(readline):
     """wrap generate_tokens to catch EOF errors"""
     try:
         for token in tokenize.generate_tokens(readline):
             yield token
     except tokenize.TokenError:
         # catch EOF error
         return
 
 def line_at_cursor(cell, cursor_pos=0):
     """Return the line in a cell at a given cursor position
-    
+
     Used for calling line-based APIs that don't support multi-line input, yet.
-    
+
     Parameters
     ----------
-    
-    cell: str
+    cell : str
         multiline block of text
-    cursor_pos: integer
+    cursor_pos : integer
         the cursor position
-    
+
     Returns
     -------
-    
     (line, offset): (string, integer)
         The line with the current cursor, and the character offset of the start of the line.
     """
     offset = 0
     lines = cell.splitlines(True)
     for line in lines:
         next_offset = offset + len(line)
         if not line.endswith('\n'):
             # If the last line doesn't have a trailing newline, treat it as if
             # it does so that the cursor at the end of the line still counts
             # as being on that line.
             next_offset += 1
         if next_offset > cursor_pos:
             break
         offset = next_offset
     else:
         line = ""
     return (line, offset)
 
 def token_at_cursor(cell, cursor_pos=0):
     """Get the token at a given cursor
-    
+
     Used for introspection.
-    
+
     Function calls are prioritized, so the token for the callable will be returned
     if the cursor is anywhere inside the call.
-    
+
     Parameters
     ----------
-    
     cell : unicode
         A block of Python code
     cursor_pos : int
         The location of the cursor in the block where the token should be found
     """
     names = []
     tokens = []
     call_names = []
     
     offsets = {1: 0} # lines start at 1
     for tup in generate_tokens(StringIO(cell).readline):
         
         tok = Token(*tup)
         
         # token, text, start, end, line = tup
         start_line, start_col = tok.start
         end_line, end_col = tok.end
         if end_line + 1 not in offsets:
             # keep track of offsets for each line
             lines = tok.line.splitlines(True)
             for lineno, line in enumerate(lines, start_line + 1):
                 if lineno not in offsets:
                     offsets[lineno] = offsets[lineno-1] + len(line)
         
         offset = offsets[start_line]
         # allow '|foo' to find 'foo' at the beginning of a line
         boundary = cursor_pos + 1 if start_col == 0 else cursor_pos
         if offset + start_col >= boundary:
             # current token starts after the cursor,
             # don't consume it
             break
         
         if tok.token == tokenize.NAME and not iskeyword(tok.text):
             if names and tokens and tokens[-1].token == tokenize.OP and tokens[-1].text == '.':
                 names[-1] = "%s.%s" % (names[-1], tok.text)
             else:
                 names.append(tok.text)
         elif tok.token == tokenize.OP:
             if tok.text == '=' and names:
                 # don't inspect the lhs of an assignment
                 names.pop(-1)
             if tok.text == '(' and names:
                 # if we are inside a function call, inspect the function
                 call_names.append(names[-1])
             elif tok.text == ')' and call_names:
                 call_names.pop(-1)
         
         tokens.append(tok)
         
         if offsets[end_line] + end_col > cursor_pos:
             # we found the cursor, stop reading
             break
         
     if call_names:
         return call_names[-1]
     elif names:
         return names[-1]
     else:
         return ''