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

Merge pull request #37 from beelit94/develop

Browse source 
release 0.10.0
This commit is contained in:
beelit94 2018-04-13 14:27:11 -07:00 committed by GitHub
commit 813d23d759
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 154 additions and 21 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
.bumpversion.cfg
View file

@ -1,5 +1,5 @@
[bumpversion]
current_version = 0.9.1
current_version = 0.10.0
commit = True
tag = False

3
.gitignore vendored
View file

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

8
CHANGELOG.md
View file

@ -7,4 +7,10 @@
### Fixed
1. [#12] Output function doesn't accept parameter 'module'
1. [#16] Handle empty space/special characters when passing string to command line options
1. Tested with terraform 0.10.0
1. Tested with terraform 0.10.0
## [0.10.0]
### Fixed
1. [#27] No interaction for apply function
1. [#18] Return access to the subprocess so output can be handled as desired
1. [#24] Full support for output(); support for raise_on_error

80
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
@ -84,7 +91,7 @@ class Terraform(object):
return wrapper
def apply(self, dir_or_plan=None, input=False, no_color=IsFlagged,
def apply(self, dir_or_plan=None, input=False, skip_plan=False, no_color=IsFlagged,
**kwargs):
"""
refer to https://terraform.io/docs/commands/apply.html
@ -92,12 +99,14 @@ class Terraform(object):
:param no_color: disable color of stdout
:param input: disable prompt for a missing variable
:param dir_or_plan: folder relative to working folder
:param skip_plan: force apply without plan (default: false)
:param kwargs: same as kwags in method 'cmd'
:returns return_code, stdout, stderr
"""
default = kwargs
default['input'] = input
default['no_color'] = no_color
default['auto-approve'] = (skip_plan == True)
option_dict = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan)
return self.cmd('apply', *args, **option_dict)
@ -252,10 +261,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
@ -274,6 +290,11 @@ class Terraform(object):
p = subprocess.Popen(cmds, stdout=stdout, stderr=stderr,
cwd=working_folder, env=environ_vars)
synchronous = kwargs.pop('synchronous', True)
if not synchronous:
return p, None, None
out, err = p.communicate()
ret_code = p.returncode
log.debug('output: {o}'.format(o=out))
@ -285,27 +306,62 @@ 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
if not kwargs.get('capture_output', True) is True:
raise ValueError('capture_output is required for this method')
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
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):
"""

2
setup.py
View file

@ -20,7 +20,7 @@ except IOError:
setup(
name=module_name,
version='0.9.1',
version='0.10.0',
url='https://github.com/beelit94/python-terraform',
license='MIT',
author='Freddy Tan',

80
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,15 @@ 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",
# Expected output varies by terraform version
["doesn't need to do anything", # Terraform < 0.10.7 (used in travis env)
"no\nactions need to be performed"], # Terraform >= 0.10.7
0,
False,
'',
'var_to_output'
],
@ -44,6 +48,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 +66,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,13 +138,30 @@ 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
if isinstance(expected_output, list):
ok = False
for xo in expected_output:
if xo in out:
ok = True
break
if not ok:
assert expected_output[0] in out
else:
assert expected_output in out
assert expected_ret_code == ret
assert expected_logs in logs
@ -223,6 +255,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')