diff --git a/python_terraform/terraform.py b/python_terraform/terraform.py index 3f61a1b..7d62720 100644 --- a/python_terraform/terraform.py +++ b/python_terraform/terraform.py @@ -52,6 +52,7 @@ class Terraform: var_file: Optional[str] = None, terraform_bin_path: Optional[str] = None, is_env_vars_included: bool = True, + terraform_version: Optional[float] = 1.0 ): """ :param working_dir: the folder of the working folder, if not given, @@ -78,6 +79,7 @@ class Terraform: self.terraform_bin_path = ( terraform_bin_path if terraform_bin_path else "terraform" ) + self.terraform_version = terraform_version self.var_file = var_file self.temp_var_files = VariableFiles() @@ -87,16 +89,6 @@ class Terraform: 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( self, dir_or_plan: Optional[str] = None, @@ -117,13 +109,14 @@ class Terraform: """ if not skip_plan: return self.plan(dir_or_plan=dir_or_plan, **kwargs) + global_opts = self._generate_default_general_options(dir_or_plan) default = kwargs.copy() default["input"] = input default["no_color"] = no_color default["auto-approve"] = True # a False value will require an input option_dict = self._generate_default_options(default) 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( self, @@ -141,29 +134,13 @@ class Terraform: :param kwargs: same as kwags in method 'cmd' :returns return_code, stdout, stderr """ + global_opts = self._generate_default_general_options(dir_or_plan) default = kwargs.copy() default["input"] = input default["no_color"] = no_color option_dict = self._generate_default_options(default) args = self._generate_default_args(dir_or_plan) - return self.cmd("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, - } + return self.cmd(global_opts, "refresh", *args, **option_dict) def destroy( self, @@ -176,11 +153,12 @@ class Terraform: force/no-color option is flagged by default :return: ret_code, stdout, stderr """ + global_opts = self._generate_default_general_options(dir_or_plan) default = kwargs.copy() default["force"] = force options = self._generate_default_options(default) args = self._generate_default_args(dir_or_plan) - return self.cmd("destroy", *args, **options) + return self.cmd(global_opts, "destroy", *args, **options) def plan( self, @@ -195,11 +173,12 @@ class Terraform: :param kwargs: options :return: ret_code, stdout, stderr """ + global_opts = self._generate_default_general_options(dir_or_plan) options = kwargs.copy() options["detailed_exitcode"] = detailed_exitcode options = self._generate_default_options(options) args = self._generate_default_args(dir_or_plan) - return self.cmd("plan", *args, **options) + return self.cmd(global_opts, "plan", *args, **options) def init( self, @@ -233,10 +212,11 @@ class Terraform: } ) options = self._generate_default_options(options) + global_opts = self._generate_default_general_options(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 examples: @@ -259,51 +239,15 @@ class Terraform: :param kwargs: same as kwags in method 'cmd' :return: string of valid terraform command """ - cmds = cmd.split() - cmds = [self.terraform_bin_path] + cmds + cmds = [self.terraform_bin_path] + cmds += self._generate_cmd_options(**global_options) + cmds += cmd.split() 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: - 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 += self._generate_cmd_options(**kwargs) cmds += args self.latest_cmd = ' '.join(cmds) @@ -311,6 +255,7 @@ class Terraform: def cmd( self, + global_opts: Dict[str, Any], cmd: str, *args, capture_output: Union[bool, str] = True, @@ -354,7 +299,7 @@ class Terraform: stderr = sys.stderr 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)) working_folder = self.working_dir if self.working_dir else None @@ -482,8 +427,91 @@ class Terraform: """ 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: 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: diff --git a/test/test_terraform.py b/test/test_terraform.py index f274031..6d1e05b 100644 --- a/test/test_terraform.py +++ b/test/test_terraform.py @@ -37,7 +37,7 @@ STRING_CASES = [ ], ] -CMD_CASES = [ +CMD_CASES_0_x = [ [ "method", "expected_output", @@ -49,6 +49,7 @@ CMD_CASES = [ [ [ lambda x: x.cmd( + {}, "plan", "var_to_output", no_color=IsFlagged, @@ -65,6 +66,7 @@ CMD_CASES = [ # try import aws instance [ lambda x: x.cmd( + {}, "import", "aws_instance.foo", "i-abcd1234", @@ -80,6 +82,7 @@ CMD_CASES = [ # test with space and special character in file path [ lambda x: x.cmd( + {}, "plan", "var_to_output", out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS, @@ -94,7 +97,79 @@ CMD_CASES = [ # test workspace command (commands with subcommand) [ 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, @@ -184,7 +259,7 @@ class TestTerraform: for s in strs: 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( self, method: Callable[..., str], @@ -194,7 +269,7 @@ class TestTerraform: expected_logs: str, caplog: LogCaptureFixture, folder: str, - ): + ): with caplog.at_level(logging.INFO): tf = Terraform(working_dir=current_path) tf.init(folder) @@ -210,6 +285,7 @@ class TestTerraform: assert expected_ret_code == ret assert expected_logs in caplog.text + @pytest.mark.parametrize( ("folder", "variables", "var_files", "expected_output", "options"), [