How to use get_compliance_details_by_config_rule method in localstack

Best Python code snippet using localstack_python

configservice.py

Source:configservice.py

1""" Utilities for writing integration tests around AWS Config service """2import time3import json4MAX_ATTEMPTS = 455WAIT_PERIOD = 206def all_rule_results(configservice, rule_name):7    """ Return details for the given config rule, and deal with slurping all the results8    :param configservice: boto client for AWS Config9    :param rule_name: name of rule to get compliance details for10    :returns: slurped version of get_compliance_details_by_config_rule response """11    paginator = configservice.get_paginator('get_compliance_details_by_config_rule')12    page_iterator = paginator.paginate(13        ConfigRuleName=rule_name,14        ComplianceTypes=[15            'NON_COMPLIANT',16            'COMPLIANT',17            'NOT_APPLICABLE'18        ]19    )20    return [21        evaluation_result22        for page in page_iterator23        for evaluation_result in page['EvaluationResults']24    ]25def _remove_missing_resource_ids(config_records, resource_ids):26    """27    Remove resource_ids found in config_results and return any remaining resource_ids28    29    :param config_records: config compliance records30    :param resource_ids: list of resource ids31    :returns: list of resource IDs found in compliance records32    """33    resources_in_config = []34    for config_record in config_records:35        config_record_id = config_record['EvaluationResultIdentifier'][36            'EvaluationResultQualifier']['ResourceId']37        if config_record_id in resource_ids:38            resources_in_config.append(config_record_id)39    return resources_in_config40def config_rule_wait_for_absent_resources(configservice, rule_name, resource_ids,41                                          wait_period=WAIT_PERIOD, max_attempts=MAX_ATTEMPTS, evaluate=False):42    """43    Wait for resource_ids to be removed from AWS Config results.44    Default timeout is 15 minutes45    :param configservice: boto client for interfacing with AWS Config service46    :param rule_name: config rule to evaluate47    :param wait_period: period to wait between checks48    :return: empty list if all resource_ids are absent. If timeout, return list of remaining ids.49    :param wait_period: length of wait period (optional)50    :param max_attempts: number of attempts before timeout (optional)51    :param evaluate: If True, initiate a config rule evaluation. Use for periodic rules. (optional)52    """53    if evaluate:54        _start_evaluations(configservice, rule_name)55    for _ in range(max_attempts):56        config_records = all_rule_results(configservice, rule_name)57        remaining_ids = _remove_missing_resource_ids(config_records, resource_ids)58        if not remaining_ids:59            return []60        time.sleep(wait_period)61    print(f'TIMEOUT waiting for these resources to disappear: {remaining_ids}')62    return remaining_ids63def _present_config_results(config_records, resource_ids):64    """ 65    If resource_id is in config_records add to dictionary and return dictionary66    67    :param config_records: config compliance records68    :param resource_ids: list of resource ids69    :returns: dictionary of resource_id: compliance_type70    """71    found_records = {}72    for config_records in config_records:73        config_record_id = config_records['EvaluationResultIdentifier'][74            'EvaluationResultQualifier']['ResourceId']75        if config_record_id in resource_ids:76            found_records[config_record_id] = config_records["ComplianceType"]77    return found_records78def config_rule_wait_for_compliance_results(configservice, rule_name, expected_results,79                                            wait_period=WAIT_PERIOD, max_attempts=MAX_ATTEMPTS,80                                            evaluate=False):81    """ 82    Wait for resources to show up in config results and validate that the results are what are expected.83    Splits the expected_results in to those that should be present (COMPLIANT/NON_COMPLIANT) and those that should be 84    absent (NOT_APPLICABLE).  Wait for the IDs of present expected results to show up in the rule's compliance details. 85    Then compare complinace details to expected results. Present results must have the same evaluation and absent results 86    must not be present.87    Default timeout is 15 minutes88    :param configservice: boto client for interfacing with AWS Config service89    :param rule_name: config rule to evaluate90    :param expected_results: dictionary of expected results in format resource_id: COMPLIANT|NON_COMPLIANT|NOT_APPLICABLE91    :return: test results compared to actual results. If timeout results are partial.92    :param wait_period: length of wait period (optional)93    :param max_attempts: number of attempts before timeout (optional)94    :param evaluate: If True, initiate a config rule evaluation. Use for periodic rules. (optional)95    """96    if evaluate:97        _start_evaluations(configservice, rule_name)98    expected_absent_ids = []99    expected_present_ids = []100    expected_present_results = {}101    for resource_id, compliance in expected_results.items():102        if compliance == "NOT_APPLICABLE":103            expected_absent_ids.append(resource_id)104        else:105            expected_present_results[resource_id] = compliance106            expected_present_ids.append(resource_id)107    expected_present_count = len(expected_present_ids)108    for _ in range(max_attempts):109        config_records = all_rule_results(configservice, rule_name)110        actual_present_results = _present_config_results(config_records, expected_present_ids)111        actual_absent_results = _present_config_results(config_records, expected_absent_ids)112        if len(actual_present_results) == expected_present_count:113            break114        time.sleep(wait_period)115    print(f'absent resources = {expected_absent_ids}')116    print(f'absent actual_results = {actual_absent_results}')117    print(f'present actual_results = {json.dumps(actual_present_results, indent=4)}')118    print(f'present expected_results = {json.dumps(expected_present_results, indent=4)}')119    return actual_present_results == expected_present_results and actual_absent_results == {}120def config_rule_wait_for_resource(configservice, resource_id, rule_name):121    """ wait for a resource_id to show up in config rule results.122    It's up to you to ensure that the rule and resource are relevant to each other... if not123    this thing will loop for a godawful long time.124    For example - eip-attached rule and eipalloc-xxxx go together.125    The point here is that the eval times for the rules ARE NOT RELIABLE for determining126    whether the rule has inspected a recent resource you created.  Therefore, this "wait" approach127    is to poll until the given resource shows up somewhere in the details - be it compliant or not.128    An irrelevant resource will never show so.... you'll be waiting... and finally get a None result129    :param configservice: boto client for interfacing with AWS Config service130    :param resource_id: resource id to wait for in the details of the call to get_compliance_details_by_config_rule131    :param rule_name: config rule to evaluate132    :return: None if resource never shows up, otherwise the EvaluationResult from call to133             get_compliance_details_by_config_rule134    """135    attempts = 0136    while True:137        compliance_result = [138            result139            for result in all_rule_results(configservice, rule_name)140            if result['EvaluationResultIdentifier']['EvaluationResultQualifier']['ResourceId'] == resource_id141        ]142        if compliance_result:143            return compliance_result[0]144        else:145            attempts += 1146            if attempts == MAX_ATTEMPTS:147                return None148            else:149                time.sleep(WAIT_PERIOD)150def _start_evaluations(configservice, rule_name):151    """ Start configuration rule evaluations """152    try:153        _ = configservice.start_config_rules_evaluation(154            ConfigRuleNames=[155                rule_name156            ]157        )158    except configservice.exceptions.LimitExceededException:159        # if throttled, just wait anyways160        pass161def evaluate_config_rule_and_wait_for_resource(configservice, resource_id,162                                               rule_name):163    """ Kick off the specified rule and wait for the resource_id to show up in the results.164    It's up to you to ensure that the rule and resource are relevant to each other... if not165    this thing will loop for a godawful long time.166    For example - eip-attached rule and eipalloc-xxxx go together.167    The point here is that the eval times for the rules ARE NOT RELIABLE for determining168    whether the rule has inspected a recent resource you created.  Therefore, this "wait" approach169    is to poll until the given resource shows up somewhere in the details - be it compliant or not.170    An irrelevant resource will never show so.... you'll be waiting... and finally get a None result171    :param configservice: boto client for interfacing with AWS Config service172    :param resource_id: resource id to wait for in the details of the call to get_compliance_details_by_config_rule173    :param rule_name: config rule to evaluate174    :return: None if resource never shows up, otherwise the EvaluationResult from call to175             get_compliance_details_by_config_rule176    """177    _start_evaluations(configservice, rule_name)...

compliance-reporter.py

Source:compliance-reporter.py

...34    for rule in rules["ConfigRules"]:35        rule_name = rule["ConfigRuleName"]36        message += "Non-Compliance Result of [" + rule_name + "]: \n"37        #message += "\n========== NON_COMPLIANT ============\n"38        response = config.get_compliance_details_by_config_rule(ConfigRuleName=rule_name, ComplianceTypes=["NON_COMPLIANT"])39        if len(response["EvaluationResults"]) > 0:40            for result in response["EvaluationResults"]:41                message += result["EvaluationResultIdentifier"]["EvaluationResultQualifier"]["ResourceId"]42                if "Annotation" in result:43                    message += ": " + result["Annotation"] + "\n"44                else:45                    message += "\n"46        else:47            message += "Nil\n"48        message += "\n\n"49#        message += "========== NON_COMPLIANT ============\n"50#        response = config.get_compliance_details_by_config_rule(ConfigRuleName=rule_name, ComplianceTypes=["COMPLIANT"])51#        for result in response["EvaluationResults"]:52#            message += result["EvaluationResultIdentifier"]["EvaluationResultQualifier"]["ResourceId"]53#        message += "\n\n"54    print ("Message: " + message)55    subject = "[" + ACCOUNT_ID + "] compliance report"56    sns = boto3.client('sns')57    topic_arn = "arn:aws:sns:ap-southeast-1:" + ACCOUNT_ID + ":compliance-reporter-topic"58    try:59        sns.publish(TopicArn=topic_arn, Subject=subject, Message=message)60    except botocore.exceptions.ClientError as e:61        print ("Fail to send message to topic " + topic_arn)...

aws_aws_compliance_test.py

Source:aws_aws_compliance_test.py

...10# aws_secret_access_key = <access-secret>11#12import boto313client = boto3.client("config")14response = client.get_compliance_details_by_config_rule(15    ConfigRuleName="ec2-stopped-instance"16)...

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.