|
|
|
@ -6,6 +6,7 @@ import shutil
|
|
|
|
|
from contextlib import contextmanager
|
|
|
|
|
from io import StringIO
|
|
|
|
|
from typing import Callable
|
|
|
|
|
from packaging import version
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
from _pytest.logging import LogCaptureFixture, caplog
|
|
|
|
@ -17,8 +18,7 @@ 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
|
|
|
|
|
semantic_version = os.environ.get("TFVER")
|
|
|
|
|
|
|
|
|
|
FILE_PATH_WITH_SPACE_AND_SPACIAL_CHARS = "test 'test.out!"
|
|
|
|
|
STRING_CASES = [
|
|
|
|
@ -60,7 +60,7 @@ CMD_CASES_0_x = [
|
|
|
|
|
var={"test_var": "test"},
|
|
|
|
|
raise_on_error=False,
|
|
|
|
|
),
|
|
|
|
|
# Expected output varies by terraform version
|
|
|
|
|
# Expected output varies by terraform semantic_version
|
|
|
|
|
"Plan: 0 to add, 0 to change, 0 to destroy.",
|
|
|
|
|
0,
|
|
|
|
|
False,
|
|
|
|
@ -131,7 +131,7 @@ CMD_CASES_1_x = [
|
|
|
|
|
var={"test_var": "test"},
|
|
|
|
|
raise_on_error=False,
|
|
|
|
|
),
|
|
|
|
|
# Expected output varies by terraform version
|
|
|
|
|
# Expected output varies by terraform semantic_version
|
|
|
|
|
"Changes to Outputs:",
|
|
|
|
|
0,
|
|
|
|
|
False,
|
|
|
|
@ -236,7 +236,7 @@ def workspace_setup_teardown():
|
|
|
|
|
|
|
|
|
|
@contextmanager
|
|
|
|
|
def wrapper(workspace_name, create=True, delete=True, *args, **kwargs):
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
|
|
|
|
|
tf.init()
|
|
|
|
|
if create:
|
|
|
|
|
tf.create_workspace(workspace_name, *args, **kwargs)
|
|
|
|
@ -271,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, terraform_version=version)
|
|
|
|
|
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 >= 1.0 else CMD_CASES_0_x))
|
|
|
|
|
@pytest.mark.parametrize(*(CMD_CASES_1_x if version.parse(semantic_version) >= version.parse("1.0.0") else CMD_CASES_0_x))
|
|
|
|
|
def test_cmd(
|
|
|
|
|
self,
|
|
|
|
|
method: Callable[..., str],
|
|
|
|
@ -290,7 +290,7 @@ class TestTerraform:
|
|
|
|
|
folder: str,
|
|
|
|
|
):
|
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
|
|
|
|
|
tf.init(folder)
|
|
|
|
|
try:
|
|
|
|
|
ret, out, _ = method(tf)
|
|
|
|
@ -305,10 +305,10 @@ class TestTerraform:
|
|
|
|
|
assert expected_logs in caplog.text
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(*(APPLY_CASES_1_x if version >= 1.0 else APPLY_CASES_0_x))
|
|
|
|
|
@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):
|
|
|
|
|
tf = Terraform(
|
|
|
|
|
working_dir=current_path, variables=variables, var_file=var_files, terraform_version=version
|
|
|
|
|
working_dir=current_path, variables=variables, var_file=var_files, terraform_semantic_version=semantic_version
|
|
|
|
|
)
|
|
|
|
|
tf.init(folder)
|
|
|
|
|
ret, out, err = tf.apply(folder, **options)
|
|
|
|
@ -318,7 +318,7 @@ class TestTerraform:
|
|
|
|
|
|
|
|
|
|
def test_apply_with_var_file(self, caplog: LogCaptureFixture):
|
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
|
|
|
|
|
folder = "var_to_output"
|
|
|
|
|
tf.init(folder)
|
|
|
|
|
tf.apply(
|
|
|
|
@ -338,7 +338,7 @@ class TestTerraform:
|
|
|
|
|
],
|
|
|
|
|
)
|
|
|
|
|
def test_options(self, cmd, args, options, fmt_test_file):
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=current_path, terraform_semantic_version=semantic_version)
|
|
|
|
|
ret, out, err = getattr(tf, cmd)(*args, **options)
|
|
|
|
|
assert ret == 0
|
|
|
|
|
assert out == ""
|
|
|
|
@ -346,26 +346,26 @@ class TestTerraform:
|
|
|
|
|
def test_state_data(self):
|
|
|
|
|
cwd = os.path.join(current_path, "test_tfstate_file")
|
|
|
|
|
tf = Terraform(working_dir=cwd, state="tfstate.test",
|
|
|
|
|
terraform_version=version)
|
|
|
|
|
terraform_semantic_version=semantic_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, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=cwd, terraform_semantic_version=semantic_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, terraform_version=version)
|
|
|
|
|
tf = Terraform(working_dir=cwd, terraform_semantic_version=semantic_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",
|
|
|
|
|
terraform_version=version)
|
|
|
|
|
terraform_semantic_version=semantic_version)
|
|
|
|
|
assert tf.tfstate.modules[0]["path"] == ["root"]
|
|
|
|
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
|
|
@ -373,7 +373,7 @@ class TestTerraform:
|
|
|
|
|
)
|
|
|
|
|
def test_override_default(self, folder, variables):
|
|
|
|
|
tf = Terraform(working_dir=current_path,
|
|
|
|
|
variables=variables, terraform_version=version)
|
|
|
|
|
variables=variables, terraform_semantic_version=semantic_version)
|
|
|
|
|
tf.init(folder)
|
|
|
|
|
ret, out, err = tf.apply(
|
|
|
|
|
folder, var={"test_var": "test2"}, no_color=IsNotFlagged,
|
|
|
|
@ -389,7 +389,7 @@ class TestTerraform:
|
|
|
|
|
required_output = "test_output"
|
|
|
|
|
with caplog.at_level(logging.INFO):
|
|
|
|
|
tf = Terraform(
|
|
|
|
|
working_dir=current_path, variables={"test_var": expected_value}, terraform_version=version
|
|
|
|
|
working_dir=current_path, variables={"test_var": expected_value}, terraform_semantic_version=semantic_version
|
|
|
|
|
)
|
|
|
|
|
tf.init("var_to_output")
|
|
|
|
|
tf.apply("var_to_output")
|
|
|
|
@ -403,7 +403,7 @@ class TestTerraform:
|
|
|
|
|
|
|
|
|
|
def test_destroy(self):
|
|
|
|
|
tf = Terraform(working_dir=current_path, variables={
|
|
|
|
|
"test_var": "test"}, terraform_version=version)
|
|
|
|
|
"test_var": "test"}, terraform_semantic_version=semantic_version)
|
|
|
|
|
tf.init("var_to_output")
|
|
|
|
|
ret, out, err = tf.destroy("var_to_output")
|
|
|
|
|
assert ret == 0
|
|
|
|
@ -414,7 +414,7 @@ class TestTerraform:
|
|
|
|
|
)
|
|
|
|
|
def test_plan(self, plan, variables, expected_ret):
|
|
|
|
|
tf = Terraform(working_dir=current_path,
|
|
|
|
|
variables=variables, terraform_version=version)
|
|
|
|
|
variables=variables, terraform_semantic_version=semantic_version)
|
|
|
|
|
tf.init(plan)
|
|
|
|
|
with pytest.raises(TerraformCommandError) as e:
|
|
|
|
|
tf.plan(plan)
|
|
|
|
@ -424,7 +424,7 @@ class TestTerraform:
|
|
|
|
|
|
|
|
|
|
def test_fmt(self, fmt_test_file):
|
|
|
|
|
tf = Terraform(working_dir=current_path, variables={
|
|
|
|
|
"test_var": "test"}, terraform_version=version)
|
|
|
|
|
"test_var": "test"}, terraform_semantic_version=semantic_version)
|
|
|
|
|
ret, out, err = tf.fmt(diff=True)
|
|
|
|
|
assert ret == 0
|
|
|
|
|
|
|
|
|
|