(arg_list, **kwargs)
| 87 | return redacted, secrets |
| 88 | |
| 89 | def _exec_command(arg_list, **kwargs): |
| 90 | silent = False |
| 91 | if 'silent' in kwargs: |
| 92 | silent = True |
| 93 | del kwargs['silent'] |
| 94 | |
| 95 | redact_flags = kwargs.pop('redact_flags', ['--password', '--passphrase', '--token', '--secret', '--api-key', '--client-secret']) |
| 96 | if isinstance(redact_flags, str): |
| 97 | redact_flags = [redact_flags] |
| 98 | secrets = _to_str_set(kwargs.pop('redacted_values', None)) |
| 99 | |
| 100 | arg_str = arg_list |
| 101 | if not isinstance(arg_str, str): |
| 102 | arg_list = [_to_str(x) for x in arg_list] |
| 103 | redacted_args, secrets = _redact_args(arg_list, redact_flags, secrets) |
| 104 | arg_str = ' '.join(redacted_args) |
| 105 | else: |
| 106 | arg_str = _sanitize_text(arg_str, secrets) |
| 107 | if not silent: log('[exec] %s' % arg_str) |
| 108 | |
| 109 | if sys.stdout.isatty(): |
| 110 | # If not on CI, we want the colored output, and we get the output as it runs, in order to preserve the colors |
| 111 | if not 'stdout' in kwargs: |
| 112 | kwargs['stdout'] = subprocess.PIPE # Only way to get output from the command |
| 113 | process = subprocess.Popen(arg_list, **kwargs) |
| 114 | output = process.communicate()[0] |
| 115 | if process.returncode != 0: |
| 116 | if not silent: log(_sanitize_text(output, secrets)) |
| 117 | else: |
| 118 | # On the CI machines, we make sure we produce a steady stream of output |
| 119 | # However, this also makes us lose the color information |
| 120 | if 'stdout' in kwargs: |
| 121 | del kwargs['stdout'] |
| 122 | process = subprocess.Popen(arg_list, stdout = subprocess.PIPE, stderr = subprocess.STDOUT, **kwargs) |
| 123 | |
| 124 | output = '' |
| 125 | while True: |
| 126 | line = process.stdout.readline().decode(errors='replace') |
| 127 | if line != '': |
| 128 | output += line |
| 129 | if not silent: log(_sanitize_text(line, secrets).rstrip()) |
| 130 | else: |
| 131 | break |
| 132 | |
| 133 | if process.wait() != 0: |
| 134 | e = ExecException(process.returncode) |
| 135 | e.output = _sanitize_text(output, secrets) |
| 136 | log('[exec] %s' % arg_str) |
| 137 | log("Error: %s" % _sanitize_text(output, secrets)) |
| 138 | raise e |
| 139 | |
| 140 | output = _to_str(output) |
| 141 | return output.strip() |
| 142 | |
| 143 | def command(args, **kwargs): |
| 144 | if kwargs.get("shell") is None: |
no test coverage detected