Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/pip/_internal/utils/subprocess.py: 17%
100 statements
« prev ^ index » next coverage.py v7.2.7, created at 2023-06-07 06:48 +0000
1import logging
2import os
3import shlex
4import subprocess
5from typing import (
6 TYPE_CHECKING,
7 Any,
8 Callable,
9 Iterable,
10 List,
11 Mapping,
12 Optional,
13 Union,
14)
16from pip._vendor.rich.markup import escape
18from pip._internal.cli.spinners import SpinnerInterface, open_spinner
19from pip._internal.exceptions import InstallationSubprocessError
20from pip._internal.utils.logging import VERBOSE, subprocess_logger
21from pip._internal.utils.misc import HiddenText
23if TYPE_CHECKING:
24 # Literal was introduced in Python 3.8.
25 #
26 # TODO: Remove `if TYPE_CHECKING` when dropping support for Python 3.7.
27 from typing import Literal
29CommandArgs = List[Union[str, HiddenText]]
32def make_command(*args: Union[str, HiddenText, CommandArgs]) -> CommandArgs:
33 """
34 Create a CommandArgs object.
35 """
36 command_args: CommandArgs = []
37 for arg in args:
38 # Check for list instead of CommandArgs since CommandArgs is
39 # only known during type-checking.
40 if isinstance(arg, list):
41 command_args.extend(arg)
42 else:
43 # Otherwise, arg is str or HiddenText.
44 command_args.append(arg)
46 return command_args
49def format_command_args(args: Union[List[str], CommandArgs]) -> str:
50 """
51 Format command arguments for display.
52 """
53 # For HiddenText arguments, display the redacted form by calling str().
54 # Also, we don't apply str() to arguments that aren't HiddenText since
55 # this can trigger a UnicodeDecodeError in Python 2 if the argument
56 # has type unicode and includes a non-ascii character. (The type
57 # checker doesn't ensure the annotations are correct in all cases.)
58 return " ".join(
59 shlex.quote(str(arg)) if isinstance(arg, HiddenText) else shlex.quote(arg)
60 for arg in args
61 )
64def reveal_command_args(args: Union[List[str], CommandArgs]) -> List[str]:
65 """
66 Return the arguments in their raw, unredacted form.
67 """
68 return [arg.secret if isinstance(arg, HiddenText) else arg for arg in args]
71def call_subprocess(
72 cmd: Union[List[str], CommandArgs],
73 show_stdout: bool = False,
74 cwd: Optional[str] = None,
75 on_returncode: 'Literal["raise", "warn", "ignore"]' = "raise",
76 extra_ok_returncodes: Optional[Iterable[int]] = None,
77 extra_environ: Optional[Mapping[str, Any]] = None,
78 unset_environ: Optional[Iterable[str]] = None,
79 spinner: Optional[SpinnerInterface] = None,
80 log_failed_cmd: Optional[bool] = True,
81 stdout_only: Optional[bool] = False,
82 *,
83 command_desc: str,
84) -> str:
85 """
86 Args:
87 show_stdout: if true, use INFO to log the subprocess's stderr and
88 stdout streams. Otherwise, use DEBUG. Defaults to False.
89 extra_ok_returncodes: an iterable of integer return codes that are
90 acceptable, in addition to 0. Defaults to None, which means [].
91 unset_environ: an iterable of environment variable names to unset
92 prior to calling subprocess.Popen().
93 log_failed_cmd: if false, failed commands are not logged, only raised.
94 stdout_only: if true, return only stdout, else return both. When true,
95 logging of both stdout and stderr occurs when the subprocess has
96 terminated, else logging occurs as subprocess output is produced.
97 """
98 if extra_ok_returncodes is None:
99 extra_ok_returncodes = []
100 if unset_environ is None:
101 unset_environ = []
102 # Most places in pip use show_stdout=False. What this means is--
103 #
104 # - We connect the child's output (combined stderr and stdout) to a
105 # single pipe, which we read.
106 # - We log this output to stderr at DEBUG level as it is received.
107 # - If DEBUG logging isn't enabled (e.g. if --verbose logging wasn't
108 # requested), then we show a spinner so the user can still see the
109 # subprocess is in progress.
110 # - If the subprocess exits with an error, we log the output to stderr
111 # at ERROR level if it hasn't already been displayed to the console
112 # (e.g. if --verbose logging wasn't enabled). This way we don't log
113 # the output to the console twice.
114 #
115 # If show_stdout=True, then the above is still done, but with DEBUG
116 # replaced by INFO.
117 if show_stdout:
118 # Then log the subprocess output at INFO level.
119 log_subprocess: Callable[..., None] = subprocess_logger.info
120 used_level = logging.INFO
121 else:
122 # Then log the subprocess output using VERBOSE. This also ensures
123 # it will be logged to the log file (aka user_log), if enabled.
124 log_subprocess = subprocess_logger.verbose
125 used_level = VERBOSE
127 # Whether the subprocess will be visible in the console.
128 showing_subprocess = subprocess_logger.getEffectiveLevel() <= used_level
130 # Only use the spinner if we're not showing the subprocess output
131 # and we have a spinner.
132 use_spinner = not showing_subprocess and spinner is not None
134 log_subprocess("Running command %s", command_desc)
135 env = os.environ.copy()
136 if extra_environ:
137 env.update(extra_environ)
138 for name in unset_environ:
139 env.pop(name, None)
140 try:
141 proc = subprocess.Popen(
142 # Convert HiddenText objects to the underlying str.
143 reveal_command_args(cmd),
144 stdin=subprocess.PIPE,
145 stdout=subprocess.PIPE,
146 stderr=subprocess.STDOUT if not stdout_only else subprocess.PIPE,
147 cwd=cwd,
148 env=env,
149 errors="backslashreplace",
150 )
151 except Exception as exc:
152 if log_failed_cmd:
153 subprocess_logger.critical(
154 "Error %s while executing command %s",
155 exc,
156 command_desc,
157 )
158 raise
159 all_output = []
160 if not stdout_only:
161 assert proc.stdout
162 assert proc.stdin
163 proc.stdin.close()
164 # In this mode, stdout and stderr are in the same pipe.
165 while True:
166 line: str = proc.stdout.readline()
167 if not line:
168 break
169 line = line.rstrip()
170 all_output.append(line + "\n")
172 # Show the line immediately.
173 log_subprocess(line)
174 # Update the spinner.
175 if use_spinner:
176 assert spinner
177 spinner.spin()
178 try:
179 proc.wait()
180 finally:
181 if proc.stdout:
182 proc.stdout.close()
183 output = "".join(all_output)
184 else:
185 # In this mode, stdout and stderr are in different pipes.
186 # We must use communicate() which is the only safe way to read both.
187 out, err = proc.communicate()
188 # log line by line to preserve pip log indenting
189 for out_line in out.splitlines():
190 log_subprocess(out_line)
191 all_output.append(out)
192 for err_line in err.splitlines():
193 log_subprocess(err_line)
194 all_output.append(err)
195 output = out
197 proc_had_error = proc.returncode and proc.returncode not in extra_ok_returncodes
198 if use_spinner:
199 assert spinner
200 if proc_had_error:
201 spinner.finish("error")
202 else:
203 spinner.finish("done")
204 if proc_had_error:
205 if on_returncode == "raise":
206 error = InstallationSubprocessError(
207 command_description=command_desc,
208 exit_code=proc.returncode,
209 output_lines=all_output if not showing_subprocess else None,
210 )
211 if log_failed_cmd:
212 subprocess_logger.error("[present-rich] %s", error)
213 subprocess_logger.verbose(
214 "[bold magenta]full command[/]: [blue]%s[/]",
215 escape(format_command_args(cmd)),
216 extra={"markup": True},
217 )
218 subprocess_logger.verbose(
219 "[bold magenta]cwd[/]: %s",
220 escape(cwd or "[inherit]"),
221 extra={"markup": True},
222 )
224 raise error
225 elif on_returncode == "warn":
226 subprocess_logger.warning(
227 'Command "%s" had error code %s in %s',
228 command_desc,
229 proc.returncode,
230 cwd,
231 )
232 elif on_returncode == "ignore":
233 pass
234 else:
235 raise ValueError(f"Invalid value: on_returncode={on_returncode!r}")
236 return output
239def runner_with_spinner_message(message: str) -> Callable[..., None]:
240 """Provide a subprocess_runner that shows a spinner message.
242 Intended for use with for BuildBackendHookCaller. Thus, the runner has
243 an API that matches what's expected by BuildBackendHookCaller.subprocess_runner.
244 """
246 def runner(
247 cmd: List[str],
248 cwd: Optional[str] = None,
249 extra_environ: Optional[Mapping[str, Any]] = None,
250 ) -> None:
251 with open_spinner(message) as spinner:
252 call_subprocess(
253 cmd,
254 command_desc=message,
255 cwd=cwd,
256 extra_environ=extra_environ,
257 spinner=spinner,
258 )
260 return runner