Merge pull request #43 from CMInformatik/main

merge upstream

.vscode/extensions.json

@@ -0,0 +1,5 @@
+{
+    "recommendations": [
+        "ms-vscode-remote.remote-containers"
+    ]
+}

README.md

@@ -18,6 +18,8 @@ The CLI works by scanning your .tf files for versioned providers and modules and
     - [Authentication](#authentication-1)
       - [.terraformrc file:](#terraformrc-file)
       - [infrapatch\_credentials.json file:](#infrapatch_credentialsjson-file)
+    - [Setup Development Environment for InfraPatch](#setup-development-environment-for-infrapatch)
+    - [Contributing](#contributing)
 
 
 ## GitHub Action
@@ -182,3 +184,16 @@ You can also specify the path to the credentials file with the `--credentials-fi
 infrapatch --credentials-file-path "path/to/credentials/file" update
 ```
 
+### Setup Development Environment for InfraPatch
+
+This repository contains a devcontainer configuration for VSCode. To use it, you need to install the following tools:
+* ["Dev Containers VSCode Extension"](https://marketplace.visualstudio.com/items?itemName=ms-vscode-remote.remote-containers) for VSCode.
+* A local Docker installation like [Docker Desktop](https://www.docker.com/products/docker-desktop).
+
+After installation, you can open the repository in the devcontainer by clicking on the green "Open in Container" button in the bottom left corner of VSCode.
+During the first start, the devcontainer will build the container image and install all dependencies.
+
+### Contributing
+
+If you have any ideas for improvements or find any bugs, feel free to open an issue or create a pull request.
+

infrapatch/core/models/tests/test_versioned_terraform_resource.py

@@ -55,7 +55,12 @@ def test_find():
 
 def test_to_dict():
     module = TerraformModule(name="test_resource", current_version="1.0.0", _source_file="test_file.py", _source="test/test_module/test_provider")
-    provider = TerraformProvider(name="test_resource", current_version="1.0.0", _source_file="test_file.py", _source="test_provider/test_provider")
+    provider = TerraformProvider(
+        name="test_resource",
+        current_version="1.0.0",
+        _source_file="test_file.py",
+        _source="test_provider/test_provider",
+    )
 
     module_dict = module.to_dict()
     provider_dict = provider.to_dict()

infrapatch/core/utils/terraform/hcl_edit_cli.py

@@ -39,10 +39,13 @@ def update_hcl_value(self, resource: str, file: Path, value: str):
         self._run_hcl_edit_command("update", resource, file, value)
 
     def get_hcl_value(self, resource: str, file: Path) -> str:
-        result = self._run_hcl_edit_command("get", resource, file)
-        if result is None:
+        result = self._run_hcl_edit_command("read", resource, file)
+        if result is None or result == "":
             raise HclEditCliException(f"Could not get value for resource '{resource}' from file '{file}'.")
-        return result
+        resource_id, value = result.split(" ")
+        if resource_id != resource:
+            raise HclEditCliException(f"Could not get value for resource '{resource}' from file '{file}'.")
+        return value
 
     def _run_hcl_edit_command(self, action: str, resource: str, file: Path, value: Union[str, None] = None) -> Optional[str]:
         command = [self._binary_path.absolute().as_posix(), action, resource]

infrapatch/core/utils/terraform/hcl_handler.py

@@ -7,7 +7,7 @@
 import pygohcl
 
 from infrapatch.core.models.versioned_terraform_resources import TerraformModule, TerraformProvider, VersionedTerraformResource
-from infrapatch.core.utils.terraform.hcl_edit_cli import HclEditCli
+from infrapatch.core.utils.terraform.hcl_edit_cli import HclEditCliInterface
 
 
 class HclParserException(Exception):
@@ -29,7 +29,7 @@ def get_credentials_form_user_rc_file(self) -> dict[str, str]:
 
 
 class HclHandler(HclHandlerInterface):
-    def __init__(self, hcl_edit_cli: HclEditCli):
+    def __init__(self, hcl_edit_cli: HclEditCliInterface):
         self.hcl_edit_cli = hcl_edit_cli
         pass
 

infrapatch/core/utils/terraform/tests/test_hcl_edit_cli.py

@@ -0,0 +1,84 @@
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from infrapatch.core.utils.terraform.hcl_edit_cli import HclEditCli, HclEditCliException
+
+
+@pytest.fixture
+def hcl_edit_cli():
+    return HclEditCli()
+
+
+def test_init_with_existing_binary_path(hcl_edit_cli):
+    assert hcl_edit_cli._binary_path.exists()
+
+
+def test_get_binary_path_windows():
+    with patch("platform.system", return_value="Windows"):
+        hcl_edit_cli = HclEditCli()
+        assert hcl_edit_cli._get_binary_path().name == "hcledit_windows.exe"
+
+
+def test_get_binary_path_linux():
+    with patch("platform.system", return_value="Linux"):
+        hcl_edit_cli = HclEditCli()
+        assert hcl_edit_cli._get_binary_path().name == "hcledit_linux"
+
+
+def test_get_binary_path_darwin():
+    with patch("platform.system", return_value="Darwin"):
+        hcl_edit_cli = HclEditCli()
+        assert hcl_edit_cli._get_binary_path().name == "hcledit_darwin"
+
+
+def test_get_binary_path_unsupported_platform():
+    with patch("platform.system", return_value="Unsupported"):
+        with pytest.raises(Exception):
+            HclEditCli()
+
+
+def test_update_hcl_value(hcl_edit_cli, tmp_path):
+    file_path = tmp_path / "test_file.hcl"
+    file_path.write_text('resource "test_resource" {\n  value = "old_value"\n}')
+
+    hcl_edit_cli.update_hcl_value("resource.test_resource.value", file_path, "new_value")
+
+    assert file_path.read_text() == 'resource "test_resource" {\n  value = "new_value"\n}'
+
+
+def test_get_hcl_value(hcl_edit_cli, tmp_path):
+    file_path = tmp_path / "test_file.hcl"
+    file_path.write_text('resource "test_resource" {\n  value = "test_value"\n}')
+
+    value = hcl_edit_cli.get_hcl_value("resource.test_resource.value", file_path)
+
+    assert value == "test_value"
+
+
+def test_get_hcl_value_non_existing_resource(hcl_edit_cli, tmp_path):
+    file_path = tmp_path / "test_file.hcl"
+    file_path.write_text('resource "test_resource" {\n  value = "test_value"\n}')
+
+    with pytest.raises(HclEditCliException):
+        hcl_edit_cli.get_hcl_value("non_existing_resource.value", file_path)
+
+
+def test_run_hcl_edit_command_success(hcl_edit_cli):
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value.returncode = 0
+        mock_run.return_value.stdout = "command_output"
+
+        result = hcl_edit_cli._run_hcl_edit_command("get", "test_resource.value", Path("test_file.hcl"))
+
+        assert result == "command_output"
+
+
+def test_run_hcl_edit_command_failure(hcl_edit_cli):
+    with patch("subprocess.run") as mock_run:
+        mock_run.return_value.returncode = 1
+        mock_run.return_value.stdout = "command_output"
+
+        with pytest.raises(HclEditCliException):
+            hcl_edit_cli._run_hcl_edit_command("get", "test_resource.value", Path("test_file.hcl"))

infrapatch/core/utils/terraform/tests/test_hcl_handler.py

@@ -0,0 +1,217 @@
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from infrapatch.core.models.versioned_terraform_resources import TerraformModule, TerraformProvider
+from infrapatch.core.utils.terraform.hcl_edit_cli import HclEditCli
+from infrapatch.core.utils.terraform.hcl_handler import HclHandler, HclParserException
+
+
+@pytest.fixture
+def tmp_user_home(tmp_path: Path):
+    return tmp_path
+
+
+@pytest.fixture
+def hcl_handler(tmp_user_home: Path):
+    return HclHandler(hcl_edit_cli=HclEditCli())
+
+
+@pytest.fixture
+def valid_terraform_code():
+    return """
+        terraform {
+            required_providers {
+                test_provider = {
+                    source = "test_provider/test_provider"
+                    version = ">1.0.0"
+                }
+                test_provider2 = {
+                    source = "spacelift.io/test_provider/test_provider2"
+                    version = "1.0.5"
+                }
+            }
+        }
+        module "test_module" {
+            source = "test/test_module/test_provider"
+            version = "2.0.0"
+            name = "Test_module"
+        }
+        module "test_module2" {
+            source = "spacelift.io/test/test_module/test_provider"
+            version = "1.0.2"
+            name = "Test_module2"
+        }
+        # This module should be ignored since it has no version
+        module "test_module3" {
+            source = "C:/test/test_module/test_provider"
+            name = "Test_module3"
+        }
+    """
+
+
+@pytest.fixture
+def invalid_terraform_code():
+    return """
+        terraform {
+            required_providers {
+                test_provider = {
+                    source = "test_provider/test_provider"
+                    version = ">1.0.0"
+                }
+                test_provider = {
+                    source = "spacelift.io/test_provider/test_provider2"
+                    version = "1.0.5
+                }
+            }
+        }
+        module "test_module" {
+            source = "test/test_module/test_provider"
+            version = "2.0.0"
+            name = Test_module"
+            }
+        }
+    """
+
+
+def test_get_terraform_resources_from_file(hcl_handler: HclHandler, valid_terraform_code: str, tmp_path: Path):
+    # Create a temporary Terraform file for testing
+    tf_file = tmp_path.joinpath("test_file.tf")
+    tf_file.write_text(valid_terraform_code)
+    resouces = hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=True, get_providers=True)
+    modules = hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=True, get_providers=False)
+    providers = hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=False, get_providers=True)
+
+    modules_filtered = [resource for resource in resouces if isinstance(resource, TerraformModule)]
+    providers_filtered = [resource for resource in resouces if isinstance(resource, TerraformProvider)]
+
+    assert len(resouces) == 4
+    assert len(modules) == 2
+    assert len(providers) == 2
+    assert len(modules_filtered) == len(modules)
+    assert len(providers_filtered) == len(providers)
+
+    for resource in resouces:
+        assert resource._source_file == tf_file.absolute().as_posix()
+        if resource.name == "test_module":
+            assert isinstance(resource, TerraformModule)
+            assert resource.current_version == "2.0.0"
+            assert resource.source == "test/test_module/test_provider"
+            assert resource.identifier == "test/test_module/test_provider"
+            assert resource.base_domain is None
+        elif resource.name == "test_module2":
+            assert isinstance(resource, TerraformModule)
+            assert resource.current_version == "1.0.2"
+            assert resource.source == "spacelift.io/test/test_module/test_provider"
+            assert resource.identifier == "test/test_module/test_provider"
+            assert resource.base_domain == "spacelift.io"
+        elif resource.name == "test_provider":
+            assert isinstance(resource, TerraformProvider)
+            assert resource.current_version == ">1.0.0"
+            assert resource.source == "test_provider/test_provider"
+            assert resource.identifier == "test_provider/test_provider"
+            assert resource.base_domain is None
+        elif resource.name == "test_provider2":
+            assert isinstance(resource, TerraformProvider)
+            assert resource.current_version == "1.0.5"
+            assert resource.source == "spacelift.io/test_provider/test_provider2"
+            assert resource.identifier == "test_provider/test_provider2"
+            assert resource.base_domain == "spacelift.io"
+        else:
+            raise Exception(f"Unknown resource '{resource.name}'.")
+
+
+def test_invalid_terraform_code_parse_error(hcl_handler: HclHandler, invalid_terraform_code: str, tmp_path: Path):
+    # Create a temporary Terraform file for testing
+    tf_file = tmp_path.joinpath("test_file.tf")
+    tf_file.write_text(invalid_terraform_code)
+    with pytest.raises(HclParserException):
+        hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=True, get_providers=True)
+
+
+def test_bump_resource_version(hcl_handler, valid_terraform_code: str, tmp_path: Path):
+    # Create a TerraformModule resource for testing
+    tf_file = tmp_path.joinpath("test_file.tf")
+    tf_file.write_text(valid_terraform_code)
+    resouces = hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=True, get_providers=True)
+
+    # bump versions
+    for resource in resouces:
+        if resource.name == "test_module":
+            resource.newest_version = "4.0.1"
+        elif resource.name == "test_module2":
+            resource.newest_version = "4.0.2"
+
+        elif resource.name == "test_provider":
+            resource.newest_version = "4.0.3"
+        elif resource.name == "test_provider2":
+            resource.newest_version = "4.0.4"
+        hcl_handler.bump_resource_version(resource)
+
+    resouces = hcl_handler.get_terraform_resources_from_file(tf_file, get_modules=True, get_providers=True)
+    # check if versions are bumped
+    for resource in resouces:
+        if resource.name == "test_module":
+            assert resource.current_version == "4.0.1"
+        elif resource.name == "test_module2":
+            assert resource.current_version == "4.0.2"
+        elif resource.name == "test_provider":
+            assert resource.current_version == ">1.0.0"  # should not be bumped since it defines newer version
+        elif resource.name == "test_provider2":
+            assert resource.current_version == "4.0.4"
+
+
+def test_get_all_terraform_files(hcl_handler):
+    # Create a temporary directory with Terraform files for testing
+    root_dir = Path("test_dir")
+    root_dir.mkdir()
+    tf_file1 = root_dir / "file1.tf"
+    tf_file1.touch()
+    tf_file2 = root_dir / "file2.tf"
+    tf_file2.touch()
+
+    # Test getting all Terraform files in the directory
+    files = hcl_handler.get_all_terraform_files(root_dir)
+    assert len(files) == 2
+    assert tf_file1 in files
+    assert tf_file2 in files
+
+    # Clean up the temporary directory
+    tf_file1.unlink()
+    tf_file2.unlink()
+    root_dir.rmdir()
+
+
+def test_get_credentials_form_user_rc_file(hcl_handler, tmp_user_home: Path):
+    # Create a temporary terraformrc file for testing
+
+    # Create an instance of HclHandler with a mock HclEditCli
+    with patch("pathlib.Path.home", return_value=tmp_user_home):
+        # test without file
+        credentials = hcl_handler.get_credentials_form_user_rc_file()
+        assert len(credentials) == 0
+
+        # Create a temporary terraformrc file for testing
+        terraform_rc_file = tmp_user_home.joinpath(".terraformrc")
+        terraform_rc_file.write_text(
+            """
+            credentials {
+                test1 = {
+                    token = "token1"
+                }
+                test2 = {
+                    token = "token2"
+                }
+            }
+        """
+        )
+
+        # Test getting credentials from the terraformrc file
+        credentials = hcl_handler.get_credentials_form_user_rc_file()
+        assert len(credentials) == 2
+        assert credentials["test1"] == "token1"
+        assert credentials["test2"] == "token2"
+
+        # Clean up the temporary file
+        terraform_rc_file.unlink()