fixed a bug with var-file argument, allow workspace commands to pass flags and options

This commit is contained in:
BNMetrics 2019-06-25 19:53:39 +01:00
parent a63e9dad24
commit f39b7e237e
4 changed files with 155 additions and 42 deletions

8
.gitignore vendored
View file

@ -9,3 +9,11 @@
.dropbox
Icon
/pytestdebug.log
.DS_Store
# virtualenv
.virtualenv/
# Intellij
.idea/

View file

@ -21,6 +21,7 @@ except ImportError:
log = logging.getLogger(__name__)
log.addHandler(NullHandler())
COMMAND_WITH_SUBCOMMANDS = ['workspace']
class IsFlagged:
pass
@ -36,6 +37,7 @@ class TerraformCommandError(subprocess.CalledProcessError):
self.out = out
self.err = err
class Terraform(object):
"""
Wrapper of terraform command line tool
@ -205,6 +207,10 @@ class Terraform(object):
"""
cmds = cmd.split()
cmds = [self.terraform_bin_path] + cmds
if cmd in COMMAND_WITH_SUBCOMMANDS:
args = list(args)
subcommand = args.pop(0)
cmds.append(subcommand)
for option, value in kwargs.items():
if '_' in option:
@ -223,9 +229,13 @@ class Terraform(object):
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
else:
elif option == 'var':
# We do not create empty var-files if there is no var passed.
# An empty var-file would result in an error: An argument or block definition is required here
if value:
filename = self.temp_var_files.create(value)
cmds += ['-var-file={0}'.format(filename)]
continue
# simple flag,
@ -319,7 +329,6 @@ class Terraform(object):
return ret_code, out, err
def output(self, *args, **kwargs):
"""
https://www.terraform.io/docs/commands/output.html
@ -388,40 +397,40 @@ class Terraform(object):
self.tfstate = Tfstate.load_file(file_path)
def set_workspace(self, workspace):
def set_workspace(self, workspace, *args, **kwargs):
"""
set workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace' ,'select', workspace)
return self.cmd('workspace', 'select', workspace, *args, **kwargs)
def create_workspace(self, workspace):
def create_workspace(self, workspace, *args, **kwargs):
"""
create workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace', 'new', workspace)
return self.cmd('workspace', 'new', workspace, *args, **kwargs)
def delete_workspace(self, workspace):
def delete_workspace(self, workspace, *args, **kwargs):
"""
delete workspace
:param workspace: the desired workspace.
:return: status
"""
return self.cmd('workspace', 'delete', workspace)
return self.cmd('workspace', 'delete', workspace, *args, **kwargs)
def show_workspace(self):
def show_workspace(self, **kwargs):
"""
show workspace
show workspace, this command does not need the [DIR] part
:return: workspace
"""
return self.cmd('workspace', 'show')
return self.cmd('workspace', 'show', **kwargs)
def __exit__(self, exc_type, exc_value, traceback):
self.temp_var_files.clean_up()

View file

@ -3,6 +3,7 @@ try:
except ImportError:
from io import StringIO
from python_terraform import *
from contextlib import contextmanager
import pytest
import os
import logging
@ -34,7 +35,7 @@ CMD_CASES = [
['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'}) ,
lambda x: x.cmd('plan', 'var_to_output', no_color=IsFlagged, var={'test_var': 'test'}),
# 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
@ -69,7 +70,16 @@ CMD_CASES = [
False,
'',
'var_to_output'
]
],
# test workspace command (commands with subcommand)
[
lambda x: x.cmd('workspace', 'show', no_color=IsFlagged),
'',
0,
False,
'command: terraform workspace show -no-color',
''
],
]
]
@ -102,6 +112,28 @@ def string_logger(request):
return lambda: str(log_stream.getvalue())
@pytest.fixture()
def workspace_setup_teardown():
"""
Fixture used in workspace related tests
Create and tear down a workspace
*Use as a contextmanager*
"""
@contextmanager
def wrapper(workspace_name, create=True, delete=True, *args, **kwargs):
tf = Terraform(working_dir=current_path)
tf.init()
if create:
tf.create_workspace(workspace_name, *args, **kwargs)
yield tf
if delete:
tf.set_workspace('default')
tf.delete_workspace(workspace_name)
yield wrapper
class TestTerraform(object):
def teardown_method(self, method):
""" teardown any state that was previously setup with a setup_method
@ -184,6 +216,17 @@ class TestTerraform(object):
assert expected_output in out.replace('\n', '').replace(' ', '')
assert err == ''
def test_apply_with_var_file(self, string_logger):
tf = Terraform(working_dir=current_path)
var_file = os.path.join(current_path, 'tfvar_file', 'test.tfvars')
tf.init()
tf.apply(var_file=os.path.join(current_path, 'tfvar_file', 'test.tfvars'))
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform apply -var-file={} -no-color -input=false -auto-approve=false'.format(var_file)
assert expected_log in logs
@pytest.mark.parametrize(
['cmd', 'args', 'options'],
[
@ -321,41 +364,93 @@ class TestTerraform(object):
tf.import_cmd('aws_instance.foo', 'i-abc1234', no_color=IsFlagged)
assert 'command: terraform import -no-color aws_instance.foo i-abc1234' in string_logger()
def test_create_workspace(self):
tf = Terraform(working_dir=current_path)
tf.init()
def test_create_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name, create=False) as tf:
ret, out, err = tf.create_workspace('test')
tf.set_workspace('default')
tf.delete_workspace('test')
assert ret == 0
assert err == ''
def test_set_workspace(self):
tf = Terraform(working_dir=current_path)
tf.init()
tf.create_workspace('test')
tf.set_workspace('test')
tf.set_workspace('default')
ret, out, err = tf.delete_workspace('test')
def test_create_workspace_with_args(
self, workspace_setup_teardown, string_logger
):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name, create=False) as tf:
ret, out, err = tf.create_workspace('test', current_path, lock=True, no_color=IsFlagged)
assert ret == 0
assert err == ''
def test_show_workspace(self):
tf = Terraform(working_dir=current_path)
tf.init()
tf.create_workspace('test')
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace new -lock=true -no-color test {}'.format(current_path)
assert expected_log in logs
def test_set_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.set_workspace(workspace_name)
assert ret == 0
assert err == ''
def test_set_workspace_with_args(
self, workspace_setup_teardown, string_logger):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.set_workspace(workspace_name, current_path, no_color=IsFlagged)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace select -no-color test {}'.format(current_path)
assert expected_log in logs
def test_show_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.show_workspace()
tf.set_workspace('default')
tf.delete_workspace('test')
assert ret == 0
assert err == ''
def test_delete_workspace(self):
tf = Terraform(working_dir=current_path)
tf.init()
tf.create_workspace('test')
tf.set_workspace('default')
ret, out, err = tf.delete_workspace('test')
tf.show_workspace()
def test_show_workspace_with_no_color(
self, workspace_setup_teardown, string_logger
):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.show_workspace(no_color=IsFlagged)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace show -no-color'
assert expected_log in logs
def test_delete_workspace(self, workspace_setup_teardown):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name, delete=False) as tf:
tf.set_workspace('default')
ret, out, err = tf.delete_workspace(workspace_name)
assert ret == 0
assert err == ''
def test_delete_workspace_with_args(
self, workspace_setup_teardown, string_logger
):
workspace_name = 'test'
with workspace_setup_teardown(workspace_name, delete=False) as tf:
tf.set_workspace('default')
ret, out, err = tf.delete_workspace(
workspace_name, current_path, force=IsFlagged, no_color=IsFlagged, lock=True,
)
assert ret == 0
assert err == ''
logs = string_logger()
logs = logs.replace('\n', '')
expected_log = 'command: terraform workspace delete -force -no-color -lock=true test {}'.format(current_path)
assert expected_log in logs

View file

@ -0,0 +1 @@
test_var = "True"