cloud-init-dev team mailing list archive

[Ryan Harper <ryan.harper@xxxxxxxxxxxxx>]

Replaced my pop() with next() in the in-line comments.

Diff comments:

> diff --git a/tests/cloud_tests/platforms/ec2/platform.py b/tests/cloud_tests/platforms/ec2/platform.py
> new file mode 100644
> index 0000000..8ff2716
> --- /dev/null
> +++ b/tests/cloud_tests/platforms/ec2/platform.py
> @@ -0,0 +1,255 @@
> +# This file is part of cloud-init. See LICENSE file for license information.
> +
> +"""Base EC2 platform."""
> +import os
> +import time
> +import uuid
> +
> +import boto3
> +import botocore
> +
> +from ..platforms import Platform
> +from .image import EC2Image
> +from .instance import EC2Instance
> +from tests.cloud_tests import util
> +
> +
> +class EC2Platform(Platform):
> +    """EC2 test platform."""
> +
> +    platform_name = 'ec2'
> +    ipv4_cidr = '192.168.1.0/20'
> +
> +    def __init__(self, config):
> +        """Set up platform."""
> +        super(EC2Platform, self).__init__(config)
> +        # Used with SSH key and custom AMI generation naming with EC2
> +        self.uuid = str(uuid.uuid1())[0:8]
> +        self.tag = config['tag']
> +
> +        self.ec2_client = boto3.client('ec2')
> +        self.ec2_resource = boto3.resource('ec2')
> +        self.instance_type = config['instance-type']
> +        self.key_name = self._upload_public_key(config)
> +        self.vpc = self._setup_vpc()
> +        self.vpc_security_group = self._get_security_group()
> +        self.vpc_subnet = self._get_subnet()
> +
> +    def create_instance(self, properties, config, features,
> +                        image_ami, user_data):
> +        """Create an instance
> +
> +        @param src_img_path: image path to launch from
> +        @param properties: image properties
> +        @param config: image configuration
> +        @param features: image features
> +        @param image_ami: string of image ami ID
> +        @param user_data: test user-data to pass to instance
> +        @return_value: cloud_tests.instances instance
> +        """
> +        return EC2Instance(self, properties, config, features,
> +                           image_ami, user_data)
> +
> +    def destroy(self):
> +        """Delete platform."""
> +        if self.key_name:
> +            self.ec2_client.delete_key_pair(KeyName=self.key_name)
> +
> +    def get_image(self, img_conf):
> +        """Get image using specified image configuration.
> +
> +        @param img_conf: configuration for image
> +        @return_value: cloud_tests.images instance
> +        """
> +        if img_conf['root-store'] == 'ebs':
> +            image_type = 'hvm-ssd'
> +        elif img_conf['root-store'] == 'instance-store':
> +            image_type = 'hvm-instance'
> +        else:
> +            raise RuntimeError('Unknown root-store type: %s' %
> +                               (img_conf['root-store']))
> +
> +        image_filter = {
> +            'Name': 'name',
> +            'Values': ['ubuntu/images-testing/%s/ubuntu-%s-daily-%s-server-*'
> +                       % (image_type, img_conf['release'], img_conf['arch'])]
> +        }
> +        response = self.ec2_client.describe_images(Filters=[image_filter])
> +        images = sorted(response['Images'], key=lambda k: k['CreationDate'])
> +
> +        try:
> +            image_ami = images[-1]['ImageId']
> +        except IndexError:
> +            raise RuntimeError('No images found for %s!' % img_conf['release'])
> +
> +        image = EC2Image(self, img_conf, image_ami)
> +        return image
> +
> +    @staticmethod
> +    def wait_for_state(resource, goal_state):
> +        """Wait up to 5 minutes (5 * 60) for a specific state.
> +
> +        @param: resource to check with resource.state
> +        @param: string goal state to be in
> +        """
> +        attempts = 0
> +
> +        while attempts < 60:
> +            try:
> +                resource.reload()
> +                current_state = resource.state
> +                # Sometimes get {'Code': 0, 'Name': 'pending'}
> +                if isinstance(current_state, dict):
> +                    current_state = current_state['Name']
> +
> +                if current_state == goal_state:
> +                    return
> +            # will occur if instance is not up yet
> +            except botocore.exception.BotoSeverError:
> +                pass
> +
> +            time.sleep(5)
> +            attempts += 1
> +
> +        raise util.InTargetExecuteError(
> +            b'', b'', 1, goal_state, 'wait_for_state',
> +            reason='%s: took too long to get to %s' % (resource, goal_state)
> +        )
> +
> +    def _create_internet_gateway(self, vpc):
> +        """Create Internet Gateway and assign to VPC.
> +
> +        @param vpc: VPC to create internet gateway on
> +        """
> +        internet_gateway = self.ec2_resource.create_internet_gateway()
> +        internet_gateway.attach_to_vpc(VpcId=vpc.vpc_id)
> +        self._tag_resource(internet_gateway)
> +        return internet_gateway
> +
> +    def _create_subnet(self, vpc):
> +        """Generate IPv4 and IPv6 subnets for use.
> +
> +        @param vpc: VPC to create subnets on
> +        """
> +        ipv6_block = vpc.ipv6_cidr_block_association_set[0]['Ipv6CidrBlock']
> +        ipv6_cidr = ipv6_block[:-2] + '64'
> +
> +        subnet = vpc.create_subnet(CidrBlock=self.ipv4_cidr,
> +                                   Ipv6CidrBlock=ipv6_cidr)
> +        modify_subnet = subnet.meta.client.modify_subnet_attribute
> +        modify_subnet(SubnetId=subnet.id,
> +                      MapPublicIpOnLaunch={'Value': True})
> +        self._tag_resource(subnet)
> +
> +    def _get_security_group(self):
> +        """Return default security group from VPC.
> +
> +        For now there should only be one.
> +
> +        @return_value: Return first security group object
> +        """
> +        return list(self.vpc.security_groups.all())[0]

I mean next(); we just need to consume the first element from the collection.

> +
> +    def _get_subnet(self):
> +        """Return first subnet from VPC.
> +
> +        For now there should only be one.
> +
> +        @return_value: Return subnet object
> +        """
> +        return list(self.vpc.subnets.all())[0]

next();

> +
> +    def _setup_vpc(self):
> +        """Setup AWS EC2 VPC or return existing VPC."""
> +        for vpc in self.ec2_resource.vpcs.all():
> +            if not vpc.tags:
> +                continue
> +            for tag in vpc.tags:
> +                if tag['Value'] == self.tag:
> +                    return vpc

next()

> +
> +        vpc = self.ec2_resource.create_vpc(
> +            CidrBlock=self.ipv4_cidr,
> +            AmazonProvidedIpv6CidrBlock=True
> +        )
> +        vpc.wait_until_available()
> +        self._tag_resource(vpc)
> +
> +        internet_gateway = self._create_internet_gateway(vpc)
> +        self._create_subnet(vpc)
> +        self._update_routing_table(vpc, internet_gateway.id)
> +        self._update_security_group(vpc)
> +
> +        return vpc
> +
> +    def _tag_resource(self, resource):
> +        """Tag a resouce with the specified tag.
> +
> +        This makes finding and deleting resources specific to this testing
> +        much easier to find.
> +
> +        @param resource: resource to tag"""
> +        tag = {
> +            'Key': 'Name',
> +            'Value': self.tag
> +        }
> +        resource.create_tags(Tags=[tag])
> +
> +    def _update_routing_table(self, vpc, internet_gateway_id):
> +        """Update default routing table with internet gateway.
> +
> +        This sets up internet access between the VPC via the internet gateway
> +        by configuring routing tables for IPv4 and IPv6.
> +
> +        @param vpc: VPC containing routing table to update
> +        @param internet_gateway_id: gateway ID to use for routing
> +        """
> +        route_table = list(vpc.route_tables.all())[0]
> +        route_table.create_route(DestinationCidrBlock='0.0.0.0/0',
> +                                 GatewayId=internet_gateway_id)
> +        route_table.create_route(DestinationIpv6CidrBlock='::/0',
> +                                 GatewayId=internet_gateway_id)
> +        self._tag_resource(route_table)
> +
> +    def _update_security_group(self, vpc):
> +        """Allow only SSH inbound to default VPC security group.
> +
> +        This revokes the initial accept all permissions and only allows
> +        the SSH inbound.
> +
> +        @param vpc: VPC containing security group to update
> +        """
> +        ssh = {
> +            'IpProtocol': 'TCP',
> +            'FromPort': 22,
> +            'ToPort': 22,
> +            'IpRanges': [{'CidrIp': '0.0.0.0/0'}],
> +            'Ipv6Ranges': [{'CidrIpv6': '::/0'}]
> +        }
> +
> +        security_group = list(vpc.security_groups.all())[0]
> +        security_group.revoke_ingress(
> +            IpPermissions=security_group.ip_permissions
> +        )
> +        security_group.authorize_ingress(
> +            IpPermissions=[ssh]
> +        )
> +        self._tag_resource(security_group)
> +
> +    def _upload_public_key(self, config):
> +        """Generate random name and upload SSH key with that name.
> +
> +        @param config: platform config
> +        @return: string of ssh key name
> +        """
> +        key_file = os.path.join(config['data_dir'], config['public_key'])
> +        with open(key_file, 'r') as file:
> +            public_key = file.read().strip('\n')
> +
> +        name = '%s-%s' % (self.tag, self.uuid)
> +        self.ec2_client.import_key_pair(KeyName=name,
> +                                        PublicKeyMaterial=public_key)
> +
> +        return name
> +
> +# vi: ts=4 expandtab


-- 
https://code.launchpad.net/~powersj/cloud-init/+git/cloud-init/+merge/335186
Your team cloud-init commiters is requested to review the proposed merge of ~powersj/cloud-init:cii-enable-ec2 into cloud-init:master.