Handle parameter conversions for different transforms

simonrw · simonrw · commit a95c522f1d79 · 2025-08-20T22:05:09.000+01:00
diff --git a/localstack-core/localstack/services/cloudformation/engine/transformers.py b/localstack-core/localstack/services/cloudformation/engine/transformers.py
@@ -14,6 +14,7 @@
 
 from localstack.aws.api import CommonServiceException
 from localstack.aws.connect import connect_to
+from localstack.services.cloudformation.engine.parameters import StackParameter
 from localstack.services.cloudformation.engine.policy_loader import create_policy_loader
 from localstack.services.cloudformation.engine.template_deployer import resolve_refs_recursively
 from localstack.services.cloudformation.engine.validations import ValidationError
@@ -39,7 +40,7 @@ class ResolveRefsRecursivelyContext:
     resources: dict
     mappings: dict
     conditions: dict
-    parameters: dict
+    parameters: dict[str, StackParameter]
 
     def resolve(self, value: Any) -> Any:
         return resolve_refs_recursively(
diff --git a/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_transform.py b/localstack-core/localstack/services/cloudformation/engine/v2/change_set_model_transform.py
@@ -42,7 +42,7 @@
 from localstack.services.cloudformation.engine.validations import ValidationError
 from localstack.services.cloudformation.stores import get_cloudformation_store
 from localstack.services.cloudformation.v2.entities import ChangeSet
-from localstack.services.cloudformation.v2.types import engine_parameter_value
+from localstack.services.cloudformation.v2.types import EngineParameter, engine_parameter_value
 from localstack.utils import testutil
 from localstack.utils.strings import long_uid
 
@@ -56,6 +56,20 @@
 _SCOPE_TRANSFORM_TEMPLATE_OUTCOME: Final[Scope] = Scope("TRANSFORM_TEMPLATE_OUTCOME")
 
 
+def engine_parameters_to_stack_parameters(
+    engine_parameters: dict[str, EngineParameter],
+) -> dict[str, StackParameter]:
+    out = {}
+    for name, engine_param in engine_parameters.items():
+        out[name] = StackParameter(
+            ParameterKey=name,
+            ParameterValue=engine_parameter_value(engine_param),
+            ResolvedValue=engine_param.get("resolved_value"),
+            ParameterType=engine_param["type_"],
+        )
+    return out
+
+
 # TODO: evaluate the use of subtypes to represent and validate types of transforms
 class GlobalTransform:
     name: str
@@ -74,8 +88,8 @@ class TransformPreprocParameter(TypedDict):
 
 
 class ChangeSetModelTransform(ChangeSetModelPreproc):
-    _before_parameters: Final[dict]
-    _after_parameters: Final[dict]
+    _before_parameters: Final[dict[str, EngineParameter] | None]
+    _after_parameters: Final[dict[str, EngineParameter] | None]
     _before_template: Final[Maybe[dict]]
     _after_template: Final[Maybe[dict]]
 
@@ -153,7 +167,10 @@ def _compute_include_transform(self, parameters: dict, fragment: dict) -> dict:
         return {**fragment, **template_to_include}
 
     def _apply_global_transform(
-        self, global_transform: GlobalTransform, template: dict, parameters: dict
+        self,
+        global_transform: GlobalTransform,
+        template: dict,
+        parameters: dict[str, EngineParameter],
     ) -> dict:
         transform_name = global_transform.name
         if transform_name == EXTENSIONS_TRANSFORM:
@@ -168,14 +185,20 @@ def _apply_global_transform(
                 resources,
                 mappings,
                 conditions,
-                parameters,
+                parameters=engine_parameters_to_stack_parameters(parameters),
             )
             transformed_template = apply_language_extensions_transform(template, resolve_context)
         elif transform_name == SERVERLESS_TRANSFORM:
+            # serverless transform just requires the key/value pairs
+            serverless_parameters = {}
+            for name, param in parameters.items():
+                serverless_parameters[name] = param.get("resolved_value") or engine_parameter_value(
+                    param
+                )
             transformed_template = self._apply_global_serverless_transformation(
                 region_name=self._change_set.region_name,
                 template=template,
-                parameters=parameters,
+                parameters=serverless_parameters,
             )
         elif transform_name == SECRETSMANAGER_TRANSFORM:
             # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/transform-aws-secretsmanager.html
@@ -217,11 +240,9 @@ def _execute_global_transforms(self) -> tuple[dict, dict]:
             else:
                 for before_global_transform in transform_before:
                     if not is_nothing(before_global_transform.name):
-                        before_parameters = {}
-
                         transformed_before_template = self._apply_global_transform(
                             global_transform=before_global_transform,
-                            parameters=before_parameters,
+                            parameters=self._before_parameters,
                             template=transformed_before_template,
                         )
 
@@ -235,17 +256,9 @@ def _execute_global_transforms(self) -> tuple[dict, dict]:
             transformed_after_template = self._after_template
             for after_global_transform in transform_after:
                 if not is_nothing(after_global_transform.name):
-                    after_parameters: dict[str, StackParameter] = {}
-                    for name, engine_param in self._after_parameters.items():
-                        after_parameters[name] = StackParameter(
-                            ParameterKey=name,
-                            ParameterValue=engine_parameter_value(engine_param),
-                            ResolvedValue=engine_param.get("resolved_value"),
-                            ParameterType=engine_param["type_"],
-                        )
                     transformed_after_template = self._apply_global_transform(
                         global_transform=after_global_transform,
-                        parameters=after_parameters,
+                        parameters=self._after_parameters,
                         template=transformed_after_template,
                     )
             # Macro transformations won't remove the transform from the template
