diff --git a/integration/config/service_names.py b/integration/config/service_names.py index 1bb88e5de..7be20e4f1 100644 --- a/integration/config/service_names.py +++ b/integration/config/service_names.py @@ -22,7 +22,6 @@ SNS = "SNS" SQS = "SQS" CUSTOM_DOMAIN = "CustomDomain" -ARM = "ARM" EFS = "EFS" S3_EVENTS = "S3Events" SELF_MANAGED_KAFKA = "SelfManagedKafka" @@ -36,3 +35,4 @@ EPHEMERAL_STORAGE = "EphemeralStorage" API_KEY = "ApiKey" APP_SYNC = "AppSync" +SNS_FILTER_POLICY_SCOPE = "SnsFilterPolicyScope" diff --git a/integration/helpers/resource.py b/integration/helpers/resource.py index ed1adb4e2..dccdbb43c 100644 --- a/integration/helpers/resource.py +++ b/integration/helpers/resource.py @@ -2,7 +2,7 @@ import random import re import string # pylint: disable=deprecated-module -from typing import Any, Callable, Dict, Set +from typing import Any, Callable, Dict, Iterator, Set from integration.config.service_names import ( APP_SYNC, @@ -12,6 +12,7 @@ REST_API, S3_EVENTS, SCHEDULE_EVENT, + SNS_FILTER_POLICY_SCOPE, SQS, STATE_MACHINE_INLINE_DEFINITION, ) @@ -210,6 +211,17 @@ def _resource_using_s3_events(resource: Dict[str, Any]) -> bool: return resource_type == "AWS::S3::Bucket" and properties.get("NotificationConfiguration") +def _get_all_event_sources(template_dict: Dict[str, Any]) -> Iterator[Dict[str, Any]]: + resources = template_dict.get("Resources", {}).values() + for resource in resources: + for event in resource.get("Properties", {}).get("Events", {}).values(): + yield event + + +def _event_using_sns_filter_policy_scope(event: Dict[str, Any]) -> bool: + return event["Type"] == "SNS" and "FilterPolicyScope" in event.get("Properties", {}) + + SERVICE_DETECTORS: Dict[str, Callable[[Dict[str, Any], Set[str]], bool]] = { HTTP_API: lambda template_dict, cfn_resource_types: "AWS::ApiGatewayV2::Api" in cfn_resource_types, REST_API: lambda template_dict, cfn_resource_types: "AWS::ApiGateway::RestApi" in cfn_resource_types, @@ -225,6 +237,9 @@ def _resource_using_s3_events(resource: Dict[str, Any]) -> bool: ), LOCATION: lambda template_dict, cfn_resource_types: "AWS::Location::PlaceIndex" in cfn_resource_types, APP_SYNC: lambda template_dict, cfn_resource_types: "AWS::AppSync::GraphQLApi" in cfn_resource_types, + SNS_FILTER_POLICY_SCOPE: lambda template_dict, cfn_resource_types: any( + _event_using_sns_filter_policy_scope(event) for event in _get_all_event_sources(template_dict) + ), }