diff --git a/test-data/stdlib-samples/3.2/subprocess.py b/test-data/stdlib-samples/3.2/subprocess.py deleted file mode 100644 index 32c2ee729791..000000000000 --- a/test-data/stdlib-samples/3.2/subprocess.py +++ /dev/null @@ -1,1705 +0,0 @@ -# subprocess - Subprocesses with accessible I/O streams -# -# For more information about this module, see PEP 324. -# -# Copyright (c) 2003-2005 by Peter Astrand -# -# Licensed to PSF under a Contributor Agreement. -# See http://www.python.org/2.4/license for licensing details. - -r"""subprocess - Subprocesses with accessible I/O streams - -This module allows you to spawn processes, connect to their -input/output/error pipes, and obtain their return codes. This module -intends to replace several other, older modules and functions, like: - -os.system -os.spawn* - -Information about how the subprocess module can be used to replace these -modules and functions can be found below. - - - -Using the subprocess module -=========================== -This module defines one class called Popen: - -class Popen(args, bufsize=0, executable=None, - stdin=None, stdout=None, stderr=None, - preexec_fn=None, close_fds=True, shell=False, - cwd=None, env=None, universal_newlines=False, - startupinfo=None, creationflags=0, - restore_signals=True, start_new_session=False, pass_fds=()): - - -Arguments are: - -args should be a string, or a sequence of program arguments. The -program to execute is normally the first item in the args sequence or -string, but can be explicitly set by using the executable argument. - -On POSIX, with shell=False (default): In this case, the Popen class -uses os.execvp() to execute the child program. args should normally -be a sequence. A string will be treated as a sequence with the string -as the only item (the program to execute). - -On POSIX, with shell=True: If args is a string, it specifies the -command string to execute through the shell. If args is a sequence, -the first item specifies the command string, and any additional items -will be treated as additional shell arguments. - -On Windows: the Popen class uses CreateProcess() to execute the child -program, which operates on strings. If args is a sequence, it will be -converted to a string using the list2cmdline method. Please note that -not all MS Windows applications interpret the command line the same -way: The list2cmdline is designed for applications using the same -rules as the MS C runtime. - -bufsize, if given, has the same meaning as the corresponding argument -to the built-in open() function: 0 means unbuffered, 1 means line -buffered, any other positive value means use a buffer of -(approximately) that size. A negative bufsize means to use the system -default, which usually means fully buffered. The default value for -bufsize is 0 (unbuffered). - -stdin, stdout and stderr specify the executed programs' standard -input, standard output and standard error file handles, respectively. -Valid values are PIPE, an existing file descriptor (a positive -integer), an existing file object, and None. PIPE indicates that a -new pipe to the child should be created. With None, no redirection -will occur; the child's file handles will be inherited from the -parent. Additionally, stderr can be STDOUT, which indicates that the -stderr data from the applications should be captured into the same -file handle as for stdout. - -On POSIX, if preexec_fn is set to a callable object, this object will be -called in the child process just before the child is executed. The use -of preexec_fn is not thread safe, using it in the presence of threads -could lead to a deadlock in the child process before the new executable -is executed. - -If close_fds is true, all file descriptors except 0, 1 and 2 will be -closed before the child process is executed. The default for close_fds -varies by platform: Always true on POSIX. True when stdin/stdout/stderr -are None on Windows, false otherwise. - -pass_fds is an optional sequence of file descriptors to keep open between the -parent and child. Providing any pass_fds implicitly sets close_fds to true. - -if shell is true, the specified command will be executed through the -shell. - -If cwd is not None, the current directory will be changed to cwd -before the child is executed. - -On POSIX, if restore_signals is True all signals that Python sets to -SIG_IGN are restored to SIG_DFL in the child process before the exec. -Currently this includes the SIGPIPE, SIGXFZ and SIGXFSZ signals. This -parameter does nothing on Windows. - -On POSIX, if start_new_session is True, the setsid() system call will be made -in the child process prior to executing the command. - -If env is not None, it defines the environment variables for the new -process. - -If universal_newlines is true, the file objects stdout and stderr are -opened as a text files, but lines may be terminated by any of '\n', -the Unix end-of-line convention, '\r', the old Macintosh convention or -'\r\n', the Windows convention. All of these external representations -are seen as '\n' by the Python program. Note: This feature is only -available if Python is built with universal newline support (the -default). Also, the newlines attribute of the file objects stdout, -stdin and stderr are not updated by the communicate() method. - -The startupinfo and creationflags, if given, will be passed to the -underlying CreateProcess() function. They can specify things such as -appearance of the main window and priority for the new process. -(Windows only) - - -This module also defines some shortcut functions: - -call(*popenargs, **kwargs): - Run command with arguments. Wait for command to complete, then - return the returncode attribute. - - The arguments are the same as for the Popen constructor. Example: - - >>> retcode = subprocess.call(["ls", "-l"]) - -check_call(*popenargs, **kwargs): - Run command with arguments. Wait for command to complete. If the - exit code was zero then return, otherwise raise - CalledProcessError. The CalledProcessError object will have the - return code in the returncode attribute. - - The arguments are the same as for the Popen constructor. Example: - - >>> subprocess.check_call(["ls", "-l"]) - 0 - -getstatusoutput(cmd): - Return (status, output) of executing cmd in a shell. - - Execute the string 'cmd' in a shell with os.popen() and return a 2-tuple - (status, output). cmd is actually run as '{ cmd ; } 2>&1', so that the - returned output will contain output or error messages. A trailing newline - is stripped from the output. The exit status for the command can be - interpreted according to the rules for the C function wait(). Example: - - >>> subprocess.getstatusoutput('ls /bin/ls') - (0, '/bin/ls') - >>> subprocess.getstatusoutput('cat /bin/junk') - (256, 'cat: /bin/junk: No such file or directory') - >>> subprocess.getstatusoutput('/bin/junk') - (256, 'sh: /bin/junk: not found') - -getoutput(cmd): - Return output (stdout or stderr) of executing cmd in a shell. - - Like getstatusoutput(), except the exit status is ignored and the return - value is a string containing the command's output. Example: - - >>> subprocess.getoutput('ls /bin/ls') - '/bin/ls' - -check_output(*popenargs, **kwargs): - Run command with arguments and return its output as a byte string. - - If the exit code was non-zero it raises a CalledProcessError. The - CalledProcessError object will have the return code in the returncode - attribute and output in the output attribute. - - The arguments are the same as for the Popen constructor. Example: - - >>> output = subprocess.check_output(["ls", "-l", "/dev/null"]) - - -Exceptions ----------- -Exceptions raised in the child process, before the new program has -started to execute, will be re-raised in the parent. Additionally, -the exception object will have one extra attribute called -'child_traceback', which is a string containing traceback information -from the childs point of view. - -The most common exception raised is OSError. This occurs, for -example, when trying to execute a non-existent file. Applications -should prepare for OSErrors. - -A ValueError will be raised if Popen is called with invalid arguments. - -check_call() and check_output() will raise CalledProcessError, if the -called process returns a non-zero return code. - - -Security --------- -Unlike some other popen functions, this implementation will never call -/bin/sh implicitly. This means that all characters, including shell -metacharacters, can safely be passed to child processes. - - -Popen objects -============= -Instances of the Popen class have the following methods: - -poll() - Check if child process has terminated. Returns returncode - attribute. - -wait() - Wait for child process to terminate. Returns returncode attribute. - -communicate(input=None) - Interact with process: Send data to stdin. Read data from stdout - and stderr, until end-of-file is reached. Wait for process to - terminate. The optional input argument should be a string to be - sent to the child process, or None, if no data should be sent to - the child. - - communicate() returns a tuple (stdout, stderr). - - Note: The data read is buffered in memory, so do not use this - method if the data size is large or unlimited. - -The following attributes are also available: - -stdin - If the stdin argument is PIPE, this attribute is a file object - that provides input to the child process. Otherwise, it is None. - -stdout - If the stdout argument is PIPE, this attribute is a file object - that provides output from the child process. Otherwise, it is - None. - -stderr - If the stderr argument is PIPE, this attribute is file object that - provides error output from the child process. Otherwise, it is - None. - -pid - The process ID of the child process. - -returncode - The child return code. A None value indicates that the process - hasn't terminated yet. A negative value -N indicates that the - child was terminated by signal N (POSIX only). - - -Replacing older functions with the subprocess module -==================================================== -In this section, "a ==> b" means that b can be used as a replacement -for a. - -Note: All functions in this section fail (more or less) silently if -the executed program cannot be found; this module raises an OSError -exception. - -In the following examples, we assume that the subprocess module is -imported with "from subprocess import *". - - -Replacing /bin/sh shell backquote ---------------------------------- -output=`mycmd myarg` -==> -output = Popen(["mycmd", "myarg"], stdout=PIPE).communicate()[0] - - -Replacing shell pipe line -------------------------- -output=`dmesg | grep hda` -==> -p1 = Popen(["dmesg"], stdout=PIPE) -p2 = Popen(["grep", "hda"], stdin=p1.stdout, stdout=PIPE) -output = p2.communicate()[0] - - -Replacing os.system() ---------------------- -sts = os.system("mycmd" + " myarg") -==> -p = Popen("mycmd" + " myarg", shell=True) -pid, sts = os.waitpid(p.pid, 0) - -Note: - -* Calling the program through the shell is usually not required. - -* It's easier to look at the returncode attribute than the - exitstatus. - -A more real-world example would look like this: - -try: - retcode = call("mycmd" + " myarg", shell=True) - if retcode < 0: - print("Child was terminated by signal", -retcode, file=sys.stderr) - else: - print("Child returned", retcode, file=sys.stderr) -except OSError as e: - print("Execution failed:", e, file=sys.stderr) - - -Replacing os.spawn* -------------------- -P_NOWAIT example: - -pid = os.spawnlp(os.P_NOWAIT, "/bin/mycmd", "mycmd", "myarg") -==> -pid = Popen(["/bin/mycmd", "myarg"]).pid - - -P_WAIT example: - -retcode = os.spawnlp(os.P_WAIT, "/bin/mycmd", "mycmd", "myarg") -==> -retcode = call(["/bin/mycmd", "myarg"]) - - -Vector example: - -os.spawnvp(os.P_NOWAIT, path, args) -==> -Popen([path] + args[1:]) - - -Environment example: - -os.spawnlpe(os.P_NOWAIT, "/bin/mycmd", "mycmd", "myarg", env) -==> -Popen(["/bin/mycmd", "myarg"], env={"PATH": "/usr/bin"}) -""" - -import sys -mswindows = (sys.platform == "win32") - -import io -import os -import traceback -import gc -import signal -import builtins -import warnings -import errno - -from typing import ( - Any, Tuple, List, Sequence, Callable, Mapping, cast, Set, Dict, IO, - TextIO, AnyStr -) -from typing_extensions import Literal -from types import TracebackType - -# Exception classes used by this module. -class CalledProcessError(Exception): - """This exception is raised when a process run by check_call() or - check_output() returns a non-zero exit status. - The exit status will be stored in the returncode attribute; - check_output() will also store the output in the output attribute. - """ - def __init__(self, returncode: int, cmd: str, output: Any = None) -> None: - self.returncode = returncode - self.cmd = cmd - self.output = output - def __str__(self) -> str: - return "Command '%s' returned non-zero exit status %d" % (self.cmd, self.returncode) - - -if mswindows: - import threading - import msvcrt - import _subprocess - class STARTUPINFO: - dwFlags = 0 - hStdInput = cast(Any, None) - hStdOutput = cast(Any, None) - hStdError = cast(Any, None) - wShowWindow = 0 - class pywintypes: - error = IOError -else: - import select - _has_poll = hasattr(select, 'poll') - import fcntl - import pickle - - try: - import _posixsubprocess - have_posixsubprocess = True - except ImportError: - have_posixsubprocess = False - warnings.warn("The _posixsubprocess module is not being used. " - "Child process reliability may suffer if your " - "program uses threads.", RuntimeWarning) - - # When select or poll has indicated that the file is writable, - # we can write up to _PIPE_BUF bytes without risk of blocking. - # POSIX defines PIPE_BUF as >= 512. - _PIPE_BUF = getattr(select, 'PIPE_BUF', 512) # type: int - - _FD_CLOEXEC = getattr(fcntl, 'FD_CLOEXEC', 1) # type: int - - def _set_cloexec(fd: int, cloexec: bool) -> None: - old = fcntl.fcntl(fd, fcntl.F_GETFD) - if cloexec: - fcntl.fcntl(fd, fcntl.F_SETFD, old | _FD_CLOEXEC) - else: - fcntl.fcntl(fd, fcntl.F_SETFD, old & ~_FD_CLOEXEC) - - if have_posixsubprocess: - _create_pipe = _posixsubprocess.cloexec_pipe - else: - def __create_pipe() -> Tuple[int, int]: - fds = os.pipe() - _set_cloexec(fds[0], True) - _set_cloexec(fds[1], True) - return fds - _create_pipe = __create_pipe - -__all__ = ["Popen", "PIPE", "STDOUT", "call", "check_call", "getstatusoutput", - "getoutput", "check_output", "CalledProcessError"] - -if mswindows: - from _subprocess import (CREATE_NEW_CONSOLE, CREATE_NEW_PROCESS_GROUP, - STD_INPUT_HANDLE, STD_OUTPUT_HANDLE, - STD_ERROR_HANDLE, SW_HIDE, - STARTF_USESTDHANDLES, STARTF_USESHOWWINDOW) - - __all__.extend(["CREATE_NEW_CONSOLE", "CREATE_NEW_PROCESS_GROUP", - "STD_INPUT_HANDLE", "STD_OUTPUT_HANDLE", - "STD_ERROR_HANDLE", "SW_HIDE", - "STARTF_USESTDHANDLES", "STARTF_USESHOWWINDOW"]) -try: - MAXFD = os.sysconf("SC_OPEN_MAX") -except: - MAXFD = 256 - -# This lists holds Popen instances for which the underlying process had not -# exited at the time its __del__ method got called: those processes are wait()ed -# for synchronously from _cleanup() when a new Popen object is created, to avoid -# zombie processes. -_active = [] # type: List[Popen] - -def _cleanup() -> None: - for inst in _active[:]: - res = inst._internal_poll(_deadstate=sys.maxsize) - if res is not None: - try: - _active.remove(inst) - except ValueError: - # This can happen if two threads create a new Popen instance. - # It's harmless that it was already removed, so ignore. - pass - -PIPE = -1 -STDOUT = -2 - - -def _eintr_retry_call(func: Any, *args: Any) -> Any: - while True: - try: - return func(*args) - except (OSError, IOError) as e: - if e.errno == errno.EINTR: - continue - raise - - -def call(*popenargs: Any, **kwargs: Any) -> int: - """Run command with arguments. Wait for command to complete, then - return the returncode attribute. - - The arguments are the same as for the Popen constructor. Example: - - retcode = call(["ls", "-l"]) - """ - return Popen(*popenargs, **kwargs).wait() - - -def check_call(*popenargs: Any, **kwargs: Any) -> int: - """Run command with arguments. Wait for command to complete. If - the exit code was zero then return, otherwise raise - CalledProcessError. The CalledProcessError object will have the - return code in the returncode attribute. - - The arguments are the same as for the Popen constructor. Example: - - check_call(["ls", "-l"]) - """ - retcode = call(*popenargs, **kwargs) - if retcode: - cmd = kwargs.get("args") - if cmd is None: - cmd = popenargs[0] - raise CalledProcessError(retcode, cmd) - return 0 - - -def check_output(*popenargs: Any, **kwargs: Any) -> bytes: - r"""Run command with arguments and return its output as a byte string. - - If the exit code was non-zero it raises a CalledProcessError. The - CalledProcessError object will have the return code in the returncode - attribute and output in the output attribute. - - The arguments are the same as for the Popen constructor. Example: - - >>> check_output(["ls", "-l", "/dev/null"]) - b'crw-rw-rw- 1 root root 1, 3 Oct 18 2007 /dev/null\n' - - The stdout argument is not allowed as it is used internally. - To capture standard error in the result, use stderr=STDOUT. - - >>> check_output(["/bin/sh", "-c", - ... "ls -l non_existent_file ; exit 0"], - ... stderr=STDOUT) - b'ls: non_existent_file: No such file or directory\n' - """ - if 'stdout' in kwargs: - raise ValueError('stdout argument not allowed, it will be overridden.') - kwargs['stdout'] = PIPE - process = Popen(*popenargs, **kwargs) - output, unused_err = process.communicate() - retcode = process.poll() - if retcode: - cmd = kwargs.get("args") - if cmd is None: - cmd = popenargs[0] - raise CalledProcessError(retcode, cmd, output=output) - return output - - -def list2cmdline(seq: Sequence[str]) -> str: - """ - Translate a sequence of arguments into a command line - string, using the same rules as the MS C runtime: - - 1) Arguments are delimited by white space, which is either a - space or a tab. - - 2) A string surrounded by double quotation marks is - interpreted as a single argument, regardless of white space - contained within. A quoted string can be embedded in an - argument. - - 3) A double quotation mark preceded by a backslash is - interpreted as a literal double quotation mark. - - 4) Backslashes are interpreted literally, unless they - immediately precede a double quotation mark. - - 5) If backslashes immediately precede a double quotation mark, - every pair of backslashes is interpreted as a literal - backslash. If the number of backslashes is odd, the last - backslash escapes the next double quotation mark as - described in rule 3. - """ - - # See - # http://msdn.microsoft.com/en-us/library/17w5ykft.aspx - # or search http://msdn.microsoft.com for - # "Parsing C++ Command-Line Arguments" - result = [] # type: List[str] - needquote = False - for arg in seq: - bs_buf = [] # type: List[str] - - # Add a space to separate this argument from the others - if result: - result.append(' ') - - needquote = (" " in arg) or ("\t" in arg) or not arg - if needquote: - result.append('"') - - for c in arg: - if c == '\\': - # Don't know if we need to double yet. - bs_buf.append(c) - elif c == '"': - # Double backslashes. - result.append('\\' * len(bs_buf)*2) - bs_buf = [] - result.append('\\"') - else: - # Normal char - if bs_buf: - result.extend(bs_buf) - bs_buf = [] - result.append(c) - - # Add remaining backslashes, if any. - if bs_buf: - result.extend(bs_buf) - - if needquote: - result.extend(bs_buf) - result.append('"') - - return ''.join(result) - - -# Various tools for executing commands and looking at their output and status. -# -# NB This only works (and is only relevant) for POSIX. - -def getstatusoutput(cmd: str) -> Tuple[int, str]: - """Return (status, output) of executing cmd in a shell. - - Execute the string 'cmd' in a shell with os.popen() and return a 2-tuple - (status, output). cmd is actually run as '{ cmd ; } 2>&1', so that the - returned output will contain output or error messages. A trailing newline - is stripped from the output. The exit status for the command can be - interpreted according to the rules for the C function wait(). Example: - - >>> import subprocess - >>> subprocess.getstatusoutput('ls /bin/ls') - (0, '/bin/ls') - >>> subprocess.getstatusoutput('cat /bin/junk') - (256, 'cat: /bin/junk: No such file or directory') - >>> subprocess.getstatusoutput('/bin/junk') - (256, 'sh: /bin/junk: not found') - """ - pipe = os.popen('{ ' + cmd + '; } 2>&1', 'r') - text = pipe.read() - sts = pipe.close() - if sts is None: sts = 0 - if text[-1:] == '\n': text = text[:-1] - return sts, text - - -def getoutput(cmd: str) -> str: - """Return output (stdout or stderr) of executing cmd in a shell. - - Like getstatusoutput(), except the exit status is ignored and the return - value is a string containing the command's output. Example: - - >>> import subprocess - >>> subprocess.getoutput('ls /bin/ls') - '/bin/ls' - """ - return getstatusoutput(cmd)[1] - - -_PLATFORM_DEFAULT_CLOSE_FDS = object() - - -class Popen(object): - def __init__(self, args: Sequence[Any], bufsize: int = 0, - executable: str = None, stdin: Any = None, - stdout: Any = None, stderr: Any = None, - preexec_fn: Callable[[], Any] = None, - close_fds: Any = _PLATFORM_DEFAULT_CLOSE_FDS, - shell: int = False, cwd: str = None, - env: Mapping[str, str] = None, - universal_newlines: int = False, - startupinfo: 'STARTUPINFO' = None, creationflags: int = 0, - restore_signals: bool = True, start_new_session: bool = False, - pass_fds: Any = ()) -> None: - """Create new Popen instance.""" - _cleanup() - - self._child_created = False - if bufsize is None: - bufsize = 0 # Restore default - if not isinstance(bufsize, int): - raise TypeError("bufsize must be an integer") - - if mswindows: - if preexec_fn is not None: - raise ValueError("preexec_fn is not supported on Windows " - "platforms") - any_stdio_set = (stdin is not None or stdout is not None or - stderr is not None) - if close_fds is _PLATFORM_DEFAULT_CLOSE_FDS: - if any_stdio_set: - close_fds = False - else: - close_fds = True - elif close_fds and any_stdio_set: - raise ValueError( - "close_fds is not supported on Windows platforms" - " if you redirect stdin/stdout/stderr") - else: - # POSIX - if close_fds is _PLATFORM_DEFAULT_CLOSE_FDS: - close_fds = True - if pass_fds and not close_fds: - warnings.warn("pass_fds overriding close_fds.", RuntimeWarning) - close_fds = True - if startupinfo is not None: - raise ValueError("startupinfo is only supported on Windows " - "platforms") - if creationflags != 0: - raise ValueError("creationflags is only supported on Windows " - "platforms") - - self.stdin = None # type: IO[Any] - self.stdout = None # type: IO[Any] - self.stderr = None # type: IO[Any] - self.pid = None # type: int - self.returncode = None # type: int - self.universal_newlines = universal_newlines - - # Input and output objects. The general principle is like - # this: - # - # Parent Child - # ------ ----- - # p2cwrite ---stdin---> p2cread - # c2pread <--stdout--- c2pwrite - # errread <--stderr--- errwrite - # - # On POSIX, the child objects are file descriptors. On - # Windows, these are Windows file handles. The parent objects - # are file descriptors on both platforms. The parent objects - # are -1 when not using PIPEs. The child objects are -1 - # when not redirecting. - - (p2cread, p2cwrite, - c2pread, c2pwrite, - errread, errwrite) = self._get_handles(stdin, stdout, stderr) - - # We wrap OS handles *before* launching the child, otherwise a - # quickly terminating child could make our fds unwrappable - # (see #8458). - - if mswindows: - if p2cwrite != -1: - p2cwrite = msvcrt.open_osfhandle(p2cwrite.Detach(), 0) - if c2pread != -1: - c2pread = msvcrt.open_osfhandle(c2pread.Detach(), 0) - if errread != -1: - errread = msvcrt.open_osfhandle(errread.Detach(), 0) - - if p2cwrite != -1: - self.stdin = io.open(p2cwrite, 'wb', bufsize) - if self.universal_newlines: - self.stdin = io.TextIOWrapper(self.stdin, write_through=True) - if c2pread != -1: - self.stdout = io.open(c2pread, 'rb', bufsize) - if universal_newlines: - self.stdout = io.TextIOWrapper(self.stdout) - if errread != -1: - self.stderr = io.open(errread, 'rb', bufsize) - if universal_newlines: - self.stderr = io.TextIOWrapper(self.stderr) - - try: - self._execute_child(args, executable, preexec_fn, close_fds, - pass_fds, cwd, env, universal_newlines, - startupinfo, creationflags, shell, - p2cread, p2cwrite, - c2pread, c2pwrite, - errread, errwrite, - restore_signals, start_new_session) - except: - # Cleanup if the child failed starting - for f in filter(None, [self.stdin, self.stdout, self.stderr]): - try: - f.close() - except EnvironmentError: - # Ignore EBADF or other errors - pass - raise - - - def _translate_newlines(self, data: bytes, encoding: str) -> str: - data = data.replace(b"\r\n", b"\n").replace(b"\r", b"\n") - return data.decode(encoding) - - def __enter__(self) -> 'Popen': - return self - - def __exit__(self, type: type, value: BaseException, - traceback: TracebackType) -> Literal[False]: - if self.stdout: - self.stdout.close() - if self.stderr: - self.stderr.close() - if self.stdin: - self.stdin.close() - # Wait for the process to terminate, to avoid zombies. - self.wait() - return False - - def __del__(self, _maxsize: int = sys.maxsize, - _active: List['Popen'] = _active) -> None: - # If __init__ hasn't had a chance to execute (e.g. if it - # was passed an undeclared keyword argument), we don't - # have a _child_created attribute at all. - if not getattr(self, '_child_created', False): - # We didn't get to successfully create a child process. - return - # In case the child hasn't been waited on, check if it's done. - self._internal_poll(_deadstate=_maxsize) - if self.returncode is None and _active is not None: - # Child is still running, keep us alive until we can wait on it. - _active.append(self) - - - def communicate(self, input: Any = None) -> Tuple[Any, Any]: - """Interact with process: Send data to stdin. Read data from - stdout and stderr, until end-of-file is reached. Wait for - process to terminate. The optional input argument should be a - string to be sent to the child process, or None, if no data - should be sent to the child. - - communicate() returns a tuple (stdout, stderr).""" - - # Optimization: If we are only using one pipe, or no pipe at - # all, using select() or threads is unnecessary. - if [self.stdin, self.stdout, self.stderr].count(None) >= 2: - stdout = None # type: IO[Any] - stderr = None # type: IO[Any] - if self.stdin: - if input: - try: - self.stdin.write(input) - except IOError as e: - if e.errno != errno.EPIPE and e.errno != errno.EINVAL: - raise - self.stdin.close() - elif self.stdout: - stdout = _eintr_retry_call(self.stdout.read) - self.stdout.close() - elif self.stderr: - stderr = _eintr_retry_call(self.stderr.read) - self.stderr.close() - self.wait() - return (stdout, stderr) - - return self._communicate(input) - - - def poll(self) -> int: - return self._internal_poll() - - - if mswindows: - # - # Windows methods - # - def _get_handles(self, stdin: Any, stdout: Any, - stderr: Any) -> Tuple[Any, Any, Any, Any, Any, Any]: - """Construct and return tuple with IO objects: - p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite - """ - if stdin is None and stdout is None and stderr is None: - return (-1, -1, -1, -1, -1, -1) - - p2cread, p2cwrite = -1, -1 # type: (Any, Any) - c2pread, c2pwrite = -1, -1 # type: (Any, Any) - errread, errwrite = -1, -1 # type: (Any, Any) - - if stdin is None: - p2cread = _subprocess.GetStdHandle(_subprocess.STD_INPUT_HANDLE) - if p2cread is None: - p2cread, _ = _subprocess.CreatePipe(None, 0) - elif stdin == PIPE: - p2cread, p2cwrite = _subprocess.CreatePipe(None, 0) - elif isinstance(stdin, int): - p2cread = msvcrt.get_osfhandle(stdin) - else: - # Assuming file-like object - p2cread = msvcrt.get_osfhandle(stdin.fileno()) - p2cread = self._make_inheritable(p2cread) - - if stdout is None: - c2pwrite = _subprocess.GetStdHandle(_subprocess.STD_OUTPUT_HANDLE) - if c2pwrite is None: - _, c2pwrite = _subprocess.CreatePipe(None, 0) - elif stdout == PIPE: - c2pread, c2pwrite = _subprocess.CreatePipe(None, 0) - elif isinstance(stdout, int): - c2pwrite = msvcrt.get_osfhandle(stdout) - else: - # Assuming file-like object - c2pwrite = msvcrt.get_osfhandle(stdout.fileno()) - c2pwrite = self._make_inheritable(c2pwrite) - - if stderr is None: - errwrite = _subprocess.GetStdHandle(_subprocess.STD_ERROR_HANDLE) - if errwrite is None: - _, errwrite = _subprocess.CreatePipe(None, 0) - elif stderr == PIPE: - errread, errwrite = _subprocess.CreatePipe(None, 0) - elif stderr == STDOUT: - errwrite = c2pwrite - elif isinstance(stderr, int): - errwrite = msvcrt.get_osfhandle(stderr) - else: - # Assuming file-like object - errwrite = msvcrt.get_osfhandle(stderr.fileno()) - errwrite = self._make_inheritable(errwrite) - - return (p2cread, p2cwrite, - c2pread, c2pwrite, - errread, errwrite) - - - def _make_inheritable(self, handle: _subprocess.Handle) -> int: - """Return a duplicate of handle, which is inheritable""" - return _subprocess.DuplicateHandle(_subprocess.GetCurrentProcess(), - handle, _subprocess.GetCurrentProcess(), 0, 1, - _subprocess.DUPLICATE_SAME_ACCESS) - - - def _find_w9xpopen(self) -> str: - """Find and return absolut path to w9xpopen.exe""" - w9xpopen = os.path.join( - os.path.dirname(_subprocess.GetModuleFileName(0)), - "w9xpopen.exe") - if not os.path.exists(w9xpopen): - # Eeek - file-not-found - possibly an embedding - # situation - see if we can locate it in sys.exec_prefix - w9xpopen = os.path.join(os.path.dirname(sys.exec_prefix), - "w9xpopen.exe") - if not os.path.exists(w9xpopen): - raise RuntimeError("Cannot locate w9xpopen.exe, which is " - "needed for Popen to work with your " - "shell or platform.") - return w9xpopen - - - def _execute_child(self, args: Sequence[str], executable: str, - preexec_fn: Callable[[], Any], close_fds: Any, - pass_fds: Any, cwd: str, env: Mapping[str, str], - universal_newlines: int, - startupinfo: STARTUPINFO, creationflags: int, - shell: int, - p2cread: Any, p2cwrite: Any, - c2pread: Any, c2pwrite: Any, - errread: Any, errwrite: Any, - restore_signals: bool, - start_new_session: bool) -> None: - """Execute program (MS Windows version)""" - - assert not pass_fds, "pass_fds not supported on Windows." - - if not isinstance(args, str): - args = list2cmdline(args) - - # Process startup details - if startupinfo is None: - startupinfo = STARTUPINFO() - if -1 not in (p2cread, c2pwrite, errwrite): - startupinfo.dwFlags |= _subprocess.STARTF_USESTDHANDLES - startupinfo.hStdInput = p2cread - startupinfo.hStdOutput = c2pwrite - startupinfo.hStdError = errwrite - - if shell: - startupinfo.dwFlags |= _subprocess.STARTF_USESHOWWINDOW - startupinfo.wShowWindow = _subprocess.SW_HIDE - comspec = os.environ.get("COMSPEC", "cmd.exe") - args = '{} /c "{}"'.format (comspec, args) - if (_subprocess.GetVersion() >= 0x80000000 or - os.path.basename(comspec).lower() == "command.com"): - # Win9x, or using command.com on NT. We need to - # use the w9xpopen intermediate program. For more - # information, see KB Q150956 - # (http://web.archive.org/web/20011105084002/http://support.microsoft.com/support/kb/articles/Q150/9/56.asp) - w9xpopen = self._find_w9xpopen() - args = '"%s" %s' % (w9xpopen, args) - # Not passing CREATE_NEW_CONSOLE has been known to - # cause random failures on win9x. Specifically a - # dialog: "Your program accessed mem currently in - # use at xxx" and a hopeful warning about the - # stability of your system. Cost is Ctrl+C won't - # kill children. - creationflags |= _subprocess.CREATE_NEW_CONSOLE - - # Start the process - try: - hp, ht, pid, tid = _subprocess.CreateProcess(executable, - cast(str, args), - # no special security - None, None, - int(not close_fds), - creationflags, - env, - cwd, - startupinfo) - except pywintypes.error as e: - # Translate pywintypes.error to WindowsError, which is - # a subclass of OSError. FIXME: We should really - # translate errno using _sys_errlist (or similar), but - # how can this be done from Python? - raise WindowsError(*e.args) - finally: - # Child is launched. Close the parent's copy of those pipe - # handles that only the child should have open. You need - # to make sure that no handles to the write end of the - # output pipe are maintained in this process or else the - # pipe will not close when the child process exits and the - # ReadFile will hang. - if p2cread != -1: - p2cread.Close() - if c2pwrite != -1: - c2pwrite.Close() - if errwrite != -1: - errwrite.Close() - - # Retain the process handle, but close the thread handle - self._child_created = True - self._handle = hp - self.pid = pid - ht.Close() - - def _internal_poll(self, _deadstate: int = None) -> int: - """Check if child process has terminated. Returns returncode - attribute. - - This method is called by __del__, so it can only refer to objects - in its local scope. - - """ - return self._internal_poll_win(_deadstate) - - from _subprocess import Handle - - def _internal_poll_win(self, _deadstate: int = None, - _WaitForSingleObject: Callable[[Handle, int], int] = - _subprocess.WaitForSingleObject, - _WAIT_OBJECT_0: int = _subprocess.WAIT_OBJECT_0, - _GetExitCodeProcess: Callable[[Handle], int] = - _subprocess.GetExitCodeProcess) -> int: - if self.returncode is None: - if _WaitForSingleObject(self._handle, 0) == _WAIT_OBJECT_0: - self.returncode = _GetExitCodeProcess(self._handle) - return self.returncode - - - def wait(self) -> int: - """Wait for child process to terminate. Returns returncode - attribute.""" - if self.returncode is None: - _subprocess.WaitForSingleObject(self._handle, - _subprocess.INFINITE) - self.returncode = _subprocess.GetExitCodeProcess(self._handle) - return self.returncode - - - def _readerthread(self, fh: IO[AnyStr], buffer: List[AnyStr]) -> None: - buffer.append(fh.read()) - fh.close() - - - def _communicate(self, input: Any) -> Tuple[Any, Any]: - stdout = cast(Any, None) # Return - stderr = cast(Any, None) # Return - - if self.stdout: - stdout = [] - stdout_thread = threading.Thread(target=self._readerthread, - args=(self.stdout, stdout)) - stdout_thread.daemon = True - stdout_thread.start() - if self.stderr: - stderr = [] - stderr_thread = threading.Thread(target=self._readerthread, - args=(self.stderr, stderr)) - stderr_thread.daemon = True - stderr_thread.start() - - if self.stdin: - if input is not None: - try: - self.stdin.write(input) - except IOError as e: - if e.errno != errno.EPIPE: - raise - self.stdin.close() - - if self.stdout: - stdout_thread.join() - if self.stderr: - stderr_thread.join() - - # All data exchanged. Translate lists into strings. - if stdout is not None: - stdout = stdout[0] - if stderr is not None: - stderr = stderr[0] - - self.wait() - return (stdout, stderr) - - def send_signal(self, sig: int) -> None: - """Send a signal to the process - """ - if sig == signal.SIGTERM: - self.terminate() - elif sig == signal.CTRL_C_EVENT: - os.kill(self.pid, signal.CTRL_C_EVENT) - elif sig == signal.CTRL_BREAK_EVENT: - os.kill(self.pid, signal.CTRL_BREAK_EVENT) - else: - raise ValueError("Unsupported signal: {}".format(sig)) - - def terminate(self) -> None: - """Terminates the process - """ - _subprocess.TerminateProcess(self._handle, 1) - - def kill(self) -> None: - """Terminates the process - """ - self.terminate() - - else: - # - # POSIX methods - # - def _get_handles(self, stdin: Any, stdout: Any, - stderr: Any) -> Tuple[Any, Any, Any, Any, Any, Any]: - """Construct and return tuple with IO objects: - p2cread, p2cwrite, c2pread, c2pwrite, errread, errwrite - """ - p2cread, p2cwrite = -1, -1 - c2pread, c2pwrite = -1, -1 - errread, errwrite = -1, -1 - - if stdin is None: - pass - elif stdin == PIPE: - p2cread, p2cwrite = _create_pipe() - elif isinstance(stdin, int): - p2cread = stdin - else: - # Assuming file-like object - p2cread = stdin.fileno() - - if stdout is None: - pass - elif stdout == PIPE: - c2pread, c2pwrite = _create_pipe() - elif isinstance(stdout, int): - c2pwrite = stdout - else: - # Assuming file-like object - c2pwrite = stdout.fileno() - - if stderr is None: - pass - elif stderr == PIPE: - errread, errwrite = _create_pipe() - elif stderr == STDOUT: - errwrite = c2pwrite - elif isinstance(stderr, int): - errwrite = stderr - else: - # Assuming file-like object - errwrite = stderr.fileno() - - return (p2cread, p2cwrite, - c2pread, c2pwrite, - errread, errwrite) - - - def _close_fds(self, fds_to_keep: Set[int]) -> None: - start_fd = 3 - for fd in sorted(fds_to_keep): - if fd >= start_fd: - os.closerange(start_fd, fd) - start_fd = fd + 1 - if start_fd <= MAXFD: - os.closerange(start_fd, MAXFD) - - - def _execute_child(self, args: Sequence[str], executable: str, - preexec_fn: Callable[[], Any], close_fds: Any, - pass_fds: Any, cwd: str, env: Mapping[str, str], - universal_newlines: int, - startupinfo: 'STARTUPINFO', creationflags: int, - shell: int, - p2cread: Any, p2cwrite: Any, - c2pread: Any, c2pwrite: Any, - errread: Any, errwrite: Any, - restore_signals: bool, - start_new_session: bool) -> None: - """Execute program (POSIX version)""" - - if isinstance(args, str): - args = [args] - else: - args = list(args) - - if shell: - args = ["/bin/sh", "-c"] + args - if executable: - args[0] = executable - - if executable is None: - executable = args[0] - - # For transferring possible exec failure from child to parent. - # Data format: "exception name:hex errno:description" - # Pickle is not used; it is complex and involves memory allocation. - errpipe_read, errpipe_write = _create_pipe() - try: - try: - - if have_posixsubprocess: - # We must avoid complex work that could involve - # malloc or free in the child process to avoid - # potential deadlocks, thus we do all this here. - # and pass it to fork_exec() - - if env is not None: - env_list = [os.fsencode(k) + b'=' + os.fsencode(v) - for k, v in env.items()] - else: - env_list = None # Use execv instead of execve. - executable_enc = os.fsencode(executable) - if os.path.dirname(executable_enc): - executable_list = (executable_enc,) # type: tuple - else: - # This matches the behavior of os._execvpe(). - executable_list = tuple( - os.path.join(os.fsencode(dir), executable_enc) - for dir in os.get_exec_path(env)) - fds_to_keep = set(pass_fds) - fds_to_keep.add(errpipe_write) - self.pid = _posixsubprocess.fork_exec( - args, executable_list, - close_fds, sorted(fds_to_keep), cwd, env_list, - p2cread, p2cwrite, c2pread, c2pwrite, - errread, errwrite, - errpipe_read, errpipe_write, - restore_signals, start_new_session, preexec_fn) - self._child_created = True - else: - # Pure Python implementation: It is not thread safe. - # This implementation may deadlock in the child if your - # parent process has any other threads running. - - gc_was_enabled = gc.isenabled() - # Disable gc to avoid bug where gc -> file_dealloc -> - # write to stderr -> hang. See issue1336 - gc.disable() - try: - self.pid = os.fork() - except: - if gc_was_enabled: - gc.enable() - raise - self._child_created = True - if self.pid == 0: - # Child - try: - # Close parent's pipe ends - if p2cwrite != -1: - os.close(p2cwrite) - if c2pread != -1: - os.close(c2pread) - if errread != -1: - os.close(errread) - os.close(errpipe_read) - - # When duping fds, if there arises a situation - # where one of the fds is either 0, 1 or 2, it - # is possible that it is overwritten (#12607). - if c2pwrite == 0: - c2pwrite = os.dup(c2pwrite) - if errwrite == 0 or errwrite == 1: - errwrite = os.dup(errwrite) - - # Dup fds for child - def _dup2(a: int, b: int) -> None: - # dup2() removes the CLOEXEC flag but - # we must do it ourselves if dup2() - # would be a no-op (issue #10806). - if a == b: - _set_cloexec(a, False) - elif a != -1: - os.dup2(a, b) - _dup2(p2cread, 0) - _dup2(c2pwrite, 1) - _dup2(errwrite, 2) - - # Close pipe fds. Make sure we don't close the - # same fd more than once, or standard fds. - closed = set() # type: Set[int] - for fd in [p2cread, c2pwrite, errwrite]: - if fd > 2 and fd not in closed: - os.close(fd) - closed.add(fd) - - # Close all other fds, if asked for - if close_fds: - fds_to_keep = set(pass_fds) - fds_to_keep.add(errpipe_write) - self._close_fds(fds_to_keep) - - - if cwd is not None: - os.chdir(cwd) - - # This is a copy of Python/pythonrun.c - # _Py_RestoreSignals(). If that were exposed - # as a sys._py_restoresignals func it would be - # better.. but this pure python implementation - # isn't likely to be used much anymore. - if restore_signals: - signals = ('SIGPIPE', 'SIGXFZ', 'SIGXFSZ') - for sig in signals: - if hasattr(signal, sig): - signal.signal(getattr(signal, sig), - signal.SIG_DFL) - - if start_new_session and hasattr(os, 'setsid'): - os.setsid() - - if preexec_fn: - preexec_fn() - - if env is None: - os.execvp(executable, args) - else: - os.execvpe(executable, args, env) - - except: - try: - exc_type, exc_value = sys.exc_info()[:2] - if isinstance(exc_value, OSError): - errno_num = exc_value.errno - else: - errno_num = 0 - message = '%s:%x:%s' % (exc_type.__name__, - errno_num, exc_value) - messageb = message.encode(errors="surrogatepass") - os.write(errpipe_write, messageb) - except Exception: - # We MUST not allow anything odd happening - # above to prevent us from exiting below. - pass - - # This exitcode won't be reported to applications - # so it really doesn't matter what we return. - os._exit(255) - - # Parent - if gc_was_enabled: - gc.enable() - finally: - # be sure the FD is closed no matter what - os.close(errpipe_write) - - if p2cread != -1 and p2cwrite != -1: - os.close(p2cread) - if c2pwrite != -1 and c2pread != -1: - os.close(c2pwrite) - if errwrite != -1 and errread != -1: - os.close(errwrite) - - # Wait for exec to fail or succeed; possibly raising an - # exception (limited in size) - data = bytearray() - while True: - part = _eintr_retry_call(os.read, errpipe_read, 50000) - data += part - if not part or len(data) > 50000: - break - finally: - # be sure the FD is closed no matter what - os.close(errpipe_read) - - if data: - try: - _eintr_retry_call(os.waitpid, self.pid, 0) - except OSError as e: - if e.errno != errno.ECHILD: - raise - try: - (exception_name, hex_errno, - err_msg_b) = bytes(data).split(b':', 2) - except ValueError: - print('Bad exception data:', repr(data)) - exception_name = b'RuntimeError' - hex_errno = b'0' - err_msg_b = b'Unknown' - child_exception_type = getattr( - builtins, exception_name.decode('ascii'), - RuntimeError) - for fd in (p2cwrite, c2pread, errread): - if fd != -1: - os.close(fd) - err_msg = err_msg_b.decode(errors="surrogatepass") - if issubclass(child_exception_type, OSError) and hex_errno: - errno_num = int(hex_errno, 16) - if errno_num != 0: - err_msg = os.strerror(errno_num) - if errno_num == errno.ENOENT: - err_msg += ': ' + repr(args[0]) - raise child_exception_type(errno_num, err_msg) - raise child_exception_type(err_msg) - - - def _handle_exitstatus( - self, sts: int, - _WIFSIGNALED: Callable[[int], bool] = os.WIFSIGNALED, - _WTERMSIG: Callable[[int], int] = os.WTERMSIG, - _WIFEXITED: Callable[[int], bool] = os.WIFEXITED, - _WEXITSTATUS: Callable[[int], int] = os.WEXITSTATUS) -> None: - # This method is called (indirectly) by __del__, so it cannot - # refer to anything outside of its local scope.""" - if _WIFSIGNALED(sts): - self.returncode = -_WTERMSIG(sts) - elif _WIFEXITED(sts): - self.returncode = _WEXITSTATUS(sts) - else: - # Should never happen - raise RuntimeError("Unknown child exit status!") - - - def _internal_poll(self, _deadstate: int = None) -> int: - """Check if child process has terminated. Returns returncode - attribute. - - This method is called by __del__, so it cannot reference anything - outside of the local scope (nor can any methods it calls). - - """ - return self._internal_poll_posix(_deadstate) - - def _internal_poll_posix(self, _deadstate: int = None, - _waitpid: Callable[[int, int], - Tuple[int, int]] = os.waitpid, - _WNOHANG: int = os.WNOHANG, - _os_error: Any = os.error) -> int: - if self.returncode is None: - try: - pid, sts = _waitpid(self.pid, _WNOHANG) - if pid == self.pid: - self._handle_exitstatus(sts) - except _os_error: - if _deadstate is not None: - self.returncode = _deadstate - return self.returncode - - - def wait(self) -> int: - """Wait for child process to terminate. Returns returncode - attribute.""" - if self.returncode is None: - try: - pid, sts = _eintr_retry_call(os.waitpid, self.pid, 0) - except OSError as e: - if e.errno != errno.ECHILD: - raise - # This happens if SIGCLD is set to be ignored or waiting - # for child processes has otherwise been disabled for our - # process. This child is dead, we can't get the status. - sts = 0 - self._handle_exitstatus(sts) - return self.returncode - - - def _communicate(self, input: Any) -> Tuple[Any, Any]: - if self.stdin: - # Flush stdio buffer. This might block, if the user has - # been writing to .stdin in an uncontrolled fashion. - self.stdin.flush() - if not input: - self.stdin.close() - - if _has_poll: - stdout, stderr = self._communicate_with_poll(input) - else: - stdout, stderr = self._communicate_with_select(input) - - # All data exchanged. Translate lists into strings. - if stdout is not None: - stdout2 = b''.join(stdout) - else: - stdout2 = None - if stderr is not None: - stderr2 = b''.join(stderr) - else: - stderr2 = None - - # Translate newlines, if requested. - # This also turns bytes into strings. - stdout3 = cast(Any, stdout2) - stderr3 = cast(Any, stderr2) - if self.universal_newlines: - if stdout is not None: - stdout3 = self._translate_newlines( - stdout2, cast(TextIO, self.stdout).encoding) - if stderr is not None: - stderr3 = self._translate_newlines( - stderr2, cast(TextIO, self.stderr).encoding) - - self.wait() - return (stdout3, stderr3) - - - def _communicate_with_poll(self, input: Any) -> Tuple[List[bytes], - List[bytes]]: - stdout = None # type: List[bytes] # Return - stderr = None # type: List[bytes] # Return - fd2file = {} # type: Dict[int, Any] - fd2output = {} # type: Dict[int, List[bytes]] - - poller = select.poll() - def register_and_append(file_obj: IO[Any], eventmask: int) -> None: - poller.register(file_obj.fileno(), eventmask) - fd2file[file_obj.fileno()] = file_obj - - def close_unregister_and_remove(fd: int) -> None: - poller.unregister(fd) - fd2file[fd].close() - fd2file.pop(fd) - - if self.stdin and input: - register_and_append(self.stdin, select.POLLOUT) - - select_POLLIN_POLLPRI = select.POLLIN | select.POLLPRI - if self.stdout: - register_and_append(self.stdout, select_POLLIN_POLLPRI) - fd2output[self.stdout.fileno()] = stdout = [] - if self.stderr: - register_and_append(self.stderr, select_POLLIN_POLLPRI) - fd2output[self.stderr.fileno()] = stderr = [] - - input_offset = 0 - while fd2file: - try: - ready = poller.poll() - except select.error as e: - if e.args[0] == errno.EINTR: - continue - raise - - # XXX Rewrite these to use non-blocking I/O on the - # file objects; they are no longer using C stdio! - - for fd, mode in ready: - if mode & select.POLLOUT: - chunk = input[input_offset : input_offset + _PIPE_BUF] - try: - input_offset += os.write(fd, chunk) - except OSError as e2: - if e2.errno == errno.EPIPE: - close_unregister_and_remove(fd) - else: - raise - else: - if input_offset >= len(input): - close_unregister_and_remove(fd) - elif mode & select_POLLIN_POLLPRI: - data = os.read(fd, 4096) - if not data: - close_unregister_and_remove(fd) - fd2output[fd].append(data) - else: - # Ignore hang up or errors. - close_unregister_and_remove(fd) - - return (stdout, stderr) - - - def _communicate_with_select(self, input: Any) -> Tuple[List[bytes], - List[bytes]]: - read_set = [] # type: List[IO[Any]] - write_set = [] # type: List[IO[Any]] - stdout = None # type: List[bytes] # Return - stderr = None # type: List[bytes] # Return - - if self.stdin and input: - write_set.append(self.stdin) - if self.stdout: - read_set.append(self.stdout) - stdout = [] - if self.stderr: - read_set.append(self.stderr) - stderr = [] - - input_offset = 0 - while read_set or write_set: - try: - rlist, wlist, xlist = select.select(read_set, write_set, []) - except select.error as e: - if e.args[0] == errno.EINTR: - continue - raise - - # XXX Rewrite these to use non-blocking I/O on the - # file objects; they are no longer using C stdio! - - if self.stdin in wlist: - chunk = input[input_offset : input_offset + _PIPE_BUF] - try: - bytes_written = os.write(self.stdin.fileno(), chunk) - except OSError as oe: - if oe.errno == errno.EPIPE: - self.stdin.close() - write_set.remove(self.stdin) - else: - raise - else: - input_offset += bytes_written - if input_offset >= len(input): - self.stdin.close() - write_set.remove(self.stdin) - - if self.stdout in rlist: - data = os.read(self.stdout.fileno(), 1024) - if not data: - self.stdout.close() - read_set.remove(self.stdout) - stdout.append(data) - - if self.stderr in rlist: - data = os.read(self.stderr.fileno(), 1024) - if not data: - self.stderr.close() - read_set.remove(self.stderr) - stderr.append(data) - - return (stdout, stderr) - - - def send_signal(self, sig: int) -> None: - """Send a signal to the process - """ - os.kill(self.pid, sig) - - def terminate(self) -> None: - """Terminate the process with SIGTERM - """ - self.send_signal(signal.SIGTERM) - - def kill(self) -> None: - """Kill the process with SIGKILL - """ - self.send_signal(signal.SIGKILL) - - -def _demo_posix() -> None: - # - # Example 1: Simple redirection: Get process list - # - plist = Popen(["ps"], stdout=PIPE).communicate()[0] - print("Process list:") - print(plist) - - # - # Example 2: Change uid before executing child - # - if os.getuid() == 0: - p = Popen(["id"], preexec_fn=lambda: os.setuid(100)) - p.wait() - - # - # Example 3: Connecting several subprocesses - # - print("Looking for 'hda'...") - p1 = Popen(["dmesg"], stdout=PIPE) - p2 = Popen(["grep", "hda"], stdin=p1.stdout, stdout=PIPE) - print(repr(p2.communicate()[0])) - - # - # Example 4: Catch execution error - # - print() - print("Trying a weird file...") - try: - print(Popen(["/this/path/does/not/exist"]).communicate()) - except OSError as e: - if e.errno == errno.ENOENT: - print("The file didn't exist. I thought so...") - else: - print("Error", e.errno) - else: - print("Gosh. No error.", file=sys.stderr) - - -def _demo_windows() -> None: - # - # Example 1: Connecting several subprocesses - # - print("Looking for 'PROMPT' in set output...") - p1 = Popen("set", stdout=PIPE, shell=True) - p2 = Popen('find "PROMPT"', stdin=p1.stdout, stdout=PIPE) - print(repr(p2.communicate()[0])) - - # - # Example 2: Simple execution of program - # - print("Executing calc...") - p = Popen("calc") - p.wait() - - -if __name__ == "__main__": - if mswindows: - _demo_windows() - else: - _demo_posix() diff --git a/test-data/stdlib-samples/3.2/test/test_subprocess.py b/test-data/stdlib-samples/3.2/test/test_subprocess.py deleted file mode 100644 index 772d8cc416e4..000000000000 --- a/test-data/stdlib-samples/3.2/test/test_subprocess.py +++ /dev/null @@ -1,1764 +0,0 @@ -import unittest -from test import support -import subprocess -import sys -import signal -import io -import os -import errno -import tempfile -import time -import re -import sysconfig -import warnings -import select -import shutil -import gc - -import resource - -from typing import Any, Dict, Callable, Iterable, List, Set, Tuple, cast - -mswindows = (sys.platform == "win32") - -# -# Depends on the following external programs: Python -# - -if mswindows: - SETBINARY = ('import msvcrt; msvcrt.setmode(sys.stdout.fileno(), ' - 'os.O_BINARY);') -else: - SETBINARY = '' - - -try: - mkstemp = tempfile.mkstemp -except AttributeError: - # tempfile.mkstemp is not available - def _mkstemp() -> Tuple[int, str]: - """Replacement for mkstemp, calling mktemp.""" - fname = tempfile.mktemp() - return os.open(fname, os.O_RDWR|os.O_CREAT), fname - mkstemp = cast(Any, _mkstemp) - - -class BaseTestCase(unittest.TestCase): - def setUp(self) -> None: - # Try to minimize the number of children we have so this test - # doesn't crash on some buildbots (Alphas in particular). - support.reap_children() - - def tearDown(self) -> None: - for inst in subprocess._active: - inst.wait() - subprocess._cleanup() - self.assertFalse(subprocess._active, "subprocess._active not empty") - - def assertStderrEqual(self, stderr: bytes, expected: bytes, - msg: object = None) -> None: - # In a debug build, stuff like "[6580 refs]" is printed to stderr at - # shutdown time. That frustrates tests trying to check stderr produced - # from a spawned Python process. - actual = support.strip_python_stderr(stderr) - self.assertEqual(actual, expected, msg) - - -class ProcessTestCase(BaseTestCase): - - def test_call_seq(self) -> None: - # call() function with sequence argument - rc = subprocess.call([sys.executable, "-c", - "import sys; sys.exit(47)"]) - self.assertEqual(rc, 47) - - def test_check_call_zero(self) -> None: - # check_call() function with zero return code - rc = subprocess.check_call([sys.executable, "-c", - "import sys; sys.exit(0)"]) - self.assertEqual(rc, 0) - - def test_check_call_nonzero(self) -> None: - # check_call() function with non-zero return code - with self.assertRaises(subprocess.CalledProcessError) as c: - subprocess.check_call([sys.executable, "-c", - "import sys; sys.exit(47)"]) - self.assertEqual(c.exception.returncode, 47) - - def test_check_output(self) -> None: - # check_output() function with zero return code - output = subprocess.check_output( - [sys.executable, "-c", "print('BDFL')"]) - self.assertIn(b'BDFL', cast(Any, output)) # see #39 - - def test_check_output_nonzero(self) -> None: - # check_call() function with non-zero return code - with self.assertRaises(subprocess.CalledProcessError) as c: - subprocess.check_output( - [sys.executable, "-c", "import sys; sys.exit(5)"]) - self.assertEqual(c.exception.returncode, 5) - - def test_check_output_stderr(self) -> None: - # check_output() function stderr redirected to stdout - output = subprocess.check_output( - [sys.executable, "-c", "import sys; sys.stderr.write('BDFL')"], - stderr=subprocess.STDOUT) - self.assertIn(b'BDFL', cast(Any, output)) # see #39 - - def test_check_output_stdout_arg(self) -> None: - # check_output() function stderr redirected to stdout - with self.assertRaises(ValueError) as c: - output = subprocess.check_output( - [sys.executable, "-c", "print('will not be run')"], - stdout=sys.stdout) - self.fail("Expected ValueError when stdout arg supplied.") - self.assertIn('stdout', c.exception.args[0]) - - def test_call_kwargs(self) -> None: - # call() function with keyword args - newenv = os.environ.copy() - newenv["FRUIT"] = "banana" - rc = subprocess.call([sys.executable, "-c", - 'import sys, os;' - 'sys.exit(os.getenv("FRUIT")=="banana")'], - env=newenv) - self.assertEqual(rc, 1) - - def test_invalid_args(self) -> None: - # Popen() called with invalid arguments should raise TypeError - # but Popen.__del__ should not complain (issue #12085) - with support.captured_stderr() as s: - self.assertRaises(TypeError, subprocess.Popen, invalid_arg_name=1) - argcount = subprocess.Popen.__init__.__code__.co_argcount - too_many_args = [0] * (argcount + 1) - self.assertRaises(TypeError, subprocess.Popen, *too_many_args) - self.assertEqual(s.getvalue(), '') - - def test_stdin_none(self) -> None: - # .stdin is None when not redirected - p = subprocess.Popen([sys.executable, "-c", 'print("banana")'], - stdout=subprocess.PIPE, stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - p.wait() - self.assertEqual(p.stdin, None) - - def test_stdout_none(self) -> None: - # .stdout is None when not redirected - p = subprocess.Popen([sys.executable, "-c", - 'print(" this bit of output is from a ' - 'test of stdout in a different ' - 'process ...")'], - stdin=subprocess.PIPE, stderr=subprocess.PIPE) - self.addCleanup(p.stdin.close) - self.addCleanup(p.stderr.close) - p.wait() - self.assertEqual(p.stdout, None) - - def test_stderr_none(self) -> None: - # .stderr is None when not redirected - p = subprocess.Popen([sys.executable, "-c", 'print("banana")'], - stdin=subprocess.PIPE, stdout=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stdin.close) - p.wait() - self.assertEqual(p.stderr, None) - - def test_executable_with_cwd(self) -> None: - python_dir = os.path.dirname(os.path.realpath(sys.executable)) - p = subprocess.Popen(["somethingyoudonthave", "-c", - "import sys; sys.exit(47)"], - executable=sys.executable, cwd=python_dir) - p.wait() - self.assertEqual(p.returncode, 47) - - @unittest.skipIf(sysconfig.is_python_build(), - "need an installed Python. See #7774") - def test_executable_without_cwd(self) -> None: - # For a normal installation, it should work without 'cwd' - # argument. For test runs in the build directory, see #7774. - p = subprocess.Popen(["somethingyoudonthave", "-c", - "import sys; sys.exit(47)"], - executable=sys.executable) - p.wait() - self.assertEqual(p.returncode, 47) - - def test_stdin_pipe(self) -> None: - # stdin redirection - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.exit(sys.stdin.read() == "pear")'], - stdin=subprocess.PIPE) - p.stdin.write(b"pear") - p.stdin.close() - p.wait() - self.assertEqual(p.returncode, 1) - - def test_stdin_filedes(self) -> None: - # stdin is set to open file descriptor - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - d = tf.fileno() - os.write(d, b"pear") - os.lseek(d, 0, 0) - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.exit(sys.stdin.read() == "pear")'], - stdin=d) - p.wait() - self.assertEqual(p.returncode, 1) - - def test_stdin_fileobj(self) -> None: - # stdin is set to open file object - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - tf.write(b"pear") - tf.seek(0) - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.exit(sys.stdin.read() == "pear")'], - stdin=tf) - p.wait() - self.assertEqual(p.returncode, 1) - - def test_stdout_pipe(self) -> None: - # stdout redirection - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdout.write("orange")'], - stdout=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.read(), b"orange") - - def test_stdout_filedes(self) -> None: - # stdout is set to open file descriptor - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - d = tf.fileno() - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdout.write("orange")'], - stdout=d) - p.wait() - os.lseek(d, 0, 0) - self.assertEqual(os.read(d, 1024), b"orange") - - def test_stdout_fileobj(self) -> None: - # stdout is set to open file object - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdout.write("orange")'], - stdout=tf) - p.wait() - tf.seek(0) - self.assertEqual(tf.read(), b"orange") - - def test_stderr_pipe(self) -> None: - # stderr redirection - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stderr.write("strawberry")'], - stderr=subprocess.PIPE) - self.addCleanup(p.stderr.close) - self.assertStderrEqual(p.stderr.read(), b"strawberry") - - def test_stderr_filedes(self) -> None: - # stderr is set to open file descriptor - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - d = tf.fileno() - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stderr.write("strawberry")'], - stderr=d) - p.wait() - os.lseek(d, 0, 0) - self.assertStderrEqual(os.read(d, 1024), b"strawberry") - - def test_stderr_fileobj(self) -> None: - # stderr is set to open file object - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stderr.write("strawberry")'], - stderr=tf) - p.wait() - tf.seek(0) - self.assertStderrEqual(tf.read(), b"strawberry") - - def test_stdout_stderr_pipe(self) -> None: - # capture stdout and stderr to the same pipe - p = subprocess.Popen([sys.executable, "-c", - 'import sys;' - 'sys.stdout.write("apple");' - 'sys.stdout.flush();' - 'sys.stderr.write("orange")'], - stdout=subprocess.PIPE, - stderr=subprocess.STDOUT) - self.addCleanup(p.stdout.close) - self.assertStderrEqual(p.stdout.read(), b"appleorange") - - def test_stdout_stderr_file(self) -> None: - # capture stdout and stderr to the same open file - tf = tempfile.TemporaryFile() - self.addCleanup(tf.close) - p = subprocess.Popen([sys.executable, "-c", - 'import sys;' - 'sys.stdout.write("apple");' - 'sys.stdout.flush();' - 'sys.stderr.write("orange")'], - stdout=tf, - stderr=tf) - p.wait() - tf.seek(0) - self.assertStderrEqual(tf.read(), b"appleorange") - - def test_stdout_filedes_of_stdout(self) -> None: - # stdout is set to 1 (#1531862). - cmd = r"import sys, os; sys.exit(os.write(sys.stdout.fileno(), b'.\n'))" - rc = subprocess.call([sys.executable, "-c", cmd], stdout=1) - self.assertEqual(rc, 2) - - def test_cwd(self) -> None: - tmpdir = tempfile.gettempdir() - # We cannot use os.path.realpath to canonicalize the path, - # since it doesn't expand Tru64 {memb} strings. See bug 1063571. - cwd = os.getcwd() - os.chdir(tmpdir) - tmpdir = os.getcwd() - os.chdir(cwd) - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stdout.write(os.getcwd())'], - stdout=subprocess.PIPE, - cwd=tmpdir) - self.addCleanup(p.stdout.close) - normcase = os.path.normcase - self.assertEqual(normcase(p.stdout.read().decode("utf-8")), - normcase(tmpdir)) - - def test_env(self) -> None: - newenv = os.environ.copy() - newenv["FRUIT"] = "orange" - with subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stdout.write(os.getenv("FRUIT"))'], - stdout=subprocess.PIPE, - env=newenv) as p: - stdout, stderr = p.communicate() - self.assertEqual(stdout, b"orange") - - # Windows requires at least the SYSTEMROOT environment variable to start - # Python - @unittest.skipIf(sys.platform == 'win32', - 'cannot test an empty env on Windows') - @unittest.skipIf(sysconfig.get_config_var('Py_ENABLE_SHARED') is not None, - 'the python library cannot be loaded ' - 'with an empty environment') - def test_empty_env(self) -> None: - with subprocess.Popen([sys.executable, "-c", - 'import os; ' - 'print(list(os.environ.keys()))'], - stdout=subprocess.PIPE, - env={}) as p: - stdout, stderr = p.communicate() - self.assertIn(stdout.strip(), - [b"[]", - # Mac OS X adds __CF_USER_TEXT_ENCODING variable to an empty - # environment - b"['__CF_USER_TEXT_ENCODING']"]) - - def test_communicate_stdin(self) -> None: - p = subprocess.Popen([sys.executable, "-c", - 'import sys;' - 'sys.exit(sys.stdin.read() == "pear")'], - stdin=subprocess.PIPE) - p.communicate(b"pear") - self.assertEqual(p.returncode, 1) - - def test_communicate_stdout(self) -> None: - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stdout.write("pineapple")'], - stdout=subprocess.PIPE) - (stdout, stderr) = p.communicate() - self.assertEqual(stdout, b"pineapple") - self.assertEqual(stderr, None) - - def test_communicate_stderr(self) -> None: - p = subprocess.Popen([sys.executable, "-c", - 'import sys; sys.stderr.write("pineapple")'], - stderr=subprocess.PIPE) - (stdout, stderr) = p.communicate() - self.assertEqual(stdout, None) - self.assertStderrEqual(stderr, b"pineapple") - - def test_communicate(self) -> None: - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stderr.write("pineapple");' - 'sys.stdout.write(sys.stdin.read())'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - self.addCleanup(p.stdin.close) - (stdout, stderr) = p.communicate(b"banana") - self.assertEqual(stdout, b"banana") - self.assertStderrEqual(stderr, b"pineapple") - - # Test for the fd leak reported in http://bugs.python.org/issue2791. - def test_communicate_pipe_fd_leak(self) -> None: - for stdin_pipe in (False, True): - for stdout_pipe in (False, True): - for stderr_pipe in (False, True): - options = {} # type: Dict[str, Any] - if stdin_pipe: - options['stdin'] = subprocess.PIPE - if stdout_pipe: - options['stdout'] = subprocess.PIPE - if stderr_pipe: - options['stderr'] = subprocess.PIPE - if not options: - continue - p = subprocess.Popen([sys.executable, "-c", "pass"], **options) - p.communicate() - if p.stdin is not None: - self.assertTrue(p.stdin.closed) - if p.stdout is not None: - self.assertTrue(p.stdout.closed) - if p.stderr is not None: - self.assertTrue(p.stderr.closed) - - def test_communicate_returns(self) -> None: - # communicate() should return None if no redirection is active - p = subprocess.Popen([sys.executable, "-c", - "import sys; sys.exit(47)"]) - (stdout, stderr) = p.communicate() - self.assertEqual(stdout, None) - self.assertEqual(stderr, None) - - def test_communicate_pipe_buf(self) -> None: - # communicate() with writes larger than pipe_buf - # This test will probably deadlock rather than fail, if - # communicate() does not work properly. - x, y = os.pipe() - if mswindows: - pipe_buf = 512 - else: - pipe_buf = os.fpathconf(x, "PC_PIPE_BUF") - os.close(x) - os.close(y) - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stdout.write(sys.stdin.read(47));' - 'sys.stderr.write("xyz"*%d);' - 'sys.stdout.write(sys.stdin.read())' % pipe_buf], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - self.addCleanup(p.stdin.close) - string_to_write = b"abc"*pipe_buf - (stdout, stderr) = p.communicate(string_to_write) - self.assertEqual(stdout, string_to_write) - - def test_writes_before_communicate(self) -> None: - # stdin.write before communicate() - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stdout.write(sys.stdin.read())'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - self.addCleanup(p.stdin.close) - p.stdin.write(b"banana") - (stdout, stderr) = p.communicate(b"split") - self.assertEqual(stdout, b"bananasplit") - self.assertStderrEqual(stderr, b"") - - def test_universal_newlines(self) -> None: - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' + SETBINARY + - 'sys.stdout.write(sys.stdin.readline());' - 'sys.stdout.flush();' - 'sys.stdout.write("line2\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write(sys.stdin.read());' - 'sys.stdout.flush();' - 'sys.stdout.write("line4\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write("line5\\r\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write("line6\\r");' - 'sys.stdout.flush();' - 'sys.stdout.write("\\nline7");' - 'sys.stdout.flush();' - 'sys.stdout.write("\\nline8");'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - universal_newlines=1) - p.stdin.write("line1\n") - self.assertEqual(p.stdout.readline(), "line1\n") - p.stdin.write("line3\n") - p.stdin.close() - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.readline(), - "line2\n") - self.assertEqual(p.stdout.read(6), - "line3\n") - self.assertEqual(p.stdout.read(), - "line4\nline5\nline6\nline7\nline8") - - def test_universal_newlines_communicate(self) -> None: - # universal newlines through communicate() - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' + SETBINARY + - 'sys.stdout.write("line2\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write("line4\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write("line5\\r\\n");' - 'sys.stdout.flush();' - 'sys.stdout.write("line6\\r");' - 'sys.stdout.flush();' - 'sys.stdout.write("\\nline7");' - 'sys.stdout.flush();' - 'sys.stdout.write("\\nline8");'], - stderr=subprocess.PIPE, - stdout=subprocess.PIPE, - universal_newlines=1) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - # BUG: can't give a non-empty stdin because it breaks both the - # select- and poll-based communicate() implementations. - (stdout, stderr) = p.communicate() - self.assertEqual(stdout, - "line2\nline4\nline5\nline6\nline7\nline8") - - def test_universal_newlines_communicate_stdin(self) -> None: - # universal newlines through communicate(), with only stdin - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' + SETBINARY + '''\nif True: - s = sys.stdin.readline() - assert s == "line1\\n", repr(s) - s = sys.stdin.read() - assert s == "line3\\n", repr(s) - '''], - stdin=subprocess.PIPE, - universal_newlines=1) - (stdout, stderr) = p.communicate("line1\nline3\n") - self.assertEqual(p.returncode, 0) - - def test_no_leaking(self) -> None: - # Make sure we leak no resources - if not mswindows: - max_handles = 1026 # too much for most UNIX systems - else: - max_handles = 2050 # too much for (at least some) Windows setups - handles = [] # type: List[int] - tmpdir = tempfile.mkdtemp() - try: - for i in range(max_handles): - try: - tmpfile = os.path.join(tmpdir, support.TESTFN) - handles.append(os.open(tmpfile, os.O_WRONLY|os.O_CREAT)) - except OSError as e: - if e.errno != errno.EMFILE: - raise - break - else: - self.skipTest("failed to reach the file descriptor limit " - "(tried %d)" % max_handles) - # Close a couple of them (should be enough for a subprocess) - for i in range(10): - os.close(handles.pop()) - # Loop creating some subprocesses. If one of them leaks some fds, - # the next loop iteration will fail by reaching the max fd limit. - for i in range(15): - p = subprocess.Popen([sys.executable, "-c", - "import sys;" - "sys.stdout.write(sys.stdin.read())"], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - data = p.communicate(b"lime")[0] - self.assertEqual(data, b"lime") - finally: - for h in handles: - os.close(h) - shutil.rmtree(tmpdir) - - def test_list2cmdline(self) -> None: - self.assertEqual(subprocess.list2cmdline(['a b c', 'd', 'e']), - '"a b c" d e') - self.assertEqual(subprocess.list2cmdline(['ab"c', '\\', 'd']), - 'ab\\"c \\ d') - self.assertEqual(subprocess.list2cmdline(['ab"c', ' \\', 'd']), - 'ab\\"c " \\\\" d') - self.assertEqual(subprocess.list2cmdline(['a\\\\\\b', 'de fg', 'h']), - 'a\\\\\\b "de fg" h') - self.assertEqual(subprocess.list2cmdline(['a\\"b', 'c', 'd']), - 'a\\\\\\"b c d') - self.assertEqual(subprocess.list2cmdline(['a\\\\b c', 'd', 'e']), - '"a\\\\b c" d e') - self.assertEqual(subprocess.list2cmdline(['a\\\\b\\ c', 'd', 'e']), - '"a\\\\b\\ c" d e') - self.assertEqual(subprocess.list2cmdline(['ab', '']), - 'ab ""') - - - def test_poll(self) -> None: - p = subprocess.Popen([sys.executable, - "-c", "import time; time.sleep(1)"]) - count = 0 - while p.poll() is None: - time.sleep(0.1) - count += 1 - # We expect that the poll loop probably went around about 10 times, - # but, based on system scheduling we can't control, it's possible - # poll() never returned None. It "should be" very rare that it - # didn't go around at least twice. - self.assertGreaterEqual(count, 2) - # Subsequent invocations should just return the returncode - self.assertEqual(p.poll(), 0) - - - def test_wait(self) -> None: - p = subprocess.Popen([sys.executable, - "-c", "import time; time.sleep(2)"]) - self.assertEqual(p.wait(), 0) - # Subsequent invocations should just return the returncode - self.assertEqual(p.wait(), 0) - - - def test_invalid_bufsize(self) -> None: - # an invalid type of the bufsize argument should raise - # TypeError. - with self.assertRaises(TypeError): - subprocess.Popen([sys.executable, "-c", "pass"], cast(Any, "orange")) - - def test_bufsize_is_none(self) -> None: - # bufsize=None should be the same as bufsize=0. - p = subprocess.Popen([sys.executable, "-c", "pass"], None) - self.assertEqual(p.wait(), 0) - # Again with keyword arg - p = subprocess.Popen([sys.executable, "-c", "pass"], bufsize=None) - self.assertEqual(p.wait(), 0) - - def test_leaking_fds_on_error(self) -> None: - # see bug #5179: Popen leaks file descriptors to PIPEs if - # the child fails to execute; this will eventually exhaust - # the maximum number of open fds. 1024 seems a very common - # value for that limit, but Windows has 2048, so we loop - # 1024 times (each call leaked two fds). - for i in range(1024): - # Windows raises IOError. Others raise OSError. - with self.assertRaises(EnvironmentError) as c: - subprocess.Popen(['nonexisting_i_hope'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - # ignore errors that indicate the command was not found - if c.exception.errno not in (errno.ENOENT, errno.EACCES): - raise c.exception - - def test_issue8780(self) -> None: - # Ensure that stdout is inherited from the parent - # if stdout=PIPE is not used - code = ';'.join([ - 'import subprocess, sys', - 'retcode = subprocess.call(' - "[sys.executable, '-c', 'print(\"Hello World!\")'])", - 'assert retcode == 0']) - output = subprocess.check_output([sys.executable, '-c', code]) - self.assertTrue(output.startswith(b'Hello World!'), ascii(output)) - - def test_handles_closed_on_exception(self) -> None: - # If CreateProcess exits with an error, ensure the - # duplicate output handles are released - ifhandle, ifname = mkstemp() - ofhandle, ofname = mkstemp() - efhandle, efname = mkstemp() - try: - subprocess.Popen (["*"], stdin=ifhandle, stdout=ofhandle, - stderr=efhandle) - except OSError: - os.close(ifhandle) - os.remove(ifname) - os.close(ofhandle) - os.remove(ofname) - os.close(efhandle) - os.remove(efname) - self.assertFalse(os.path.exists(ifname)) - self.assertFalse(os.path.exists(ofname)) - self.assertFalse(os.path.exists(efname)) - - def test_communicate_epipe(self) -> None: - # Issue 10963: communicate() should hide EPIPE - p = subprocess.Popen([sys.executable, "-c", 'pass'], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - self.addCleanup(p.stdin.close) - p.communicate(b"x" * 2**20) - - def test_communicate_epipe_only_stdin(self) -> None: - # Issue 10963: communicate() should hide EPIPE - p = subprocess.Popen([sys.executable, "-c", 'pass'], - stdin=subprocess.PIPE) - self.addCleanup(p.stdin.close) - time.sleep(2) - p.communicate(b"x" * 2**20) - - @unittest.skipUnless(hasattr(signal, 'SIGALRM'), - "Requires signal.SIGALRM") - def test_communicate_eintr(self) -> None: - # Issue #12493: communicate() should handle EINTR - def handler(signum, frame): - pass - old_handler = signal.signal(signal.SIGALRM, handler) - self.addCleanup(signal.signal, signal.SIGALRM, old_handler) - - # the process is running for 2 seconds - args = [sys.executable, "-c", 'import time; time.sleep(2)'] - for stream in ('stdout', 'stderr'): - kw = {stream: subprocess.PIPE} # type: Dict[str, Any] - with subprocess.Popen(args, **kw) as process: - signal.alarm(1) - # communicate() will be interrupted by SIGALRM - process.communicate() - - -# context manager -class _SuppressCoreFiles(object): - """Try to prevent core files from being created.""" - old_limit = None # type: Tuple[int, int] - - def __enter__(self) -> None: - """Try to save previous ulimit, then set it to (0, 0).""" - if resource is not None: - try: - self.old_limit = resource.getrlimit(resource.RLIMIT_CORE) - resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) - except (ValueError, resource.error): - pass - - if sys.platform == 'darwin': - # Check if the 'Crash Reporter' on OSX was configured - # in 'Developer' mode and warn that it will get triggered - # when it is. - # - # This assumes that this context manager is used in tests - # that might trigger the next manager. - value = subprocess.Popen(['/usr/bin/defaults', 'read', - 'com.apple.CrashReporter', 'DialogType'], - stdout=subprocess.PIPE).communicate()[0] - if value.strip() == b'developer': - print("this tests triggers the Crash Reporter, " - "that is intentional", end='') - sys.stdout.flush() - - def __exit__(self, *args: Any) -> None: - """Return core file behavior to default.""" - if self.old_limit is None: - return - if resource is not None: - try: - resource.setrlimit(resource.RLIMIT_CORE, self.old_limit) - except (ValueError, resource.error): - pass - - -@unittest.skipIf(mswindows, "POSIX specific tests") -class POSIXProcessTestCase(BaseTestCase): - - def test_exceptions(self) -> None: - nonexistent_dir = "/_this/pa.th/does/not/exist" - try: - os.chdir(nonexistent_dir) - except OSError as e: - # This avoids hard coding the errno value or the OS perror() - # string and instead capture the exception that we want to see - # below for comparison. - desired_exception = e - desired_exception.strerror += ': ' + repr(sys.executable) - else: - self.fail("chdir to nonexistant directory %s succeeded." % - nonexistent_dir) - - # Error in the child re-raised in the parent. - try: - p = subprocess.Popen([sys.executable, "-c", ""], - cwd=nonexistent_dir) - except OSError as e: - # Test that the child process chdir failure actually makes - # it up to the parent process as the correct exception. - self.assertEqual(desired_exception.errno, e.errno) - self.assertEqual(desired_exception.strerror, e.strerror) - else: - self.fail("Expected OSError: %s" % desired_exception) - - def test_restore_signals(self) -> None: - # Code coverage for both values of restore_signals to make sure it - # at least does not blow up. - # A test for behavior would be complex. Contributions welcome. - subprocess.call([sys.executable, "-c", ""], restore_signals=True) - subprocess.call([sys.executable, "-c", ""], restore_signals=False) - - def test_start_new_session(self) -> None: - # For code coverage of calling setsid(). We don't care if we get an - # EPERM error from it depending on the test execution environment, that - # still indicates that it was called. - try: - output = subprocess.check_output( - [sys.executable, "-c", - "import os; print(os.getpgid(os.getpid()))"], - start_new_session=True) - except OSError as e: - if e.errno != errno.EPERM: - raise - else: - parent_pgid = os.getpgid(os.getpid()) - child_pgid = int(output) - self.assertNotEqual(parent_pgid, child_pgid) - - def test_run_abort(self) -> None: - # returncode handles signal termination - with _SuppressCoreFiles(): - p = subprocess.Popen([sys.executable, "-c", - 'import os; os.abort()']) - p.wait() - self.assertEqual(-p.returncode, signal.SIGABRT) - - def test_preexec(self) -> None: - # DISCLAIMER: Setting environment variables is *not* a good use - # of a preexec_fn. This is merely a test. - p = subprocess.Popen([sys.executable, "-c", - 'import sys,os;' - 'sys.stdout.write(os.getenv("FRUIT"))'], - stdout=subprocess.PIPE, - preexec_fn=lambda: os.putenv("FRUIT", "apple")) - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.read(), b"apple") - - def test_preexec_exception(self) -> None: - def raise_it(): - raise ValueError("What if two swallows carried a coconut?") - try: - p = subprocess.Popen([sys.executable, "-c", ""], - preexec_fn=raise_it) - except RuntimeError as e: - self.assertTrue( - subprocess._posixsubprocess, - "Expected a ValueError from the preexec_fn") - except ValueError as e2: - self.assertIn("coconut", e2.args[0]) - else: - self.fail("Exception raised by preexec_fn did not make it " - "to the parent process.") - - def test_preexec_gc_module_failure(self) -> None: - # This tests the code that disables garbage collection if the child - # process will execute any Python. - def raise_runtime_error(): - raise RuntimeError("this shouldn't escape") - enabled = gc.isenabled() - orig_gc_disable = gc.disable - orig_gc_isenabled = gc.isenabled - try: - gc.disable() - self.assertFalse(gc.isenabled()) - subprocess.call([sys.executable, '-c', ''], - preexec_fn=lambda: None) - self.assertFalse(gc.isenabled(), - "Popen enabled gc when it shouldn't.") - - gc.enable() - self.assertTrue(gc.isenabled()) - subprocess.call([sys.executable, '-c', ''], - preexec_fn=lambda: None) - self.assertTrue(gc.isenabled(), "Popen left gc disabled.") - - setattr(gc, 'disable', raise_runtime_error) - self.assertRaises(RuntimeError, subprocess.Popen, - [sys.executable, '-c', ''], - preexec_fn=lambda: None) - - del gc.isenabled # force an AttributeError - self.assertRaises(AttributeError, subprocess.Popen, - [sys.executable, '-c', ''], - preexec_fn=lambda: None) - finally: - setattr(gc, 'disable', orig_gc_disable) - setattr(gc, 'isenabled', orig_gc_isenabled) - if not enabled: - gc.disable() - - def test_args_string(self) -> None: - # args is a string - fd, fname = mkstemp() - # reopen in text mode - with open(fd, "w", errors=cast(Any, "surrogateescape")) as fobj: # see #260 - fobj.write("#!/bin/sh\n") - fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" % - sys.executable) - os.chmod(fname, 0o700) - p = subprocess.Popen(fname) - p.wait() - os.remove(fname) - self.assertEqual(p.returncode, 47) - - def test_invalid_args(self) -> None: - # invalid arguments should raise ValueError - self.assertRaises(ValueError, subprocess.call, - [sys.executable, "-c", - "import sys; sys.exit(47)"], - startupinfo=47) - self.assertRaises(ValueError, subprocess.call, - [sys.executable, "-c", - "import sys; sys.exit(47)"], - creationflags=47) - - def test_shell_sequence(self) -> None: - # Run command through the shell (sequence) - newenv = os.environ.copy() - newenv["FRUIT"] = "apple" - p = subprocess.Popen(["echo $FRUIT"], shell=1, - stdout=subprocess.PIPE, - env=newenv) - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") - - def test_shell_string(self) -> None: - # Run command through the shell (string) - newenv = os.environ.copy() - newenv["FRUIT"] = "apple" - p = subprocess.Popen("echo $FRUIT", shell=1, - stdout=subprocess.PIPE, - env=newenv) - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.read().strip(b" \t\r\n\f"), b"apple") - - def test_call_string(self) -> None: - # call() function with string argument on UNIX - fd, fname = mkstemp() - # reopen in text mode - with open(fd, "w", errors=cast(Any, "surrogateescape")) as fobj: # see #260 - fobj.write("#!/bin/sh\n") - fobj.write("exec '%s' -c 'import sys; sys.exit(47)'\n" % - sys.executable) - os.chmod(fname, 0o700) - rc = subprocess.call(fname) - os.remove(fname) - self.assertEqual(rc, 47) - - def test_specific_shell(self) -> None: - # Issue #9265: Incorrect name passed as arg[0]. - shells = [] # type: List[str] - for prefix in ['/bin', '/usr/bin/', '/usr/local/bin']: - for name in ['bash', 'ksh']: - sh = os.path.join(prefix, name) - if os.path.isfile(sh): - shells.append(sh) - if not shells: # Will probably work for any shell but csh. - self.skipTest("bash or ksh required for this test") - sh = '/bin/sh' - if os.path.isfile(sh) and not os.path.islink(sh): - # Test will fail if /bin/sh is a symlink to csh. - shells.append(sh) - for sh in shells: - p = subprocess.Popen("echo $0", executable=sh, shell=True, - stdout=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.assertEqual(p.stdout.read().strip(), bytes(sh, 'ascii')) - - def _kill_process(self, method: str, *args: Any) -> subprocess.Popen: - # Do not inherit file handles from the parent. - # It should fix failures on some platforms. - p = subprocess.Popen([sys.executable, "-c", """if 1: - import sys, time - sys.stdout.write('x\\n') - sys.stdout.flush() - time.sleep(30) - """], - close_fds=True, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - # Wait for the interpreter to be completely initialized before - # sending any signal. - p.stdout.read(1) - getattr(p, method)(*args) - return p - - def test_send_signal(self) -> None: - p = self._kill_process('send_signal', signal.SIGINT) - _, stderr = p.communicate() - self.assertIn(b'KeyboardInterrupt', stderr) - self.assertNotEqual(p.wait(), 0) - - def test_kill(self) -> None: - p = self._kill_process('kill') - _, stderr = p.communicate() - self.assertStderrEqual(stderr, b'') - self.assertEqual(p.wait(), -signal.SIGKILL) - - def test_terminate(self) -> None: - p = self._kill_process('terminate') - _, stderr = p.communicate() - self.assertStderrEqual(stderr, b'') - self.assertEqual(p.wait(), -signal.SIGTERM) - - def check_close_std_fds(self, fds: Iterable[int]) -> None: - # Issue #9905: test that subprocess pipes still work properly with - # some standard fds closed - stdin = 0 - newfds = [] # type: List[int] - for a in fds: - b = os.dup(a) - newfds.append(b) - if a == 0: - stdin = b - try: - for fd in fds: - os.close(fd) - out, err = subprocess.Popen([sys.executable, "-c", - 'import sys;' - 'sys.stdout.write("apple");' - 'sys.stdout.flush();' - 'sys.stderr.write("orange")'], - stdin=stdin, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE).communicate() - err = support.strip_python_stderr(err) - self.assertEqual((out, err), (b'apple', b'orange')) - finally: - for b, a in zip(newfds, fds): - os.dup2(b, a) - for b in newfds: - os.close(b) - - def test_close_fd_0(self) -> None: - self.check_close_std_fds([0]) - - def test_close_fd_1(self) -> None: - self.check_close_std_fds([1]) - - def test_close_fd_2(self) -> None: - self.check_close_std_fds([2]) - - def test_close_fds_0_1(self) -> None: - self.check_close_std_fds([0, 1]) - - def test_close_fds_0_2(self) -> None: - self.check_close_std_fds([0, 2]) - - def test_close_fds_1_2(self) -> None: - self.check_close_std_fds([1, 2]) - - def test_close_fds_0_1_2(self) -> None: - # Issue #10806: test that subprocess pipes still work properly with - # all standard fds closed. - self.check_close_std_fds([0, 1, 2]) - - def test_remapping_std_fds(self) -> None: - # open up some temporary files - temps = [mkstemp() for i in range(3)] - try: - temp_fds = [fd for fd, fname in temps] - - # unlink the files -- we won't need to reopen them - for fd, fname in temps: - os.unlink(fname) - - # write some data to what will become stdin, and rewind - os.write(temp_fds[1], b"STDIN") - os.lseek(temp_fds[1], 0, 0) - - # move the standard file descriptors out of the way - saved_fds = [os.dup(fd) for fd in range(3)] - try: - # duplicate the file objects over the standard fd's - for fd, temp_fd in enumerate(temp_fds): - os.dup2(temp_fd, fd) - - # now use those files in the "wrong" order, so that subprocess - # has to rearrange them in the child - p = subprocess.Popen([sys.executable, "-c", - 'import sys; got = sys.stdin.read();' - 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'], - stdin=temp_fds[1], - stdout=temp_fds[2], - stderr=temp_fds[0]) - p.wait() - finally: - # restore the original fd's underneath sys.stdin, etc. - for std, saved in enumerate(saved_fds): - os.dup2(saved, std) - os.close(saved) - - for fd in temp_fds: - os.lseek(fd, 0, 0) - - out = os.read(temp_fds[2], 1024) - err = support.strip_python_stderr(os.read(temp_fds[0], 1024)) - self.assertEqual(out, b"got STDIN") - self.assertEqual(err, b"err") - - finally: - for fd in temp_fds: - os.close(fd) - - def check_swap_fds(self, stdin_no: int, stdout_no: int, - stderr_no: int) -> None: - # open up some temporary files - temps = [mkstemp() for i in range(3)] - temp_fds = [fd for fd, fname in temps] - try: - # unlink the files -- we won't need to reopen them - for fd, fname in temps: - os.unlink(fname) - - # save a copy of the standard file descriptors - saved_fds = [os.dup(fd) for fd in range(3)] - try: - # duplicate the temp files over the standard fd's 0, 1, 2 - for fd, temp_fd in enumerate(temp_fds): - os.dup2(temp_fd, fd) - - # write some data to what will become stdin, and rewind - os.write(stdin_no, b"STDIN") - os.lseek(stdin_no, 0, 0) - - # now use those files in the given order, so that subprocess - # has to rearrange them in the child - p = subprocess.Popen([sys.executable, "-c", - 'import sys; got = sys.stdin.read();' - 'sys.stdout.write("got %s"%got); sys.stderr.write("err")'], - stdin=stdin_no, - stdout=stdout_no, - stderr=stderr_no) - p.wait() - - for fd in temp_fds: - os.lseek(fd, 0, 0) - - out = os.read(stdout_no, 1024) - err = support.strip_python_stderr(os.read(stderr_no, 1024)) - finally: - for std, saved in enumerate(saved_fds): - os.dup2(saved, std) - os.close(saved) - - self.assertEqual(out, b"got STDIN") - self.assertEqual(err, b"err") - - finally: - for fd in temp_fds: - os.close(fd) - - # When duping fds, if there arises a situation where one of the fds is - # either 0, 1 or 2, it is possible that it is overwritten (#12607). - # This tests all combinations of this. - def test_swap_fds(self) -> None: - self.check_swap_fds(0, 1, 2) - self.check_swap_fds(0, 2, 1) - self.check_swap_fds(1, 0, 2) - self.check_swap_fds(1, 2, 0) - self.check_swap_fds(2, 0, 1) - self.check_swap_fds(2, 1, 0) - - def test_surrogates_error_message(self) -> None: - def prepare() -> None: - raise ValueError("surrogate:\uDCff") - - try: - subprocess.call( - [sys.executable, "-c", "pass"], - preexec_fn=prepare) - except ValueError as err: - # Pure Python implementations keeps the message - self.assertIsNone(subprocess._posixsubprocess) - self.assertEqual(str(err), "surrogate:\uDCff") - except RuntimeError as err2: - # _posixsubprocess uses a default message - self.assertIsNotNone(subprocess._posixsubprocess) - self.assertEqual(str(err2), "Exception occurred in preexec_fn.") - else: - self.fail("Expected ValueError or RuntimeError") - - def test_undecodable_env(self) -> None: - for key, value in (('test', 'abc\uDCFF'), ('test\uDCFF', '42')): - # test str with surrogates - script = "import os; print(ascii(os.getenv(%s)))" % repr(key) - env = os.environ.copy() - env[key] = value - # Use C locale to get ascii for the locale encoding to force - # surrogate-escaping of \xFF in the child process; otherwise it can - # be decoded as-is if the default locale is latin-1. - env['LC_ALL'] = 'C' - stdout = subprocess.check_output( - [sys.executable, "-c", script], - env=env) - stdout = stdout.rstrip(b'\n\r') - self.assertEqual(stdout.decode('ascii'), ascii(value)) - - # test bytes - keyb = key.encode("ascii", "surrogateescape") - valueb = value.encode("ascii", "surrogateescape") - script = "import os; print(ascii(os.getenvb(%s)))" % repr(keyb) - envb = dict(os.environ.copy().items()) # type: Dict[Any, Any] - envb[keyb] = valueb - stdout = subprocess.check_output( - [sys.executable, "-c", script], - env=envb) - stdout = stdout.rstrip(b'\n\r') - self.assertEqual(stdout.decode('ascii'), ascii(valueb)) - - def test_bytes_program(self) -> None: - abs_program = os.fsencode(sys.executable) - path, programs = os.path.split(sys.executable) - program = os.fsencode(programs) - - # absolute bytes path - exitcode = subprocess.call([abs_program, "-c", "pass"]) - self.assertEqual(exitcode, 0) - - # bytes program, unicode PATH - env = os.environ.copy() - env["PATH"] = path - exitcode = subprocess.call([program, "-c", "pass"], env=env) - self.assertEqual(exitcode, 0) - - # bytes program, bytes PATH - envb = os.environb.copy() - envb[b"PATH"] = os.fsencode(path) - exitcode = subprocess.call([program, "-c", "pass"], env=envb) - self.assertEqual(exitcode, 0) - - def test_pipe_cloexec(self) -> None: - sleeper = support.findfile("input_reader.py", subdir="subprocessdata") - fd_status = support.findfile("fd_status.py", subdir="subprocessdata") - - p1 = subprocess.Popen([sys.executable, sleeper], - stdin=subprocess.PIPE, stdout=subprocess.PIPE, - stderr=subprocess.PIPE, close_fds=False) - - self.addCleanup(p1.communicate, b'') - - p2 = subprocess.Popen([sys.executable, fd_status], - stdout=subprocess.PIPE, close_fds=False) - - output, error = p2.communicate() - result_fds = set(map(int, output.split(b','))) - unwanted_fds = set([p1.stdin.fileno(), p1.stdout.fileno(), - p1.stderr.fileno()]) - - self.assertFalse(result_fds & unwanted_fds, - "Expected no fds from %r to be open in child, " - "found %r" % - (unwanted_fds, result_fds & unwanted_fds)) - - def test_pipe_cloexec_real_tools(self) -> None: - qcat = support.findfile("qcat.py", subdir="subprocessdata") - qgrep = support.findfile("qgrep.py", subdir="subprocessdata") - - subdata = b'zxcvbn' - data = subdata * 4 + b'\n' - - p1 = subprocess.Popen([sys.executable, qcat], - stdin=subprocess.PIPE, stdout=subprocess.PIPE, - close_fds=False) - - p2 = subprocess.Popen([sys.executable, qgrep, subdata], - stdin=p1.stdout, stdout=subprocess.PIPE, - close_fds=False) - - self.addCleanup(p1.wait) - self.addCleanup(p2.wait) - def kill_p1() -> None: - #try: - p1.terminate() - #except ProcessLookupError: - # pass - def kill_p2() -> None: - #try: - p2.terminate() - #except ProcessLookupError: - # pass - self.addCleanup(kill_p1) - self.addCleanup(kill_p2) - - p1.stdin.write(data) - p1.stdin.close() - - readfiles, ignored1, ignored2 = select.select([p2.stdout], [], [], 10) - - self.assertTrue(readfiles, "The child hung") - self.assertEqual(p2.stdout.read(), data) - - p1.stdout.close() - p2.stdout.close() - - def test_close_fds(self) -> None: - fd_status = support.findfile("fd_status.py", subdir="subprocessdata") - - fds = os.pipe() - self.addCleanup(os.close, fds[0]) - self.addCleanup(os.close, fds[1]) - - open_fds = set([fds[0], fds[1]]) - # add a bunch more fds - for _ in range(9): - fd = os.open("/dev/null", os.O_RDONLY) - self.addCleanup(os.close, fd) - open_fds.add(fd) - - p = subprocess.Popen([sys.executable, fd_status], - stdout=subprocess.PIPE, close_fds=False) - output, ignored = p.communicate() - remaining_fds = set(map(int, output.split(b','))) - - self.assertEqual(remaining_fds & open_fds, open_fds, - "Some fds were closed") - - p = subprocess.Popen([sys.executable, fd_status], - stdout=subprocess.PIPE, close_fds=True) - output, ignored = p.communicate() - remaining_fds = set(map(int, output.split(b','))) - - self.assertFalse(remaining_fds & open_fds, - "Some fds were left open") - self.assertIn(1, remaining_fds, "Subprocess failed") - - # Keep some of the fd's we opened open in the subprocess. - # This tests _posixsubprocess.c's proper handling of fds_to_keep. - fds_to_keep = set(open_fds.pop() for _ in range(8)) - p = subprocess.Popen([sys.executable, fd_status], - stdout=subprocess.PIPE, close_fds=True, - pass_fds=()) - output, ignored = p.communicate() - remaining_fds = set(map(int, output.split(b','))) - - self.assertFalse(remaining_fds & fds_to_keep & open_fds, - "Some fds not in pass_fds were left open") - self.assertIn(1, remaining_fds, "Subprocess failed") - - # Mac OS X Tiger (10.4) has a kernel bug: sometimes, the file - # descriptor of a pipe closed in the parent process is valid in the - # child process according to fstat(), but the mode of the file - # descriptor is invalid, and read or write raise an error. - @support.requires_mac_ver(10, 5) - def test_pass_fds(self) -> None: - fd_status = support.findfile("fd_status.py", subdir="subprocessdata") - - open_fds = set() # type: Set[int] - - for x in range(5): - fds = os.pipe() - self.addCleanup(os.close, fds[0]) - self.addCleanup(os.close, fds[1]) - open_fds.update([fds[0], fds[1]]) - - for fd in open_fds: - p = subprocess.Popen([sys.executable, fd_status], - stdout=subprocess.PIPE, close_fds=True, - pass_fds=(fd, )) - output, ignored = p.communicate() - - remaining_fds = set(map(int, output.split(b','))) - to_be_closed = open_fds - {fd} - - self.assertIn(fd, remaining_fds, "fd to be passed not passed") - self.assertFalse(remaining_fds & to_be_closed, - "fd to be closed passed") - - # pass_fds overrides close_fds with a warning. - with self.assertWarns(RuntimeWarning) as context: - self.assertFalse(subprocess.call( - [sys.executable, "-c", "import sys; sys.exit(0)"], - close_fds=False, pass_fds=(fd, ))) - self.assertIn('overriding close_fds', str(context.warning)) - - def test_stdout_stdin_are_single_inout_fd(self) -> None: - with io.open(os.devnull, "r+") as inout: - p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], - stdout=inout, stdin=inout) - p.wait() - - def test_stdout_stderr_are_single_inout_fd(self) -> None: - with io.open(os.devnull, "r+") as inout: - p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], - stdout=inout, stderr=inout) - p.wait() - - def test_stderr_stdin_are_single_inout_fd(self) -> None: - with io.open(os.devnull, "r+") as inout: - p = subprocess.Popen([sys.executable, "-c", "import sys; sys.exit(0)"], - stderr=inout, stdin=inout) - p.wait() - - def test_wait_when_sigchild_ignored(self) -> None: - # NOTE: sigchild_ignore.py may not be an effective test on all OSes. - sigchild_ignore = support.findfile("sigchild_ignore.py", - subdir="subprocessdata") - p = subprocess.Popen([sys.executable, sigchild_ignore], - stdout=subprocess.PIPE, stderr=subprocess.PIPE) - stdout, stderr = p.communicate() - self.assertEqual(0, p.returncode, "sigchild_ignore.py exited" - " non-zero with this error:\n%s" % - stderr.decode('utf8')) - - def test_select_unbuffered(self) -> None: - # Issue #11459: bufsize=0 should really set the pipes as - # unbuffered (and therefore let select() work properly). - select = support.import_module("select") - p = subprocess.Popen([sys.executable, "-c", - 'import sys;' - 'sys.stdout.write("apple")'], - stdout=subprocess.PIPE, - bufsize=0) - f = p.stdout - self.addCleanup(f.close) - try: - self.assertEqual(f.read(4), b"appl") - self.assertIn(f, select.select([f], [], [], 0.0)[0]) - finally: - p.wait() - - def test_zombie_fast_process_del(self) -> None: - # Issue #12650: on Unix, if Popen.__del__() was called before the - # process exited, it wouldn't be added to subprocess._active, and would - # remain a zombie. - # spawn a Popen, and delete its reference before it exits - p = subprocess.Popen([sys.executable, "-c", - 'import sys, time;' - 'time.sleep(0.2)'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - ident = id(p) - pid = p.pid - del p - # check that p is in the active processes list - self.assertIn(ident, [id(o) for o in subprocess._active]) - - def test_leak_fast_process_del_killed(self) -> None: - # Issue #12650: on Unix, if Popen.__del__() was called before the - # process exited, and the process got killed by a signal, it would never - # be removed from subprocess._active, which triggered a FD and memory - # leak. - # spawn a Popen, delete its reference and kill it - p = subprocess.Popen([sys.executable, "-c", - 'import time;' - 'time.sleep(3)'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - ident = id(p) - pid = p.pid - del p - os.kill(pid, signal.SIGKILL) - # check that p is in the active processes list - self.assertIn(ident, [id(o) for o in subprocess._active]) - - # let some time for the process to exit, and create a new Popen: this - # should trigger the wait() of p - time.sleep(0.2) - with self.assertRaises(EnvironmentError) as c: - with subprocess.Popen(['nonexisting_i_hope'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) as proc: - pass - # p should have been wait()ed on, and removed from the _active list - self.assertRaises(OSError, os.waitpid, pid, 0) - self.assertNotIn(ident, [id(o) for o in subprocess._active]) - - -@unittest.skipUnless(mswindows, "Windows specific tests") -class Win32ProcessTestCase(BaseTestCase): - - def test_startupinfo(self) -> None: - # startupinfo argument - # We uses hardcoded constants, because we do not want to - # depend on win32all. - STARTF_USESHOWWINDOW = 1 - SW_MAXIMIZE = 3 - startupinfo = subprocess.STARTUPINFO() - startupinfo.dwFlags = STARTF_USESHOWWINDOW - startupinfo.wShowWindow = SW_MAXIMIZE - # Since Python is a console process, it won't be affected - # by wShowWindow, but the argument should be silently - # ignored - subprocess.call([sys.executable, "-c", "import sys; sys.exit(0)"], - startupinfo=startupinfo) - - def test_creationflags(self) -> None: - # creationflags argument - CREATE_NEW_CONSOLE = 16 - sys.stderr.write(" a DOS box should flash briefly ...\n") - subprocess.call(sys.executable + - ' -c "import time; time.sleep(0.25)"', - creationflags=CREATE_NEW_CONSOLE) - - def test_invalid_args(self) -> None: - # invalid arguments should raise ValueError - self.assertRaises(ValueError, subprocess.call, - [sys.executable, "-c", - "import sys; sys.exit(47)"], - preexec_fn=lambda: 1) - self.assertRaises(ValueError, subprocess.call, - [sys.executable, "-c", - "import sys; sys.exit(47)"], - stdout=subprocess.PIPE, - close_fds=True) - - def test_close_fds(self) -> None: - # close file descriptors - rc = subprocess.call([sys.executable, "-c", - "import sys; sys.exit(47)"], - close_fds=True) - self.assertEqual(rc, 47) - - def test_shell_sequence(self) -> None: - # Run command through the shell (sequence) - newenv = os.environ.copy() - newenv["FRUIT"] = "physalis" - p = subprocess.Popen(["set"], shell=1, - stdout=subprocess.PIPE, - env=newenv) - self.addCleanup(p.stdout.close) - self.assertIn(b"physalis", p.stdout.read()) - - def test_shell_string(self) -> None: - # Run command through the shell (string) - newenv = os.environ.copy() - newenv["FRUIT"] = "physalis" - p = subprocess.Popen("set", shell=1, - stdout=subprocess.PIPE, - env=newenv) - self.addCleanup(p.stdout.close) - self.assertIn(b"physalis", p.stdout.read()) - - def test_call_string(self) -> None: - # call() function with string argument on Windows - rc = subprocess.call(sys.executable + - ' -c "import sys; sys.exit(47)"') - self.assertEqual(rc, 47) - - def _kill_process(self, method: str, *args: Any) -> None: - # Some win32 buildbot raises EOFError if stdin is inherited - p = subprocess.Popen([sys.executable, "-c", """if 1: - import sys, time - sys.stdout.write('x\\n') - sys.stdout.flush() - time.sleep(30) - """], - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) - self.addCleanup(p.stdout.close) - self.addCleanup(p.stderr.close) - self.addCleanup(p.stdin.close) - # Wait for the interpreter to be completely initialized before - # sending any signal. - p.stdout.read(1) - getattr(p, method)(*args) - _, stderr = p.communicate() - self.assertStderrEqual(stderr, b'') - returncode = p.wait() - self.assertNotEqual(returncode, 0) - - def test_send_signal(self) -> None: - self._kill_process('send_signal', signal.SIGTERM) - - def test_kill(self) -> None: - self._kill_process('kill') - - def test_terminate(self) -> None: - self._kill_process('terminate') - - -# The module says: -# "NB This only works (and is only relevant) for UNIX." -# -# Actually, getoutput should work on any platform with an os.popen, but -# I'll take the comment as given, and skip this suite. -@unittest.skipUnless(os.name == 'posix', "only relevant for UNIX") -class CommandTests(unittest.TestCase): - def test_getoutput(self) -> None: - self.assertEqual(subprocess.getoutput('echo xyzzy'), 'xyzzy') - self.assertEqual(subprocess.getstatusoutput('echo xyzzy'), - (0, 'xyzzy')) - - # we use mkdtemp in the next line to create an empty directory - # under our exclusive control; from that, we can invent a pathname - # that we _know_ won't exist. This is guaranteed to fail. - dir = None # type: str - try: - dir = tempfile.mkdtemp() - name = os.path.join(dir, "foo") - - status, output = subprocess.getstatusoutput('cat ' + name) - self.assertNotEqual(status, 0) - finally: - if dir is not None: - os.rmdir(dir) - - -@unittest.skipUnless(getattr(subprocess, '_has_poll', False), - "poll system call not supported") -class ProcessTestCaseNoPoll(ProcessTestCase): - def setUp(self) -> None: - subprocess._has_poll = False - ProcessTestCase.setUp(self) - - def tearDown(self) -> None: - subprocess._has_poll = True - ProcessTestCase.tearDown(self) - - -#@unittest.skipUnless(getattr(subprocess, '_posixsubprocess', False), -# "_posixsubprocess extension module not found.") -#class ProcessTestCasePOSIXPurePython(ProcessTestCase, POSIXProcessTestCase): -# @classmethod -# def setUpClass(cls): -# global subprocess -# assert subprocess._posixsubprocess -# # Reimport subprocess while forcing _posixsubprocess to not exist. -# with support.check_warnings(('.*_posixsubprocess .* not being used.*', -# RuntimeWarning)): -# subprocess = support.import_fresh_module( -# 'subprocess', blocked=['_posixsubprocess']) -# assert not subprocess._posixsubprocess -# -# @classmethod -# def tearDownClass(cls): -# global subprocess -# # Reimport subprocess as it should be, restoring order to the universe#. -# subprocess = support.import_fresh_module('subprocess') -# assert subprocess._posixsubprocess - - -class HelperFunctionTests(unittest.TestCase): - @unittest.skipIf(mswindows, "errno and EINTR make no sense on windows") - def test_eintr_retry_call(self) -> None: - record_calls = [] # type: List[Any] - def fake_os_func(*args: Any) -> tuple: - record_calls.append(args) - if len(record_calls) == 2: - raise OSError(errno.EINTR, "fake interrupted system call") - return tuple(reversed(args)) - - self.assertEqual((999, 256), - subprocess._eintr_retry_call(fake_os_func, 256, 999)) - self.assertEqual([(256, 999)], record_calls) - # This time there will be an EINTR so it will loop once. - self.assertEqual((666,), - subprocess._eintr_retry_call(fake_os_func, 666)) - self.assertEqual([(256, 999), (666,), (666,)], record_calls) - - -@unittest.skipUnless(mswindows, "Windows-specific tests") -class CommandsWithSpaces (BaseTestCase): - - def setUp(self) -> None: - super().setUp() - f, fname = mkstemp(".py", "te st") - self.fname = fname.lower () - os.write(f, b"import sys;" - b"sys.stdout.write('%d %s' % (len(sys.argv), [a.lower () for a in sys.argv]))" - ) - os.close(f) - - def tearDown(self) -> None: - os.remove(self.fname) - super().tearDown() - - def with_spaces(self, *args: Any, **kwargs: Any) -> None: - kwargs['stdout'] = subprocess.PIPE - p = subprocess.Popen(*args, **kwargs) - self.addCleanup(p.stdout.close) - self.assertEqual( - p.stdout.read ().decode("mbcs"), - "2 [%r, 'ab cd']" % self.fname - ) - - def test_shell_string_with_spaces(self) -> None: - # call() function with string argument with spaces on Windows - self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname, - "ab cd"), shell=1) - - def test_shell_sequence_with_spaces(self) -> None: - # call() function with sequence argument with spaces on Windows - self.with_spaces([sys.executable, self.fname, "ab cd"], shell=1) - - def test_noshell_string_with_spaces(self) -> None: - # call() function with string argument with spaces on Windows - self.with_spaces('"%s" "%s" "%s"' % (sys.executable, self.fname, - "ab cd")) - - def test_noshell_sequence_with_spaces(self) -> None: - # call() function with sequence argument with spaces on Windows - self.with_spaces([sys.executable, self.fname, "ab cd"]) - - -class ContextManagerTests(BaseTestCase): - - def test_pipe(self) -> None: - with subprocess.Popen([sys.executable, "-c", - "import sys;" - "sys.stdout.write('stdout');" - "sys.stderr.write('stderr');"], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) as proc: - self.assertEqual(proc.stdout.read(), b"stdout") - self.assertStderrEqual(proc.stderr.read(), b"stderr") - - self.assertTrue(proc.stdout.closed) - self.assertTrue(proc.stderr.closed) - - def test_returncode(self) -> None: - with subprocess.Popen([sys.executable, "-c", - "import sys; sys.exit(100)"]) as proc: - pass - # __exit__ calls wait(), so the returncode should be set - self.assertEqual(proc.returncode, 100) - - def test_communicate_stdin(self) -> None: - with subprocess.Popen([sys.executable, "-c", - "import sys;" - "sys.exit(sys.stdin.read() == 'context')"], - stdin=subprocess.PIPE) as proc: - proc.communicate(b"context") - self.assertEqual(proc.returncode, 1) - - def test_invalid_args(self) -> None: - with self.assertRaises(EnvironmentError) as c: - with subprocess.Popen(['nonexisting_i_hope'], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE) as proc: - pass - - if c.exception.errno != errno.ENOENT: # ignore "no such file" - raise c.exception - - -def test_main(): - unit_tests = (ProcessTestCase, - POSIXProcessTestCase, - Win32ProcessTestCase, - #ProcessTestCasePOSIXPurePython, - CommandTests, - ProcessTestCaseNoPoll, - HelperFunctionTests, - CommandsWithSpaces, - ContextManagerTests, - ) - - support.run_unittest(*unit_tests) - support.reap_children() - -if __name__ == "__main__": - unittest.main()