KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
How to use describe_aggregate_id_format method in localstack
Best Python code snippet using localstack_python
fetch.py
Source:fetch.py 
1import logging2import os3from pathlib import Path4import time5from typing import Any, Callable, Dict, Iterator, List, Optional, Tuple6import yaml7import botocore.exceptions8from botocore.session import Session as Boto9from introspector.error import GFInternal10from introspector.models.raw_import import ERROR_KEY11_log = logging.getLogger(__name__)12_THIS_DIR: Path = Path(os.path.dirname(__file__))13KeyFilter = Callable[[str], bool]14class ClientProxy(object):15 def __init__(self, client, service: str):16 self._client = client17 self._patch_client()18 self._service = service19 @property20 def _is_gov(self) -> bool:21 return '-gov-' in self._client.meta.region_name22 def _patch_client(self):23 pass24 def _should_import(self, key: str) -> bool:25 return True26 def resource_names(self) -> Iterator[str]:27 return filter(self._should_import, dir(self._client))28 def _list_args(self, key: str) -> Dict:29 return {}30 def _paginate_args(self, key: str) -> Dict:31 return {}32 def canonical_name(self, py_name: str) -> str:33 return self._client.meta.method_to_api_mapping[py_name]34 def _map_error_code(self, code: str, resource_name: str) -> Optional[str]:35 return None36 def list(self, key: str, kwargs,37 retry_on_throttle) -> Optional[Tuple[str, Any]]:38 prefix = len(key.split('_')[0])39 resource_name = self._client._PY_TO_OP_NAME[key][prefix:]40 extra_kwargs = dict(self._list_args(key), **kwargs)41 try:42 if self._client.can_paginate(key):43 paginator = self._client.get_paginator(key)44 method_args = dict(self._paginate_args(key), **extra_kwargs)45 iterator = paginator.paginate(**method_args)46 result = iterator.build_full_result()47 else:48 op = self._client.meta.method_to_api_mapping[key]49 op_model = self._client.meta.service_model.operation_model(op)50 output = op_model.output_shape51 attr = getattr(self._client, key)52 full_result = attr(**extra_kwargs)53 result = {54 result_key: full_result[result_key]55 for result_key in output.members.keys()56 if result_key in full_result57 }58 return resource_name, result59 except KeyError as e:60 raise GFInternal(61 f'Pagination Exception raise {self._client._PY_TO_OP_NAME[key]}')62 except botocore.exceptions.ParamValidationError as e:63 # TODO: fix this64 _log.debug(f'{key} Needs param input {str(e)}')65 return resource_name, {ERROR_KEY: 'needs param input'}66 except botocore.exceptions.ClientError as e:67 code = e.response.get('Error', {}).get('Code')68 if code == 'UnsupportedOperation':69 _log.info(f'{resource_name} Not supported in this region')70 return resource_name, {ERROR_KEY: 'unsupported in region'}71 elif code == 'MissingParameter':72 return resource_name, {ERROR_KEY: 'missing parameter'}73 elif code == 'OptInRequired':74 return resource_name, {ERROR_KEY: 'missing opt-in'}75 elif code in ('AuthFailure', 'AccessDeniedException',76 'UnauthorizedOperation', 'AccessDenied'):77 _log.warn(f'Missing permissions for {self._service} {key}')78 return resource_name, {ERROR_KEY: 'auth failure'}79 elif code == 'InvalidClientTokenId':80 return resource_name, {ERROR_KEY: 'invalid token'}81 elif code is not None and code.startswith('NoSuch'):82 # No results, return nothing83 return None84 elif code == 'Throttling' or code == 'ThrottlingException':85 if retry_on_throttle:86 _log.warn(f'Throttled for {key}, retrying')87 time.sleep(3)88 return self.list(key, kwargs, retry_on_throttle=False)89 else:90 _log.error(f'Throttled for {key}, not retrying')91 raise e92 else:93 mapped = self._map_error_code(code, resource_name)94 if mapped is not None:95 return resource_name, {ERROR_KEY: mapped}96 raise e97 def get(self, key: str, kwargs):98 try:99 api_call = getattr(self._client, key)100 full_result = api_call(**kwargs)101 op = self._client.meta.method_to_api_mapping[key]102 op_model = self._client.meta.service_model.operation_model(op)103 output = op_model.output_shape104 attr_result = {}105 saw_result = False106 for result_key in output.members.keys():107 value = full_result.get(result_key)108 if value is not None:109 saw_result = True110 attr_result[result_key] = value111 if saw_result:112 return attr_result113 except botocore.exceptions.ClientError as e:114 error = e.response.get('Error', {}).get('Code', '')115 if error.startswith('NoSuch') \116 or error.endswith('NotFoundError') \117 or error.endswith('NotFound') \118 or error == 'ResourceNotFoundException':119 # No results, nothing to return120 return None121 elif error in ('AuthFailure', 'MethodNotAllowed',122 'AccessDeniedException', 'AccessDenied'):123 # Auth failure124 _log.warn(f'Missing permissions for {self._service} {key}')125 else:126 raise e127class EC2ClientProxy(ClientProxy):128 MISSING_PAGINATION = {129 'describe_coip_pools':130 'PoolIds',131 'describe_elastic_gpus':132 'ElasticGpuSet',133 'describe_instance_type_offerings':134 'InstanceTypeOfferings',135 'describe_local_gateway_route_table_virtual_interface_group_associations':136 'LocalGatewayRouteTableVirtualInterfaceGroupAssociationIds',137 'describe_local_gateway_route_table_vpc_associations':138 'LocalGatewayRouteTableVpcAssociations',139 'describe_local_gateway_route_tables':140 'LocalGatewayRouteTableSet',141 'describe_local_gateway_virtual_interface_groups':142 'LocalGatewayVirtualInterfaceGroups',143 'describe_local_gateway_virtual_interfaces':144 'LocalGatewayVirtualInterfaces',145 'describe_local_gateways':146 'LocalGateways',147 'describe_transit_gateway_peering_attachments':148 'TransitGatewayPeeringAttachments',149 'describe_instance_types':150 'InstanceTypes',151 'describe_transit_gateway_multicast_domains':152 'TransitGatewayMulticastDomains'153 }154 SKIPLIST = [155 'describe_reserved_instances_offerings',156 # unsupported157 'describe_transit_gateway_connect_peers',158 'describe_transit_gateway_connects',159 'describe_spot_price_history',160 # Describes services that *can* have VPC endpoints, not ones that do161 'describe_vpc_endpoint_services',162 # TODO: verify this, i think it's about regional support for long ids163 'describe_aggregate_id_format',164 # TODO: look into this165 'describe_moving_addresses',166 # Failing in some cases, and we don't map167 'describe_id_format',168 # not needed, it's on most return values169 'describe_tags',170 # Not top level171 'describe_instance_attribute'172 ]173 GOV_SKIPLIST = [174 'describe_client_vpn_endpoints', 'describe_managed_prefix_lists',175 'describe_network_insights_analyses', 'describe_network_insights_paths'176 ]177 EXTRA_ARGS = {178 'describe_images': {179 'Filters': [{180 'Name': 'is-public',181 'Values': ['false']182 }]183 },184 'describe_snapshots': {185 'OwnerIds': ['self']186 }187 }188 PAGE_SIZES = {189 'describe_ipv6_pools': 10,190 'describe_public_ipv4_pools': 10,191 'describe_reserved_instances_modifications': None192 }193 INVALID_ACTIONS = [194 'FpgaImages', 'TransitGatewayMulticastDomains', 'ByoipCidrs',195 'Ipv6Pools', 'CoipPools',196 'LocalGatewayRouteTableVirtualInterfaceGroupAssociations',197 'LocalGatewayRouteTableVpcAssociations', 'LocalGatewayRouteTables',198 'LocalGatewayVirtualInterfaceGroups', 'LocalGatewayVirtualInterfaces',199 'LocalGateways'200 ]201 def __init__(self, *args, **kwargs):202 super().__init__(*args, **kwargs)203 ec2_svc_file = _THIS_DIR / 'svcs' / 'ec2.yml'204 with ec2_svc_file.open('r') as f:205 self._spec = yaml.safe_load(f)206 def _patch_client(self):207 # Force loading of the pagination config208 self._client.can_paginate('describe_instances')209 assert 'page_config' in self._client._cache210 for py_name, result_key in self.MISSING_PAGINATION.items():211 op_name = self._client._PY_TO_OP_NAME[py_name]212 self._client._cache['page_config'][op_name] = {213 'input_token': 'NextToken',214 'output_token': 'NextToken',215 'limit_key': 'MaxResults',216 'result_key': result_key217 }218 def _should_import(self, key: str) -> bool:219 return key.startswith('describe_') and key not in self.SKIPLIST and not (220 self._is_gov and key in self.GOV_SKIPLIST)221 def _list_args(self, key: str) -> Dict:222 return self.EXTRA_ARGS.get(key, {})223 def _paginate_args(self, key: str) -> Dict:224 # TODO: key this off of output shape225 page_size = self.PAGE_SIZES.get(key, 100)226 return {'PaginationConfig': {'PageSize': page_size}}227 def _map_error_code(self, code: str, resource_name: str) -> Optional[str]:228 if code == 'InvalidSpotDatafeed.NotFound':229 return 'Missing subscription'230 elif code == 'InvalidAction' and resource_name in self.INVALID_ACTIONS:231 return 'Missing Input params'232 return None233class S3ClientProxy(ClientProxy):234 MISSING_PAGINATION = {235 'list_bucket_analytics_configurations': 'AnalyticsConfigurationList',236 'list_bucket_inventory_configurations': 'InventoryConfigurationList',237 'list_bucket_metrics_configurations': 'MetricsConfigurationList'238 }239 GET_PREFIX = 'get_bucket_'240 LIST_PREFIX = 'list_bucket_'241 SKIPLIST = [242 # deprecated methods covered by other calls243 'get_bucket_notification',244 'get_bucket_lifecycle',245 'list_bucket_intelligent_tiering_configurations'246 ]247 GOV_SKIPLIST = ['get_bucket_accelerate_configuration']248 def _patch_client(self):249 # Force loading of the pagination config250 self._client.can_paginate('list_bucket_analytics_configurations')251 for py_name, result_key in self.MISSING_PAGINATION.items():252 op_name = self._client._PY_TO_OP_NAME[py_name]253 self._client._cache['page_config'][op_name] = {254 'input_token': 'ContinuationToken',255 'output_token': 'NextContinuationToken',256 'result_key': result_key257 }258 def _should_import(self, key: str) -> bool:259 if key in self.SKIPLIST:260 return False261 elif self._is_gov and key in self.GOV_SKIPLIST:262 return False263 elif key.startswith(self.LIST_PREFIX):264 return True265 elif key.startswith(self.GET_PREFIX):266 # Only return true if there is not a corresponding list call267 item = key[len(self.GET_PREFIX):]268 list_op = f'list_bucket_{item}s'269 return not hasattr(self._client, list_op)270 else:271 return False272class ELBClientProxy(ClientProxy):273 SKIPLIST = [274 'describe_load_balancers', 'describe_account_limits',275 'describe_instance_health', 'describe_load_balancer_policy_types'276 ]277 def _should_import(self, key: str) -> bool:278 return key.startswith('describe_') and key not in self.SKIPLIST279class LambdaClientProxy(ClientProxy):280 SKIPLIST: List[str] = []281 def _should_import(self, key: str) -> bool:282 return key.startswith('list_') and key not in self.SKIPLIST283class CloudtrailClientProxy(ClientProxy):284 SKIPLIST: List[str] = []285 def _should_import(self, key: str) -> bool:286 return key.startswith('describe_') and key not in self.SKIPLIST287class RDSClientProxy(ClientProxy):288 SKIPLIST = [289 'describe_custom_availability_zones',290 'describe_installation_media',291 'describe_export_tasks',292 'describe_reserved_db_instances_offerings',293 'describe_db_engine_versions',294 'describe_valid_db_instance_modifications',295 'describe_option_group_options',296 'describe_db_proxies', # Not sure why this comes back with unknown297 'describe_global_clusters',298 # migrating away299 'describe_db_snapshots',300 'describe_db_cluster_snapshots'301 ]302 def _should_import(self, key: str) -> bool:303 return key.startswith('describe_') and key not in self.SKIPLIST304class IAMClientProxy(ClientProxy):305 SKIPLIST: List[str] = []306 MISSING_PAGINATION = {'list_user_tags': 'Tags', 'list_role_tags': 'Tags'}307 def _should_import(self, key: str) -> bool:308 return key.startswith('list_') and key not in self.SKIPLIST309 def _patch_client(self):310 # Force loading of the pagination config311 self._client.can_paginate('list_users')312 for py_name, result_key in self.MISSING_PAGINATION.items():313 op_name = self._client._PY_TO_OP_NAME[py_name]314 self._client._cache['page_config'][op_name] = {315 'input_token': 'Marker',316 'limit_key': 'MaxItems',317 'more_results': 'IsTruncated',318 'output_token': 'Marker',319 'result_key': result_key320 }321class AWSFetch(object):322 SVC_CLASS = {323 'ec2': EC2ClientProxy,324 's3': S3ClientProxy,325 'elb': ELBClientProxy,326 'iam': IAMClientProxy,327 'rds': RDSClientProxy,328 'lambda': LambdaClientProxy,329 'cloudtrail': CloudtrailClientProxy330 }331 def __init__(self, boto: Boto):332 self._boto = boto333 def client(self,334 service: str,335 region: Optional[str] = None) -> 'ClientProxy':336 client_class = self.SVC_CLASS.get(service, ClientProxy)337 kwargs = {}338 if region is not None:339 kwargs['region_name'] = region340 client = self._boto.create_client(service, **kwargs)341 return client_class(client, service)342class ServiceProxy(object):343 def __init__(self, impl: ClientProxy):344 self._impl = impl345 def resource_names(self) -> Iterator[str]:346 return self._impl.resource_names()347 def list(self, resource: str, **kwargs) -> Optional[Tuple[str, Any]]:348 _log.debug(f'calling list {resource} {kwargs}')349 return self._impl.list(resource, kwargs, retry_on_throttle=True)350 def get(self, resource: str, **kwargs):351 _log.debug(f'calling get {resource} {kwargs}')352 return self._impl.get(resource, kwargs)353 def canonical_name(self, py_name: str) -> str:354 return self._impl.canonical_name(py_name)355class Proxy(object):356 @classmethod357 def build(cls, boto: Boto, patch_id: Optional[int] = None) -> 'Proxy':358 return cls(AWSFetch(boto))359 @classmethod360 def dummy(cls, boto: Boto) -> 'Proxy':361 return cls(AWSFetch(boto))362 def __init__(self, aws: AWSFetch):363 self._aws = aws364 def service(self,365 service: str,366 region: Optional[str] = None) -> ServiceProxy:...
not_used_operations.py
Source:not_used_operations.py
1not_used_operations = {2 'ec2': [3 'describe_load_balancer_policies',4 'describe_vpc_endpoint_services',5 'describe_aggregate_id_format',6 'describe_load_balancer_policy_types',7 'describe_endpoints',8 'describe_availability_zones',9 'describe_account_attributes',10 'describe_dhcp_options',11 'describe_fpga_images',12 'describe_host_reservation_offerings',13 'describe_id_format',14 'describe_instance_event_notification_attributes',15 'describe_instance_type_offerings',16 'describe_instance_types',17 'describe_managed_prefix_lists',18 'describe_prefix_lists',19 'describe_regions',20 'describe_launch_template_versions',21 'describe_principal_id_format',22 'describe_spot_price_history',23 'describe_flow_logs',24 ### The blow 2 require extra parameters.25 'describe_images',26 'describe_snapshots',27 ],28 'ecr': [29 'get_authorization_token',30 ],31 'ecs': [32 'describe_capacity_providers',33 ],34 'efs': [35 'describe_account_preferences',36 ],37 'acm': [38 'get_account_configuration',39 ],40 'athena': [41 'list_data_catalogs',42 ],43 'apigateway': [44 'get_account',45 'get_sdk_types',46 ],47 'xray': [48 'get_encryption_config',49 'get_sample_rules',50 ],51 'waf': [52 'get_change_token',53 ],54 'waf-regional': [55 'get_change_token',56 ],57 'sts': [58 'get_caller_identity',59 'get_session_token',60 ],61 'ssm': [62 'describe_available_patches',63 'describe_patch_baselines',64 'get_inventory_schema',65 ],66 'ses': [67 'get_account_sending_enabled',68 'get_send_quota',69 ],70 'sesv2': [71 'get_account',72 'get_deliverability_dashboard_options',73 ],74 'signer': [75 'list_signing_platforms',76 ],77 'shield': [78 'describe_attack_statistics',79 'get_subscription_state',80 ],81 'lex-models': [82 'get_builtin_intents',83 'get_builtin_slot_types',84 ],85 'dynamodb': [86 'describe_limits',87 ],88 'lightsail': [89 'get_blueprints',90 'get_container_service_powers',91 'get_container_api_metadata',92 'get_bundles',93 'get_relational_database_blueprints',94 'get_relational_database_bundles',95 'get_regions',96 'get_bucket_bundles',97 ],98 'inspector': [99 'describe_cross_account_access_role',100 'list_rules_packages',101 ],102 'lambda': [103 'get_account_settings',104 ],105 'kinesis': [106 'describe_limits',107 ],108 'kafka': [109 'list_kafka_versions',110 'get_compatible_kafka_versions',111 ],112 'iot': [113 'describe_account_audit_configuration',114 'describe_event_configurations',115 'get_indexing_configuration',116 'get_registration_code',117 'list_domain_configurations'118 ],119 'iotsitewise': [120 'describe_logging_options',121 'describe_storage_configuration'122 ],123 'guardduty': [124 'get_invitations_count',125 ],126 'gamelift': [127 'describe_ec2_instance_limits',128 ],129 'elb': [130 'describe_account_limits',131 'describe_load_balancer_policies',132 ],133 'elbv2': [134 'describe_account_limits',135 'describe_ssl_policies',136 ],137 'elasticache': [138 'describe_cache_parameter_groups',139 'describe_service_updates',140 'list_allowed_node_type_modifications',141 'describe_cache_security_groups',142 'describe_users',143 ],144 'elasticbeanstalk': [145 'describe_account_attributes',146 'list_available_solution_stacks'147 ],148 'clouddirectory': [149 'list_managed_schema_arns',150 ],151 'codebuild': [152 'list_builds',153 'list_curated_environment_images'154 ],155 'cloudtrail': [156 'list_public_keys',157 'get_event_selectors',158 ],159 'cloudwatch': [160 'describe_alarm_history',161 'list_metrics',162 ],163 'snowball': [164 'get_snowball_usage',165 'list_compatible_images'166 ],167 'sms': [168 'get_servers'169 ],170 'sagemaker': [171 'get_sagemaker_servicecatalog_portfolio_status',172 ],173 'transfer': [174 'list_security_policies'175 ],176 'schemas': [177 'list_registries',178 ],179 'securityhub': [180 'describe_standards',181 ],182 'secretsmanager': [183 'get_random_password',184 ],185 'service-quotes': [186 'list_services',187 ],188 'wellarchitected': [189 'list_lenses',190 ],191 'route53': [192 'get_checker_ip_ranges',193 'get_geo_location',194 'get_traffic_policy_instance_count',195 'get_hosted_zone_count',196 'get_health_check_count',197 'list_geo_locations',198 ],199 'resourcegroupstaggingapi': [200 'get_resources',201 'get_tag_keys',202 ],203 'events': [204 'describe_event_bus',205 'list_event_buses',206 ],207 'fis': [208 'list_actions',209 ],210 'frauddetector': [211 'get_kms_encryption_key',212 ],213 'glacier': [214 'get_data_retrieval_policy',215 ],216 'glue': [217 'get_catalog_import_status',218 ],219 'apprunner': [220 'list_auto_scaling_configurations',221 ],222 'rds': [223 'describe_account_attributes',224 'describe_certificates',225 'describe_source_regions',226 ],227 'ram': [228 'list_permissions',229 ],230 'redshift': [231 'describe_account_attributes',232 'describe_cluster_tracks',233 'describe_orderable_cluster_options',234 'describe_storage',235 ],236 'polly': [237 'describe_voices',238 ],239 'opsworks': [240 'describe_my_user_profile',241 'describe_operating_systems',242 'describe_user_profiles',243 ],244 'opsworkscm': [245 'describe_account_attributes',246 ],247 'emr': [248 'get_block_public_access_configuration',249 'list_release_labels',250 ],251 'elastictranscoder': [252 'list_presets',253 ],254 'dms': [255 'describe_account_attributes',256 'describe_applicable_individual_assessments',257 'describe_orderable_replication_instances',258 ],259 'docdb': [260 'describe_certificates',261 'describe_db_subnet_groups',262 ],263 'ds': [264 'get_directory_limits',265 ],266 'compute-optimizer': [267 'get_enrollment_status',268 ],269 'config': [270 'get_compliance_summary_by_config_rule',271 'get_compliance_summary_by_resource_type',272 'get_discovered_resource_counts',273 ],274 'dax': [275 'describe_parameter_groups'276 ],277 'codedeploy': [278 'list_deployment_configs',279 ],280 'backup': [281 'list_backup_plan_templates',282 ],283 'autoscaling': [284 'describe_account_limits',285 ],286 'auditmanger': [287 'get_account_status',288 ],289 'cloudformation': [290 'describe_account_limits',291 ],292 'nimble': [293 'list_eulas',294 ],295 'cloudfront': [296 'list_cache_polocies',297 ]...
practice2.py
Source:practice2.py
1import boto32client = boto3.client('ec2')3# t1 = client.describe_addresses()4# print(t1)5t1 =client.describe_aggregate_id_format(6 DryRun=True|False7)...
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run localstack automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
