Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Vespene subprocess example
def execute_command(build, command, input_text=None, env=None, log_command=True, output_log=True, message_log=False, timeout=None):
"""
Execute a command (a list or string) with input_text as input, appending
the output of all commands to the build log.
"""
timeout_cmd = get_timeout()
shell = True
if type(command) == list:
if timeout and timeout_cmd:
command.insert(0, timeout)
command.insert(0, timeout_cmd)
shell = False
else:
if timeout and timeout_cmd:
command = "%s %s %s" % (timeout_cmd, timeout, command)
sock = os.environ.get('SSH_AUTH_SOCK', None)
if env and sock:
env['SSH_AUTH_SOCK'] = sock
if log_command:
LOG.debug("executing: %s" % command)
if build:
build.append_message(command)
process = subprocess.Popen(command, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, shell=shell, env=env)
if input_text is None:
input_text = ""
stdin = io.TextIOWrapper(
process.stdin,
encoding='utf-8',
line_buffering=True,
)
stdout = io.TextIOWrapper(
process.stdout,
encoding='utf-8',
)
stdin.write(input_text)
stdin.close()
out = ""
for line in stdout:
line = ansi_escape.sub('', line)
if build:
check_if_can_continue(build)
if output_log:
build.append_output(line)
if message_log:
build.append_message(line)
out = "" + line
if line.startswith("vespene/set"):
handle_output_variables(build, line)
process.wait()
if process.returncode != 0:
build.append_message("build failed with exit code %s" % process.returncode)
build.status = FAILURE
build.return_code = process.returncode
build.save(force_update=True)
raise Exception("Failed")
return out
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment