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.