feat: implement function to construct Modules out of the terraform configuration

samcli/hook_packages/terraform/hooks/prepare/resource_linking.py

@@ -69,6 +69,115 @@ def full_address(self) -> str:
         return self.address
 
 
+def _build_module(
+    module_name: Optional[str],
+    module_configuration: Dict,
+    input_variables: Dict[str, Expression],
+    parent_module_address: Optional[str],
+) -> TFModule:
+    module = TFModule(None, None, {}, [], {}, {})
+
+    module.full_address = _build_module_full_address(module_name, parent_module_address)
+    module.variables = _build_module_variables_from_configuration(module_configuration, input_variables)
+    module.resources = _build_module_resources_from_configuration(module_configuration, module)
+    module.outputs = _build_module_outputs_from_configuration(module_configuration)
+    module.child_modules = _build_child_modules_from_configuration(module_configuration, module)
+
+    return module
+
+
+def _build_module_full_address(module_name: Optional[str], parent_module_address: Optional[str]) -> Optional[str]:
+    full_address = None
+    if module_name:
+        full_address = f"module.{module_name}"
+        if parent_module_address:
+            full_address = f"{parent_module_address}.{full_address}"
+
+    return full_address
+
+
+def _build_module_variables_from_configuration(
+    module_configuration: Dict, input_variables: Dict[str, Expression]
+) -> Dict[str, Expression]:
+    module_variables: Dict[str, Expression] = {}
+
+    default_variables = module_configuration.get("variables", {})
+    for variable_name, variable_value in default_variables.items():
+        module_variables[variable_name] = ConstantValue(variable_value.get("default"))
+    module_variables.update(input_variables)
+
+    return module_variables
+
+
+def _build_module_resources_from_configuration(module_configuration: Dict, module: TFModule) -> List[TFResource]:
+    module_resources = []
+
+    config_resources = module_configuration.get("resources", [])
+    for config_resource in config_resources:
+        resource_attributes: Dict[str, Expression] = {}
+
+        expressions = config_resource.get("expressions", {})
+        print(expressions)
+        for expression_name, expression_value in expressions.items():
+            parsed_expression = _build_expression_from_configuration(expression_value)
+            resource_attributes[expression_name] = parsed_expression
+
+        resource_address = config_resource.get("address")
+        resource_type = config_resource.get("type")
+        module_resources.append(TFResource(resource_address, resource_type, module, resource_attributes))
+
+    return module_resources
+
+
+def _build_module_outputs_from_configuration(module_configuration: Dict) -> Dict[str, Expression]:
+    module_outputs = {}
+
+    config_outputs = module_configuration.get("outputs", {})
+    for output_name, output_value in config_outputs.items():
+        expression = output_value.get("expression", {})
+        parsed_expression = _build_expression_from_configuration(expression)
+        module_outputs[output_name] = parsed_expression
+
+    return module_outputs
+
+
+def _build_child_modules_from_configuration(module_configuration: Dict, module: TFModule) -> Dict[str, TFModule]:
+    child_modules = {}
+
+    module_calls = module_configuration.get("module_calls", {})
+    for module_call_name, module_call_value in module_calls.items():
+        module_call_input_variables: Dict[str, Expression] = {}
+
+        expressions = module_call_value.get("expressions", {})
+        for expression_name, expression_value in expressions.items():
+            parsed_expression = _build_expression_from_configuration(expression_value)
+            module_call_input_variables[expression_name] = parsed_expression
+
+        module_call_module_config = module_call_value.get("module", {})
+        module_call_built_module = _build_module(
+            module_call_name, module_call_module_config, module_call_input_variables, module.full_address
+        )
+
+        module_call_built_module.parent_module = module
+        child_modules[module_call_name] = module_call_built_module
+
+    return child_modules
+
+
+def _build_expression_from_configuration(expression_configuration: Dict) -> Expression:
+    constant_value = expression_configuration.get("constant_value")
+    references = expression_configuration.get("references")
+
+    parsed_expression: Expression
+
+    if constant_value:
+        parsed_expression = ConstantValue(constant_value)
+    elif references:
+        parsed_expression = References(references)
+
+    return parsed_expression
+
+
 def _clean_references_list(references: List[str]) -> List[str]:
     """
     Return a new copy of the complete references list.

tests/unit/hook_packages/terraform/hooks/prepare/test_resource_linking.py

@@ -1,6 +1,6 @@
 from copy import deepcopy
 from unittest import TestCase
-from unittest.mock import Mock, patch
+from unittest.mock import Mock, patch, call, create_autospec
 
 from parameterized import parameterized
 
@@ -13,6 +13,13 @@
     ConstantValue,
     References,
     _resolve_module_variable,
+    _build_module,
+    _build_expression_from_configuration,
+    _build_module_full_address,
+    _build_child_modules_from_configuration,
+    _build_module_outputs_from_configuration,
+    _build_module_resources_from_configuration,
+    _build_module_variables_from_configuration,
 )
 
 
@@ -350,3 +357,227 @@ def test_resolve_module_variable_invalid_module_reference(
             ex.exception.args[0],
             "An error occurred when attempting to link two resources: Couldn't find child module layer_module.",
         )
+
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_module_full_address")
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_module_variables_from_configuration")
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_module_resources_from_configuration")
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_module_outputs_from_configuration")
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_child_modules_from_configuration")
+    def test_build_module(
+        self,
+        patched_build_child_modules_from_configuration,
+        patched_build_module_outputs_from_configuration,
+        patched_build_module_resources_from_configuration,
+        patched_build_module_variables_from_configuration,
+        patched_build_module_full_address,
+    ):
+        mock_full_address = Mock()
+        patched_build_module_full_address.return_value = mock_full_address
+
+        mock_variables = Mock()
+        patched_build_module_variables_from_configuration.return_value = mock_variables
+
+        mock_resources = Mock()
+        patched_build_module_resources_from_configuration.return_value = mock_resources
+
+        mock_outputs = Mock()
+        patched_build_module_outputs_from_configuration.return_value = mock_outputs
+
+        mock_child_modules = Mock()
+        patched_build_child_modules_from_configuration.return_value = mock_child_modules
+
+        result = _build_module(Mock(), Mock(), Mock(), Mock())
+        expected_module = TFModule(
+            mock_full_address, None, mock_variables, mock_resources, mock_child_modules, mock_outputs
+        )
+
+        self.assertEqual(result, expected_module)
+
+    @parameterized.expand(
+        [
+            (None, None, None),
+            ("some_module", None, "module.some_module"),
+            ("some_module", "parent_module_address", "parent_module_address.module.some_module"),
+        ]
+    )
+    def test_build_module_full_address(self, module_name, parent_module_address, expected_full_address):
+        result = _build_module_full_address(module_name, parent_module_address)
+        self.assertEqual(result, expected_full_address)
+
+    def test_build_module_variables_from_configuration(self):
+        module_configuration = {
+            "variables": {
+                "var1": {"default": "var1_default_value"},
+                "var2": {"default": "var2_default_value"},
+                "var3": {},
+                "var4": {},
+            },
+        }
+
+        input_variables = {
+            "var2": ConstantValue("var2_input_value"),
+            "var4": ConstantValue("var4_input_value"),
+        }
+
+        result = _build_module_variables_from_configuration(module_configuration, input_variables)
+
+        self.assertEqual(result["var1"], ConstantValue("var1_default_value"))
+        self.assertEqual(result["var2"], input_variables["var2"])
+        self.assertEqual(result["var3"], ConstantValue(None))
+        self.assertEqual(result["var4"], ConstantValue("var4_input_value"))
+
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_expression_from_configuration")
+    def test_build_module_resources_from_configuration(
+        self,
+        patched_build_expression_from_configuration,
+    ):
+        mock_parsed_expression = Mock()
+        patched_build_expression_from_configuration.return_value = mock_parsed_expression
+
+        module_configuration = {
+            "resources": [
+                {
+                    "address": "resource1_address",
+                    "type": "resource1_type",
+                    "expressions": {
+                        "expression1": Mock(),
+                        "expression2": Mock(),
+                    },
+                },
+                {
+                    "address": "resource2_address",
+                    "type": "resource2_type",
+                    "expressions": {
+                        "expression3": Mock(),
+                        "expression4": Mock(),
+                    },
+                },
+            ]
+        }
+
+        mock_module = Mock()
+
+        result = _build_module_resources_from_configuration(module_configuration, mock_module)
+
+        expected_resources = [
+            TFResource(
+                "resource1_address",
+                "resource1_type",
+                mock_module,
+                {"expression1": mock_parsed_expression, "expression2": mock_parsed_expression},
+            ),
+            TFResource(
+                "resource2_address",
+                "resource2_type",
+                mock_module,
+                {"expression3": mock_parsed_expression, "expression4": mock_parsed_expression},
+            ),
+        ]
+
+        self.assertEqual(result, expected_resources)
+
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_expression_from_configuration")
+    def test_build_module_outputs_from_configuration(
+        self,
+        patched_build_expression_from_configuration,
+    ):
+        parsed_expression = Mock()
+        patched_build_expression_from_configuration.return_value = parsed_expression
+
+        module_configuration = {
+            "outputs": {
+                "output1": {
+                    "expression": Mock(),
+                },
+                "output2": {
+                    "expression": Mock(),
+                },
+            }
+        }
+
+        result = _build_module_outputs_from_configuration(module_configuration)
+
+        expected_outputs = {
+            "output1": parsed_expression,
+            "output2": parsed_expression,
+        }
+
+        self.assertEqual(result, expected_outputs)
+
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_expression_from_configuration")
+    @patch("samcli.hook_packages.terraform.hooks.prepare.resource_linking._build_module")
+    def test_build_child_modules_from_configuration(
+        self,
+        patched_build_module,
+        patched_build_expression_from_configuration,
+    ):
+        mock_parsed_expression = Mock()
+        patched_build_expression_from_configuration.return_value = mock_parsed_expression
+
+        child_built_modules = [Mock(), Mock()]
+        patched_build_module.side_effect = child_built_modules
+
+        mock_child_config1 = Mock()
+        mock_child_config2 = Mock()
+        module_configuration = {
+            "module_calls": {
+                "module1": {
+                    "expressions": {
+                        "expression1": Mock(),
+                        "expression2": Mock(),
+                    },
+                    "module": mock_child_config1,
+                },
+                "module2": {
+                    "expressions": {
+                        "expression3": Mock(),
+                        "expression4": Mock(),
+                    },
+                    "module": mock_child_config2,
+                },
+            },
+        }
+
+        mock_module = Mock(full_address="module.some_address")
+
+        result = _build_child_modules_from_configuration(module_configuration, mock_module)
+
+        # check it builds child modules
+        patched_build_module.assert_has_calls(
+            [
+                call(
+                    "module1",
+                    mock_child_config1,
+                    {"expression1": mock_parsed_expression, "expression2": mock_parsed_expression},
+                    "module.some_address",
+                ),
+                call(
+                    "module2",
+                    mock_child_config2,
+                    {"expression3": mock_parsed_expression, "expression4": mock_parsed_expression},
+                    "module.some_address",
+                ),
+            ]
+        )
+
+        # check it sets parent module of each child
+        for child in child_built_modules:
+            self.assertEqual(child.parent_module, mock_module)
+
+        # check return result
+        self.assertCountEqual(list(result.keys()), ["module1", "module2"])
+        self.assertCountEqual(list(result.values()), child_built_modules)
+
+    @parameterized.expand(
+        [
+            ({"constant_value": "hello"}, ConstantValue("hello")),
+            ({"references": ["hello", "world"]}, References(["hello", "world"])),
+        ]
+    )
+    def test_build_expression_from_configuration(
+        self,
+        expression_configuration,
+        expected_parsed_expression,
+    ):
+        result = _build_expression_from_configuration(expression_configuration)
+        self.assertEqual(result, expected_parsed_expression)