Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_args():
assert tftest.parse_args() == []
for kwargs, expected in ARGS_TESTS:
assert tftest.parse_args(**kwargs) == expected
def test_targets():
assert tftest.parse_args(targets=['one', 'two']) == sorted(
['-target=one', '-target=two'])
def test_var_args():
assert sorted(tftest.parse_args(init_vars={'a': 1, 'b': '["2"]'})) == sorted(
["-backend-config='a=1'", '-backend-config=\'b=["2"]\''])
assert sorted(tftest.parse_args(tf_vars={'a': 1, 'b': '["2"]'})) == sorted(
['-var', 'b=["2"]', '-var', 'a=1'])
def test_args():
assert tftest.parse_args() == []
for kwargs, expected in ARGS_TESTS:
assert tftest.parse_args(**kwargs) == expected
def test_var_args():
assert sorted(tftest.parse_args(init_vars={'a': 1, 'b': '["2"]'})) == sorted(
["-backend-config='a=1'", '-backend-config=\'b=["2"]\''])
assert sorted(tftest.parse_args(tf_vars={'a': 1, 'b': '["2"]'})) == sorted(
['-var', 'b=["2"]', '-var', 'a=1'])
def output(self, name=None, color=False, json_format=True):
"""Run Terraform output command."""
cmd_args = []
if name:
cmd_args.append(name)
cmd_args += parse_args(color=color, json_format=json_format)
output = self.execute_command('output', *cmd_args).out
_LOGGER.debug('output %s', output)
if json_format:
try:
output = TerraformOutputs(json.loads(output))
except json.JSONDecodeError as e:
_LOGGER.warn('error decoding output: {}'.format(e))
self.last_output = output
return output
def __init__(self, path, raw):
self._raw = raw
self.path = path
self.outputs = TerraformOutputs(raw['outputs'])
self.depends_on = raw['depends_on']
# key type provider attributes depends_on
self.resources = {}
for k, v in raw['resources'].items():
self.resources[k] = TerraformStateResource(
k, v['provider'], v['type'],
v.get('primary', {}).get('attributes', {}), v['depends_on'], v)
def __init__(self, raw):
super(TerraformOutputs, self).__init__(raw)
self.sensitive = tuple(k for k, v in raw.items() if v.get('sensitive'))
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()