KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use select_aggregate_resource_config method in localstack

Best Python code snippet using localstack_python

app.py

Source:app.py

...14http = urllib3.PoolManager()15import boto316client = boto3.client('config')17#handle NextToken, or you would have incomplete results18def select_aggregate_resource_config(Expression, ConfigurationAggregatorName, NextToken=None, current=None):19    #https://www.toptal.com/python/top-10-mistakes-that-python-programmers-make20    if current is None:21        current = {'Results': []}22    #print(len(current['Results']))23    kwargs = {'NextToken': NextToken} if NextToken else {}24    try:25        page = client.select_aggregate_resource_config(26            Expression=Expression,27            ConfigurationAggregatorName=ConfigurationAggregatorName,28            **kwargs29            #Limit=230        )31        current.get('Results').extend(page.get('Results'))32    except:33        print(sys.exc_info())34        print('Error: something is wrong. Result may possibly be incomplete')35        return current36    if page.get('NextToken') is None:37        return current38    else:39        return select_aggregate_resource_config(Expression, ConfigurationAggregatorName, page.get('NextToken'), current)40def audit(context):41    violations = []42    with open(context, 'r') as json_file:43        json_text = json.dumps(json.load(json_file))44        #https://api.slack.com/messaging/webhooks45        #https://developers.mattermost.com/integrate/incoming-webhooks/46        #http://ykng0.hatenablog.com/entry/2016/12/25/22542447        #https://gist.github.com/rantav/c096294f6f35c45155b448        #https://slack.dev/python-slackclient/basic_usage.html49    contextObject = json.loads(json_text)50    configRuleName = contextObject['attachments'][0]['props']['rvt']['configservice']['rule']['configRuleName']51    print(configRuleName)52    #53    #Generate Summary54    #55    summary_template_location = str(os.environ['LAMBDA_TASK_ROOT'] + "/function/summarytemplates/" + configRuleName + '.json')56    with open(summary_template_location, 'r') as summary_template_file:57        summary_template = json.dumps(json.load(summary_template_file))58    summaryResponse = select_aggregate_resource_config(59        #https://docs.aws.amazon.com/config/latest/developerguide/example-query.html60        Expression=f'''61            SELECT62              accountId,63              COUNT(*)64            WHERE65              configuration.complianceType = 'NON_COMPLIANT'66              AND configuration.configRuleList.configRuleName = '{configRuleName}'67            GROUP BY68              accountId69        ''',70        ConfigurationAggregatorName='ConfigurationAggregator'71    )72    #for testing73    #summaryResponse['Results'] = []74    if len(summaryResponse['Results']) == 0:75        print('Excellent. Nothing is marked as noncompliant. Abort')76        return violations77    #for testing78    #summaryResponse['Results'].extend({'{"COUNT(*)":16,"accountId":"888888888888"}'})79    #summaryResponse['Results'].extend({'{"COUNT(*)":20,"accountId":"999999999999"}'})80    total = functools.reduce(lambda a,b:a+b, [ json.loads(o)['COUNT(*)'] for o in summaryResponse['Results'] ])81    if total > int(os.environ['MaxViolationDetailsSendTo']):82        toomanyMsg = '*[Warning]* Individuals to this channel have been omitted since number of noncompliant resources has exceeded threshold *' + os.environ['MaxViolationDetailsSendTo'] + '*. Go to <https://docs.aws.amazon.com/config/latest/developerguide/aggregate-data.html|Aggregated View> of the Compliance Account instead.'83    else:84        toomanyMsg = 'Check the previous messages for Individuals. They are also available on <https://docs.aws.amazon.com/config/latest/developerguide/aggregate-data.html|Aggregated View> of the Compliance Account'85    sumList = '\\\\n'.join( [ str(json.loads(o)['accountId'])+' '+str(json.loads(o)['COUNT(*)']) for o in summaryResponse['Results'] ] )86    #https://api.slack.com/reference/surfaces/formatting#line-breaks87    # prepare template88    summaryReportMapping = [89        summary_template,90        [ 'value.ViolationCountsByAccount', sumList ],91        [ 'value.Totalling', total ],92        [ 'value.toomanyMsg', toomanyMsg ],93    ]94    summaryItem = json.loads(functools.reduce(lambda a,b :re.sub(b[0], str(b[1]), a) ,summaryReportMapping))95    if total > int(os.environ['MaxViolationDetailsSendTo']):96        print('[Warning] Number of Violations has exceeded threshold. Summary only')97        violations.append(summaryItem)98        return violations99    #100    #Generate Individuals101    #102    #https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/config.html#ConfigService.Client.select_aggregate_resource_config103    #Note: you may want to limit length for the response since enormous number of violations would have returned104    #An idea to restrict the length is here. Absent from handling NextToken and provide argument "Limit=N"105    #response = client.select_aggregate_resource_config(106    response = select_aggregate_resource_config(107        #https://github.com/awslabs/aws-config-resource-schema/blob/master/config/properties/AWS.properties.json108        #https://stackoverflow.com/a/50550283109        Expression=f'''110            SELECT111              accountId,112              awsRegion,113              configuration.targetResourceId,114              configuration.targetResourceType,115              configuration.complianceType,116              configuration.configRuleList117            WHERE118              configuration.complianceType = 'NON_COMPLIANT'119              AND configuration.configRuleList.configRuleName = '{configRuleName}'120        ''',...

aws_config.py

Source:aws_config.py

...17            "aws_config.configuration_aggregator.name"18        ).format(region=config.region)19        if not configuration_aggregator_name:20            raise MissingConfigurationValue("Invalid configuration for aws_config")21        response = config_client.select_aggregate_resource_config(22            Expression=query,23            ConfigurationAggregatorName=configuration_aggregator_name,24            Limit=100,25        )26        for r in response.get("Results", []):27            resources.append(json.loads(r))28        while response.get("NextToken"):29            response = config_client.select_aggregate_resource_config(30                Expression=query,31                ConfigurationAggregatorName=configuration_aggregator_name,32                Limit=100,33                NextToken=response["NextToken"],34            )35            for r in response.get("Results", []):36                resources.append(json.loads(r))37        return resources38    else:  # Don't use Config aggregator and instead query all the regions on an account39        session = boto3.Session()40        available_regions = session.get_available_regions("config")41        excluded_regions = config.get(42            "api_protect.exclude_regions",43            ["af-south-1", "ap-east-1", "eu-south-1", "me-south-1"],...

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.