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

How to use describe_classic_link_instances method in localstack

Best Python code snippet using localstack_python

volume_cleanup.py

Source:volume_cleanup.py

...33    """ Function to retrieve all instance details """34    print "Retrieving Non-Classic Instances"35    my_instances = client.describe_instances()36    print "Retrieving Classic Instances"37    list_classic_instances = client.describe_classic_link_instances()38    for x in my_instances['Reservations']:39        i_name = x['Instances'][0]['InstanceId']40        try:41            for tag in x['Instances'][0]['Tags']:42                if tag['Key'] == "Name":43                    i_name = tag['Value']44                    # used_instances[x['Instances'][0]['InstanceId']].update({'name': tag['Value']})45        except Exception:46            i_name = x['Instances'][0]['InstanceId']47        if (x['Instances'][0]['State']['Name'] == "running" or x['Instances'][0]['InstanceId'] in instance_whitelist) and 'VpcId' in x['Instances'][0]:48            used_instances[x['Instances'][0]['InstanceId']] = {49                'id': x['Instances'][0]['InstanceId'],50                'state': x['Instances'][0]['State']['Name'],51                'type': x['Instances'][0]['InstanceType'],...

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:...

main.py

Source:main.py

...5ec2 = session.client('ec2')6def get_classic_link_resources():7    results = {}8    instance_name = 'NoName'9    resp = ec2.describe_classic_link_instances(MaxResults=1000)10    for instance in resp['Instances']:11        for tag in instance['Tags']:12            if tag['Key'] == 'Name':13                instance_name = tag['Value']14               15        if instance_name not in results.keys():16            results[instance_name] = [instance['InstanceId']]17        else:18            results[instance_name].append(instance['InstanceId'])19        print(instance_name)20            21    with open('results.json', 'w') as f:22        json.dump(results, f)23if __name__ == '__main__':...

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.