dda-python-terraform/test/test_terraform.py

538 lines
18 KiB
Python
Raw Normal View History

2020-10-19 22:30:02 +00:00
import fnmatch
import logging
2020-10-19 22:30:02 +00:00
import os
import re
2016-12-20 16:55:55 +00:00
import shutil
2020-10-19 22:30:02 +00:00
from contextlib import contextmanager
2020-10-20 22:08:58 +00:00
from io import StringIO
2021-01-21 10:07:00 +00:00
from typing import Callable
from packaging import version
2020-10-19 22:30:02 +00:00
import pytest
2021-02-28 13:44:58 +00:00
from _pytest.logging import LogCaptureFixture, caplog
2021-01-21 10:07:00 +00:00
2022-01-28 08:17:28 +00:00
from dda_python_terraform import IsFlagged, IsNotFlagged, Terraform, TerraformCommandError
logging.basicConfig(level=logging.DEBUG)
2017-05-09 09:47:36 +00:00
root_logger = logging.getLogger()
current_path = os.path.dirname(os.path.realpath(__file__))
semantic_version = os.environ.get("TFVER")
2021-10-23 14:22:59 +00:00
FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS = "test 'test.out!"
STRING_CASES = [
2020-10-19 22:30:02 +00:00
[
2022-01-19 15:51:07 +00:00
lambda x: x.generate_cmd_string(
{}, "apply", "the_folder", no_color=IsFlagged),
2020-10-19 22:30:02 +00:00
"terraform apply -no-color the_folder",
],
[
2022-01-19 15:51:07 +00:00
lambda x: x.generate_cmd_string({},
"push", "path", vcs=True, token="token", atlas_address="url"
),
2020-10-19 22:30:02 +00:00
"terraform push -vcs=true -token=token -atlas-address=url path",
],
2021-06-25 11:36:20 +00:00
[
2022-01-19 15:51:07 +00:00
lambda x: x.generate_cmd_string({},
"refresh", "path", token="token"
),
2021-06-25 11:36:20 +00:00
"terraform refresh -token=token path",
],
2020-10-19 22:30:02 +00:00
]
2021-10-23 13:37:29 +00:00
CMD_CASES_0_x = [
2020-10-19 22:30:02 +00:00
[
"method",
"expected_output",
"expected_ret_code",
"expected_exception",
"expected_logs",
"folder",
],
[
[
2020-10-19 22:30:02 +00:00
lambda x: x.cmd(
2021-10-23 13:37:29 +00:00
{},
2021-02-28 13:44:58 +00:00
"plan",
"var_to_output",
no_color=IsFlagged,
var={"test_var": "test"},
raise_on_error=False,
2020-10-19 22:30:02 +00:00
),
# Expected output varies by terraform semantic_version
2020-10-20 22:08:58 +00:00
"Plan: 0 to add, 0 to change, 0 to destroy.",
2017-05-09 09:47:36 +00:00
0,
False,
2020-10-19 22:30:02 +00:00
"",
"var_to_output",
],
# try import aws instance
[
2020-10-19 22:30:02 +00:00
lambda x: x.cmd(
2021-10-23 13:37:29 +00:00
{},
2020-10-19 22:30:02 +00:00
"import",
"aws_instance.foo",
"i-abcd1234",
no_color=IsFlagged,
2021-02-28 13:44:58 +00:00
raise_on_error=False,
2020-10-19 22:30:02 +00:00
),
"",
1,
2021-02-28 13:44:58 +00:00
False,
"Error: No Terraform configuration files",
2020-10-19 22:30:02 +00:00
"",
],
# test with space and special character in file path
[
2020-10-19 22:30:02 +00:00
lambda x: x.cmd(
2021-10-23 13:37:29 +00:00
{},
2021-02-28 13:44:58 +00:00
"plan",
"var_to_output",
out=FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS,
raise_on_error=False,
2020-10-19 22:30:02 +00:00
),
"",
0,
False,
2020-10-19 22:30:02 +00:00
"",
"var_to_output",
],
# test workspace command (commands with subcommand)
[
2021-02-28 13:44:58 +00:00
lambda x: x.cmd(
2021-10-23 13:37:29 +00:00
{}, "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 semantic_version
2022-01-12 12:55:11 +00:00
"Changes to Outputs:",
2021-10-23 13:37:29 +00:00
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
2021-02-28 13:44:58 +00:00
),
2020-10-19 22:30:02 +00:00
"",
0,
False,
2020-10-19 22:30:02 +00:00
"Command: terraform workspace show -no-color",
"",
],
2020-10-19 22:30:02 +00:00
],
]
2015-12-31 07:15:51 +00:00
2022-01-19 15:51:07 +00:00
APPLY_CASES_0_x = [
["folder", "variables", "var_files", "expected_output", "options"],
2022-01-19 16:06:14 +00:00
[("var_to_output", {"test_var": "test"}, None, 'test_output=test', {}),
2022-01-19 15:51:07 +00:00
("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"}', {},),
2022-01-19 16:15:59 +00:00
("var_to_output", {"test_map_var": {"c": "c", "d": "d"}}, "var_to_output/test_map_var.json", 'test_map_output={"e"="e""f"="f"}', {},),
2022-01-19 15:51:07 +00:00
("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},), ]]
2020-10-19 22:30:02 +00:00
@pytest.fixture(scope="function")
2016-12-20 16:55:55 +00:00
def fmt_test_file(request):
2020-10-19 22:30:02 +00:00
target = os.path.join(current_path, "bad_fmt", "test.backup")
orgin = os.path.join(current_path, "bad_fmt", "test.tf")
shutil.copy(orgin, target)
2016-12-20 16:55:55 +00:00
def td():
shutil.move(target, orgin)
request.addfinalizer(td)
return
2015-12-31 07:15:51 +00:00
2021-02-28 13:44:58 +00:00
# @pytest.fixture()
# def string_logger(request) -> Callable[..., str]:
# log_stream = StringIO()
# handler = logging.StreamHandler(log_stream)
# root_logger.addHandler(handler)
2017-05-09 09:47:36 +00:00
2021-02-28 13:44:58 +00:00
# def td():
# root_logger.removeHandler(handler)
# log_stream.close()
2017-05-09 09:47:36 +00:00
2021-02-28 13:44:58 +00:00
# request.addfinalizer(td)
# return lambda: str(log_stream.getvalue())
2017-05-09 09:47:36 +00:00
@pytest.fixture()
def workspace_setup_teardown():
2021-02-28 13:44:58 +00:00
"""Fixture used in workspace related tests.
Create and tear down a workspace
*Use as a contextmanager*
"""
2020-10-19 22:30:02 +00:00
@contextmanager
def wrapper(workspace_name, create=True, delete=True, *args, **kwargs):
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
tf.init()
if create:
tf.create_workspace(workspace_name, *args, **kwargs)
yield tf
if delete:
2020-10-19 22:30:02 +00:00
tf.set_workspace("default")
tf.delete_workspace(workspace_name)
yield wrapper
2021-02-28 13:44:58 +00:00
class TestTerraform:
2021-01-21 10:07:00 +00:00
def teardown_method(self, _) -> None:
2021-02-28 13:44:58 +00:00
"""Teardown any state that was previously setup with a setup_method call."""
2022-01-19 15:51:07 +00:00
exclude = ["test_tfstate_file",
"test_tfstate_file2", "test_tfstate_file3"]
2021-01-21 10:07:00 +00:00
def purge(dir: str, pattern: str) -> None:
2017-08-04 23:02:49 +00:00
for root, dirnames, filenames in os.walk(dir):
dirnames[:] = [d for d in dirnames if d not in exclude]
2017-08-04 23:02:49 +00:00
for filename in fnmatch.filter(filenames, pattern):
f = os.path.join(root, filename)
os.remove(f)
for dirname in fnmatch.filter(dirnames, pattern):
d = os.path.join(root, dirname)
shutil.rmtree(d)
2020-10-19 22:30:02 +00:00
purge(".", "*.tfstate")
purge(".", "*.tfstate.backup")
purge(".", "*.terraform")
purge(".", FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS)
2020-10-19 22:30:02 +00:00
@pytest.mark.parametrize(["method", "expected"], STRING_CASES)
2021-01-21 10:07:00 +00:00
def test_generate_cmd_string(self, method: Callable[..., str], expected: str):
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
result = method(tf)
strs = expected.split()
for s in strs:
assert s in result
@pytest.mark.parametrize(*(CMD_CASES_1_x if version.parse(semantic_version) >= version.parse("1.0.0") else CMD_CASES_0_x))
2020-10-19 22:30:02 +00:00
def test_cmd(
self,
2021-01-21 10:07:00 +00:00
method: Callable[..., str],
expected_output: str,
expected_ret_code: int,
expected_exception: bool,
expected_logs: str,
2021-02-28 13:44:58 +00:00
caplog: LogCaptureFixture,
folder: str,
2022-01-19 15:51:07 +00:00
):
2021-02-28 13:44:58 +00:00
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
2021-02-28 13:44:58 +00:00
tf.init(folder)
try:
ret, out, _ = method(tf)
assert not expected_exception
except TerraformCommandError as e:
assert expected_exception
ret = e.returncode
out = e.out
2021-01-21 10:07:00 +00:00
assert expected_output in out
assert expected_ret_code == ret
2021-02-28 13:44:58 +00:00
assert expected_logs in caplog.text
2015-12-31 07:15:51 +00:00
2022-01-19 16:06:14 +00:00
@pytest.mark.parametrize(*(APPLY_CASES_1_x if version.parse(semantic_version) >= version.parse("1.0.0") else APPLY_CASES_0_x))
def test_apply(self, folder, variables, var_files, expected_output, options):
2020-10-19 22:30:02 +00:00
tf = Terraform(
working_dir=current_path, variables=variables, var_file=var_files, terraform_semantic_version=semantic_version
2020-10-19 22:30:02 +00:00
)
2017-08-04 23:06:55 +00:00
tf.init(folder)
ret, out, err = tf.apply(folder, **options)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert expected_output in out.replace("\n", "").replace(" ", "")
assert err == ""
2021-02-28 13:44:58 +00:00
def test_apply_with_var_file(self, caplog: LogCaptureFixture):
with caplog.at_level(logging.INFO):
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
2021-02-28 13:44:58 +00:00
folder = "var_to_output"
tf.init(folder)
tf.apply(
folder,
2022-01-19 15:51:07 +00:00
var_file=os.path.join(
current_path, "tfvar_files", "test.tfvars"),
2021-02-28 13:44:58 +00:00
)
for log in caplog.messages:
if log.startswith("Command: terraform apply"):
2020-10-19 22:30:02 +00:00
assert log.count("-var-file=") == 1
2016-12-20 16:43:44 +00:00
@pytest.mark.parametrize(
2020-10-19 22:30:02 +00:00
["cmd", "args", "options"],
2016-12-20 16:43:44 +00:00
[
# bool value
2020-10-19 22:30:02 +00:00
("fmt", ["bad_fmt"], {"list": False, "diff": False})
],
2016-12-20 16:43:44 +00:00
)
2016-12-20 16:55:55 +00:00
def test_options(self, cmd, args, options, fmt_test_file):
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
2016-12-20 16:43:44 +00:00
ret, out, err = getattr(tf, cmd)(*args, **options)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert out == ""
def test_state_data(self):
2020-10-19 22:30:02 +00:00
cwd = os.path.join(current_path, "test_tfstate_file")
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=cwd, state="tfstate.test",
terraform_semantic_version=semantic_version)
tf.read_state_file()
2020-10-19 22:30:02 +00:00
assert tf.tfstate.modules[0]["path"] == ["root"]
2015-12-31 07:15:51 +00:00
def test_state_default(self):
2020-10-19 22:30:02 +00:00
cwd = os.path.join(current_path, "test_tfstate_file2")
tf = Terraform(working_dir=cwd, terraform_semantic_version=semantic_version)
tf.read_state_file()
2020-10-19 22:30:02 +00:00
assert tf.tfstate.modules[0]["path"] == ["default"]
def test_state_default_backend(self):
2020-10-19 22:30:02 +00:00
cwd = os.path.join(current_path, "test_tfstate_file3")
tf = Terraform(working_dir=cwd, terraform_semantic_version=semantic_version)
tf.read_state_file()
2020-10-19 22:30:02 +00:00
assert tf.tfstate.modules[0]["path"] == ["default_backend"]
def test_pre_load_state_data(self):
2020-10-19 22:30:02 +00:00
cwd = os.path.join(current_path, "test_tfstate_file")
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=cwd, state="tfstate.test",
terraform_semantic_version=semantic_version)
2020-10-19 22:30:02 +00:00
assert tf.tfstate.modules[0]["path"] == ["root"]
2015-12-31 07:15:51 +00:00
@pytest.mark.parametrize(
2020-10-19 22:30:02 +00:00
("folder", "variables"), [("var_to_output", {"test_var": "test"})]
)
def test_override_default(self, folder, variables):
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=current_path,
variables=variables, terraform_semantic_version=semantic_version)
2017-08-04 23:17:52 +00:00
tf.init(folder)
2020-10-19 22:30:02 +00:00
ret, out, err = tf.apply(
2020-10-20 22:08:58 +00:00
folder, var={"test_var": "test2"}, no_color=IsNotFlagged,
2020-10-19 22:30:02 +00:00
)
out = out.replace("\n", "")
assert "\x1b[0m\x1b[1m\x1b[32mApply" in out
2022-01-19 15:33:06 +00:00
out = tf.output(folder, "test_output")
2020-10-19 22:30:02 +00:00
assert "test2" in out
2021-02-28 13:44:58 +00:00
@pytest.mark.parametrize("output_all", [True, False])
def test_output(self, caplog: LogCaptureFixture, output_all: bool):
expected_value = "test"
required_output = "test_output"
with caplog.at_level(logging.INFO):
tf = Terraform(
working_dir=current_path, variables={"test_var": expected_value}, terraform_semantic_version=semantic_version
2021-02-28 13:44:58 +00:00
)
tf.init("var_to_output")
tf.apply("var_to_output")
params = tuple() if output_all else (required_output,)
2022-01-12 12:07:46 +00:00
result = tf.output("var_to_output", *params)
2021-02-28 13:44:58 +00:00
if output_all:
assert result[required_output]["value"] == expected_value
else:
2021-02-28 13:44:58 +00:00
assert result == expected_value
assert expected_value in caplog.messages[-1]
def test_destroy(self):
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=current_path, variables={
"test_var": "test"}, terraform_semantic_version=semantic_version)
2020-10-19 22:30:02 +00:00
tf.init("var_to_output")
ret, out, err = tf.destroy("var_to_output")
assert ret == 0
2020-10-19 22:30:02 +00:00
assert "Destroy complete! Resources: 0 destroyed." in out
@pytest.mark.parametrize(
2020-10-19 22:30:02 +00:00
("plan", "variables", "expected_ret"), [("vars_require_input", {}, 1)]
)
def test_plan(self, plan, variables, expected_ret):
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=current_path,
variables=variables, terraform_semantic_version=semantic_version)
2021-02-28 13:44:58 +00:00
tf.init(plan)
with pytest.raises(TerraformCommandError) as e:
tf.plan(plan)
assert (
2022-01-19 16:19:05 +00:00
"\nError:" in e.value.err
2021-02-28 13:44:58 +00:00
)
def test_fmt(self, fmt_test_file):
2022-01-19 15:51:07 +00:00
tf = Terraform(working_dir=current_path, variables={
"test_var": "test"}, terraform_semantic_version=semantic_version)
ret, out, err = tf.fmt(diff=True)
assert ret == 0
def test_create_workspace(self, workspace_setup_teardown):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
with workspace_setup_teardown(workspace_name, create=False) as tf:
2020-10-19 22:30:02 +00:00
ret, out, err = tf.create_workspace("test")
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2022-01-12 12:28:04 +00:00
# The directory flag is no longer supported in v1.0.8
# this should either be done with -chdir or probably just be removed
"""
2021-02-28 13:44:58 +00:00
def test_create_workspace_with_args(self, workspace_setup_teardown, caplog):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
state_file_path = os.path.join(
current_path, "test_tfstate_file2", "terraform.tfstate"
)
2021-02-28 13:44:58 +00:00
with workspace_setup_teardown(
workspace_name, create=False
) as tf, caplog.at_level(logging.INFO):
2020-10-19 22:30:02 +00:00
ret, out, err = tf.create_workspace(
"test", current_path, no_color=IsFlagged
)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2021-02-28 13:44:58 +00:00
assert (
f"Command: terraform workspace new -no-color test {current_path}"
in caplog.messages
2020-10-19 22:30:02 +00:00
)
2022-01-12 12:28:04 +00:00
"""
def test_set_workspace(self, workspace_setup_teardown):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.set_workspace(workspace_name)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2019-05-20 18:34:42 +00:00
2022-01-12 12:28:04 +00:00
# see comment on test_create_workspace_with_args
"""
2021-02-28 13:44:58 +00:00
def test_set_workspace_with_args(self, workspace_setup_teardown, caplog):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
2021-02-28 13:44:58 +00:00
with workspace_setup_teardown(workspace_name) as tf, caplog.at_level(
logging.INFO
):
2020-10-19 22:30:02 +00:00
ret, out, err = tf.set_workspace(
workspace_name, current_path, no_color=IsFlagged
)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2021-02-28 13:44:58 +00:00
assert (
f"Command: terraform workspace select -no-color test {current_path}"
in caplog.messages
2020-10-19 22:30:02 +00:00
)
2022-01-12 12:28:04 +00:00
"""
2022-01-19 15:51:07 +00:00
def test_show_workspace(self, workspace_setup_teardown):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
with workspace_setup_teardown(workspace_name) as tf:
ret, out, err = tf.show_workspace()
2019-05-20 18:34:42 +00:00
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2021-02-28 13:44:58 +00:00
def test_show_workspace_with_no_color(self, workspace_setup_teardown, caplog):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
2021-02-28 13:44:58 +00:00
with workspace_setup_teardown(workspace_name) as tf, caplog.at_level(
logging.INFO
):
ret, out, err = tf.show_workspace(no_color=IsFlagged)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2021-02-28 13:44:58 +00:00
assert "Command: terraform workspace show -no-color" in caplog.messages
def test_delete_workspace(self, workspace_setup_teardown):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
with workspace_setup_teardown(workspace_name, delete=False) as tf:
2020-10-19 22:30:02 +00:00
tf.set_workspace("default")
ret, out, err = tf.delete_workspace(workspace_name)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2022-01-12 12:28:04 +00:00
# see above comments
"""
2021-02-28 13:44:58 +00:00
def test_delete_workspace_with_args(self, workspace_setup_teardown, caplog):
2020-10-19 22:30:02 +00:00
workspace_name = "test"
2021-02-28 13:44:58 +00:00
with workspace_setup_teardown(
workspace_name, delete=False
) as tf, caplog.at_level(logging.INFO):
2020-10-19 22:30:02 +00:00
tf.set_workspace("default")
ret, out, err = tf.delete_workspace(
2019-06-25 21:12:41 +00:00
workspace_name, current_path, force=IsFlagged,
)
assert ret == 0
2020-10-19 22:30:02 +00:00
assert err == ""
2021-02-28 13:44:58 +00:00
assert (
f"Command: terraform workspace delete -force test {current_path}"
in caplog.messages
2020-10-19 22:30:02 +00:00
)
2022-01-12 12:28:04 +00:00
"""
2022-08-05 16:09:17 +00:00
def test_list_workspace(self):
tf = Terraform(working_dir=current_path)
workspaces = tf.list_workspace()
assert len(workspaces) > 0
assert 'default' in workspaces