Merge branch 'morosi-feat' into 'master'

Add update_if_exist option to add_output/parameter

See merge request it/e3-aws!91

Author: adanaja

src/e3/aws/troposphere/__init__.py

@@ -118,7 +118,10 @@ def resources(self, stack: Stack) -> list[AWSObject | Construct]:
         # Add the parameter during template creation even if the S3 key may not
         # be known yet. This is useful if creating a stack from code, so that
         # the exported template contains the parameter
-        stack.add_parameter(self.s3_key_parameter)
+        stack.add_parameter(
+            self.s3_key_parameter,
+            update_if_exist=True,
+        )
         # Adding the Output can be useful for Lambda function with versioning.
         # The exported value can be retrieved by the lambda without the need to update
         # the lambda version.
@@ -128,7 +131,8 @@ def resources(self, stack: Stack) -> list[AWSObject | Construct]:
                 Description=f"S3 URI for the Asset {self.name}",
                 Export=Export(name=self.s3_uri_output_name),
                 Value=f"s3://{stack.s3_bucket}/{stack.s3_assets_key}{self.s3_key}",
-            )
+            ),
+            update_if_exist=True,
         )
         return []
 
@@ -276,7 +280,7 @@ def add(self, element: AWSObject | Construct | Stack) -> Stack:
                 # Special case to keep track of Assets and generate parameters
                 # for the S3 keys
                 if isinstance(construct, Asset):
-                    self.add_parameter(construct.s3_key_parameter)
+                    self.add_parameter(construct.s3_key_parameter, update_if_exist=True)
                     self.assets[construct.name] = construct
 
                 constructs_to_objects.extend(construct.resources(stack=self))
@@ -295,26 +299,41 @@ def extend(self, elements: Iterable[AWSObject | Construct | Stack]) -> Stack:
 
         return self
 
-    def add_parameter(self, parameter: Parameter | list[Parameter]) -> None:
+    def add_parameter(
+        self, parameter: Parameter | list[Parameter], update_if_exist: bool = False
+    ) -> None:
         """Add parameters to stack template.
 
         :param parameter: parameter to add to the template
+        :param update_if_exist: update the parameter if already exists, avoiding
+            the duplicate key exception
         """
         if not isinstance(parameter, list):
             parameter = [parameter]
 
         for param in parameter:
-            if param.title in self.template.parameters:
+            if update_if_exist and param.title in self.template.parameters:
                 self.template.parameters[param.title] = param
             else:
                 self.template.add_parameter(param)
 
-    def add_output(self, output: Output | list[Output]) -> None:
+    def add_output(
+        self, output: Output | list[Output], update_if_exist: bool = False
+    ) -> None:
         """Add outputs to stack template.
 
         :param output: output to add to the template
+        :param update_if_exist: update the output if already exists, avoiding
+            the duplicate key exception
         """
-        self.template.add_output(output)
+        if not isinstance(output, list):
+            output = [output]
+
+        for out in output:
+            if update_if_exist and out.title in self.template.outputs:
+                self.template.outputs[out.title] = out
+            else:
+                self.template.add_output(out)
 
     def add_condition(self, condition_name: str, condition: ConditionFunction) -> None:
         """Add condition to stack template.

tests/tests_e3_aws/troposphere/stack/stack_test.py

@@ -2,7 +2,7 @@
 
 import json
 from pathlib import Path
-
+import pytest
 from troposphere import Parameter, Output, Export
 
 from e3.aws.troposphere.s3.bucket import Bucket
@@ -59,6 +59,28 @@ def test_add_parameters() -> None:
     assert stack.export()["Parameters"] == expected_template
 
 
+def test_update_parameter() -> None:
+    """Test updating an already existing parameter."""
+    p = Parameter("Parameter", Description="My parameter", Type="String")
+    stack = Stack("test-stack", "this is a test stack")
+    stack.add_parameter(p)
+
+    # This one should fail because of the duplicate key
+    with pytest.raises(match='duplicate key "Parameter" detected'):
+        stack.add_parameter(p)
+
+    # This one should update the parameter
+    p.Description = "Updated parameter"
+    stack.add_parameter(p, update_if_exist=True)
+
+    assert stack.export()["Parameters"] == {
+        "Parameter": {
+            "Description": "Updated parameter",
+            "Type": "String",
+        },
+    }
+
+
 def test_add_outputs() -> None:
     """Test adding outputs to a stack."""
     stack = Stack("test-stack", "this is a test stack")
@@ -85,6 +107,30 @@ def test_add_outputs() -> None:
     assert stack.export()["Resources"] == expected_template
 
 
+def test_update_output() -> None:
+    """Test updating an already existing output."""
+    o = Output("Output", Description="My output", Value=Export(name="Output"))
+    stack = Stack("test-stack", "this is a test stack")
+    stack.add_output(o)
+
+    # This one should fail because of the duplicate key
+    with pytest.raises(match='duplicate key "Output" detected'):
+        stack.add_output(o)
+
+    # This one should update the output
+    o.Description = "Updated output"
+    stack.add_output(o, update_if_exist=True)
+
+    assert stack.export()["Outputs"] == {
+        "Output": {
+            "Description": "Updated output",
+            "Value": {
+                "Name": "Output",
+            },
+        },
+    }
+
+
 def test_extend() -> None:
     """Test adding multiple construct and retrieving an AWSObject from a stack."""
     stack = Stack("test-stack", "this is a test stack")