How to use describe_configuration_aggregators method in localstack

Best Python code snippet using localstack_python

test_config.py

Source:test_config.py

...373        ],374        'AllAwsRegions': False375    }376@mock_config377def test_describe_configuration_aggregators():378    client = boto3.client('config', region_name='us-west-2')379    # Without any config aggregators:380    assert not client.describe_configuration_aggregators()['ConfigurationAggregators']381    # Make 10 config aggregators:382    for x in range(0, 10):383        client.put_configuration_aggregator(384            ConfigurationAggregatorName='testing{}'.format(x),385            AccountAggregationSources=[386                {387                    'AccountIds': [388                        '012345678910',389                    ],390                    'AllAwsRegions': True391                }392            ]393        )394    # Describe with an incorrect name:395    with assert_raises(ClientError) as ce:396        client.describe_configuration_aggregators(ConfigurationAggregatorNames=['DoesNotExist'])397    assert 'The configuration aggregator does not exist.' in ce.exception.response['Error']['Message']398    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationAggregatorException'399    # Error describe with more than 1 item in the list:400    with assert_raises(ClientError) as ce:401        client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing0', 'DoesNotExist'])402    assert 'At least one of the configuration aggregators does not exist.' in ce.exception.response['Error']['Message']403    assert ce.exception.response['Error']['Code'] == 'NoSuchConfigurationAggregatorException'404    # Get the normal list:405    result = client.describe_configuration_aggregators()406    assert not result.get('NextToken')407    assert len(result['ConfigurationAggregators']) == 10408    # Test filtered list:409    agg_names = ['testing0', 'testing1', 'testing2']410    result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=agg_names)411    assert not result.get('NextToken')412    assert len(result['ConfigurationAggregators']) == 3413    assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == agg_names414    # Test Pagination:415    result = client.describe_configuration_aggregators(Limit=4)416    assert len(result['ConfigurationAggregators']) == 4417    assert result['NextToken'] == 'testing4'418    assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \419           ['testing{}'.format(x) for x in range(0, 4)]420    result = client.describe_configuration_aggregators(Limit=4, NextToken='testing4')421    assert len(result['ConfigurationAggregators']) == 4422    assert result['NextToken'] == 'testing8'423    assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \424           ['testing{}'.format(x) for x in range(4, 8)]425    result = client.describe_configuration_aggregators(Limit=4, NextToken='testing8')426    assert len(result['ConfigurationAggregators']) == 2427    assert not result.get('NextToken')428    assert [agg['ConfigurationAggregatorName'] for agg in result['ConfigurationAggregators']] == \429           ['testing{}'.format(x) for x in range(8, 10)]430    # Test Pagination with Filtering:431    result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing2', 'testing4'], Limit=1)432    assert len(result['ConfigurationAggregators']) == 1433    assert result['NextToken'] == 'testing4'434    assert result['ConfigurationAggregators'][0]['ConfigurationAggregatorName'] == 'testing2'435    result = client.describe_configuration_aggregators(ConfigurationAggregatorNames=['testing2', 'testing4'], Limit=1, NextToken='testing4')436    assert not result.get('NextToken')437    assert result['ConfigurationAggregators'][0]['ConfigurationAggregatorName'] == 'testing4'438    # Test with an invalid filter:439    with assert_raises(ClientError) as ce:440        client.describe_configuration_aggregators(NextToken='WRONG')441    assert 'The nextToken provided is invalid' == ce.exception.response['Error']['Message']442    assert ce.exception.response['Error']['Code'] == 'InvalidNextTokenException'443@mock_config444def test_put_aggregation_authorization():445    client = boto3.client('config', region_name='us-west-2')446    # Too many tags (>50):447    with assert_raises(ClientError) as ce:448        client.put_aggregation_authorization(449            AuthorizedAccountId='012345678910',450            AuthorizedAwsRegion='us-west-2',451            Tags=[{'Key': '{}'.format(x), 'Value': '{}'.format(x)} for x in range(0, 51)]452        )453    assert 'Member must have length less than or equal to 50' in ce.exception.response['Error']['Message']454    assert ce.exception.response['Error']['Code'] == 'ValidationException'...

aws_config_aggregator.py

Source:aws_config_aggregator.py

...76from ansible.module_utils.aws.core import AnsibleAWSModule, is_boto3_error_code77from ansible.module_utils.ec2 import AWSRetry, camel_dict_to_snake_dict78def resource_exists(client, module, params):79    try:80        aggregator = client.describe_configuration_aggregators(81            ConfigurationAggregatorNames=[params['name']]82        )83        return aggregator['ConfigurationAggregators'][0]84    except is_boto3_error_code('NoSuchConfigurationAggregatorException'):85        return86    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:  # pylint: disable=duplicate-except87        module.fail_json_aws(e)88def create_resource(client, module, params, result):89    try:90        client.put_configuration_aggregator(91            ConfigurationAggregatorName=params['ConfigurationAggregatorName'],92            AccountAggregationSources=params['AccountAggregationSources'],93            OrganizationAggregationSource=params['OrganizationAggregationSource']94        )95        result['changed'] = True96        result['aggregator'] = camel_dict_to_snake_dict(resource_exists(client, module, params))97        return result98    except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:99        module.fail_json_aws(e, msg="Couldn't create AWS Config configuration aggregator")100def update_resource(client, module, resource_type, params, result):101    current_params = client.describe_configuration_aggregators(102        ConfigurationAggregatorNames=[params['name']]103    )104    del current_params['ConfigurationAggregatorArn']105    del current_params['CreationTime']106    del current_params['LastUpdatedTime']107    if params != current_params['ConfigurationAggregators'][0]:108        try:109            client.put_configuration_aggregator(110                ConfigurationAggregatorName=params['ConfigurationAggregatorName'],111                AccountAggregationSources=params['AccountAggregationSources'],112                OrganizationAggregationSource=params['OrganizationAggregationSource']113            )114            result['changed'] = True115            result['aggregator'] = camel_dict_to_snake_dict(resource_exists(client, module, params))...

responses.py

Source:responses.py

...11    def put_configuration_aggregator(self):12        aggregator = self.config_backend.put_configuration_aggregator(json.loads(self.body), self.region)13        schema = {'ConfigurationAggregator': aggregator}14        return json.dumps(schema)15    def describe_configuration_aggregators(self):16        aggregators = self.config_backend.describe_configuration_aggregators(self._get_param('ConfigurationAggregatorNames'),17                                                                             self._get_param('NextToken'),18                                                                             self._get_param('Limit'))19        return json.dumps(aggregators)20    def delete_configuration_aggregator(self):21        self.config_backend.delete_configuration_aggregator(self._get_param('ConfigurationAggregatorName'))22        return ""23    def put_aggregation_authorization(self):24        agg_auth = self.config_backend.put_aggregation_authorization(self.region,25                                                                     self._get_param('AuthorizedAccountId'),26                                                                     self._get_param('AuthorizedAwsRegion'),27                                                                     self._get_param('Tags'))28        schema = {'AggregationAuthorization': agg_auth}29        return json.dumps(schema)30    def describe_aggregation_authorizations(self):...

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.