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

adjust cmd to new 1.0 release

Browse source
This commit is contained in:
jem 2021-10-23 15:37:29 +02:00
parent 9c73aaec5e
commit 607cdaf408
2 changed files with 182 additions and 78 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

176
python_terraform/terraform.py
View file

@ -52,6 +52,7 @@ class Terraform:
var_file: Optional[str] = None, var_file: Optional[str] = None,
terraform_bin_path: Optional[str] = None, terraform_bin_path: Optional[str] = None,
is_env_vars_included: bool = True, is_env_vars_included: bool = True,
terraform_version: Optional[float] = 1.0
): ):
""" """
:param working_dir: the folder of the working folder, if not given, :param working_dir: the folder of the working folder, if not given,
@ -78,6 +79,7 @@ class Terraform:
self.terraform_bin_path = ( self.terraform_bin_path = (
terraform_bin_path if terraform_bin_path else "terraform" terraform_bin_path if terraform_bin_path else "terraform"
) )
self.terraform_version = terraform_version
self.var_file = var_file self.var_file = var_file
self.temp_var_files = VariableFiles() self.temp_var_files = VariableFiles()
@ -87,16 +89,6 @@ class Terraform:
self.latest_cmd = '' self.latest_cmd = ''
def __getattr__(self, item: str) -> Callable:
def wrapper(*args, **kwargs):
cmd_name = str(item)
if cmd_name.endswith("_cmd"):
cmd_name = cmd_name[:-4]
logger.debug("called with %r and %r", args, kwargs)
return self.cmd(cmd_name, *args, **kwargs)
return wrapper
def apply( def apply(
self, self,
dir_or_plan: Optional[str] = None, dir_or_plan: Optional[str] = None,
@ -117,13 +109,14 @@ class Terraform:
""" """
if not skip_plan: if not skip_plan:
return self.plan(dir_or_plan=dir_or_plan, **kwargs) return self.plan(dir_or_plan=dir_or_plan, **kwargs)
global_opts = self._generate_default_general_options(dir_or_plan)
default = kwargs.copy() default = kwargs.copy()
default["input"] = input default["input"] = input
default["no_color"] = no_color default["no_color"] = no_color
default["auto-approve"] = True # a False value will require an input default["auto-approve"] = True # a False value will require an input
option_dict = self._generate_default_options(default) option_dict = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan) args = self._generate_default_args(dir_or_plan)
return self.cmd("apply", *args, **option_dict) return self.cmd(global_opts, "apply", *args, **option_dict)
def refresh( def refresh(
self, self,
@ -141,29 +134,13 @@ class Terraform:
:param kwargs: same as kwags in method 'cmd' :param kwargs: same as kwags in method 'cmd'
:returns return_code, stdout, stderr :returns return_code, stdout, stderr
""" """
global_opts = self._generate_default_general_options(dir_or_plan)
default = kwargs.copy() default = kwargs.copy()
default["input"] = input default["input"] = input
default["no_color"] = no_color default["no_color"] = no_color
option_dict = self._generate_default_options(default) option_dict = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan) args = self._generate_default_args(dir_or_plan)
return self.cmd("refresh", *args, **option_dict) return self.cmd(global_opts, "refresh", *args, **option_dict)
def _generate_default_args(self, dir_or_plan: Optional[str]) -> Sequence[str]:
return [dir_or_plan] if dir_or_plan else []
def _generate_default_options(
self, input_options: Dict[str, Any]
) -> Dict[str, Any]:
return {
"state": self.state,
"target": self.targets,
"var": self.variables,
"var_file": self.var_file,
"parallelism": self.parallelism,
"no_color": IsFlagged,
"input": False,
**input_options,
}
def destroy( def destroy(
self, self,
@ -176,11 +153,12 @@ class Terraform:
force/no-color option is flagged by default force/no-color option is flagged by default
:return: ret_code, stdout, stderr :return: ret_code, stdout, stderr
""" """
global_opts = self._generate_default_general_options(dir_or_plan)
default = kwargs.copy() default = kwargs.copy()
default["force"] = force default["force"] = force
options = self._generate_default_options(default) options = self._generate_default_options(default)
args = self._generate_default_args(dir_or_plan) args = self._generate_default_args(dir_or_plan)
return self.cmd("destroy", *args, **options) return self.cmd(global_opts, "destroy", *args, **options)
def plan( def plan(
self, self,
@ -195,11 +173,12 @@ class Terraform:
:param kwargs: options :param kwargs: options
:return: ret_code, stdout, stderr :return: ret_code, stdout, stderr
""" """
global_opts = self._generate_default_general_options(dir_or_plan)
options = kwargs.copy() options = kwargs.copy()
options["detailed_exitcode"] = detailed_exitcode options["detailed_exitcode"] = detailed_exitcode
options = self._generate_default_options(options) options = self._generate_default_options(options)
args = self._generate_default_args(dir_or_plan) args = self._generate_default_args(dir_or_plan)
return self.cmd("plan", *args, **options) return self.cmd(global_opts, "plan", *args, **options)
def init( def init(
self, self,
@ -233,10 +212,11 @@ class Terraform:
} }
) )
options = self._generate_default_options(options) options = self._generate_default_options(options)
global_opts = self._generate_default_general_options(dir_or_plan)
args = self._generate_default_args(dir_or_plan) args = self._generate_default_args(dir_or_plan)
return self.cmd("init", *args, **options) return self.cmd(global_opts, "init", *args, **options)
def generate_cmd_string(self, cmd: str, *args, **kwargs) -> List[str]: def generate_cmd_string(self, global_options: Dict[str, Any], cmd: str, *args, **kwargs) -> List[str]:
"""For any generate_cmd_string doesn't written as public method of Terraform """For any generate_cmd_string doesn't written as public method of Terraform
examples: examples:
@ -259,51 +239,15 @@ class Terraform:
:param kwargs: same as kwags in method 'cmd' :param kwargs: same as kwags in method 'cmd'
:return: string of valid terraform command :return: string of valid terraform command
""" """
cmds = cmd.split() cmds = [self.terraform_bin_path]
cmds = [self.terraform_bin_path] + cmds cmds += self._generate_cmd_options(**global_options)
cmds += cmd.split()
if cmd in COMMAND_WITH_SUBCOMMANDS: if cmd in COMMAND_WITH_SUBCOMMANDS:
args = list(args) args = list(args)
subcommand = args.pop(0) subcommand = args.pop(0)
cmds.append(subcommand) cmds.append(subcommand)
for option, value in kwargs.items(): cmds += self._generate_cmd_options(**kwargs)
if "_" in option:
option = option.replace("_", "-")
if isinstance(value, list):
for sub_v in value:
cmds += [f"-{option}={sub_v}"]
continue
if isinstance(value, dict):
if "backend-config" in option:
for bk, bv in value.items():
cmds += [f"-backend-config={bk}={bv}"]
continue
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
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 += [f"-var-file={filename}"]
continue
# simple flag,
if value is IsFlagged:
cmds += [f"-{option}"]
continue
if value is None or value is IsNotFlagged:
continue
if isinstance(value, bool):
value = "true" if value else "false"
cmds += [f"-{option}={value}"]
cmds += args cmds += args
self.latest_cmd = ' '.join(cmds) self.latest_cmd = ' '.join(cmds)
@ -311,6 +255,7 @@ class Terraform:
def cmd( def cmd(
self, self,
global_opts: Dict[str, Any],
cmd: str, cmd: str,
*args, *args,
capture_output: Union[bool, str] = True, capture_output: Union[bool, str] = True,
@ -354,7 +299,7 @@ class Terraform:
stderr = sys.stderr stderr = sys.stderr
stdout = sys.stdout stdout = sys.stdout
cmds = self.generate_cmd_string(cmd, *args, **kwargs) cmds = self.generate_cmd_string(global_opts, cmd, *args, **kwargs)
logger.info("Command: %s", " ".join(cmds)) logger.info("Command: %s", " ".join(cmds))
working_folder = self.working_dir if self.working_dir else None working_folder = self.working_dir if self.working_dir else None
@ -482,9 +427,92 @@ class Terraform:
""" """
return self.cmd("workspace", "show", **kwargs) return self.cmd("workspace", "show", **kwargs)
def _generate_default_args(self, dir_or_plan: Optional[str]) -> Sequence[str]:
if (self.terraform_version < 1.0 and dir_or_plan):
return [dir_or_plan]
else:
return []
def _generate_default_general_options(self, dir_or_plan: Optional[str]) -> Dict[str, Any]:
if (self.terraform_version >= 1.0 and dir_or_plan):
return {"chdir": dir_or_plan}
else:
return {}
def _generate_default_options(
self, input_options: Dict[str, Any]
) -> Dict[str, Any]:
return {
"state": self.state,
"target": self.targets,
"var": self.variables,
"var_file": self.var_file,
"parallelism": self.parallelism,
"no_color": IsFlagged,
"input": False,
**input_options,
}
def _generate_cmd_options(self, **kwargs) -> List[str]:
"""
"""
result = []
for option, value in kwargs.items():
if "_" in option:
option = option.replace("_", "-")
if isinstance(value, list):
for sub_v in value:
result += [f"-{option}={sub_v}"]
continue
if isinstance(value, dict):
if "backend-config" in option:
for bk, bv in value.items():
result += [f"-backend-config={bk}={bv}"]
continue
# since map type sent in string won't work, create temp var file for
# variables, and clean it up later
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)
result += [f"-var-file={filename}"]
continue
# simple flag,
if value is IsFlagged:
result += [f"-{option}"]
continue
if value is None or value is IsNotFlagged:
continue
if isinstance(value, bool):
value = "true" if value else "false"
result += [f"-{option}={value}"]
return result
def __exit__(self, exc_type, exc_value, traceback) -> None: def __exit__(self, exc_type, exc_value, traceback) -> None:
self.temp_var_files.clean_up() self.temp_var_files.clean_up()
def __getattr__(self, item: str) -> Callable:
def wrapper(*args, **kwargs):
cmd_name = str(item)
if cmd_name.endswith("_cmd"):
cmd_name = cmd_name[:-4]
logger.debug("called with %r and %r", args, kwargs)
return self.cmd(cmd_name, *args, **kwargs)
return wrapper
class VariableFiles: class VariableFiles:
def __init__(self): def __init__(self):

82
test/test_terraform.py
View file

@ -37,7 +37,7 @@ STRING_CASES = [
], ],
] ]
CMD_CASES = [ CMD_CASES_0_x = [
[ [
"method", "method",
"expected_output", "expected_output",
@ -49,6 +49,7 @@ CMD_CASES = [
[ [
[ [
lambda x: x.cmd( lambda x: x.cmd(
{},
"plan", "plan",
"var_to_output", "var_to_output",
no_color=IsFlagged, no_color=IsFlagged,
@ -65,6 +66,7 @@ CMD_CASES = [
# try import aws instance # try import aws instance
[ [
lambda x: x.cmd( lambda x: x.cmd(
{},
"import", "import",
"aws_instance.foo", "aws_instance.foo",
"i-abcd1234", "i-abcd1234",
@ -80,6 +82,7 @@ CMD_CASES = [
# test with space and special character in file path # test with space and special character in file path
[ [
lambda x: x.cmd( lambda x: x.cmd(
{},
"plan", "plan",
"var_to_output", "var_to_output",
out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS, out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS,
@ -94,7 +97,79 @@ CMD_CASES = [
# test workspace command (commands with subcommand) # test workspace command (commands with subcommand)
[ [
lambda x: x.cmd( lambda x: x.cmd(
"workspace", "show", no_color=IsFlagged, raise_on_error=False {}, "workspace", "show", no_color=IsFlagged, raise_on_error=False
),
"",
0,
False,
"Command: terraform workspace show -no-color",
"",
],
],
]
CMD_CASES_1_x = [
[
"method",
"expected_output",
"expected_ret_code",
"expected_exception",
"expected_logs",
"folder",
],
[
[
lambda x: x.cmd(
{"chdir": "var_to_output"},
"plan",
"",
no_color=IsFlagged,
var={"test_var": "test"},
raise_on_error=False,
),
# Expected output varies by terraform version
"Plan: 0 to add, 0 to change, 0 to destroy.",
0,
False,
"",
"var_to_output",
],
# try import aws instance
[
lambda x: x.cmd(
{},
"import",
"aws_instance.foo",
"i-abcd1234",
no_color=IsFlagged,
raise_on_error=False,
),
"",
1,
False,
"Error: No Terraform configuration files",
"",
],
# test with space and special character in file path
[
lambda x: x.cmd(
{"chdir": "var_to_output"},
"plan",
"",
out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS,
raise_on_error=False,
),
"",
0,
False,
"",
"var_to_output",
],
# test workspace command (commands with subcommand)
[
lambda x: x.cmd(
{}, "workspace", "show", no_color=IsFlagged, raise_on_error=False
), ),
"", "",
0, 0,
@ -184,7 +259,7 @@ class TestTerraform:
for s in strs: for s in strs:
assert s in result assert s in result
@pytest.mark.parametrize(*CMD_CASES) @pytest.mark.parametrize(*(CMD_CASES_0_x if (os.environ.get("TFVER") and os.environ.get("TFVER").startsWith("0")) else CMD_CASES_1_x))
def test_cmd( def test_cmd(
self, self,
method: Callable[..., str], method: Callable[..., str],
@ -210,6 +285,7 @@ class TestTerraform:
assert expected_ret_code == ret assert expected_ret_code == ret
assert expected_logs in caplog.text assert expected_logs in caplog.text
@pytest.mark.parametrize( @pytest.mark.parametrize(
("folder", "variables", "var_files", "expected_output", "options"), ("folder", "variables", "var_files", "expected_output", "options"),
[ [