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

Import changes from gitlab/dda/python-terraform

Browse Source
merge-requests/1/head
Th0masL 2 years ago
parent 8a3451c54c
commit faac8aea46
No known key found for this signature in database
GPG Key ID: 5111AAE5DF075B38
9 changed files with 337 additions and 157 deletions
Show Stats Download Patch File Download Diff File

22
.gitignore vendored
Unescape Escape View File

@ -1,16 +1,17 @@
# terraform
*.tfstate
*.tfstate.backup
# python
*.pyc
*.egg-info
.idea
.cache
__pycache__
/.pypirc
/.tox/
.dropbox
env/
Icon
dist/
build/
/pytestdebug.log
.DS_Store
.pytestdebug.log
# virtualenv
.virtualenv/
@ -18,8 +19,17 @@ venv/
# Intellij
.idea
.idea/
# VSCode
.vscode/
pyrightconfig.json
tmp.txt
# other
.dropbox
.DS_Store
/.tox/
env/
Icon

5
CHANGELOG.md
Unescape Escape View File

@ -17,3 +17,8 @@
## [0.10.1]
1. [#48] adding extension for temp file to adopt the change in terraform 0.12.0
1. [#49] add workspace support
## [1.0.1]
1. adding option to output the latest cmd
1. added refresh command
1. intenden to work with tf1.0

12
doc/releasing.md
Unescape Escape View File

@ -0,0 +1,12 @@
## Release
```
adjust version no in setup.py to release version no.
git commit -am "release"
git tag -am "release" release-[release version no]
git push --follow-tags
increase version no in setup.py
git commit -am "version bump"
git push
pip3 install --upgrade --user python-terraform
```

221
python_terraform/terraform.py
Unescape Escape View File

@ -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] = 0.13
):
"""
: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()
@ -85,15 +87,7 @@ class Terraform:
self.tfstate = None
self.read_state_file(self.state)
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
self.latest_cmd = ''
def apply(
self,
@ -115,30 +109,38 @@ 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 _generate_default_args(self, dir_or_plan: Optional[str]) -> Sequence[str]:
return [dir_or_plan] if dir_or_plan else []
def refresh(
self,
dir_or_plan: Optional[str] = None,
input: bool = False,
no_color: Type[TerraformFlag] = IsFlagged,
**kwargs,
) -> CommandOutput:
"""Refer to https://terraform.io/docs/commands/refresh.html
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,
}
no-color is flagged by default
: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 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(global_opts, "refresh", *args, **option_dict)
def destroy(
self,
@ -151,11 +153,15 @@ 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
# force is no longer a flag in version >= 1.0
if self.terraform_version < 1.0:
default["force"] = force
default["auto-approve"] = True
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,
@ -170,11 +176,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,
@ -208,10 +215,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:
@ -234,57 +242,23 @@ 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)
return cmds
def cmd(
self,
global_opts: Dict[str, Any],
cmd: str,
*args,
capture_output: Union[bool, str] = True,
@ -328,7 +302,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
@ -367,7 +341,7 @@ class Terraform:
return ret_code, out, err
def output(
self, *args, capture_output: bool = True, **kwargs
self, dir_or_plan: Optional[str] = None, *args, capture_output: bool = True, **kwargs
) -> Union[None, str, Dict[str, str], Dict[str, Dict[str, str]]]:
"""Refer https://www.terraform.io/docs/commands/output.html
@ -395,7 +369,10 @@ class Terraform:
if capture_output is False:
raise ValueError("capture_output is required for this method")
ret, out, _ = self.output_cmd(*args, **kwargs)
global_opts = self._generate_default_general_options(dir_or_plan)
ret, out, _ = self.cmd(global_opts, "output", *args, **kwargs)
# ret, out, _ = self.output_cmd(global_opts, *args, **kwargs)
if ret:
return None
@ -431,7 +408,8 @@ class Terraform:
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "select", workspace, *args, **kwargs)
global_opts = self._generate_default_general_options(False)
return self.cmd(global_opts, "workspace", "select", workspace, *args, **kwargs)
def create_workspace(self, workspace, *args, **kwargs) -> CommandOutput:
"""Create workspace
@ -439,7 +417,8 @@ class Terraform:
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "new", workspace, *args, **kwargs)
global_opts = self._generate_default_general_options(False)
return self.cmd(global_opts, "workspace", "new", workspace, *args, **kwargs)
def delete_workspace(self, workspace, *args, **kwargs) -> CommandOutput:
"""Delete workspace
@ -447,18 +426,20 @@ class Terraform:
:param workspace: the desired workspace.
:return: status
"""
return self.cmd("workspace", "delete", workspace, *args, **kwargs)
global_opts = self._generate_default_general_options(False)
return self.cmd(global_opts, "workspace", "delete", workspace, *args, **kwargs)
def show_workspace(self, **kwargs) -> CommandOutput:
"""Show workspace, this command does not need the [DIR] part
:return: workspace
"""
return self.cmd("workspace", "show", **kwargs)
global_opts = self._generate_default_general_options(False)
return self.cmd(global_opts, "workspace", "show", **kwargs)
def list_workspace(self) -> List[str]:
"""List of workspaces
:return: workspaces
:example:
>>> tf = Terraform()
@ -475,9 +456,93 @@ class Terraform:
)
)
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)
global_opts = self._generate_default_general_options(False)
return self.cmd(global_opts, cmd_name, *args, **kwargs)
return wrapper
class VariableFiles:
def __init__(self):

1
requirements.txt
Unescape Escape View File

@ -1 +0,0 @@
/bin/bash: q : commande introuvable

15
requirements_dev.txt
Unescape Escape View File

@ -1,4 +1,11 @@
wheel
flake8
pytest
pytest-cov
coverage==5.3
flake8==3.8.4
flake8-polyfill==1.0.2
mypy==0.790
mypy-extensions==0.4.3
pycodestyle==2.6.0
pyflakes==2.2.0
pylint==2.6.0
pytest==6.1.2
pytest-cov==2.10.1
pytest-datafiles==2.0

2
setup.py
Unescape Escape View File

@ -21,7 +21,7 @@ except IOError:
setup(
name=module_name,
version="0.14.0",
version="1.0.4-dev",
url="https://github.com/beelit94/python-terraform",
license="MIT",
author="Freddy Tan",

210
test/test_terraform.py
Unescape Escape View File

@ -17,21 +17,31 @@ root_logger = logging.getLogger()
current_path = os.path.dirname(os.path.realpath(__file__))
version = 1.0 if (os.environ.get("TFVER") and os.environ.get(
"TFVER").startswith("1")) else 0.13
FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS = "test 'test.out!"
STRING_CASES = [
[
lambda x: x.generate_cmd_string("apply", "the_folder", no_color=IsFlagged),
lambda x: x.generate_cmd_string(
{}, "apply", "the_folder", no_color=IsFlagged),
"terraform apply -no-color the_folder",
],
[
lambda x: x.generate_cmd_string(
"push", "path", vcs=True, token="token", atlas_address="url"
),
lambda x: x.generate_cmd_string({},
"push", "path", vcs=True, token="token", atlas_address="url"
),
"terraform push -vcs=true -token=token -atlas-address=url path",
],
[
lambda x: x.generate_cmd_string({},
"refresh", "path", token="token"
),
"terraform refresh -token=token path",
],
]
CMD_CASES = [
CMD_CASES_0_x = [
[
"method",
"expected_output",
@ -43,6 +53,7 @@ CMD_CASES = [
[
[
lambda x: x.cmd(
{},
"plan",
"var_to_output",
no_color=IsFlagged,
@ -59,6 +70,7 @@ CMD_CASES = [
# try import aws instance
[
lambda x: x.cmd(
{},
"import",
"aws_instance.foo",
"i-abcd1234",
@ -74,6 +86,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,
@ -88,7 +101,77 @@ 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
"Changes to Outputs:",
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,
@ -99,6 +182,22 @@ CMD_CASES = [
],
]
APPLY_CASES_0_x = [
["folder", "variables", "var_files", "expected_output", "options"],
[("var_to_output", {"test_var": "test"}, None, 'test_output=test', {}),
("var_to_output", {"test_list_var": ["c", "d"]}, None, 'test_list_output=["c","d",]', {},),
("var_to_output", {"test_map_var": {"c": "c", "d": "d"}}, None, 'test_map_output={"c"="c""d"="d"}', {},),
("var_to_output", {"test_map_var": {"c": "c", "d": "d"}}, "var_to_output/test_map_var.json", 'test_map_output={"e"="e""f"="f"}', {},),
("var_to_output", {}, None, "\x1b[0m\x1b[1m\x1b[32mApplycomplete!", {"no_color": IsNotFlagged},), ]]
APPLY_CASES_1_x = [
["folder", "variables", "var_files", "expected_output", "options"],
[("var_to_output", {"test_var": "test"}, None, 'test_output="test"', {}),
("var_to_output", {"test_list_var": ["c", "d"]}, None, 'test_list_output=tolist(["c","d",])', {},),
("var_to_output", {"test_map_var": {"c": "c", "d": "d"}}, None, 'test_map_output=tomap({"c"="c""d"="d"})', {},),
("var_to_output", {"test_map_var": {"c": "c", "d": "d"}}, "test_map_var.json", 'test_map_output=tomap({"e"="e""f"="f"})', {},),
("var_to_output", {}, None, "\x1b[0m\x1b[1m\x1b[32mApplycomplete!", {"no_color": IsNotFlagged},), ]]
@pytest.fixture(scope="function")
def fmt_test_file(request):
@ -137,7 +236,7 @@ def workspace_setup_teardown():
@contextmanager
def wrapper(workspace_name, create=True, delete=True, *args, **kwargs):
tf = Terraform(working_dir=current_path)
tf = Terraform(working_dir=current_path, terraform_version=version)
tf.init()
if create:
tf.create_workspace(workspace_name, *args, **kwargs)
@ -152,7 +251,8 @@ def workspace_setup_teardown():
class TestTerraform:
def teardown_method(self, _) -> None:
"""Teardown any state that was previously setup with a setup_method call."""
exclude = ["test_tfstate_file", "test_tfstate_file2", "test_tfstate_file3"]
exclude = ["test_tfstate_file",
"test_tfstate_file2", "test_tfstate_file3"]
def purge(dir: str, pattern: str) -> None:
for root, dirnames, filenames in os.walk(dir):
@ -171,14 +271,14 @@ class TestTerraform:
@pytest.mark.parametrize(["method", "expected"], STRING_CASES)
def test_generate_cmd_string(self, method: Callable[..., str], expected: str):
tf = Terraform(working_dir=current_path)
tf = Terraform(working_dir=current_path, terraform_version=version)
result = method(tf)
strs = expected.split()
for s in strs:
assert s in result
@pytest.mark.parametrize(*CMD_CASES)
@pytest.mark.parametrize(*(CMD_CASES_1_x if version >= 1.0 else CMD_CASES_0_x))
def test_cmd(
self,
method: Callable[..., str],
@ -190,7 +290,7 @@ class TestTerraform:
folder: str,
):
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path)
tf = Terraform(working_dir=current_path, terraform_version=version)
tf.init(folder)
try:
ret, out, _ = method(tf)
@ -204,44 +304,11 @@ class TestTerraform:
assert expected_ret_code == ret
assert expected_logs in caplog.text
@pytest.mark.parametrize(
("folder", "variables", "var_files", "expected_output", "options"),
[
("var_to_output", {"test_var": "test"}, None, "test_output=test", {}),
(
"var_to_output",
{"test_list_var": ["c", "d"]},
None,
'test_list_output=["c","d",]',
{},
),
(
"var_to_output",
{"test_map_var": {"c": "c", "d": "d"}},
None,
'test_map_output={"c"="c""d"="d"}',
{},
),
(
"var_to_output",
{"test_map_var": {"c": "c", "d": "d"}},
"var_to_output/test_map_var.json",
# Values are overriden
'test_map_output={"e"="e""f"="f"}',
{},
),
(
"var_to_output",
{},
None,
"\x1b[0m\x1b[1m\x1b[32mApplycomplete!",
{"no_color": IsNotFlagged},
),
],
)
@pytest.mark.parametrize(*(APPLY_CASES_1_x if version >= 1.0 else APPLY_CASES_0_x))
def test_apply(self, folder, variables, var_files, expected_output, options):
tf = Terraform(
working_dir=current_path, variables=variables, var_file=var_files
working_dir=current_path, variables=variables, var_file=var_files, terraform_version=version
)
tf.init(folder)
ret, out, err = tf.apply(folder, **options)
@ -251,13 +318,13 @@ class TestTerraform:
def test_apply_with_var_file(self, caplog: LogCaptureFixture):
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path)
tf = Terraform(working_dir=current_path, terraform_version=version)
folder = "var_to_output"
tf.init(folder)
tf.apply(
folder,
var_file=os.path.join(current_path, "tfvar_files", "test.tfvars"),
var_file=os.path.join(
current_path, "tfvar_files", "test.tfvars"),
)
for log in caplog.messages:
if log.startswith("Command: terraform apply"):
@ -271,46 +338,49 @@ class TestTerraform:
],
)
def test_options(self, cmd, args, options, fmt_test_file):
tf = Terraform(working_dir=current_path)
tf = Terraform(working_dir=current_path, terraform_version=version)
ret, out, err = getattr(tf, cmd)(*args, **options)
assert ret == 0
assert out == ""
def test_state_data(self):
cwd = os.path.join(current_path, "test_tfstate_file")
tf = Terraform(working_dir=cwd, state="tfstate.test")
tf = Terraform(working_dir=cwd, state="tfstate.test",
terraform_version=version)
tf.read_state_file()
assert tf.tfstate.modules[0]["path"] == ["root"]
def test_state_default(self):
cwd = os.path.join(current_path, "test_tfstate_file2")
tf = Terraform(working_dir=cwd)
tf = Terraform(working_dir=cwd, terraform_version=version)
tf.read_state_file()
assert tf.tfstate.modules[0]["path"] == ["default"]
def test_state_default_backend(self):
cwd = os.path.join(current_path, "test_tfstate_file3")
tf = Terraform(working_dir=cwd)
tf = Terraform(working_dir=cwd, terraform_version=version)
tf.read_state_file()
assert tf.tfstate.modules[0]["path"] == ["default_backend"]
def test_pre_load_state_data(self):
cwd = os.path.join(current_path, "test_tfstate_file")
tf = Terraform(working_dir=cwd, state="tfstate.test")
tf = Terraform(working_dir=cwd, state="tfstate.test",
terraform_version=version)
assert tf.tfstate.modules[0]["path"] == ["root"]
@pytest.mark.parametrize(
("folder", "variables"), [("var_to_output", {"test_var": "test"})]
)
def test_override_default(self, folder, variables):
tf = Terraform(working_dir=current_path, variables=variables)
tf = Terraform(working_dir=current_path,
variables=variables, terraform_version=version)
tf.init(folder)
ret, out, err = tf.apply(
folder, var={"test_var": "test2"}, no_color=IsNotFlagged,
)
out = out.replace("\n", "")
assert "\x1b[0m\x1b[1m\x1b[32mApply" in out
out = tf.output("test_output")
out = tf.output(folder, "test_output")
assert "test2" in out
@pytest.mark.parametrize("output_all", [True, False])
@ -319,12 +389,12 @@ class TestTerraform:
required_output = "test_output"
with caplog.at_level(logging.INFO):
tf = Terraform(
working_dir=current_path, variables={"test_var": expected_value}
working_dir=current_path, variables={"test_var": expected_value}, terraform_version=version
)
tf.init("var_to_output")
tf.apply("var_to_output")
params = tuple() if output_all else (required_output,)
result = tf.output(*params)
result = tf.output("var_to_output", *params)
if output_all:
assert result[required_output]["value"] == expected_value
else:
@ -332,7 +402,8 @@ class TestTerraform:
assert expected_value in caplog.messages[-1]
def test_destroy(self):
tf = Terraform(working_dir=current_path, variables={"test_var": "test"})
tf = Terraform(working_dir=current_path, variables={
"test_var": "test"}, terraform_version=version)
tf.init("var_to_output")
ret, out, err = tf.destroy("var_to_output")
assert ret == 0
@ -342,17 +413,18 @@ class TestTerraform:
("plan", "variables", "expected_ret"), [("vars_require_input", {}, 1)]
)
def test_plan(self, plan, variables, expected_ret):
tf = Terraform(working_dir=current_path, variables=variables)
tf = Terraform(working_dir=current_path,
variables=variables, terraform_version=version)
tf.init(plan)
with pytest.raises(TerraformCommandError) as e:
tf.plan(plan)
assert (
e.value.err
== """\nError: Missing required argument\n\nThe argument "region" is required, but was not set.\n\n"""
"\nError:" in e.value.err
)
def test_fmt(self, fmt_test_file):
tf = Terraform(working_dir=current_path, variables={"test_var": "test"})
tf = Terraform(working_dir=current_path, variables={
"test_var": "test"}, terraform_version=version)
ret, out, err = tf.fmt(diff=True)
assert ret == 0
@ -363,6 +435,9 @@ class TestTerraform:
assert ret == 0
assert err == ""
# The directory flag is no longer supported in v1.0.8
# this should either be done with -chdir or probably just be removed
"""
def test_create_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
state_file_path = os.path.join(
@ -381,6 +456,7 @@ class TestTerraform:
f"Command: terraform workspace new -no-color test {current_path}"
in caplog.messages
)
"""
def test_set_workspace(self, workspace_setup_teardown):
workspace_name = "test"
@ -389,6 +465,8 @@ class TestTerraform:
assert ret == 0
assert err == ""
# see comment on test_create_workspace_with_args
"""
def test_set_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf, caplog.at_level(
@ -404,6 +482,7 @@ class TestTerraform:
f"Command: terraform workspace select -no-color test {current_path}"
in caplog.messages
)
"""
def test_show_workspace(self, workspace_setup_teardown):
workspace_name = "test"
@ -431,6 +510,8 @@ class TestTerraform:
assert ret == 0
assert err == ""
# see above comments
"""
def test_delete_workspace_with_args(self, workspace_setup_teardown, caplog):
workspace_name = "test"
with workspace_setup_teardown(
@ -447,6 +528,7 @@ class TestTerraform:
f"Command: terraform workspace delete -force test {current_path}"
in caplog.messages
)
"""
def test_list_workspace(self):
tf = Terraform(working_dir=current_path)

6
test/vars_require_input/main.tf
Unescape Escape View File

@ -1,16 +1,16 @@
variable "ami" {
default = "foo"
type = "string"
type = string
}
variable "list" {
default = []
type = "list"
type = list
}
variable "map" {
default = {}
type = "map"
type = map
}
resource "aws_instance" "bar" {

Write Preview
Loading…
Cancel
Save