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