fix types

IPython/utils/_process_common.py

@@ -1,217 +1,218
 """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 os
 import shlex
+import subprocess
 import sys
-import os
-from typing import Callable, Optional, Union, List
+from typing import IO, Any, Callable, List, Union
+
 from IPython.utils import py3compat
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
-def read_no_interrupt(p: subprocess.Popen) -> str:
+def read_no_interrupt(stream: IO[Any]) -> bytes:
     """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()
+        return stream.read()
     except IOError as err:
         if err.errno != errno.EINTR:
             raise
 
 
 def process_handler(
     cmd: Union[str, List[str]],
     callback: Callable[[subprocess.Popen], int],
     stderr=subprocess.PIPE,
 ) -> int:
     """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.
     callback : callable
         A one-argument function that will be called with the Popen object.
     stderr : file descriptor number, optional
         By default this is set to ``subprocess.PIPE``, but you can also pass the
         value ``subprocess.STDOUT`` to force the subprocess' stderr to go into
         the same file descriptor as its stdout.  This is useful to read stdout
         and stderr combined in the order they are generated.
 
     Returns
     -------
     The return value of the provided callback is returned.
     """
     sys.stdout.flush()
     sys.stderr.flush()
     # On win32, close_fds can't be true when using pipes for stdin/out/err
     if sys.platform == "win32" and stderr != subprocess.PIPE:
         close_fds = False
     else:
         close_fds = True
     # 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.
 
     Returns
     -------
     output : str
         A string containing the combination of stdout and stderr from the
     subprocess, in whatever order the subprocess originally wrote to its
     file descriptors (so the order of the information in this string is the
     correct order as would be seen if running the command in a terminal).
     """
     out = process_handler(cmd, lambda p: p.communicate()[0], subprocess.STDOUT)
     if out is None:
         return ''
     return py3compat.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.
 
     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.
 
     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_win32.py

@@ -1,215 +1,220
 """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 ctypes
 import os
+import subprocess
 import sys
-import ctypes
 import time
-
-from ctypes import c_int, POINTER
-from ctypes.wintypes import LPCWSTR, HLOCAL
-from subprocess import STDOUT, TimeoutExpired
+from ctypes import POINTER, c_int
+from ctypes.wintypes import HLOCAL, LPCWSTR
+from subprocess import STDOUT
 from threading import Thread
-import subprocess
+from types import TracebackType
+from typing import IO, Any, List, Optional
 
-from typing import Optional, List
-import traceback
+from . import py3compat
+from ._process_common import arg_split as py_arg_split
 
 # our own imports
-from ._process_common import read_no_interrupt, process_handler, arg_split as py_arg_split
-from . import py3compat
+from ._process_common import process_handler, read_no_interrupt
 from .encoding import DEFAULT_ENCODING
 
 #-----------------------------------------------------------------------------
 # Function definitions
 #-----------------------------------------------------------------------------
 
 class AvoidUNCPath:
     """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) -> Optional[str]:
         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: Optional[type], exc_value: Optional[BaseException], traceback
+        self, exc_type: Optional[type[BaseException]], exc_value: Optional[BaseException], traceback:TracebackType
     ) -> None:
         if self.is_unc_path:
             os.chdir(self.path)
 
 
 def _system_body(p: subprocess.Popen) -> int:
     """Callback for _system."""
     enc = DEFAULT_ENCODING
 
+    # Dec 2024: in both of these functions, I'm not sure why we .splitlines()
+    # the bytes and then decode each line individually instead of just decoding
+    # the whole thing at once.
     def stdout_read() -> None:
         try:
-            for line in read_no_interrupt(p.stdout).splitlines():
-                line = line.decode(enc, "replace")
+            assert p.stdout is not None
+            for byte_line in read_no_interrupt(p.stdout).splitlines():
+                line = byte_line.decode(enc, "replace")
                 print(line, file=sys.stdout)
         except Exception as e:
             print(f"Error reading stdout: {e}", file=sys.stderr)
 
     def stderr_read() -> None:
         try:
-            for line in read_no_interrupt(p.stderr).splitlines():
-                line = line.decode(enc, "replace")
+            assert p.stderr is not None
+            for byte_line in read_no_interrupt(p.stderr).splitlines():
+                line = byte_line.decode(enc, "replace")
                 print(line, file=sys.stderr)
         except Exception as e:
             print(f"Error reading stderr: {e}", file=sys.stderr)
 
     stdout_thread = Thread(target=stdout_read)
     stderr_thread = Thread(target=stderr_read)
 
     stdout_thread.start()
     stderr_thread.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:
             break
 
     # Join the threads to ensure they complete before returning
     stdout_thread.join()
     stderr_thread.join()
 
     return result
 
 
 def system(cmd: str) -> Optional[int]:
     """Win32 version of os.system() that works with network shares.
 
     Note that this implementation returns None, as meant for use in IPython.
 
     Parameters
     ----------
     cmd : str or list
         A command to be executed in the system shell.
 
     Returns
     -------
     int : child process' exit code.
     """
     # The controller provides interactivity with both
     # stdin and stdout
     #import _process_win32_controller
     #_process_win32_controller.system(cmd)
 
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         return process_handler(cmd, _system_body)
 
 def getoutput(cmd: str) -> str:
     """Return standard output of executing cmd in a shell.
 
     Accepts the same arguments as os.system().
 
     Parameters
     ----------
     cmd : str or list
         A command to be executed in the system shell.
 
     Returns
     -------
     stdout : str
     """
 
     with AvoidUNCPath() as path:
         if path is not None:
             cmd = '"pushd %s &&"%s' % (path, cmd)
         out = process_handler(cmd, lambda p: p.communicate()[0], STDOUT)
 
     if out is None:
         out = b''
     return py3compat.decode(out)
 
 try:
     windll = ctypes.windll  # type: ignore [attr-defined]
     CommandLineToArgvW = windll.shell32.CommandLineToArgvW
     CommandLineToArgvW.arg_types = [LPCWSTR, POINTER(c_int)]
     CommandLineToArgvW.restype = POINTER(LPCWSTR)
     LocalFree = windll.kernel32.LocalFree
     LocalFree.res_type = HLOCAL
     LocalFree.arg_types = [HLOCAL]
 
     def arg_split(
         commandline: str, posix: bool = False, strict: bool = True
     ) -> List[str]:
         """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(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)
             )
             if arg is not None
         ]
-        retval = LocalFree(result_pointer)
+        LocalFree(result_pointer)
         return result
 except AttributeError:
     arg_split = py_arg_split
 
 def check_pid(pid: int) -> bool:
     # OpenProcess returns 0 if no such process (of ours) exists
     # positive int otherwise
     return bool(windll.kernel32.OpenProcess(1, 0, pid))