Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More
How to use describe_transit_gateway_connects method in localstack
Best Python code snippet using localstack_python
infraDecom.py
Source:infraDecom.py 
...176 'Values': [str(tgwId)]177 }178 ]179 listTgwConnectInfo = []180 tgwConnectId = ec2client.describe_transit_gateway_connects(181 Filters=newfilter)182 for tgwConnect in tgwConnectId['TransitGatewayConnects']:183 tgwConnectInfo = {}184 tgwConnectInfo['Id'] = tgwConnect['TransitGatewayAttachmentId']185 tgwConnectInfo['State'] = tgwConnect['State']186 listTgwConnectInfo.append(tgwConnectInfo)187 return listTgwConnectInfo188def listTgwConnectPeer(tgwConnectAttachId):189 """190 return a list of TGW connect peer of TGW Connect whose ID is tgwConnectAttachId191 """192 listTgwConnectPeerInfo = []193 newfilter = [194 {195 'Name': 'transit-gateway-attachment-id',196 'Values': [str(tgwConnectAttachId)]197 }198 ]199 tgwConnectPeerId = ec2client.describe_transit_gateway_connect_peers(200 Filters=newfilter)201 for x2 in tgwConnectPeerId['TransitGatewayConnectPeers']:202 tgwConnectPeerInfo = {}203 tgwConnectPeerInfo['Id'] = x2['TransitGatewayConnectPeerId']204 tgwConnectPeerInfo['State'] = x2['State']205 listTgwConnectPeerInfo.append(tgwConnectPeerInfo)206 return listTgwConnectPeerInfo207def listTgwPeering(tgwId):208 """209 return a list of TGW Peering Attachment of Transit Gateway whose ID is tgwId210 """211 newfilter = [212 {213 'Name': 'transit-gateway-id',214 'Values': [str(tgwId)]215 }216 ]217 listTgwPeeringInfo = []218 response = ec2client.describe_transit_gateway_peering_attachments(219 Filters=newfilter)220 tgwPeeringAttachments = response['TransitGatewayPeeringAttachments']221 for twgPeeringAttachment in tgwPeeringAttachments:222 tgwPeeringInfo = {}223 tgwPeeringInfo['Id'] = twgPeeringAttachment['TransitGatewayAttachmentId']224 tgwPeeringInfo['State'] = twgPeeringAttachment['State']225 listTgwPeeringInfo.append(tgwPeeringInfo)226 return listTgwPeeringInfo227def listInstance():228 """229 return a list of TGW has tag key=owner, value=huyen230 """231 listInstanceInfo = []232 response = ec2client.describe_instances(Filters=custom_filter)233 if response == []:234 print('No instances.')235 else:236 for x2 in response['Reservations']:237 for x3 in x2['Instances']:238 instanceInfo = {}239 instanceInfo['Id'] = x3['InstanceId']240 instanceInfo['State'] = x3['State']['Name']241 if x3['State']['Name'] != 'terminated':242 instanceInfo['VpcId'] = x3['VpcId']243 else:244 instanceInfo['VpcId'] = ''245 listInstanceInfo.append(instanceInfo)246 return listInstanceInfo247def listVpc():248 """249 return a list of VPC 250 """251 listVpcInfo = []252 vpcs = ec2client.describe_vpcs(Filters=custom_filter)253 for x2 in vpcs['Vpcs']:254 vpcInfo = {}255 vpcInfo['Id'] = x2['VpcId']256 vpcInfo['State'] = x2['State']257 listVpcInfo.append(vpcInfo)258 return listVpcInfo259def listSubnet():260 """ 261 return a list of subnet ID each has properties: ID, VPC ID and State262 """263 listSubnetInfo = []264 subnets = ec2client.describe_subnets(Filters=custom_filter)265 for x2 in subnets['Subnets']:266 subnetInfo = {}267 subnetInfo['Id'] = x2['SubnetId']268 subnetInfo['VpcId'] = x2['VpcId']269 subnetInfo['State'] = x2['State']270 listSubnetInfo.append(subnetInfo)271 return listSubnetInfo272def listSg():273 """274 return a list of security groups with properties ID, VPC ID, IngressRules, EgressRule275 """276 listSgInfo = []277 sgs = ec2client.describe_security_groups(Filters=custom_filter)278 for x2 in sgs['SecurityGroups']:279 sgInfo = {}280 sgInfo['GroupName'] = x2['GroupName']281 sgInfo['Id'] = x2['GroupId']282 sgInfo['VpcId'] = x2['VpcId']283 sgInfoIngressSecGroup = []284 sgInfoEgressSecGroup = []285 x3 = x2['IpPermissions']286 for x4 in x3:287 for x5 in x4['UserIdGroupPairs']:288 sgInfoIngressSecGroup.append(x5['GroupId'])289 sgInfo['IngressSecGroup'] = sgInfoIngressSecGroup290 y3 = x2['IpPermissionsEgress']291 for y4 in y3:292 for y5 in y4['UserIdGroupPairs']:293 sgInfoEgressSecGroup.append(y5['GroupId'])294 sgInfo['EgressSecGroup'] = sgInfoEgressSecGroup295 listSgInfo.append(sgInfo)296 return listSgInfo297def listSgRules(sgId):298 """299 return list of ingress rule list and egress rules list of a security group300 """301 ingressRules = []302 egressRules = []303 newfilter = [304 {305 'Name': 'group-id',306 'Values': [str(sgId)]307 }308 ]309 response = ec2client.describe_security_group_rules(Filters=newfilter)310 sgRules = response['SecurityGroupRules']311 for sgRule in sgRules:312 if sgRule['IsEgress']:313 egressRules.append(sgRule['SecurityGroupRuleId'])314 else:315 ingressRules.append(sgRule['SecurityGroupRuleId'])316 return ingressRules, egressRules317def listRt():318 """319 return a list of route table with properties of RouteTableId, VPC ID320 """321 listRtInfo = []322 rts = ec2client.describe_route_tables(Filters=custom_filter)323 for x2 in rts['RouteTables']:324 rtInfo = {}325 rtInfo['Id'] = x2['RouteTableId']326 rtInfo['VpcId'] = x2['VpcId']327 listRtInfo.append(rtInfo)328 return listRtInfo329def listIgw():330 """331 return a list of Internet Gateways with properties State and VPC ID332 """333 listIgwInfo = []334 igws = ec2client.describe_internet_gateways(Filters=custom_filter)335 for x2 in igws['InternetGateways']:336 igwInfo = {}337 if x2['Attachments']:338 igwInfo['Id'] = x2['InternetGatewayId']339 igwInfo['State'] = x2['Attachments'][0]['State']340 igwInfo['VpcId'] = x2['Attachments'][0]['VpcId']341 listIgwInfo.append(igwInfo)342 else:343 print('Internet Gateway is detached.')344 return listIgwInfo345def delTgwPeering(tgwPeeringId):346 ec2client.delete_transit_gateway_peering_attachment(347 TransitGatewayAttachmentId=tgwPeeringId)348def delTgwConnectPeer(connectPeerId):349 """350 delete tgw connnect peers of tgw connect attachment whose ID is connectPeerId351 """352 ec2client.delete_transit_gateway_connect_peer(353 TransitGatewayConnectPeerId=connectPeerId)354def delTgwConnect(connectAttachmentId):355 """356 check if it is possible to delete tgwConnect attachment357 only delete tgwConnect attachment if all connectPeers are in deleted state358 """359 eligibleDeletion = False360 listTgwConnectPeerInfo = listTgwConnectPeer(connectAttachmentId)361 eligibleConnectPeers = []362 # delete all connect peers who are not in deleted or deleting state.363 for tgwConnectPeerInfo in listTgwConnectPeerInfo:364 if tgwConnectPeerInfo['State'] not in ['deleted', 'deleting']:365 #print("Deleting TGW Connect Peer", tgwConnectPeerInfo['Id'], "in", connectAttachmentId)366 delTgwConnectPeer(tgwConnectPeerInfo['Id'])367 eligibleConnectPeers.append(tgwConnectPeerInfo['Id'])368 # progressive bar369 for eligibleConnectPeer in eligibleConnectPeers:370 aliveBar(1500 + randrange(100, 200), 0.05,371 "Deleting Connect Peer " + eligibleConnectPeer)372 # check if all connect peer are in delete states373 while not eligibleDeletion:374 time.sleep(30)375 listState = []376 listTgwConnectPeerInfo = listTgwConnectPeer(connectAttachmentId)377 for tgwConnectPeerInfo in listTgwConnectPeerInfo:378 listState.append(tgwConnectPeerInfo['State'])379 if listTgwConnectPeerInfo == []:380 eligibleDeletion = True381 else:382 if all(element == 'deleted' for element in listState):383 eligibleDeletion = True384 print(385 "All peers in deleted state, start deleting TGW Connect Attachment...")386 print("Deleting TGW Connect Attachment ", connectAttachmentId)387 if eligibleDeletion:388 ec2client.delete_transit_gateway_connect(389 TransitGatewayAttachmentId=connectAttachmentId)390def delTgwVpcAttachment(tgwVpcAttachmentId):391 """392 find all tgwConnectAttachment who use that vpc as transport attachmentInfoList393 if tgwConnectAttachment state is deleted then delete VPC attachment394 """395 eligibleDeletion = False396 newfilter = [397 {398 'Name': 'transport-transit-gateway-attachment-id',399 'Values': [str(tgwVpcAttachmentId)]400 }401 ]402 while not eligibleDeletion:403 time.sleep(15)404 listConnectState = []405 response = ec2client.describe_transit_gateway_connects(406 Filters=newfilter)407 if response == []:408 eligibleDeletion = True409 else:410 tgwConnectAttachments = response['TransitGatewayConnects']411 for tgwConnectAttachment in tgwConnectAttachments:412 listConnectState.append(tgwConnectAttachment['State'])413 if all(element == 'deleted' for element in listConnectState):414 eligibleDeletion = True415 print(416 "All Connect Attachments are in deleted state, start deleting VPC Attachment...")417 #print("Deleting TGW VPC Attachment", tgwVpcAttachmentId)418 if eligibleDeletion:419 ec2client.delete_transit_gateway_vpc_attachment(...
cloudwalk.py
Source:cloudwalk.py
...161 'Values': [str(tgwId)]162 }163 ]164 listTgwConnectInfo = []165 tgwConnectId = ec2client.describe_transit_gateway_connects(166 Filters=newfilter)167 for tgwConnect in tgwConnectId['TransitGatewayConnects']:168 tgwConnectInfo = {}169 tgwConnectInfo['Id'] = tgwConnect['TransitGatewayAttachmentId']170 tgwConnectInfo['State'] = tgwConnect['State']171 listTgwConnectInfo.append(tgwConnectInfo)172 return listTgwConnectInfo173def listTgwConnectPeer(tgwConnectAttachId):174 """175 return a list of TGW connect peer of TGW Connect whose ID is tgwConnectAttachId176 """177 listTgwConnectPeerInfo = []178 newfilter = [179 {...
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:...
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?
