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

How to use put_configuration_aggregator method in localstack

Best Python code snippet using localstack_python

test_config.py

Source:test_config.py

...107        })108    assert ce.exception.response['Error']['Code'] == 'MaxNumberOfConfigurationRecordersExceededException'109    assert "maximum number of configuration recorders: 1 is reached." in ce.exception.response['Error']['Message']110@mock_config111def test_put_configuration_aggregator():112    client = boto3.client('config', region_name='us-west-2')113    # With too many aggregation sources:114    with assert_raises(ClientError) as ce:115        client.put_configuration_aggregator(116            ConfigurationAggregatorName='testing',117            AccountAggregationSources=[118                {119                    'AccountIds': [120                        '012345678910',121                        '111111111111',122                        '222222222222'123                    ],124                    'AwsRegions': [125                        'us-east-1',126                        'us-west-2'127                    ]128                },129                {130                    'AccountIds': [131                        '012345678910',132                        '111111111111',133                        '222222222222'134                    ],135                    'AwsRegions': [136                        'us-east-1',137                        'us-west-2'138                    ]139                }140            ]141        )142    assert 'Member must have length less than or equal to 1' in ce.exception.response['Error']['Message']143    assert ce.exception.response['Error']['Code'] == 'ValidationException'144    # With an invalid region config (no regions defined):145    with assert_raises(ClientError) as ce:146        client.put_configuration_aggregator(147            ConfigurationAggregatorName='testing',148            AccountAggregationSources=[149                {150                    'AccountIds': [151                        '012345678910',152                        '111111111111',153                        '222222222222'154                    ],155                    'AllAwsRegions': False156                }157            ]158        )159    assert 'Your request does not specify any regions' in ce.exception.response['Error']['Message']160    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'161    with assert_raises(ClientError) as ce:162        client.put_configuration_aggregator(163            ConfigurationAggregatorName='testing',164            OrganizationAggregationSource={165                'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole'166            }167        )168    assert 'Your request does not specify any regions' in ce.exception.response['Error']['Message']169    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'170    # With both region flags defined:171    with assert_raises(ClientError) as ce:172        client.put_configuration_aggregator(173            ConfigurationAggregatorName='testing',174            AccountAggregationSources=[175                {176                    'AccountIds': [177                        '012345678910',178                        '111111111111',179                        '222222222222'180                    ],181                    'AwsRegions': [182                        'us-east-1',183                        'us-west-2'184                    ],185                    'AllAwsRegions': True186                }187            ]188        )189    assert 'You must choose one of these options' in ce.exception.response['Error']['Message']190    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'191    with assert_raises(ClientError) as ce:192        client.put_configuration_aggregator(193            ConfigurationAggregatorName='testing',194            OrganizationAggregationSource={195                'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',196                'AwsRegions': [197                    'us-east-1',198                    'us-west-2'199                ],200                'AllAwsRegions': True201            }202        )203    assert 'You must choose one of these options' in ce.exception.response['Error']['Message']204    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'205    # Name too long:206    with assert_raises(ClientError) as ce:207        client.put_configuration_aggregator(208            ConfigurationAggregatorName='a' * 257,209            AccountAggregationSources=[210                {211                    'AccountIds': [212                        '012345678910',213                    ],214                    'AllAwsRegions': True215                }216            ]217        )218    assert 'configurationAggregatorName' in ce.exception.response['Error']['Message']219    assert ce.exception.response['Error']['Code'] == 'ValidationException'220    # Too many tags (>50):221    with assert_raises(ClientError) as ce:222        client.put_configuration_aggregator(223            ConfigurationAggregatorName='testing',224            AccountAggregationSources=[225                {226                    'AccountIds': [227                        '012345678910',228                    ],229                    'AllAwsRegions': True230                }231            ],232            Tags=[{'Key': '{}'.format(x), 'Value': '{}'.format(x)} for x in range(0, 51)]233        )234    assert 'Member must have length less than or equal to 50' in ce.exception.response['Error']['Message']235    assert ce.exception.response['Error']['Code'] == 'ValidationException'236    # Tag key is too big (>128 chars):237    with assert_raises(ClientError) as ce:238        client.put_configuration_aggregator(239            ConfigurationAggregatorName='testing',240            AccountAggregationSources=[241                {242                    'AccountIds': [243                        '012345678910',244                    ],245                    'AllAwsRegions': True246                }247            ],248            Tags=[{'Key': 'a' * 129, 'Value': 'a'}]249        )250    assert 'Member must have length less than or equal to 128' in ce.exception.response['Error']['Message']251    assert ce.exception.response['Error']['Code'] == 'ValidationException'252    # Tag value is too big (>256 chars):253    with assert_raises(ClientError) as ce:254        client.put_configuration_aggregator(255            ConfigurationAggregatorName='testing',256            AccountAggregationSources=[257                {258                    'AccountIds': [259                        '012345678910',260                    ],261                    'AllAwsRegions': True262                }263            ],264            Tags=[{'Key': 'tag', 'Value': 'a' * 257}]265        )266    assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']267    assert ce.exception.response['Error']['Code'] == 'ValidationException'268    # Duplicate Tags:269    with assert_raises(ClientError) as ce:270        client.put_configuration_aggregator(271            ConfigurationAggregatorName='testing',272            AccountAggregationSources=[273                {274                    'AccountIds': [275                        '012345678910',276                    ],277                    'AllAwsRegions': True278                }279            ],280            Tags=[{'Key': 'a', 'Value': 'a'}, {'Key': 'a', 'Value': 'a'}]281        )282    assert 'Duplicate tag keys found.' in ce.exception.response['Error']['Message']283    assert ce.exception.response['Error']['Code'] == 'InvalidInput'284    # Invalid characters in the tag key:285    with assert_raises(ClientError) as ce:286        client.put_configuration_aggregator(287            ConfigurationAggregatorName='testing',288            AccountAggregationSources=[289                {290                    'AccountIds': [291                        '012345678910',292                    ],293                    'AllAwsRegions': True294                }295            ],296            Tags=[{'Key': '!', 'Value': 'a'}]297        )298    assert 'Member must satisfy regular expression pattern:' in ce.exception.response['Error']['Message']299    assert ce.exception.response['Error']['Code'] == 'ValidationException'300    # If it contains both the AccountAggregationSources and the OrganizationAggregationSource301    with assert_raises(ClientError) as ce:302        client.put_configuration_aggregator(303            ConfigurationAggregatorName='testing',304            AccountAggregationSources=[305                {306                    'AccountIds': [307                        '012345678910',308                    ],309                    'AllAwsRegions': False310                }311            ],312            OrganizationAggregationSource={313                'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',314                'AllAwsRegions': False315            }316        )317    assert 'AccountAggregationSource and the OrganizationAggregationSource' in ce.exception.response['Error']['Message']318    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'319    # If it contains neither:320    with assert_raises(ClientError) as ce:321        client.put_configuration_aggregator(322            ConfigurationAggregatorName='testing',323        )324    assert 'AccountAggregationSource or the OrganizationAggregationSource' in ce.exception.response['Error']['Message']325    assert ce.exception.response['Error']['Code'] == 'InvalidParameterValueException'326    # Just make one:327    account_aggregation_source = {328        'AccountIds': [329            '012345678910',330            '111111111111',331            '222222222222'332        ],333        'AwsRegions': [334            'us-east-1',335            'us-west-2'336        ],337        'AllAwsRegions': False338    }339    result = client.put_configuration_aggregator(340        ConfigurationAggregatorName='testing',341        AccountAggregationSources=[account_aggregation_source],342    )343    assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testing'344    assert result['ConfigurationAggregator']['AccountAggregationSources'] == [account_aggregation_source]345    assert 'arn:aws:config:us-west-2:123456789012:config-aggregator/config-aggregator-' in \346           result['ConfigurationAggregator']['ConfigurationAggregatorArn']347    assert result['ConfigurationAggregator']['CreationTime'] == result['ConfigurationAggregator']['LastUpdatedTime']348    # Update the existing one:349    original_arn = result['ConfigurationAggregator']['ConfigurationAggregatorArn']350    account_aggregation_source.pop('AwsRegions')351    account_aggregation_source['AllAwsRegions'] = True352    result = client.put_configuration_aggregator(353        ConfigurationAggregatorName='testing',354        AccountAggregationSources=[account_aggregation_source]355    )356    assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testing'357    assert result['ConfigurationAggregator']['AccountAggregationSources'] == [account_aggregation_source]358    assert result['ConfigurationAggregator']['ConfigurationAggregatorArn'] == original_arn359    # Make an org one:360    result = client.put_configuration_aggregator(361        ConfigurationAggregatorName='testingOrg',362        OrganizationAggregationSource={363            'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',364            'AwsRegions': ['us-east-1', 'us-west-2']365        }366    )367    assert result['ConfigurationAggregator']['ConfigurationAggregatorName'] == 'testingOrg'368    assert result['ConfigurationAggregator']['OrganizationAggregationSource'] == {369        'RoleArn': 'arn:aws:iam::012345678910:role/SomeRole',370        'AwsRegions': [371            'us-east-1',372            'us-west-2'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'455    # Tag key is too big (>128 chars):456    with assert_raises(ClientError) as ce:457        client.put_aggregation_authorization(458            AuthorizedAccountId='012345678910',459            AuthorizedAwsRegion='us-west-2',460            Tags=[{'Key': 'a' * 129, 'Value': 'a'}]461        )462    assert 'Member must have length less than or equal to 128' in ce.exception.response['Error']['Message']463    assert ce.exception.response['Error']['Code'] == 'ValidationException'464    # Tag value is too big (>256 chars):465    with assert_raises(ClientError) as ce:466        client.put_aggregation_authorization(467            AuthorizedAccountId='012345678910',468            AuthorizedAwsRegion='us-west-2',469            Tags=[{'Key': 'tag', 'Value': 'a' * 257}]470        )471    assert 'Member must have length less than or equal to 256' in ce.exception.response['Error']['Message']472    assert ce.exception.response['Error']['Code'] == 'ValidationException'473    # Duplicate Tags:474    with assert_raises(ClientError) as ce:475        client.put_aggregation_authorization(476            AuthorizedAccountId='012345678910',477            AuthorizedAwsRegion='us-west-2',478            Tags=[{'Key': 'a', 'Value': 'a'}, {'Key': 'a', 'Value': 'a'}]479        )480    assert 'Duplicate tag keys found.' in ce.exception.response['Error']['Message']481    assert ce.exception.response['Error']['Code'] == 'InvalidInput'482    # Invalid characters in the tag key:483    with assert_raises(ClientError) as ce:484        client.put_aggregation_authorization(485            AuthorizedAccountId='012345678910',486            AuthorizedAwsRegion='us-west-2',487            Tags=[{'Key': '!', 'Value': 'a'}]488        )489    assert 'Member must satisfy regular expression pattern:' in ce.exception.response['Error']['Message']490    assert ce.exception.response['Error']['Code'] == 'ValidationException'491    # Put a normal one there:492    result = client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-east-1',493                                                  Tags=[{'Key': 'tag', 'Value': 'a'}])494    assert result['AggregationAuthorization']['AggregationAuthorizationArn'] == 'arn:aws:config:us-west-2:123456789012:' \495                                                                                'aggregation-authorization/012345678910/us-east-1'496    assert result['AggregationAuthorization']['AuthorizedAccountId'] == '012345678910'497    assert result['AggregationAuthorization']['AuthorizedAwsRegion'] == 'us-east-1'498    assert isinstance(result['AggregationAuthorization']['CreationTime'], datetime)499    creation_date = result['AggregationAuthorization']['CreationTime']500    # And again:501    result = client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-east-1')502    assert result['AggregationAuthorization']['AggregationAuthorizationArn'] == 'arn:aws:config:us-west-2:123456789012:' \503                                                                                'aggregation-authorization/012345678910/us-east-1'504    assert result['AggregationAuthorization']['AuthorizedAccountId'] == '012345678910'505    assert result['AggregationAuthorization']['AuthorizedAwsRegion'] == 'us-east-1'506    assert result['AggregationAuthorization']['CreationTime'] == creation_date507@mock_config508def test_describe_aggregation_authorizations():509    client = boto3.client('config', region_name='us-west-2')510    # With no aggregation authorizations:511    assert not client.describe_aggregation_authorizations()['AggregationAuthorizations']512    # Make 10 account authorizations:513    for i in range(0, 10):514        client.put_aggregation_authorization(AuthorizedAccountId='{}'.format(str(i) * 12), AuthorizedAwsRegion='us-west-2')515    result = client.describe_aggregation_authorizations()516    assert len(result['AggregationAuthorizations']) == 10517    assert not result.get('NextToken')518    for i in range(0, 10):519        assert result['AggregationAuthorizations'][i]['AuthorizedAccountId'] == str(i) * 12520    # Test Pagination:521    result = client.describe_aggregation_authorizations(Limit=4)522    assert len(result['AggregationAuthorizations']) == 4523    assert result['NextToken'] == ('4' * 12) + '/us-west-2'524    assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(0, 4)]525    result = client.describe_aggregation_authorizations(Limit=4, NextToken=('4' * 12) + '/us-west-2')526    assert len(result['AggregationAuthorizations']) == 4527    assert result['NextToken'] == ('8' * 12) + '/us-west-2'528    assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(4, 8)]529    result = client.describe_aggregation_authorizations(Limit=4, NextToken=('8' * 12) + '/us-west-2')530    assert len(result['AggregationAuthorizations']) == 2531    assert not result.get('NextToken')532    assert [auth['AuthorizedAccountId'] for auth in result['AggregationAuthorizations']] == ['{}'.format(str(x) * 12) for x in range(8, 10)]533    # Test with an invalid filter:534    with assert_raises(ClientError) as ce:535        client.describe_aggregation_authorizations(NextToken='WRONG')536    assert 'The nextToken provided is invalid' == ce.exception.response['Error']['Message']537    assert ce.exception.response['Error']['Code'] == 'InvalidNextTokenException'538@mock_config539def test_delete_aggregation_authorization():540    client = boto3.client('config', region_name='us-west-2')541    client.put_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')542    # Delete it:543    client.delete_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')544    # Verify that none are there:545    assert not client.describe_aggregation_authorizations()['AggregationAuthorizations']546    # Try it again -- nothing should happen:547    client.delete_aggregation_authorization(AuthorizedAccountId='012345678910', AuthorizedAwsRegion='us-west-2')548@mock_config549def test_delete_configuration_aggregator():550    client = boto3.client('config', region_name='us-west-2')551    client.put_configuration_aggregator(552        ConfigurationAggregatorName='testing',553        AccountAggregationSources=[554            {555                'AccountIds': [556                    '012345678910',557                ],558                'AllAwsRegions': True559            }560        ]561    )562    client.delete_configuration_aggregator(ConfigurationAggregatorName='testing')563    # And again to confirm that it's deleted:564    with assert_raises(ClientError) as ce:565        client.delete_configuration_aggregator(ConfigurationAggregatorName='testing')...

aws_config_aggregator.py

Source:aws_config_aggregator.py

...86    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))116            return result117        except (botocore.exceptions.ClientError, botocore.exceptions.BotoCoreError) as e:118            module.fail_json_aws(e, msg="Couldn't create AWS Config configuration aggregator")119def delete_resource(client, module, resource_type, params, result):120    try:121        client.delete_configuration_aggregator(122            ConfigurationAggregatorName=params['ConfigurationAggregatorName']123        )...

responses.py

Source:responses.py

...7        return config_backends[self.region]8    def put_configuration_recorder(self):9        self.config_backend.put_configuration_recorder(self._get_param('ConfigurationRecorder'))10        return ""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'),...

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.