Best Python code snippet using localstack_python
test_transit_gateway.py
Source:test_transit_gateway.py
...627 )["TransitGatewayVpcAttachment"]628 table = ec2.create_transit_gateway_route_table(TransitGatewayId=gateway_id)[629 "TransitGatewayRouteTable"630 ]631 initial = ec2.get_transit_gateway_route_table_propagations(632 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]633 )634 initial["TransitGatewayRouteTablePropagations"].should.equal(635 [636 {637 "TransitGatewayAttachmentId": "",638 "ResourceId": "",639 "ResourceType": "",640 "State": "",641 }642 ]643 )644 ec2.associate_transit_gateway_route_table(645 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],646 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],647 )648 ec2.enable_transit_gateway_route_table_propagation(649 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],650 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],651 )652 enabled = ec2.get_transit_gateway_route_table_propagations(653 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]654 )655 enabled["TransitGatewayRouteTablePropagations"].should.equal(656 [657 {658 "TransitGatewayAttachmentId": attchmnt["TransitGatewayAttachmentId"],659 "ResourceId": "vpc-id",660 "ResourceType": "vpc",661 "State": "enabled",662 }663 ]664 )665@mock_ec2666def test_disable_transit_gateway_route_table_propagation_without_enabling_first():667 ec2 = boto3.client("ec2", region_name="us-west-1")668 gateway_id = ec2.create_transit_gateway(Description="g")["TransitGateway"][669 "TransitGatewayId"670 ]671 attchmnt = ec2.create_transit_gateway_vpc_attachment(672 TransitGatewayId=gateway_id, VpcId="vpc-id", SubnetIds=["sub1"]673 )["TransitGatewayVpcAttachment"]674 table = ec2.create_transit_gateway_route_table(TransitGatewayId=gateway_id)[675 "TransitGatewayRouteTable"676 ]677 initial = ec2.get_transit_gateway_route_table_propagations(678 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]679 )680 initial["TransitGatewayRouteTablePropagations"].should.equal(681 [682 {683 "TransitGatewayAttachmentId": "",684 "ResourceId": "",685 "ResourceType": "",686 "State": "",687 }688 ]689 )690 ec2.associate_transit_gateway_route_table(691 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],692 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],693 )694 try:695 ec2.disable_transit_gateway_route_table_propagation(696 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],697 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],698 )699 assert False, "Should not be able to disable before enabling it"700 except Exception:701 pass702@mock_ec2703def test_disable_transit_gateway_route_table_propagation():704 ec2 = boto3.client("ec2", region_name="us-west-1")705 gateway_id = ec2.create_transit_gateway(Description="g")["TransitGateway"][706 "TransitGatewayId"707 ]708 attchmnt = ec2.create_transit_gateway_vpc_attachment(709 TransitGatewayId=gateway_id, VpcId="vpc-id", SubnetIds=["sub1"]710 )["TransitGatewayVpcAttachment"]711 table = ec2.create_transit_gateway_route_table(TransitGatewayId=gateway_id)[712 "TransitGatewayRouteTable"713 ]714 initial = ec2.get_transit_gateway_route_table_propagations(715 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]716 )717 initial["TransitGatewayRouteTablePropagations"].should.equal(718 [719 {720 "TransitGatewayAttachmentId": "",721 "ResourceId": "",722 "ResourceType": "",723 "State": "",724 }725 ]726 )727 ec2.associate_transit_gateway_route_table(728 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],729 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],730 )731 ec2.enable_transit_gateway_route_table_propagation(732 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],733 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],734 )735 ec2.disable_transit_gateway_route_table_propagation(736 TransitGatewayAttachmentId=attchmnt["TransitGatewayAttachmentId"],737 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"],738 )739 disabled = ec2.get_transit_gateway_route_table_propagations(740 TransitGatewayRouteTableId=table["TransitGatewayRouteTableId"]741 )742 disabled["TransitGatewayRouteTablePropagations"].should.equal(743 [744 {745 "ResourceId": "",746 "ResourceType": "",747 "State": "",748 "TransitGatewayAttachmentId": "",749 }750 ]...
transit_gateway_route_tables.py
Source:transit_gateway_route_tables.py
...84 )85 return template.render(86 transit_gateway_route_table_associations=transit_gateway_route_table_associations87 )88 def get_transit_gateway_route_table_propagations(self):89 transit_gateway_route_table_id = self._get_param("TransitGatewayRouteTableId")90 filters = filters_from_querystring(self.querystring)91 transit_gateway_route_table_propagations = self.ec2_backend.get_all_transit_gateway_route_table_propagations(92 transit_gateway_route_table_id, filters93 )94 template = self.response_template(95 GET_TRANSIT_GATEWAY_ROUTE_TABLE_PROPAGATIONS_RESPONSE96 )97 return template.render(98 transit_gateway_route_table_propagations=transit_gateway_route_table_propagations99 )100CREATE_TRANSIT_GATEWAY_ROUTE_TABLE_RESPONSE = """<CreateTransitGatewayRouteTableResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/">101 <requestId>3a495d25-08d4-466d-822e-477c9b1fc606</requestId>102 <transitGatewayRouteTable>...
awshelper.py
Source:awshelper.py
1import boto32_clients = {}3_functions = {}4_session = None5# the last two keys should both be None or both be non-None6def _get_client(client_type, region):7 if _session is None:8 raise ValueError("Session has not been initialized")9 return _session.client(client_type, region_name=region)10def _aws_response(response_func, args_dict):11 response = response_func(**args_dict)12 metadata = response["ResponseMetadata"]13 if "HTTPStatusCode" not in metadata or metadata["HTTPStatusCode"] != 200:14 raise IOError("Unexpected status code " + metadata.get("HTTPStatusCode", None))15 response.pop("ResponseMetadata")16 return response17def aws_get_config(region):18 config = {}19 for key in _functions[region]:20 (response_func, args_dict) = _functions[region][key]21 try:22 response = _aws_response(response_func, args_dict)23 except Exception as e:24 print("Exception getting {} for {}: {}".format(key, region, e))25 continue26 config[key] = response27 if key == "TransitGatewayRouteTables":28 _aws_get_route_table_details(_clients[region]["ec2"], config, response["TransitGatewayRouteTables"])29 if key == "LoadBalancers":30 _aws_get_load_balancer_details(_clients[region]["elbv2"], config, response["LoadBalancers"])31 if key == "TargetGroups":32 _aws_get_elbv2_target_health(_clients[region]["elbv2"], config, response["TargetGroups"])33 return config34import json35def _aws_get_load_balancer_details(client, config, load_balancers):36 listeners = []37 attributes = []38 for load_balancer in load_balancers:39 lb_arn = load_balancer["LoadBalancerArn"]40 try:41 response = _aws_response(client.describe_listeners, dict(LoadBalancerArn=lb_arn))42 response["LoadBalancerArn"] = lb_arn43 listeners.append(response)44 except Exception as e:45 print(46 "Exception getting load balancer listeners for {} with arn {}: {}".format(47 load_balancer["LoadBalancerName"],48 lb_arn, e))49 try:50 response = _aws_response(client.describe_load_balancer_attributes, dict(LoadBalancerArn=lb_arn))51 response["LoadBalancerArn"] = lb_arn52 attributes.append(response)53 except Exception as e:54 print(55 "Exception getting load balancer attributes for {} with arn {}: {}".format(56 load_balancer["LoadBalancerName"],57 lb_arn, e))58 config["LoadBalancerListeners"] = {"LoadBalancerListeners": listeners}59 config["LoadBalancerAttributes"] = {"LoadBalancerAttributes" : attributes}60def _aws_get_elbv2_target_health(client, config, target_groups):61 healths = []62 for target_group in target_groups:63 tg_arn = target_group["TargetGroupArn"]64 try:65 response = _aws_response(client.describe_target_health, dict(TargetGroupArn=tg_arn))66 response["TargetGroupArn"] = tg_arn67 healths.append(response)68 except Exception as e:69 print(70 "Exception getting target health for target group {} with Arn {}: {}".format(71 target_group["TargetGroupName"], tg_arn, e))72 config["LoadBalancerTargetHealth"] = {"LoadBalancerTargetHealth": healths}73def _aws_get_route_table_details(client, config, route_tables):74 static_routes = []75 propogations = []76 for table in route_tables:77 table_id = table["TransitGatewayRouteTableId"]78 try:79 propogations_response = _aws_response(client.get_transit_gateway_route_table_propagations,80 dict(TransitGatewayRouteTableId=table_id))81 propogations_response["TransitGatewayRouteTableId"] = table_id82 propogations.append(propogations_response)83 except Exception as e:84 print("Exception getting Propagations for {}: {}".format(table_id, e))85 try:86 static_routes_response = _aws_response(client.search_transit_gateway_routes,87 dict(TransitGatewayRouteTableId=table_id,88 Filters=[{"Name": "type", "Values": ["static"]}]))89 if static_routes_response['AdditionalRoutesAvailable']:90 raise IOError("Could not fetch all static routes for " + table_id)91 static_routes_response["TransitGatewayRouteTableId"] = table_id92 static_routes.append(static_routes_response)93 except Exception as e:94 print("Exception getting Static routes for {}: {}".format(table_id, e))95 config["TransitGatewayPropagations"] = {"TransitGatewayPropagations": propogations}96 config["TransitGatewayStaticRoutes"] = {"TransitGatewayStaticRoutes": static_routes}97def aws_get_regions():98 return _clients.keys()99def aws_init(regions=None, vpc_ids=None, skip_data=None, profile=None):100 global _session, _clients, _functions101 _session = boto3.session.Session(profile_name=profile)102 if regions is None or len(regions) == 0:103 ec2_client = _get_client('ec2', "us-west-1")104 response = ec2_client.describe_regions()105 regions_to_get = [region["RegionName"] for region in response["Regions"]]106 else:107 regions_to_get = regions108 if vpc_ids is None or len(vpc_ids) == 0:109 vpc_filter = []110 attachment_vpc_filter = []111 else:112 vpc_filter = [{'Name': "vpc-id", 'Values': vpc_ids}]113 attachment_vpc_filter = [{'Name': "attachment.vpc-id", 'Values': vpc_ids}]114 for region in regions_to_get:115 _clients[region] = {}116 ec2_client = _get_client("ec2", region)117 _clients[region]["ec2"] = ec2_client118 _functions[region] = {}119 _functions[region]["Addresses"] = (ec2_client.describe_addresses, dict())120 _functions[region]["AvailabilityZones"] = (ec2_client.describe_availability_zones, dict())121 _functions[region]["ClassicLinkInstances"] = (122 ec2_client.describe_classic_link_instances, dict(Filters=vpc_filter))123 _functions[region]["CustomerGateways"] = (ec2_client.describe_customer_gateways, dict())124 _functions[region]["DhcpOptions"] = (ec2_client.describe_dhcp_options, dict())125 _functions[region]["Hosts"] = (ec2_client.describe_hosts, dict())126 _functions[region]["InstanceStatuses"] = (ec2_client.describe_instance_status, dict())127 _functions[region]["InternetGateways"] = (128 ec2_client.describe_internet_gateways, dict(Filters=attachment_vpc_filter))129 _functions[region]["MovingAddressStatuses"] = (ec2_client.describe_moving_addresses, dict())130 _functions[region]["NatGateways"] = (ec2_client.describe_nat_gateways, dict(Filters=vpc_filter))131 _functions[region]["NetworkAcls"] = (ec2_client.describe_network_acls, dict(Filters=vpc_filter))132 _functions[region]["NetworkInterfaces"] = (ec2_client.describe_network_interfaces, dict(Filters=vpc_filter))133 _functions[region]["PlacementGroups"] = (ec2_client.describe_placement_groups, dict())134 _functions[region]["PrefixLists"] = (ec2_client.describe_prefix_lists, dict())135 _functions[region]["Reservations"] = (ec2_client.describe_instances, dict(Filters=vpc_filter))136 _functions[region]["RouteTables"] = (ec2_client.describe_route_tables, dict(Filters=vpc_filter))137 _functions[region]["SecurityGroups"] = (ec2_client.describe_security_groups, dict(Filters=vpc_filter))138 _functions[region]["Subnets"] = (ec2_client.describe_subnets, dict(Filters=vpc_filter))139 _functions[region]["Tags"] = (ec2_client.describe_tags, dict())140 _functions[region]["TransitGatewayAttachments"] = (ec2_client.describe_transit_gateway_attachments, dict())141 _functions[region]["TransitGatewayRouteTables"] = (ec2_client.describe_transit_gateway_route_tables, dict())142 _functions[region]["TransitGatewayVpcAttachments"] = (143 ec2_client.describe_transit_gateway_vpc_attachments, dict(Filters=vpc_filter))144 _functions[region]["TransitGateways"] = (ec2_client.describe_transit_gateways, dict())145 _functions[region]["VpcEndpoints"] = (ec2_client.describe_vpc_endpoints, dict(Filters=vpc_filter))146 _functions[region]["VpcPeeringConnections"] = (ec2_client.describe_vpc_peering_connections, dict())147 _functions[region]["Vpcs"] = (ec2_client.describe_vpcs, dict(Filters=vpc_filter))148 _functions[region]["VpcClassicLink"] = (ec2_client.describe_vpc_classic_link, dict())149 _functions[region]["VpcClassicLinkDnsSupport"] = (ec2_client.describe_vpc_classic_link_dns_support, dict())150 _functions[region]["VpnConnections"] = (ec2_client.describe_vpn_connections, dict())151 _functions[region]["VpnGateways"] = (ec2_client.describe_vpn_gateways, dict(Filters=attachment_vpc_filter))152 # get all elasticsearch domain names for this account (VPC based filtering is not supported yet)153 es_client = _get_client('es', region)154 _clients[region]["es"] = es_client155 domain_names = es_client.list_domain_names()156 _functions[region]["ElasticsearchDomains"] = (es_client.describe_elasticsearch_domains, dict(157 DomainNames=[domainEntry["DomainName"] for domainEntry in domain_names["DomainNames"]]))158 # get all RDS instances (VPC based filtering is not supported yet)159 rds_client = _get_client('rds', region)160 _clients[region]["rds"] = rds_client161 _functions[region]["RdsInstances"] = (rds_client.describe_db_instances, dict())162 elbv2_client = _get_client('elbv2', region)163 _clients[region]["elbv2"] = elbv2_client164 _functions[region]["LoadBalancers"] = (elbv2_client.describe_load_balancers, dict())165 _functions[region]["TargetGroups"] = (elbv2_client.describe_target_groups, dict())166 if skip_data is not None:167 for key in skip_data:168 _functions[region].pop(key)169def aws_test_access(profile=None):170 try:171 session = boto3.session.Session(profile_name=profile)172 client = session.client("ec2", "us-west-1")173 client.describe_regions()174 print("You have access to AWS!\n")175 except Exception as e:...
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!!