fix(cfn-lang-ext): preserve inline-source intrinsics through LE-aware build merge (#9029)

tests/integration/buildcmd/test_build_cmd_language_extensions.py

@@ -111,6 +111,124 @@ def test_build_dynamic_codeuri_generates_mappings(self):
         self.assertEqual(codeuri["Fn::FindInMap"][0], mapping_name)
         self.assertEqual(codeuri["Fn::FindInMap"][1], {"Ref": "FunctionName"})
 
+    def test_build_zipfile_fnsub_preserves_intrinsic(self):
+        """Regression for #9029: `Code.ZipFile: !Sub ...` on a Lambda function under
+        AWS::LanguageExtensions must round-trip through `sam build` with the Fn::Sub
+        intact. Before the fix, the LE-aware merge step copied the LE-resolved value
+        back, baking default pseudo-parameter values (us-east-1, 123456789012) into
+        the built template."""
+        self.template_path = str(Path(self.test_data_path, "language-extensions-zipfile-fnsub", "template.yaml"))
+
+        cmdlist = self.get_command_list()
+        command_result = run_command(cmdlist, cwd=self.working_dir)
+
+        self.assertEqual(command_result.process.returncode, 0, f"Build failed: {command_result.stderr.decode('utf-8')}")
+
+        built_template_path = self.default_build_dir.joinpath("template.yaml")
+        self.assertTrue(built_template_path.exists())
+
+        with open(built_template_path, "r") as f:
+            built_template = yaml.safe_load(f)
+
+        function_props = built_template["Resources"]["MyTriggerFunction"]["Properties"]
+
+        # Code.ZipFile must remain a dict containing Fn::Sub, not a resolved string.
+        zipfile_value = function_props["Code"]["ZipFile"]
+        self.assertIsInstance(
+            zipfile_value, dict, f"Expected Code.ZipFile to remain a Fn::Sub dict, got: {zipfile_value!r}"
+        )
+        self.assertIn("Fn::Sub", zipfile_value)
+        sub_body = zipfile_value["Fn::Sub"]
+        self.assertIn("${AWS::Region}", sub_body)
+        self.assertIn("${AWS::AccountId}", sub_body)
+        # Confirm no default pseudo-param values leaked through.
+        self.assertNotIn("us-east-1", sub_body)
+        self.assertNotIn("123456789012", sub_body)
+
+        # Role's Fn::Sub should likewise be preserved (sanity check on the same merge path).
+        role_value = function_props["Role"]
+        self.assertIsInstance(role_value, dict)
+        self.assertIn("Fn::Sub", role_value)
+        self.assertIn("${AWS::AccountId}", role_value["Fn::Sub"])
+
+    def test_build_foreach_static_zipfile_fnsub_preserves_intrinsic(self):
+        """Regression for #9029 (ForEach static branch / case B): inside a
+        Fn::ForEach body, `Code.ZipFile: !Sub ...` whose body does NOT reference
+        the loop variable must round-trip through `sam build` with the Fn::Sub
+        intact. The static-branch merge has no build artifact for an inline-source
+        Lambda, so the user-authored property must be preserved verbatim."""
+        self.template_path = str(
+            Path(self.test_data_path, "language-extensions-foreach-zipfile-static", "template.yaml")
+        )
+
+        cmdlist = self.get_command_list() + ["--language-extensions"]
+        command_result = run_command(cmdlist, cwd=self.working_dir)
+
+        self.assertEqual(command_result.process.returncode, 0, f"Build failed: {command_result.stderr.decode('utf-8')}")
+
+        built_template_path = self.default_build_dir.joinpath("template.yaml")
+        self.assertTrue(built_template_path.exists())
+
+        with open(built_template_path, "r") as f:
+            built_template = yaml.safe_load(f)
+
+        # Fn::ForEach structure must be preserved
+        resources = built_template.get("Resources", {})
+        foreach_block = resources.get("Fn::ForEach::Workers")
+        self.assertIsNotNone(foreach_block, "Fn::ForEach::Workers must survive in built template")
+        self.assertEqual(foreach_block[0], "WorkerName")
+        self.assertEqual(foreach_block[1], ["Alpha", "Beta"])
+
+        # The body's Code.ZipFile must remain a Fn::Sub dict, not a resolved string,
+        # and must still reference ${AWS::Region} / ${AWS::AccountId}.
+        body = foreach_block[2]
+        worker_props = body["${WorkerName}Worker"]["Properties"]
+        zipfile_value = worker_props["Code"]["ZipFile"]
+        self.assertIsInstance(zipfile_value, dict)
+        self.assertIn("Fn::Sub", zipfile_value)
+        sub_body = zipfile_value["Fn::Sub"]
+        self.assertIn("${AWS::Region}", sub_body)
+        self.assertIn("${AWS::AccountId}", sub_body)
+        self.assertNotIn("us-east-1", sub_body)
+        self.assertNotIn("123456789012", sub_body)
+
+    def test_build_foreach_dynamic_inline_zipfile_preserved(self):
+        """Regression for #9029 (ForEach dynamic branch / case C): inside a
+        Fn::ForEach body, a property that references the loop variable but produces
+        no build artifact for any iteration (e.g. inline-source `Code.ZipFile`) is
+        passed through verbatim. CFN's LanguageExtensions transform expands the
+        ForEach at deploy time, substituting the loop variable in each per-iteration
+        copy, so the inline body works correctly without a SAM-built artifact."""
+        self.template_path = str(
+            Path(self.test_data_path, "language-extensions-foreach-zipfile-dynamic", "template.yaml")
+        )
+
+        cmdlist = self.get_command_list() + ["--language-extensions"]
+        command_result = run_command(cmdlist, cwd=self.working_dir)
+
+        self.assertEqual(command_result.process.returncode, 0, f"Build failed: {command_result.stderr.decode('utf-8')}")
+
+        built_template_path = self.default_build_dir.joinpath("template.yaml")
+        with open(built_template_path, "r") as f:
+            built_template = yaml.safe_load(f)
+
+        resources = built_template.get("Resources", {})
+        foreach_block = resources["Fn::ForEach::Workers"]
+        body = foreach_block[2]
+        worker_props = body["${WorkerName}Worker"]["Properties"]
+        zipfile_value = worker_props["Code"]["ZipFile"]
+        # The inline ZipFile must remain a Fn::Sub dict with both the loop
+        # variable and pseudo-parameter intrinsics intact — neither resolved
+        # at build time nor replaced by a Mapping lookup.
+        self.assertIsInstance(zipfile_value, dict)
+        self.assertIn("Fn::Sub", zipfile_value)
+        sub_body = zipfile_value["Fn::Sub"]
+        self.assertIn("${WorkerName}", sub_body)
+        role_value = worker_props["Role"]
+        self.assertIsInstance(role_value, dict)
+        self.assertIn("Fn::Sub", role_value)
+        self.assertIn("${AWS::AccountId}", role_value["Fn::Sub"])
+
     def test_build_nested_foreach_dynamic_codeuri_generates_mappings(self):
         """Test that nested Fn::ForEach with dynamic CodeUri generates Mappings."""
         self.template_path = str(

tests/integration/testdata/buildcmd/language-extensions-foreach-zipfile-dynamic/template.yaml

@@ -0,0 +1,31 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform:
+  - AWS::LanguageExtensions
+  - AWS::Serverless-2016-10-31
+
+Description: >
+  Regression test for #9029, ForEach dynamic-branch (case C). Each expanded
+  Lambda has `Code: {ZipFile: !Sub ...}` whose body references the loop
+  variable. There is no build artifact for `sam build` to merge, so the
+  property must be passed through verbatim. CFN's LanguageExtensions transform
+  expands the ForEach at deploy time and substitutes the loop variable in each
+  per-iteration copy, so the inline body works correctly without a SAM-built
+  artifact.
+
+Resources:
+  Fn::ForEach::Workers:
+    - WorkerName
+    - - Alpha
+      - Beta
+    - ${WorkerName}Worker:
+        Type: AWS::Lambda::Function
+        Properties:
+          FunctionName: !Sub ${WorkerName}-worker
+          Runtime: python3.11
+          Handler: index.handler
+          Role: !Sub 'arn:aws:iam::${AWS::AccountId}:role/worker-role'
+          Code:
+            ZipFile: !Sub |
+              import boto3
+              def handler(event, context):
+                  print('worker name: ${WorkerName}')

tests/integration/testdata/buildcmd/language-extensions-foreach-zipfile-static/template.yaml

@@ -0,0 +1,32 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform:
+  - AWS::LanguageExtensions
+  - AWS::Serverless-2016-10-31
+
+Description: >
+  Regression test for #9029, ForEach static-branch (case B). Each expanded
+  Lambda has `Code: {ZipFile: !Sub ...}` whose body references CFN pseudo-
+  parameters but NOT the loop variable. The static-branch merge must skip
+  these resources (no build artifact) so the user-authored `Fn::Sub`
+  survives intact in the built template.
+
+Resources:
+  Fn::ForEach::Workers:
+    - WorkerName
+    - - Alpha
+      - Beta
+    - ${WorkerName}Worker:
+        Type: AWS::Lambda::Function
+        Properties:
+          FunctionName: !Sub ${WorkerName}-worker
+          Runtime: python3.11
+          Handler: index.handler
+          Role: !Sub 'arn:aws:iam::${AWS::AccountId}:role/worker-role'
+          Code:
+            ZipFile: !Sub |
+              import boto3
+              def handler(event, context):
+                  client = boto3.client('logs')
+                  client.put_log_events(
+                      logGroupName='/aws/lambda/worker-${AWS::Region}-${AWS::AccountId}',
+                  )

tests/integration/testdata/buildcmd/language-extensions-zipfile-fnsub/template.yaml

@@ -0,0 +1,26 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Transform:
+  - AWS::LanguageExtensions
+  - AWS::Serverless-2016-10-31
+
+Description: >
+  Regression test for #9029. Inline-source `Code: {ZipFile: !Sub ...}` on a
+  Lambda function under AWS::LanguageExtensions must round-trip through
+  `sam build` with the `Fn::Sub` intact (no pseudo-parameter resolution).
+
+Resources:
+  MyTriggerFunction:
+    Type: AWS::Lambda::Function
+    Properties:
+      FunctionName: my-trigger
+      Runtime: python3.11
+      Handler: index.handler
+      Role: !Sub 'arn:aws:iam::${AWS::AccountId}:role/my-role'
+      Code:
+        ZipFile: !Sub |
+          import boto3
+          def handler(event, context):
+              client = boto3.client('stepfunctions')
+              client.start_execution(
+                  stateMachineArn='arn:aws:states:${AWS::Region}:${AWS::AccountId}:stateMachine:my-state-machine',
+              )