meissa/dda-python-terraform
6
3
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Merge pull request #1 from surround-io/sammck-develop

Browse source 
Add full support for 'output' command, and enable raise_on_error option
This commit is contained in:
Sam McKelvie 2017-10-13 13:52:38 -07:00 committed by GitHub
commit 723ab0b79e
3 changed files with 126 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
.gitignore vendored
View file

@ -7,4 +7,5 @@
/.pypirc
/.tox/
.dropbox
Icon
Icon
/pytestdebug.log

72
python_terraform/__init__.py
View file

@ -10,6 +10,7 @@ import tempfile
from python_terraform.tfstate import Tfstate
try: # Python 2.7+
from logging import NullHandler
except ImportError:
@ -29,6 +30,12 @@ class IsNotFlagged:
pass
class TerraformCommandError(subprocess.CalledProcessError):
def __init__(self, ret_code, cmd, out, err):
super(TerraformCommandError, self).__init__(ret_code, cmd)
self.out = out
self.err = err
class Terraform(object):
"""
Wrapper of terraform command line tool
@ -252,10 +259,17 @@ class Terraform(object):
if the option 'capture_output' is passed (with any value other than
True), terraform output will be printed to stdout/stderr and
"None" will be returned as out and err.
if the option 'raise_on_error' is passed (with any value that evaluates to True),
and the terraform command returns a nonzerop return code, then
a TerraformCommandError exception will be raised. The exception object will
have the following properties:
returncode: The command's return code
out: The captured stdout, or None if not captured
err: The captured stderr, or None if not captured
:return: ret_code, out, err
"""
capture_output = kwargs.pop('capture_output', True)
raise_on_error = kwargs.pop('raise_on_error', False)
if capture_output is True:
stderr = subprocess.PIPE
stdout = subprocess.PIPE
@ -285,27 +299,61 @@ class Terraform(object):
self.temp_var_files.clean_up()
if capture_output is True:
return ret_code, out.decode('utf-8'), err.decode('utf-8')
out = out.decode('utf-8')
err = err.decode('utf-8')
else:
return ret_code, None, None
out = None
err = None
def output(self, name, *args, **kwargs):
if ret_code != 0 and raise_on_error:
raise TerraformCommandError(
ret_code, ' '.join(cmds), out=out, err=err)
return ret_code, out, err
def output(self, *args, **kwargs):
"""
https://www.terraform.io/docs/commands/output.html
:param name: name of output
:return: output value
Note that this method does not conform to the (ret_code, out, err) return convention. To use
the "output" command with the standard convention, call "output_cmd" instead of
"output".
:param args: Positional arguments. There is one optional positional
argument NAME; if supplied, the returned output text
will be the json for a single named output value.
:param kwargs: Named options, passed to the command. In addition,
'full_value': If True, and NAME is provided, then
the return value will be a dict with
"value', 'type', and 'sensitive'
properties.
:return: None, if an error occured
Output value as a string, if NAME is provided and full_value
is False or not provided
Output value as a dict with 'value', 'sensitive', and 'type' if
NAME is provided and full_value is True.
dict of named dicts each with 'value', 'sensitive', and 'type',
if NAME is not provided
"""
full_value = kwargs.pop('full_value', False)
name_provided = (len(args) > 0)
kwargs['json'] = IsFlagged
kwargs['capture_output'] = True
ret, out, err = self.cmd(
'output', name, json=IsFlagged, *args, **kwargs)
ret, out, err = self.output_cmd(*args, **kwargs)
log.debug('output raw string: {0}'.format(out))
if ret != 0:
return None
return None
out = out.lstrip()
output_dict = json.loads(out)
return output_dict['value']
value = json.loads(out)
if name_provided and not full_value:
value = value['value']
return value
def read_state_file(self, file_path=None):
"""

68
test/test_terraform.py
View file

@ -12,6 +12,7 @@ import fnmatch
logging.basicConfig(level=logging.DEBUG)
root_logger = logging.getLogger()
current_path = os.path.dirname(os.path.realpath(__file__))
FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS = "test 'test.out!"
@ -30,12 +31,14 @@ STRING_CASES = [
]
CMD_CASES = [
['method', 'expected_output', 'expected_ret_code', 'expected_logs', 'folder'],
['method', 'expected_output', 'expected_ret_code', 'expected_exception', 'expected_logs', 'folder'],
[
[
lambda x: x.cmd('plan', 'var_to_output', no_color=IsFlagged, var={'test_var': 'test'}) ,
"doesn't need to do anything",
#"doesn't need to do anything",
"no\nactions need to be performed",
0,
False,
'',
'var_to_output'
],
@ -44,6 +47,16 @@ CMD_CASES = [
lambda x: x.cmd('import', 'aws_instance.foo', 'i-abcd1234', no_color=IsFlagged),
'',
1,
False,
'command: terraform import -no-color aws_instance.foo i-abcd1234',
''
],
# try import aws instance with raise_on_error
[
lambda x: x.cmd('import', 'aws_instance.foo', 'i-abcd1234', no_color=IsFlagged, raise_on_error=True),
'',
1,
True,
'command: terraform import -no-color aws_instance.foo i-abcd1234',
''
],
@ -52,6 +65,7 @@ CMD_CASES = [
lambda x: x.cmd('plan', 'var_to_output', out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS),
'',
0,
False,
'',
'var_to_output'
]
@ -123,10 +137,18 @@ class TestTerraform(object):
assert s in result
@pytest.mark.parametrize(*CMD_CASES)
def test_cmd(self, method, expected_output, expected_ret_code, expected_logs, string_logger, folder):
def test_cmd(self, method, expected_output, expected_ret_code, expected_exception, expected_logs, string_logger, folder):
tf = Terraform(working_dir=current_path)
tf.init(folder)
ret, out, err = method(tf)
try:
ret, out, err = method(tf)
assert not expected_exception
except TerraformCommandError as e:
assert expected_exception
ret = e.returncode
out = e.out
err = e.err
logs = string_logger()
logs = logs.replace('\n', '')
assert expected_output in out
@ -223,6 +245,44 @@ class TestTerraform(object):
else:
assert result == 'test'
@pytest.mark.parametrize(
("param"),
[
({}),
({'module': 'test2'}),
]
)
def test_output_full_value(self, param, string_logger):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
tf.apply('var_to_output')
result = tf.output('test_output', **dict(param, full_value=True))
regex = re.compile("terraform output (-module=test2 -json|-json -module=test2) test_output")
log_str = string_logger()
if param:
assert re.search(regex, log_str), log_str
else:
assert result['value'] == 'test'
@pytest.mark.parametrize(
("param"),
[
({}),
({'module': 'test2'}),
]
)
def test_output_all(self, param, string_logger):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')
tf.apply('var_to_output')
result = tf.output(**param)
regex = re.compile("terraform output (-module=test2 -json|-json -module=test2)")
log_str = string_logger()
if param:
assert re.search(regex, log_str), log_str
else:
assert result['test_output']['value'] == 'test'
def test_destroy(self):
tf = Terraform(working_dir=current_path, variables={'test_var': 'test'})
tf.init('var_to_output')