diff --git a/samcli/commands/_utils/template.py b/samcli/commands/_utils/template.py index aa85fe3350..2600fdbd92 100644 --- a/samcli/commands/_utils/template.py +++ b/samcli/commands/_utils/template.py @@ -148,7 +148,11 @@ def _update_relative_paths(template_dict, original_root, new_root): properties[path_prop_name] = updated_path - for _, resource in template_dict.get("Resources", {}).items(): + for resource_id, resource in template_dict.get("Resources", {}).items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue + resource_type = resource.get("Type") if resource_type not in RESOURCES_WITH_LOCAL_PATHS: diff --git a/samcli/lib/iac/cdk/utils.py b/samcli/lib/iac/cdk/utils.py index e4cab0dfe6..40d7d658df 100644 --- a/samcli/lib/iac/cdk/utils.py +++ b/samcli/lib/iac/cdk/utils.py @@ -42,7 +42,10 @@ def _resource_level_metadata_exists(resources: Dict) -> bool: Dict of resources to look through """ - for _, resource in resources.items(): + for resource_id, resource in resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue if resource.get("Type", "") == CDK_METADATA_TYPE_VALUE: return True return False @@ -58,7 +61,10 @@ def _cdk_path_metadata_exists(resources: Dict) -> bool: Dict of resources to look through """ - for _, resource in resources.items(): + for resource_id, resource in resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue metadata = resource.get("Metadata", {}) if metadata and CDK_PATH_METADATA_KEY in metadata: return True diff --git a/samcli/lib/providers/sam_function_provider.py b/samcli/lib/providers/sam_function_provider.py index 5787fb409f..7b1e3cb23a 100644 --- a/samcli/lib/providers/sam_function_provider.py +++ b/samcli/lib/providers/sam_function_provider.py @@ -202,6 +202,11 @@ def _extract_functions( result: Dict[str, Function] = {} # a dict with full_path as key and extracted function as value for stack in stacks: for name, resource in stack.resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if name.startswith("Fn::ForEach::") or not isinstance(resource, dict): + LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}") + continue + resource_type = resource.get("Type") resource_properties = resource.get("Properties", {}) resource_metadata = resource.get("Metadata", None) diff --git a/samcli/lib/providers/sam_layer_provider.py b/samcli/lib/providers/sam_layer_provider.py index ecd642f751..07184e03c9 100644 --- a/samcli/lib/providers/sam_layer_provider.py +++ b/samcli/lib/providers/sam_layer_provider.py @@ -83,6 +83,11 @@ def _extract_layers(self) -> List[LayerVersion]: layers = [] for stack in self._stacks: for name, resource in stack.resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if name.startswith("Fn::ForEach::") or not isinstance(resource, dict): + LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}") + continue + # In the list of layers that is defined within a template, you can reference a LayerVersion resource. # When running locally, we need to follow that Ref so we can extract the local path to the layer code. resource_type = resource.get("Type") diff --git a/samcli/lib/providers/sam_stack_provider.py b/samcli/lib/providers/sam_stack_provider.py index d27ad19a8f..e86ca09889 100644 --- a/samcli/lib/providers/sam_stack_provider.py +++ b/samcli/lib/providers/sam_stack_provider.py @@ -110,6 +110,11 @@ def _extract_stacks(self) -> None: """ for name, resource in self._resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if name.startswith("Fn::ForEach::") or not isinstance(resource, dict): + LOG.debug(f"Skipping Fn::ForEach construct or non-dict resource: {name}") + continue + resource_type = resource.get("Type") resource_properties = resource.get("Properties", {}) resource_metadata = resource.get("Metadata", None) diff --git a/samcli/lib/samlib/resource_metadata_normalizer.py b/samcli/lib/samlib/resource_metadata_normalizer.py index f1bb101888..47385c9df7 100644 --- a/samcli/lib/samlib/resource_metadata_normalizer.py +++ b/samcli/lib/samlib/resource_metadata_normalizer.py @@ -62,6 +62,10 @@ def normalize(template_dict, normalize_parameters=False): resources = template_dict.get(RESOURCES_KEY, {}) for logical_id, resource in resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if logical_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue + # copy metadata to another variable, change its values and assign it back in the end resource_metadata = deepcopy(resource.get(METADATA_KEY)) or {} @@ -228,6 +232,10 @@ def get_resource_id(resource_properties, logical_id): str The unique function id """ + # Skip Fn::ForEach constructs which are lists, not dicts + if not isinstance(resource_properties, dict): + return logical_id + resource_metadata = resource_properties.get("Metadata", {}) customer_defined_id = resource_metadata.get(SAM_RESOURCE_ID_KEY) diff --git a/samcli/lib/samlib/wrapper.py b/samcli/lib/samlib/wrapper.py index d4c510b5df..3e42403daf 100644 --- a/samcli/lib/samlib/wrapper.py +++ b/samcli/lib/samlib/wrapper.py @@ -26,6 +26,7 @@ from samtranslator.validator.validator import SamTemplateValidator from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException +from samcli.lib.utils.foreach_handler import filter_foreach_constructs from .local_uri_plugin import SupportLocalUriPlugin @@ -69,8 +70,19 @@ def run_plugins(self, convert_local_uris=True): # Temporarily disabling validation for DeletionPolicy and UpdateReplacePolicy when language extensions are set self._patch_language_extensions() + # Filter out Fn::ForEach constructs before parsing + # CloudFormation will handle these server-side + template_copy, foreach_constructs = filter_foreach_constructs(template_copy) + try: parser.parse(template_copy, all_plugins) # parse() will run all configured plugins + + # Add back Fn::ForEach constructs after parsing + if foreach_constructs: + if "Resources" not in template_copy: + template_copy["Resources"] = {} + template_copy["Resources"].update(foreach_constructs) + except InvalidDocumentException as e: raise InvalidSamDocumentException( functools.reduce(lambda message, error: message + " " + str(error), e.causes, str(e)) @@ -100,6 +112,8 @@ def patched_func(self): SamResource.valid = patched_func + # Removed: _filter_foreach_constructs() now uses shared utility from samcli.lib.utils.foreach_handler + @staticmethod def _check_using_language_extension(template: Dict) -> bool: """ diff --git a/samcli/lib/translate/sam_template_validator.py b/samcli/lib/translate/sam_template_validator.py index 5c2a9ed33d..5236fb48b5 100644 --- a/samcli/lib/translate/sam_template_validator.py +++ b/samcli/lib/translate/sam_template_validator.py @@ -13,6 +13,7 @@ from samtranslator.translator.translator import Translator from samcli.commands.validate.lib.exceptions import InvalidSamDocumentException +from samcli.lib.utils.foreach_handler import filter_foreach_constructs from samcli.lib.utils.packagetype import IMAGE, ZIP from samcli.lib.utils.resources import AWS_SERVERLESS_FUNCTION from samcli.yamlhelper import yaml_dump @@ -82,12 +83,24 @@ def get_translated_template_if_valid(self): self._replace_local_codeuri() self._replace_local_image() + # Filter out Fn::ForEach constructs before translation + # CloudFormation will handle these server-side + template_to_translate, foreach_constructs = filter_foreach_constructs(self.sam_template) + try: template = sam_translator.translate( - sam_template=self.sam_template, + sam_template=template_to_translate, parameter_values=self.parameter_overrides, get_managed_policy_map=self._get_managed_policy_map, ) + + # Add back Fn::ForEach constructs after translation + if foreach_constructs: + if "Resources" not in template: + template["Resources"] = {} + template["Resources"].update(foreach_constructs) + LOG.debug("Preserved %d Fn::ForEach construct(s) in template", len(foreach_constructs)) + LOG.debug("Translated template is:\n%s", yaml_dump(template)) return yaml_dump(template) except InvalidDocumentException as e: @@ -95,6 +108,8 @@ def get_translated_template_if_valid(self): functools.reduce(lambda message, error: message + " " + str(error), e.causes, str(e)) ) from e + # Removed: _filter_foreach_constructs() now uses shared utility from samcli.lib.utils.foreach_handler + @functools.lru_cache(maxsize=None) def _get_managed_policy_map(self) -> Dict[str, str]: """ @@ -130,7 +145,11 @@ def _replace_local_codeuri(self): ): SamTemplateValidator._update_to_s3_uri("CodeUri", properties) - for _, resource in all_resources.items(): + for resource_id, resource in all_resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue + resource_type = resource.get("Type") resource_dict = resource.get("Properties", {}) @@ -158,7 +177,11 @@ def _replace_local_image(self): This ensures sam validate works without having to package the app or use ImageUri. """ resources = self.sam_template.get("Resources", {}) - for _, resource in resources.items(): + for resource_id, resource in resources.items(): + # Skip Fn::ForEach constructs which are lists, not dicts + if resource_id.startswith("Fn::ForEach::") or not isinstance(resource, dict): + continue + resource_type = resource.get("Type") properties = resource.get("Properties", {}) diff --git a/samcli/lib/utils/foreach_handler.py b/samcli/lib/utils/foreach_handler.py new file mode 100644 index 0000000000..b6dc9587cd --- /dev/null +++ b/samcli/lib/utils/foreach_handler.py @@ -0,0 +1,66 @@ +""" +Utility functions for handling CloudFormation Fn::ForEach intrinsic function +""" +import copy +import logging +from typing import Dict, Tuple + +LOG = logging.getLogger(__name__) + + +def filter_foreach_constructs(template: Dict) -> Tuple[Dict, Dict]: + """ + Filter out Fn::ForEach constructs from template before SAM transformation. + CloudFormation will handle these server-side during deployment. + + Parameters + ---------- + template : Dict + The SAM/CloudFormation template dictionary + + Returns + ------- + Tuple[Dict, Dict] + (template_without_foreach, foreach_constructs_dict) + + Notes + ----- + Fn::ForEach constructs are identified by resource IDs starting with "Fn::ForEach::" + These constructs are lists, not dicts, and would cause parsing errors if processed locally. + CloudFormation expands them server-side during deployment. + """ + template_copy = copy.deepcopy(template) + resources = template_copy.get("Resources", {}) + + # If no Resources section, nothing to filter + if not resources: + return template_copy, {} + + # Separate Fn::ForEach constructs from regular resources + foreach_constructs = {} + regular_resources = {} + + for resource_id, resource in resources.items(): + if resource_id.startswith("Fn::ForEach::"): + foreach_constructs[resource_id] = resource + LOG.info( + f"Detected Fn::ForEach construct '{resource_id}'. " + "This will be expanded by CloudFormation during deployment." + ) + else: + regular_resources[resource_id] = resource + + # If template only has ForEach constructs, add a placeholder resource + # to satisfy SAM Translator's requirement for non-empty Resources section + if not regular_resources and foreach_constructs: + regular_resources["__PlaceholderForForEachOnly"] = { + "Type": "AWS::CloudFormation::WaitConditionHandle", + "Properties": {}, + } + LOG.debug("Added placeholder resource since template only contains Fn::ForEach constructs") + + # Only update Resources if there were any ForEach constructs + if foreach_constructs: + template_copy["Resources"] = regular_resources + + return template_copy, foreach_constructs \ No newline at end of file diff --git a/tests/integration/testdata/fn-foreach/basic-topics-template.yaml b/tests/integration/testdata/fn-foreach/basic-topics-template.yaml new file mode 100644 index 0000000000..e72c88bdcb --- /dev/null +++ b/tests/integration/testdata/fn-foreach/basic-topics-template.yaml @@ -0,0 +1,14 @@ +AWSTemplateFormatVersion: '2010-09-09' +Transform: + - AWS::LanguageExtensions + - AWS::Serverless-2016-10-31 + +Resources: + 'Fn::ForEach::Topics': + - TopicName + - [Success, Failure, Timeout, Unknown] + - 'SnsTopic${TopicName}': + Type: AWS::SNS::Topic + Properties: + TopicName: !Ref TopicName + FifoTopic: true diff --git a/tests/unit/lib/samlib/test_wrapper_foreach.py b/tests/unit/lib/samlib/test_wrapper_foreach.py new file mode 100644 index 0000000000..e61a373d93 --- /dev/null +++ b/tests/unit/lib/samlib/test_wrapper_foreach.py @@ -0,0 +1,417 @@ +""" +Unit tests for Fn::ForEach support in SamTranslatorWrapper and shared foreach_handler utility +""" +import copy +from unittest import TestCase + +from samcli.lib.samlib.wrapper import SamTranslatorWrapper +from samcli.lib.utils.foreach_handler import filter_foreach_constructs + + +class TestFilterForEachConstructs(TestCase): + """Test the _filter_foreach_constructs static method""" + + def test_filters_foreach_from_regular_resources(self): + """Verify Fn::ForEach constructs are filtered out while regular resources remain""" + template = { + "AWSTemplateFormatVersion": "2010-09-09", + "Transform": ["AWS::LanguageExtensions", "AWS::Serverless-2016-10-31"], + "Resources": { + "Fn::ForEach::Topics": [ + "TopicName", + ["Success", "Failure"], + {"SnsTopic${TopicName}": {"Type": "AWS::SNS::Topic"}}, + ], + "RegularFunction": { + "Type": "AWS::Serverless::Function", + "Properties": {"Handler": "app.handler"}, + }, + "AnotherResource": {"Type": "AWS::S3::Bucket", "Properties": {}}, + }, + } + + filtered, foreach = filter_foreach_constructs(template) + + # Regular resources should remain + self.assertIn("RegularFunction", filtered["Resources"]) + self.assertIn("AnotherResource", filtered["Resources"]) + + # ForEach should be removed + self.assertNotIn("Fn::ForEach::Topics", filtered["Resources"]) + + # ForEach should be in separate dict + self.assertIn("Fn::ForEach::Topics", foreach) + self.assertEqual(len(foreach), 1) + + def test_adds_placeholder_when_only_foreach(self): + """Verify placeholder resource is added when template only has ForEach""" + template = { + "Resources": { + "Fn::ForEach::OnlyThis": [ + "Item", + ["A", "B"], + {"Resource${Item}": {"Type": "AWS::SNS::Topic"}}, + ] + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # Placeholder should be added + self.assertIn("__PlaceholderForForEachOnly", filtered["Resources"]) + self.assertEqual( + filtered["Resources"]["__PlaceholderForForEachOnly"]["Type"], + "AWS::CloudFormation::WaitConditionHandle", + ) + + # ForEach should be in separate dict + self.assertIn("Fn::ForEach::OnlyThis", foreach) + + def test_preserves_multiple_foreach_constructs(self): + """Verify multiple Fn::ForEach constructs are all filtered and preserved""" + template = { + "Resources": { + "Fn::ForEach::Topics": ["X", ["A"], {"T${X}": {"Type": "AWS::SNS::Topic"}}], + "Fn::ForEach::Functions": ["Y", ["B"], {"F${Y}": {"Type": "AWS::Lambda::Function"}}], + "Fn::ForEach::Queues": ["Z", ["C"], {"Q${Z}": {"Type": "AWS::SQS::Queue"}}], + "RegularResource": {"Type": "AWS::S3::Bucket"}, + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # Regular resource remains + self.assertIn("RegularResource", filtered["Resources"]) + + # All ForEach removed from filtered + self.assertNotIn("Fn::ForEach::Topics", filtered["Resources"]) + self.assertNotIn("Fn::ForEach::Functions", filtered["Resources"]) + self.assertNotIn("Fn::ForEach::Queues", filtered["Resources"]) + + # All ForEach in separate dict + self.assertEqual(len(foreach), 3) + self.assertIn("Fn::ForEach::Topics", foreach) + self.assertIn("Fn::ForEach::Functions", foreach) + self.assertIn("Fn::ForEach::Queues", foreach) + + def test_handles_empty_resources_dict(self): + """Verify handling of template with empty Resources dict""" + template = {"Resources": {}} + + filtered, foreach = filter_foreach_constructs(template) + + # Should return empty foreach dict + self.assertEqual(foreach, {}) + # Resources should still be empty dict + self.assertEqual(filtered["Resources"], {}) + + def test_handles_template_with_no_foreach(self): + """Verify normal templates without ForEach are unchanged""" + template = { + "Resources": { + "Function1": {"Type": "AWS::Serverless::Function"}, + "Bucket1": {"Type": "AWS::S3::Bucket"}, + } + } + + original = copy.deepcopy(template) + filtered, foreach = filter_foreach_constructs(template) + + # No ForEach found + self.assertEqual(foreach, {}) + + # Resources unchanged + self.assertEqual(filtered["Resources"], original["Resources"]) + + def test_foreach_with_complex_structure(self): + """Test ForEach with nested intrinsic functions""" + template = { + "Parameters": {"EnvNames": {"Type": "CommaDelimitedList"}}, + "Resources": { + "Fn::ForEach::ComplexFunctions": [ + "Env", + {"Ref": "EnvNames"}, # Ref in iterator + { + "Function${Env}": { + "Type": "AWS::Serverless::Function", + "Properties": { + "FunctionName": {"Fn::Sub": "func-${Env}"}, # Sub in properties + }, + } + }, + ] + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # ForEach should be preserved with all its complexity + self.assertIn("Fn::ForEach::ComplexFunctions", foreach) + self.assertEqual(foreach["Fn::ForEach::ComplexFunctions"][1], {"Ref": "EnvNames"}) + + def test_function_uses_deepcopy_internally(self): + """Verify the function uses deepcopy so original is not directly modified""" + template = { + "Resources": { + "Fn::ForEach::Items": ["X", ["A"], {"R${X}": {"Type": "AWS::SNS::Topic"}}], + "RegularResource": {"Type": "AWS::S3::Bucket"}, + } + } + + template_id_before = id(template["Resources"]) + filtered, foreach = filter_foreach_constructs(template) + + # The filtered template should be a different object + self.assertNotEqual(id(filtered["Resources"]), template_id_before) + + # Foreach should contain the ForEach construct + self.assertIn("Fn::ForEach::Items", foreach) + + # Filtered should not contain ForEach + self.assertNotIn("Fn::ForEach::Items", filtered["Resources"]) + + def test_foreach_detection_is_prefix_based(self): + """Verify detection uses startswith, not exact match""" + template = { + "Resources": { + "Fn::ForEach::TopicsV1": ["X", ["A"], {}], + "Fn::ForEach::Functions": ["Y", ["B"], {}], + "NotForEach": {"Type": "AWS::S3::Bucket"}, + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # Both ForEach variants detected + self.assertEqual(len(foreach), 2) + self.assertIn("Fn::ForEach::TopicsV1", foreach) + self.assertIn("Fn::ForEach::Functions", foreach) + + +class TestForEachEdgeCases(TestCase): + """Test edge cases and error conditions""" + + def test_foreach_with_empty_collection(self): + """Test ForEach with empty list""" + template = {"Resources": {"Fn::ForEach::Empty": ["X", [], {"R${X}": {"Type": "AWS::SNS::Topic"}}]}} + + filtered, foreach = filter_foreach_constructs(template) + + # Should still be filtered (CloudFormation will handle empty list) + self.assertIn("Fn::ForEach::Empty", foreach) + self.assertIn("__PlaceholderForForEachOnly", filtered["Resources"]) + + def test_resource_named_starting_with_foreach_but_is_dict(self): + """Test resource that starts with 'Fn::ForEach' but is a regular resource dict""" + template = { + "Resources": { + "Fn::ForEach::MyCustomName": { # Dict, not list - unusual naming + "Type": "AWS::Serverless::Function", + "Properties": {}, + } + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # Current implementation filters by prefix, so this would be filtered + # This is acceptable - users shouldn't name resources like this + self.assertIn("Fn::ForEach::MyCustomName", foreach) + + def test_mixed_resources_preserves_correct_ones(self): + """Test complex mix of ForEach and regular resources""" + template = { + "Resources": { + "Bucket1": {"Type": "AWS::S3::Bucket"}, + "Fn::ForEach::Set1": ["A", ["1"], {}], + "Function1": {"Type": "AWS::Serverless::Function"}, + "Fn::ForEach::Set2": ["B", ["2"], {}], + "Table1": {"Type": "AWS::DynamoDB::Table"}, + "Fn::ForEach::Set3": ["C", ["3"], {}], + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # 3 regular resources + self.assertEqual(len(filtered["Resources"]), 3) + self.assertIn("Bucket1", filtered["Resources"]) + self.assertIn("Function1", filtered["Resources"]) + self.assertIn("Table1", filtered["Resources"]) + + # 3 ForEach constructs + self.assertEqual(len(foreach), 3) + + def test_foreach_structure_is_preserved_exactly(self): + """Verify ForEach structure is not modified during filtering""" + original_foreach = [ + "EnvName", + {"Ref": "Environments"}, + { + "Topic${EnvName}": { + "Type": "AWS::SNS::Topic", + "Properties": {"TopicName": {"Fn::Sub": "topic-${EnvName}"}}, + } + }, + ] + + template = {"Resources": {"Fn::ForEach::Topics": copy.deepcopy(original_foreach)}} + + _, foreach = filter_foreach_constructs(template) + + # Structure should be identical + self.assertEqual(foreach["Fn::ForEach::Topics"], original_foreach) + + def test_foreach_with_malformed_structure(self): + """Test ForEach with unexpected structure""" + template = { + "Resources": { + "Fn::ForEach::Malformed": "not-a-list", # Should be list, but isn't + "RegularResource": {"Type": "AWS::S3::Bucket"}, + } + } + + filtered, foreach = filter_foreach_constructs(template) + + # Malformed ForEach should still be filtered (by prefix) + self.assertIn("Fn::ForEach::Malformed", foreach) + self.assertIn("RegularResource", filtered["Resources"]) + + +class TestForEachWithTransforms(TestCase): + """Test ForEach with different Transform configurations""" + + def test_with_language_extensions_only(self): + """Test with only LanguageExtensions transform""" + template = { + "Transform": "AWS::LanguageExtensions", + "Resources": {"Fn::ForEach::Items": ["X", ["A"], {}]}, + } + + filtered, foreach = filter_foreach_constructs(template) + + # Should still work + self.assertIn("Fn::ForEach::Items", foreach) + + def test_with_serverless_only_no_foreach(self): + """Test normal Serverless template without LanguageExtensions""" + template = { + "Transform": "AWS::Serverless-2016-10-31", + "Resources": {"Function1": {"Type": "AWS::Serverless::Function"}}, + } + + filtered, foreach = filter_foreach_constructs(template) + + # No ForEach in Serverless-only template + self.assertEqual(foreach, {}) + self.assertIn("Function1", filtered["Resources"]) + + def test_with_both_transforms_in_list(self): + """Test with list of transforms (correct usage)""" + template = { + "Transform": ["AWS::LanguageExtensions", "AWS::Serverless-2016-10-31"], + "Resources": { + "Fn::ForEach::Items": ["X", ["A"], {}], + "Function1": {"Type": "AWS::Serverless::Function"}, + }, + } + + filtered, foreach = filter_foreach_constructs(template) + + self.assertEqual(len(foreach), 1) + self.assertIn("Function1", filtered["Resources"]) + + +class TestForEachRealWorldScenarios(TestCase): + """Test real-world scenarios from issue #5647""" + + def test_multi_tenant_lambda_functions(self): + """Test multi-tenant use case - multiple tenant functions""" + template = { + "Transform": ["AWS::LanguageExtensions", "AWS::Serverless-2016-10-31"], + "Parameters": {"Tenants": {"Type": "CommaDelimitedList", "Default": "tenant1,tenant2"}}, + "Resources": { + "Fn::ForEach::TenantFunctions": [ + "TenantId", + {"Fn::Split": [",", {"Ref": "Tenants"}]}, + { + "${TenantId}Function": { + "Type": "AWS::Serverless::Function", + "Properties": { + "FunctionName": {"Fn::Sub": "${TenantId}-processor"}, + "Handler": "app.handler", + "Runtime": "python3.11", + }, + } + }, + ] + }, + } + + filtered, foreach = filter_foreach_constructs(template) + + # ForEach should be filtered + self.assertIn("Fn::ForEach::TenantFunctions", foreach) + # Placeholder added since no other resources + self.assertIn("__PlaceholderForForEachOnly", filtered["Resources"]) + + def test_sns_topics_from_issue_example(self): + """Test original issue #5647 example""" + template = { + "AWSTemplateFormatVersion": "2010-09-09", + "Transform": ["AWS::LanguageExtensions", "AWS::Serverless-2016-10-31"], + "Resources": { + "Fn::ForEach::Topics": [ + "TopicName", + ["Success", "Failure", "Timeout", "Unknown"], + { + "SnsTopic${TopicName}": { + "Type": "AWS::SNS::Topic", + "Properties": {"TopicName": {"Ref": "TopicName"}, "FifoTopic": True}, + } + }, + ] + }, + } + + filtered, foreach = filter_foreach_constructs(template) + + # This should not crash (was the original bug) + self.assertIn("Fn::ForEach::Topics", foreach) + self.assertEqual(foreach["Fn::ForEach::Topics"][1], ["Success", "Failure", "Timeout", "Unknown"]) + + def test_iam_policy_statements_foreach(self): + """Test IAM policy use case - ForEach inside Policies (not top-level)""" + template = { + "Transform": ["AWS::LanguageExtensions", "AWS::Serverless-2016-10-31"], + "Parameters": {"BucketNames": {"Type": "CommaDelimitedList"}}, + "Resources": { + "MyFunction": { + "Type": "AWS::Serverless::Function", + "Properties": { + "Policies": { + "Fn::ForEach::S3Access": [ + "BucketName", + {"Ref": "BucketNames"}, + { + "Statement": [ + { + "Effect": "Allow", + "Action": "s3:GetObject", + "Resource": {"Fn::Sub": "arn:aws:s3:::${BucketName}/*"}, + } + ] + }, + ] + } + }, + } + }, + } + + filtered, foreach = filter_foreach_constructs(template) + + # Regular function should remain + self.assertIn("MyFunction", filtered["Resources"]) + # No top-level ForEach in this case (ForEach is inside Policies) + self.assertEqual(len(foreach), 0) \ No newline at end of file