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

How to use ec2_client method in localstack

Best Python code snippet using localstack_python

vpc_test_utils.py

Source:vpc_test_utils.py

1# Copyright 2017-2019 Amazon.com, Inc. or its affiliates. All Rights Reserved.2#3# Licensed under the Apache License, Version 2.0 (the "License"). You4# may not use this file except in compliance with the License. A copy of5# the License is located at6#7#     http://aws.amazon.com/apache2.0/8#9# or in the "license" file accompanying this file. This file is10# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF11# ANY KIND, either express or implied. See the License for the specific12# language governing permissions and limitations under the License.13from __future__ import absolute_import14import os15import tempfile16import tests.integ.lock as lock17VPC_NAME = "sagemaker-python-sdk-test-vpc"18LOCK_PATH = os.path.join(tempfile.gettempdir(), "sagemaker_test_vpc_lock")19LOCK_PATH_EFS = os.path.join(tempfile.gettempdir(), "sagemaker_efs_fsx_vpc_lock")20def _get_subnet_ids_by_name(ec2_client, name):21    desc = ec2_client.describe_subnets(Filters=[{"Name": "tag-value", "Values": [name]}])22    if len(desc["Subnets"]) == 0:23        return None24    else:25        return [subnet["SubnetId"] for subnet in desc["Subnets"]]26def _get_security_id_by_name(ec2_client, name):27    desc = ec2_client.describe_security_groups(Filters=[{"Name": "tag-value", "Values": [name]}])28    if len(desc["SecurityGroups"]) == 0:29        return None30    else:31        return desc["SecurityGroups"][0]["GroupId"]32def _security_group_ids_by_vpc_id(sagemaker_session, vpc_id):33    ec2_resource = sagemaker_session.boto_session.resource("ec2")34    security_group_ids = []35    vpc = ec2_resource.Vpc(vpc_id)36    for sg in vpc.security_groups.all():37        security_group_ids.append(sg.id)38    return security_group_ids39def _vpc_exists(ec2_client, name):40    desc = ec2_client.describe_vpcs(Filters=[{"Name": "tag-value", "Values": [name]}])41    return len(desc["Vpcs"]) > 042def _vpc_id_by_name(ec2_client, name):43    desc = ec2_client.describe_vpcs(Filters=[{"Name": "tag-value", "Values": [name]}])44    vpc_id = desc["Vpcs"][0]["VpcId"]45    return vpc_id46def _route_table_id(ec2_client, vpc_id):47    desc = ec2_client.describe_route_tables(Filters=[{"Name": "vpc-id", "Values": [vpc_id]}])48    return desc["RouteTables"][0]["RouteTableId"]49def check_or_create_vpc_resources_efs_fsx(sagemaker_session, name=VPC_NAME):50    # use lock to prevent race condition when tests are running concurrently51    with lock.lock(LOCK_PATH_EFS):52        ec2_client = sagemaker_session.boto_session.client("ec2")53        if _vpc_exists(ec2_client, name):54            vpc_id = _vpc_id_by_name(ec2_client, name)55            return (56                _get_subnet_ids_by_name(ec2_client, name),57                _security_group_ids_by_vpc_id(sagemaker_session, vpc_id),58            )59        else:60            return _create_vpc_with_name_efs_fsx(ec2_client, name)61def _create_vpc_with_name_efs_fsx(ec2_client, name):62    vpc_id, [subnet_id_a, subnet_id_b], security_group_id = _create_vpc_resources(ec2_client, name)63    ec2_client.modify_vpc_attribute(EnableDnsHostnames={"Value": True}, VpcId=vpc_id)64    ig = ec2_client.create_internet_gateway()65    internet_gateway_id = ig["InternetGateway"]["InternetGatewayId"]66    ec2_client.attach_internet_gateway(InternetGatewayId=internet_gateway_id, VpcId=vpc_id)67    route_table_id = _route_table_id(ec2_client, vpc_id)68    ec2_client.create_route(69        DestinationCidrBlock="0.0.0.0/0", GatewayId=internet_gateway_id, RouteTableId=route_table_id70    )71    ec2_client.associate_route_table(RouteTableId=route_table_id, SubnetId=subnet_id_a)72    ec2_client.associate_route_table(RouteTableId=route_table_id, SubnetId=subnet_id_b)73    ec2_client.authorize_security_group_ingress(74        GroupId=security_group_id,75        IpPermissions=[76            {77                "IpProtocol": "tcp",78                "FromPort": 988,79                "ToPort": 988,80                "UserIdGroupPairs": [{"GroupId": security_group_id}],81            },82            {83                "IpProtocol": "tcp",84                "FromPort": 2049,85                "ToPort": 2049,86                "UserIdGroupPairs": [{"GroupId": security_group_id}],87            },88            {89                "IpProtocol": "tcp",90                "FromPort": 22,91                "ToPort": 22,92                "IpRanges": [{"CidrIp": "0.0.0.0/0", "Description": "For SSH to EC2"}],93            },94        ],95    )96    return [subnet_id_a], [security_group_id]97def _create_vpc_resources(ec2_client, name):98    vpc_id = ec2_client.create_vpc(CidrBlock="10.0.0.0/16")["Vpc"]["VpcId"]99    ec2_client.create_tags(Resources=[vpc_id], Tags=[{"Key": "Name", "Value": name}])100    availability_zone_name = ec2_client.describe_availability_zones()["AvailabilityZones"][0][101        "ZoneName"102    ]103    subnet_id_a = ec2_client.create_subnet(104        CidrBlock="10.0.0.0/24", VpcId=vpc_id, AvailabilityZone=availability_zone_name105    )["Subnet"]["SubnetId"]106    print("created subnet: {}".format(subnet_id_a))107    subnet_id_b = ec2_client.create_subnet(108        CidrBlock="10.0.1.0/24", VpcId=vpc_id, AvailabilityZone=availability_zone_name109    )["Subnet"]["SubnetId"]110    print("created subnet: {}".format(subnet_id_b))111    s3_service = [112        s for s in ec2_client.describe_vpc_endpoint_services()["ServiceNames"] if s.endswith("s3")113    ][0]114    ec2_client.create_vpc_endpoint(115        VpcId=vpc_id, ServiceName=s3_service, RouteTableIds=[_route_table_id(ec2_client, vpc_id)]116    )117    print("created s3 vpc endpoint")118    security_group_id = ec2_client.create_security_group(119        VpcId=vpc_id, GroupName=name, Description=name120    )["GroupId"]121    print("created security group: {}".format(security_group_id))122    # multi-host vpc jobs require communication among hosts123    ec2_client.authorize_security_group_ingress(124        GroupId=security_group_id,125        IpPermissions=[126            {127                "IpProtocol": "tcp",128                "FromPort": 0,129                "ToPort": 65535,130                "UserIdGroupPairs": [{"GroupId": security_group_id}],131            }132        ],133    )134    ec2_client.create_tags(135        Resources=[subnet_id_a, subnet_id_b, security_group_id],136        Tags=[{"Key": "Name", "Value": name}],137    )138    return vpc_id, [subnet_id_a, subnet_id_b], security_group_id139def _create_vpc_with_name(ec2_client, name):140    vpc_id, [subnet_id_a, subnet_id_b], security_group_id = _create_vpc_resources(ec2_client, name)141    return [subnet_id_a, subnet_id_b], security_group_id142def get_or_create_vpc_resources(ec2_client):143    # use lock to prevent race condition when tests are running concurrently144    with lock.lock(LOCK_PATH):145        if _vpc_exists(ec2_client, VPC_NAME):146            print("using existing vpc: {}".format(VPC_NAME))147            return (148                _get_subnet_ids_by_name(ec2_client, VPC_NAME),149                _get_security_id_by_name(ec2_client, VPC_NAME),150            )151        else:152            print("creating new vpc: {}".format(VPC_NAME))153            return _create_vpc_with_name(ec2_client, VPC_NAME)154def setup_security_group_for_encryption(ec2_client, security_group_id):155    sg_desc = ec2_client.describe_security_groups(GroupIds=[security_group_id])156    ingress_perms = sg_desc["SecurityGroups"][0]["IpPermissions"]157    if len(ingress_perms) == 1:158        ec2_client.authorize_security_group_ingress(159            GroupId=security_group_id,160            IpPermissions=[161                {"IpProtocol": "50", "UserIdGroupPairs": [{"GroupId": security_group_id}]},162                {163                    "IpProtocol": "udp",164                    "FromPort": 500,165                    "ToPort": 500,166                    "UserIdGroupPairs": [{"GroupId": security_group_id}],167                },168            ],...

main.py

Source:main.py

1import click2import boto33import os4import time5import progressbar6from datetime import datetime, timedelta, timezone7from pyfiglet import Figlet89# Hello awsegy10f = Figlet(font='slant')11print(f.renderText('awsegy'))1213# Config(ec2.instnace)14ec2_client = boto3.client('ec2', region_name = 'ap-northeast-2')15ecs_client = boto3.client('ecs', region_name = 'ap-northeast-2')16171819response = ecs_client.describe_task_definition(20    taskDefinition = 'ecs-test'21)22result = response.get('taskDefinition')23last = result.get('taskDefinitionArn')2425get_task = last[-10:]2627# List(ec2.Snapshot)28# snapshots = ec2_client.snapshots.filter(OwnerIds=['self'])2930# Option(code.function)31def Search(Search): 32    split_idx = Search.split(':')33    custom_filter = [{'Name':'tag:{}'.format(split_idx[0]), 'Values': ['{}'.format(split_idx[1])]}]34    response = ec2_client.describe_instances(Filters=custom_filter)35    return response['Reservations']3637def ProgressBar(Number):38    bar = progressbar.ProgressBar(widgets=[' [', progressbar.Timer(), '] ', progressbar.Bar(), ' (', progressbar.ETA(), ') ',])39    for i in bar(range(Number)):40        time.sleep(0.1)414243# Code(main)44@click.group()45def cli():46    '''47    Interactive CLI tool 0.1.4e48    '''49    pass505152@click.command(help='Check your instance information!      ex)"awsegy list tag:Name"')53@click.argument('list')54def list(list):55    try:56        for instance in Search(list):57            print('\nInstance Id : '    + instance['Instances'][0]['InstanceId'])58            print('Publice Ip : '     + instance['Instances'][0]['PublicIpAddress'])59            print('Private Ip : '     + instance['Instances'][0]['PrivateIpAddress'])60            print('Instance State : ' + instance['Instances'][0]['State']['Name']+'\n')61    except:62        print('Oops! There''s no such name.')636465@click.command(help='The instance type to change. ex) "awsegy change tag:Name Instance Type"')66@click.argument('change', nargs=2)67def change(change):68    for instance in Search(change[0]):69        ids = []70        id = instance['Instances'][0]['InstanceId']71        ids.append(id)72        if instance['Instances'][0]['State']['Name'] == 'running':73            print(f'Instance Stop Processing : {ids}')74            ec2_client.stop_instances(InstanceIds=ids)75            waiter  = ec2_client.get_waiter('instance_stopped')76            ProgressBar(300)77            ec2_client.modify_instance_attribute(InstanceId=id, Attribute='instanceType', Value=change[1])78            ec2_client.start_instances(InstanceIds=ids)79            print(f'Instance Running : {ids}')80        else:81            ec2_client.modify_instance_attribute(InstanceId=id, Attribute='instanceType', Value=change[1])82            ec2_client.start_instances(InstanceIds=ids)83            print(f'Instance Running : {ids}')848586@click.command(help='Tagging Instance!        ex)"awsegy tag all tag:Name"')87@click.argument('tag' , nargs=2)88def tag(tag):89    split_idx = tag[1].split(':')90    response = ec2_client.describe_instances()91    instances = response['Reservations']92    instance_ids = []93    if tag[0] == 'all':94        for instance in instances:95            instance_ids.append(instance['Instances'][0]['InstanceId'])96            tage_creation = ec2_client.create_tags(97            Resources = instance_ids, 98            Tags = [{'Key' : f'{split_idx[0]}', 'Value' : f'{split_idx[1]}',}])99            print(f'Finished Instance Count : {len(instance_ids)}')100    else:101        for instance in instances:102            custom_filter = [{'Name':'tag:Name', 'Values': ['{}'.format(tag[0])]}]103            response = ec2_client.describe_instances(Filters=custom_filter)104            instance_ids.append(instance['Instances'][0]['InstanceId'])105            tage_creation = ec2_client.create_tags(106            Resources = instance_ids, 107            Tags = [{'Key' : f'{split_idx[0]}', 'Value' : f'{split_idx[1]}',}])108            print(f'Finished Instance Count : {len(instance_ids)}')109            110111        112        113        114@click.command(help='Untagging Instance!        ex)"awsegy dtag all tag:Name"')115@click.argument('dtag' , nargs=2)116def dtag(dtag):117    split_idx = dtag[1].split(':')118    response = ec2_client.describe_instances()119    instances = response['Reservations']120    instance_ids = []121    if dtag[0] == 'all':122        for instance in instances:123            instance_ids.append(instance['Instances'][0]['InstanceId'])124            tage_creation = ec2_client.delete_tags(125            Resources = instance_ids, 126            Tags = [{'Key' : f'{split_idx[0]}', 'Value' : f'{split_idx[1]}',}])127            print(f'Finished Instance Count : {len(instance_ids)}')128    else:129        for instance in instances:130            custom_filter = [{'Name':'tag:Name', 'Values': ['{}'.format(dtag[0])]}]131            response = ec2_client.describe_instances(Filters=custom_filter)132            instance_ids.append(instance['Instances'][0]['InstanceId'])133            tage_creation = ec2_client.delete_tags(134            Resources = instance_ids, 135            Tags = [{'Key' : f'{split_idx[0]}', 'Value' : f'{split_idx[1]}',}])136            print(f'Finished Instance Count : {len(instance_ids)}')137138139        140@click.command(help='ECS Continuous integration        ex)"awsegy <docker-hub/tag:name>')141@click.argument('ci')142def ci(ci):143    print(f'Build start {ci}')144    os.system(f'docker build -t {ci} .')145    os.system(f'docker push {ci}')146    os.system(f'aws ecs register-task-definition --cli-input-json file://job.json')147    ProgressBar(500)148    os.system(f'aws ecs update-service --cluster knowre --service knowre-service --task-definition {get_task}')149    print(f'Finshed Task {get_task} !!')150    151        152153154155def main():156    cli.add_command(list)157    cli.add_command(change)158    cli.add_command(tag)159    cli.add_command(dtag)160    cli.add_command(ci)161    cli()162163if __name__ == "__main__":164    main()165166167168169170171172173174
...

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.