Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use describe_subscription_filters method in localstack

Best Python code snippet using localstack_python

main.py

Source:main.py

...52    messages = []53    bar = Bar("Processing", max=len(log_groups))54    for log_group_name in log_groups:55        try:56            response = log_client.describe_subscription_filters(57                logGroupName=log_group_name,58                limit=5,59            )60            if len(response["subscriptionFilters"]) > 0:61                for subscription_filter in response["subscriptionFilters"]:62                    if subscription_filter["destinationArn"].endswith(63                        ":function:log_handler"64                    ):65                        logger.debug(66                            f"Log Group Name: {subscription_filter['logGroupName']}, Lambda: {subscription_filter['destinationArn']}"67                        )68                        messages.append(69                            f"Log Group Name: {subscription_filter['logGroupName']}, Lambda: {subscription_filter['destinationArn']}"70                        )71            bar.next()72        except log_client.exceptions.LimitExceededException:73            logger.warning("Limit exceeded exception, ignoring.")74            pass75        except log_client.exceptions.InvalidParameterException:76            logger.warning("Invalid parameter exception, ignoring.")77            pass78    bar.finish()79    print(messages)80    return81def delete_duplicate_filters():82    (83        log_handler_arn,84        aws_lambda_log_handler_arn,85        log_groups,86    ) = get_lambda_and_log_groups_details()87    logger.debug(f"Found {len(log_groups)} log groups")88    messages = []89    bar = Bar("Processing", max=len(log_groups))90    dry_run = strtobool(os.getenv("DRY_RUN", "true"))91    logger.info(f"Running DRY_RUN mode = {str(bool(dry_run))}")92    for log_group_name in log_groups:93        if log_group_name in excluded_log_group_names:94            continue95        function_name = get_function_name(log_group_name)96        subscription_filter_name = f"{function_name}-log-handler-lambda-subscription"97        try:98            response = log_client.describe_subscription_filters(99                logGroupName=log_group_name,100                limit=5,101            )102            filters = response["subscriptionFilters"]103            if (104                len(filters) == 2105                and filters[0]["destinationArn"] == filters[1]["destinationArn"]106            ):107                logger.debug(filters)108                if not dry_run:109                    messages.append(110                        f"Deleting the following filter: {subscription_filter_name} from {log_group_name}"111                    )112                    response = log_client.delete_subscription_filter(113                        logGroupName=log_group_name,114                        filterName=subscription_filter_name,115                    )116                    logger.debug(response)117                else:118                    messages.append(119                        f"We would be deleting the following filter: {subscription_filter_name} from {log_group_name}"120                    )121            bar.next()122        except log_client.exceptions.LimitExceededException:123            logger.warning("Limit exceeded exception, ignoring.")124            pass125        except log_client.exceptions.InvalidParameterException:126            logger.warning("Invalid parameter exception, ignoring.")127            pass128    bar.finish()129    print(messages)130    return131def get_formatted_subscription_filters():132    (133        log_handler_arn,134        aws_lambda_log_handler_arn,135        log_groups,136    ) = get_lambda_and_log_groups_details()137    logger.debug(f"Found {len(log_groups)} log groups")138    subscription_filter_map = {}139    bar = Bar("Processing", max=len(log_groups))140    for log_group_name in log_groups:141        if log_group_name in excluded_log_group_names:142            continue143        function_name = get_function_name(log_group_name)144        subscription_filter_name = f"{function_name}-log-handler-lambda-subscription"145        try:146            response = log_client.describe_subscription_filters(147                logGroupName=log_group_name,148                filterNamePrefix=subscription_filter_name,149                limit=5,150            )151            if len(response["subscriptionFilters"]) > 0:152                for subscription_filter in response["subscriptionFilters"]:153                    if subscription_filter["destinationArn"].endswith(154                        ":function:log_handler"155                    ):156                        subscription_filter_map[157                            log_group_name158                        ] = subscription_filter_name159            bar.next()160        except log_client.exceptions.LimitExceededException:...

log_group.py

Source:log_group.py

...49        next_token = ''50        filters = []51        while True:52            if next_token:53                result = client.describe_subscription_filters(54                    logGroupName=log_group_name,55                    nextToken=next_token56                )57            else:58                result = client.describe_subscription_filters(59                    logGroupName=log_group_name60                )61            filters.extend(result['subscriptionFilters'])62            if 'nextToken' in result:63                next_token = result['nextToken']64            else:65                break66        for item in filters:67            if item['filterName'] == subscription_filter_name:68                add_filter = False69        # Add new subscription filter if the parameter is set70        if subscription_destination_arn and add_filter:71            try:72                client.put_subscription_filter(73                    logGroupName=log_group_name,74                    filterName=subscription_filter_name,75                    filterPattern=subscription_filter_pattern,76                    destinationArn=subscription_destination_arn77                )78            except Exception as e:79                error_msg = 'Error: {}'.format(e)80                logger.error(error_msg)81        log_groups_result = client.describe_log_groups(82            logGroupNamePrefix=log_group_name,83            limit=184        )85        return {86            'Status': 'SUCCESS',87            'Reason': 'Log group created',88            'PhysicalResourceId': context.log_stream_name,89            'StackId': event['StackId'],90            'RequestId': event['RequestId'],91            'LogicalResourceId': event['LogicalResourceId'],92            'Data': {93                'Value': log_group_name,94                'Arn': log_groups_result.get('logGroups')[0].get('arn')95            }96        }97    except Exception as e:98        error_msg = 'Error: {}'.format(e)99        logger.error(error_msg)100        return {101            'Status': 'FAILED',102            'Reason': error_msg,103            'PhysicalResourceId': context.log_stream_name,104            'StackId': event['StackId'],105            'RequestId': event['RequestId'],106            'LogicalResourceId': event['LogicalResourceId']107        }108@handler.delete109def lambda_handler(event, context):110    properties = event['ResourceProperties']111    log_group_name = properties['LogGroupName']112    subscription_filter_name = properties.get('SubscriptionFilterName')113    return {114        'Status': 'SUCCESS',115        'Reason': 'Log subscription deleted',116        'PhysicalResourceId': context.log_stream_name,117        'StackId': event['StackId'],118        'RequestId': event['RequestId'],119        'LogicalResourceId': event['LogicalResourceId'],120        'Data': {121            'Value': log_group_name122        }123    }124    # try:125    #     client = boto3.client('logs')126    #     # Clear all existing subscription filters127    #     next_token = ''128    #     filters = []129    #     while True:130    #         if next_token:131    #             result = client.describe_subscription_filters(132    #                 logGroupName=log_group_name,133    #                 nextToken=next_token134    #             )135    #         else:136    #             result = client.describe_subscription_filters(137    #                 logGroupName=log_group_name138    #             )139    #         filters.extend(result['subscriptionFilters'])140    #         if 'nextToken' in result:141    #             next_token = result['nextToken']142    #         else:143    #             break144    #     for item in filters:145    #         if item['filterName'] == subscription_filter_name:146    #             client.delete_subscription_filter(147    #                 logGroupName=log_group_name,148    #                 filterName=item['filterName']149    #             )150    #     return {...

test_update.py

Source:test_update.py

1import pytest2import test_fixtures3from test_fixtures import context4from test_fixtures import log_groups5from test_fixtures import update_event6def test_update_no_existing(log_groups, update_event, context):7  return_value = log_groups.client.describe_log_groups.return_value8  log_groups.client.describe_log_groups.side_effect = [{'logGroups': []},test_fixtures.DESCRIBE_LOG_GROUPS]9  response = log_groups.handle(update_event,context)10  assert log_groups.client.create_log_group.called11  log_groups.client.put_retention_policy.assert_called_with(logGroupName=test_fixtures.LOG_GROUP_NAME, retentionInDays=30)12  assert log_groups.client.describe_subscription_filters.called13  assert not log_groups.client.delete_subscription_filter.called14  assert response == { 15    'PhysicalResourceId': test_fixtures.LOG_GROUP_NAME,16    'Data': { 'Arn': test_fixtures.FUNCTION_ARN }17  }18def test_update_retention(log_groups, update_event, context):19  update_event['ResourceProperties']['Retention'] = '7'20  response = log_groups.handle(update_event,context)21  assert not log_groups.client.create_log_group.called22  log_groups.client.put_retention_policy.assert_called_with(logGroupName=test_fixtures.LOG_GROUP_NAME, retentionInDays=7)23  assert log_groups.client.describe_subscription_filters.called24  assert not log_groups.client.delete_subscription_filter.called25  assert not log_groups.client.put_subscription_filter.called26  assert response == { 27    'PhysicalResourceId': test_fixtures.LOG_GROUP_NAME,28    'Data': { 'Arn': test_fixtures.FUNCTION_ARN }29  }30def test_update_new_subscription(log_groups, update_event, context):31  update_event['ResourceProperties']['Subscription'] = {32    'FilterName': 'Default',33    'FilterPattern': '',34    'DestinationArn': test_fixtures.SUBSCRIPTION_ARN35  }36  response = log_groups.handle(update_event,context)37  assert not log_groups.client.create_log_group.called38  log_groups.client.put_retention_policy.assert_called_with(logGroupName=test_fixtures.LOG_GROUP_NAME, retentionInDays=30)39  assert log_groups.client.describe_subscription_filters.called40  assert not log_groups.client.delete_subscription_filter.called41  assert log_groups.client.put_subscription_filter.called42  assert response == { 43    'PhysicalResourceId': test_fixtures.LOG_GROUP_NAME,44    'Data': { 'Arn': test_fixtures.FUNCTION_ARN }45  }46def test_update_existing_subscription(log_groups, update_event, context):47  update_event['ResourceProperties']['Subscription'] = {48    'FilterName': 'Default',49    'FilterPattern': '[filter]',50    'DestinationArn': test_fixtures.SUBSCRIPTION_ARN51  }52  log_groups.client.describe_subscription_filters.return_value = test_fixtures.DESCRIBE_SUBSCRIPTION_FILTERS53  response = log_groups.handle(update_event,context)54  assert not log_groups.client.create_log_group.called55  log_groups.client.put_retention_policy.assert_called_with(logGroupName=test_fixtures.LOG_GROUP_NAME, retentionInDays=30)56  assert log_groups.client.describe_subscription_filters.called57  assert log_groups.client.delete_subscription_filter.called58  assert log_groups.client.put_subscription_filter.called59  assert response == { 60    'PhysicalResourceId': test_fixtures.LOG_GROUP_NAME,61    'Data': { 'Arn': test_fixtures.FUNCTION_ARN }62  }63def test_update_existing_subscription_no_changes(log_groups, update_event, context):64  update_event['ResourceProperties']['Subscription'] = {65    'FilterName': 'Default',66    'FilterPattern': '',67    'DestinationArn': test_fixtures.SUBSCRIPTION_ARN68  }69  log_groups.client.describe_subscription_filters.return_value = test_fixtures.DESCRIBE_SUBSCRIPTION_FILTERS70  response = log_groups.handle(update_event,context)71  assert not log_groups.client.create_log_group.called72  log_groups.client.put_retention_policy.assert_called_with(logGroupName=test_fixtures.LOG_GROUP_NAME, retentionInDays=30)73  assert log_groups.client.describe_subscription_filters.called74  assert not log_groups.client.delete_subscription_filter.called75  assert not log_groups.client.put_subscription_filter.called76  assert response == { 77    'PhysicalResourceId': test_fixtures.LOG_GROUP_NAME,78    'Data': { 'Arn': test_fixtures.FUNCTION_ARN }...

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.