Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def execute_command(self, cmd, *cmd_args):
"""Run arbitrary Terraform command."""
_LOGGER.debug([cmd, cmd_args])
cmdline = [self.terraform, cmd]
cmdline += cmd_args
p = subprocess.Popen(cmdline, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
cwd=self.tfdir, env=os.environ.copy())
out, err = p.communicate()
out = out.decode('utf-8', errors='ignore')
err = err.decode('utf-8', errors='ignore')
retcode = p.returncode
if retcode == 1:
message = 'Error running command {command}: {retcode} {out} {err}'.format(
command=cmd, retcode=retcode, out=out, err=err)
_LOGGER.critical(message)
raise TerraformTestError(message)
return TerraformCommandOutput(retcode, out, err)
Args:
tf_vars: the Terraform variables to use for plan/apply/destroy
plan: run plan
output: run output
destroy: run destroy if apply raises an error
Returns:
Wrapped output if output is run from here.
"""
if plan:
self.plan(tf_vars=tf_vars)
# catch errors so we can run destroy before re-raising
try:
self.apply(tf_vars=tf_vars)
except TerraformTestError:
if destroy:
_LOGGER.warn('running teardown to clean up')
self.teardown(tf_vars)
raise
if output:
return self.output()