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):...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!