KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use list_event_source_mappings method in localstack
Best Python code snippet using localstack_python
aws_api.py
Source:aws_api.py 
1""" Methods for AWS API. """2from puresec_cli.utils import eprint3import boto34import botocore5import re6class AwsApi:7 CONFIGURATION_PROCESSORS = [8 # function(self, name, resource_id, resource_config)9 lambda self: self._process_logs_configuration,10 lambda self: self._process_vpc_configuration,11 lambda self: self._process_stream_configuration,12 ]13 def _process_logs_configuration(self, name, resource_id, resource_config):14 """15 >>> from tests.utils import normalize_dict16 >>> provider = AwsApi()17 >>> provider.default_region = 'us-east-1'18 >>> provider.default_account = '1234'19 >>> provider._function_permissions = {}20 >>> provider._process_logs_configuration('someFunction', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})21 >>> normalize_dict(provider._function_permissions)22 {'someFunction': {'arn:aws:logs:us-east-1:1234:log-group:/aws/lambda/SomeFunction:*': {'logs:CreateLogGroup', 'logs:CreateLogStream', 'logs:PutLogEvents'}}}23 """24 self._function_permissions. \25 setdefault(name, {}). \26 setdefault("arn:aws:logs:{}:{}:log-group:/aws/lambda/{}:*".format(self.default_region, self.default_account, resource_config['Properties']['FunctionName']), set()). \27 update(('logs:CreateLogGroup', 'logs:CreateLogStream', 'logs:PutLogEvents'))28 VPC_ACTIONS = (29 'ec2:DescribeNetworkInterfaces',30 'ec2:CreateNetworkInterface',31 'ec2:DeleteNetworkInterface',32 )33 def _process_vpc_configuration(self, name, resource_id, resource_config):34 """35 >>> from tests.utils import normalize_dict36 >>> provider = AwsApi()37 >>> provider._function_permissions = {}38 >>> provider._process_vpc_configuration('someFunction', 'SomeFunctionName', {'Properties': {}})39 >>> provider._function_permissions40 {}41 >>> provider._process_vpc_configuration('someFunction', 'SomeFunctionName', {'Properties': {'VpcConfig': {}}})42 >>> normalize_dict(provider._function_permissions)43 {'someFunction': {'*': {'ec2:CreateNetworkInterface', 'ec2:DeleteNetworkInterface', 'ec2:DescribeNetworkInterfaces'}}}44 """45 if 'VpcConfig' not in resource_config['Properties']:46 return47 self._function_permissions. \48 setdefault(name, {}). \49 setdefault('*', set()). \50 update(AwsApi.VPC_ACTIONS)51 # arn:aws:(kinesis):us-east-1:<account>:stream/<stream name>52 STREAM_ARN_SERVICE_PATTERN = re.compile(r"^arn:aws:([^:]*):[^:]*:[^:]*:stream/.*")53 STREAM_ACTIONS = (54 'DescribeStream',55 'GetRecords',56 'GetShardIterator',57 'ListStreams',58 )59 FUNCTION_ID_PATTERN = r"\b({}|{})\b"60 def _process_stream_configuration(self, name, resource_id, resource_config):61 """62 >>> from tests.utils import normalize_dict63 >>> from tests.mock import Mock64 >>> mock = Mock(__name__)65 >>> provider = AwsApi()66 >>> provider.default_region = 'us-east-1'67 >>> provider.default_account = '1234'68 >>> mock.mock(provider, 'get_cached_api_result', {'EventSourceMappings': []})69 >>> provider.cloudformation_template = {'Resources': {'Mapping': {'Type': 'AWS::Lambda::EventSourceMapping',70 ... 'Properties': {'FunctionName': 'SomeFunction',71 ... 'EventSourceArn': 'arn:aws:kinesis:us-east-1:1234:stream/SomeStream'}}}}72 >>> provider._function_permissions = {}73 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})74 >>> normalize_dict(provider._function_permissions)75 {'functionName': {'arn:aws:kinesis:us-east-1:1234:stream/SomeStream': {'kinesis:DescribeStream', 'kinesis:GetRecords', 'kinesis:GetShardIterator', 'kinesis:ListStreams'}}}76 >>> mock.calls_for('AwsApi.get_cached_api_result')77 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'78 >>> provider.cloudformation_template = {'Resources': {'Mapping': {'Type': 'AWS::Lambda::EventSourceMapping',79 ... 'Properties': {'FunctionName': 'arn:aws:lambda:us-east-1:1234:function:SomeFunction',80 ... 'EventSourceArn': 'arn:aws:kinesis:us-east-1:1234:stream/SomeStream'}}}}81 >>> provider._function_permissions = {}82 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})83 >>> normalize_dict(provider._function_permissions)84 {'functionName': {'arn:aws:kinesis:us-east-1:1234:stream/SomeStream': {'kinesis:DescribeStream', 'kinesis:GetRecords', 'kinesis:GetShardIterator', 'kinesis:ListStreams'}}}85 >>> mock.calls_for('AwsApi.get_cached_api_result')86 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'87 >>> provider.cloudformation_template = {'Resources': {'Mapping': {'Type': 'AWS::Lambda::EventSourceMapping',88 ... 'Properties': {'FunctionName': 'AnotherFunction',89 ... 'EventSourceArn': 'arn:aws:kinesis:us-east-1:1234:stream/SomeStream'}}}}90 >>> provider._function_permissions = {}91 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})92 >>> provider._function_permissions93 {}94 >>> mock.calls_for('AwsApi.get_cached_api_result')95 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'96 >>> provider.cloudformation_template = {'Resources': {'Mapping': {'Type': 'AWS::Lambda::EventSourceMapping',97 ... 'Properties': {'FunctionName': 'SomeFunction',98 ... 'EventSourceArn': 'arn:aws:dynamodb:us-east-1:1234:table/SomeTable'}}}}99 >>> provider._function_permissions = {}100 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})101 >>> provider._function_permissions102 {}103 >>> mock.calls_for('AwsApi.get_cached_api_result')104 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'105 >>> provider.cloudformation_template = {'Resources': {'Mapping': {'Type': 'AWS::DynamoDB::Table'}}}106 >>> provider._function_permissions = {}107 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})108 >>> provider._function_permissions109 {}110 >>> mock.calls_for('AwsApi.get_cached_api_result')111 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'112 >>> provider.cloudformation_template = None113 >>> mock.mock(provider, 'get_cached_api_result', {'EventSourceMappings': [{'EventSourceArn': 'arn:aws:kinesis:us-east-1:1234:stream/SomeStream'}]})114 >>> provider._function_permissions = {}115 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})116 >>> normalize_dict(provider._function_permissions)117 {'functionName': {'arn:aws:kinesis:us-east-1:1234:stream/SomeStream': {'kinesis:DescribeStream', 'kinesis:GetRecords', 'kinesis:GetShardIterator', 'kinesis:ListStreams'}}}118 >>> mock.calls_for('AwsApi.get_cached_api_result')119 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'120 >>> mock.mock(provider, 'get_cached_api_result', {'EventSourceMappings': [{'EventSourceArn': 'arn:aws:dynamodb:us-east-1:1234:table/SomeTable'}]})121 >>> provider._function_permissions = {}122 >>> provider._process_stream_configuration('functionName', 'SomeFunctionName', {'Properties': {'FunctionName': 'SomeFunction'}})123 >>> provider._function_permissions124 {}125 >>> mock.calls_for('AwsApi.get_cached_api_result')126 'lambda', account='1234', api_kwargs={'FunctionName': 'SomeFunction'}, api_method='list_event_source_mappings', region='us-east-1'127 """128 function_name = resource_config['Properties']['FunctionName']129 # From CloudFormation130 if self.cloudformation_template:131 function_id_pattern = re.compile(AwsApi.FUNCTION_ID_PATTERN.format(function_name, resource_id), re.IGNORECASE)132 for other_resource_id, other_resource_config in self.cloudformation_template.get('Resources', {}).items():133 if other_resource_config.get('Type') == 'AWS::Lambda::EventSourceMapping':134 # either ARN, function name, or broken intrinsic function135 target = other_resource_config.get('Properties', {}).get('FunctionName')136 if target and function_id_pattern.search(target):137 arn = other_resource_config['Properties'].get('EventSourceArn')138 if not arn:139 eprint("warn: event source mapping for `{}` missing `EventSourceArn`", name)140 continue141 service_match = AwsApi.STREAM_ARN_SERVICE_PATTERN.match(arn)142 if not service_match:143 continue144 service = service_match.group(1)145 self._function_permissions. \146 setdefault(name, {}). \147 setdefault(arn, set()). \148 update("{}:{}".format(service, action) for action in AwsApi.STREAM_ACTIONS)149 # From production environment150 event_source_mappings = self.get_cached_api_result('lambda', region=self.default_region, account=self.default_account, api_method='list_event_source_mappings', api_kwargs={'FunctionName': function_name})151 for event_source_mapping in event_source_mappings['EventSourceMappings']:152 service_match = AwsApi.STREAM_ARN_SERVICE_PATTERN.match(event_source_mapping['EventSourceArn'])153 if not service_match:154 continue155 service = service_match.group(1)156 self._function_permissions. \157 setdefault(name, {}). \158 setdefault(event_source_mapping['EventSourceArn'], set()). \159 update("{}:{}".format(service, action) for action in AwsApi.STREAM_ACTIONS)160 # Utilities161 # { (client, api_method, api_kwargs): { resource: resource_pattern } } }162 RESOURCE_CACHE = {}163 def get_cached_api_result(self, service, region, account, api_method, api_kwargs={}):164 client = self.get_client(service, region, account)165 if client is None:166 eprint("error: cannot create {} client for region: '{}', account: '{}'", service, region, account)167 return168 cache_key = (client, api_method, frozenset(api_kwargs.items()))169 result = AwsApi.RESOURCE_CACHE.get(cache_key)170 if result is None:171 try:172 result = AwsApi.RESOURCE_CACHE[cache_key] = getattr(client, api_method)(**api_kwargs)173 except (botocore.exceptions.BotoCoreError, botocore.exceptions.ClientError) as e:174 eprint("error: failed to list resources on {}:\n{}", service, e)175 raise SystemExit(-1)176 return result177 # { (service, region, account): client }178 CLIENTS_CACHE = {}179 def get_client(self, service, region, account):180 """181 >>> from pprint import pprint182 >>> from tests.mock import Mock183 >>> mock = Mock(__name__)184 >>> class Session:185 ... def client(self, *args, **kwargs):186 ... return (args, kwargs)187 >>> provider = AwsApi()188 >>> provider.config = {}189 >>> provider.session = Session()190 >>> provider.default_account = 'default_account'191 >>> class Args:192 ... pass193 >>> args = Args()194 >>> args.no_input = False195 >>> provider.args = args196 >>> pprint(provider.get_client('dynamodb', 'us-east-1', 'default_account'))197 (('dynamodb',), {'region_name': 'us-east-1'})198 >>> mock.mock(None, 'eprint')199 >>> pprint(provider.get_client('dynamodb', 'us-east-1', '*'))200 (('dynamodb',), {'region_name': 'us-east-1'})201 >>> mock.calls_for('eprint')202 "warn: unknown account ('*'), using default session"203 >>> class Session:204 ... def client(self, *args, **kwargs):205 ... return (args, kwargs)206 >>> mock.mock(boto3, 'Session', Session())207 >>> mock.mock(None, 'input', lambda message: 'dummy')208 >>> pprint(provider.get_client('dynamodb', 'us-east-1', 'another_account'))209 (('dynamodb',), {'region_name': 'us-east-1'})210 >>> mock.calls_for('boto3.Session')211 profile_name='dummy'212 >>> pprint(provider.config)213 {'aws': {'accounts': {'another_account': {'profile': 'dummy'}}}}214 """215 client = AwsApi.CLIENTS_CACHE.get((service, region, account))216 if client:217 return client # from cache218 if region == '*':219 eprint("warn: unknown region ('*'), using the default ('{}')", self.default_region)220 region = self.default_region221 if account == '*':222 eprint("warn: unknown account ('*'), using default session")223 client = self.session.client(224 service,225 region_name=region226 )227 elif account == self.default_account:228 client = self.session.client(229 service,230 region_name=region231 )232 elif self.args.no_input:233 eprint("warn: unknown account ('{}') and --no-input set, using default session", account)234 client = self.session.client(235 service,236 region_name=region237 )238 else:239 account_config = self.config.setdefault('aws', {}).setdefault('accounts', {}).setdefault(account, {})240 if not 'profile' in account_config:241 account_config['profile'] = input("Enter configured AWS profile for {}: ".format(account))242 client = boto3.Session(profile_name=account_config['profile']).client(service, region_name=region)243 AwsApi.CLIENTS_CACHE[(service, region, account)] = client...
lambda_function.py
Source:lambda_function.py
...40 except Exception as e:41 print(e)42 raise e43 print("Checking If Event Source Mapping Already Exists For SQS Queue Events -> Lambda")44 event_sources = lambda_client.list_event_source_mappings(45 EventSourceArn=sqsQueueArn,46 FunctionName=lambdaArn47 )48 if len(event_sources["EventSourceMappings"]) > 0:49 print("Event Source Mapping Already Setup Between SQS Queue And Lambda")50 print(event_sources)51 else:52 print("Linking SQS Events To Lambda")53 lambda_linking_response = lambda_client.create_event_source_mapping(54 EventSourceArn=sqsQueueArn,55 FunctionName=lambdaArn,56 )57 print("Linking SQS Events To Lambda Complete")58 print(lambda_linking_response)59 return { "Status": "SUCCESS" }60def on_update(event):61 physical_id = event["PhysicalResourceId"]62 props = event["ResourceProperties"]63 oldProps = event["OldResourceProperties"]64 print("update resource %s with props %s" % (physical_id, props))65 66 bucketArn = props["bucketArn"]67 bucketName = props["bucketName"]68 sqsQueueArn = props["sqsQueueArn"]69 lambdaArn = props["lambdaArn"]70 oldBucketArn = oldProps["bucketArn"]71 oldBucketName = oldProps["bucketName"]72 oldSqsQueueArn = oldProps["sqsQueueArn"]73 oldLambdaArn = oldProps["lambdaArn"]74 s3_client = boto3.client('s3')75 lambda_client = boto3.client('lambda')76 print("Updating Linking Bucket Events To SQS")77 try:78 bucket_linking_response = s3_client.put_bucket_notification_configuration(79 Bucket=bucketName,80 NotificationConfiguration={81 "QueueConfigurations": [82 {83 'QueueArn': sqsQueueArn,84 'Events': [85 's3:ObjectCreated:*',86 ]87 }88 ]89 }90 )91 print("Updating Linking Bucket Events To SQS Complete")92 print(bucket_linking_response)93 except botocore.exceptions.ClientError as error:94 print(error)95 print(error.response)96 raise error97 except Exception as e:98 print(e)99 raise e100 print("Deleting Old Linking SQS Events To Lambda")101 event_sources = lambda_client.list_event_source_mappings(102 EventSourceArn=oldSqsQueueArn,103 FunctionName=oldLambdaArn104 )105 for event_source in event_sources['EventSourceMappings']:106 lambda_client.delete_event_source_mapping(107 UUID=event_source["UUID"]108 )109 print("Linking SQS Events To Lambda")110 lambda_linking_response = lambda_client.create_event_source_mapping(111 EventSourceArn=sqsQueueArn,112 FunctionName=lambdaArn,113 )114 print("Linking SQS Events To Lambda Complete")115 print(lambda_linking_response)116 return { "Status": "SUCCESS" }117def on_delete(event):118 physical_id = event["PhysicalResourceId"]119 props = event["ResourceProperties"]120 print("delete resource %s" % physical_id)121 122 bucketArn = props["bucketArn"]123 bucketName = props["bucketName"]124 sqsQueueArn = props["sqsQueueArn"]125 lambdaArn = props["lambdaArn"]126 s3_client = boto3.client('s3')127 lambda_client = boto3.client('lambda')128 print("Unlinking Bucket Events To SQS")129 bucket_linking_response = s3_client.put_bucket_notification_configuration(130 Bucket=bucketName,131 NotificationConfiguration={132 "QueueConfigurations": []133 }134 )135 print("Unlinking Bucket Events To SQS Complete")136 print(bucket_linking_response)137 print("Deleting Old Linking SQS Events To Lambda")138 event_sources = lambda_client.list_event_source_mappings(139 EventSourceArn=sqsQueueArn,140 FunctionName=lambdaArn141 )142 for event_source in event_sources['EventSourceMappings']:143 lambda_client.delete_event_source_mapping(144 UUID=event_source["UUID"]145 )...
Automation Testing Tutorials
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
