Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge pull request #138 from IlyaSukhanov/master

Browse Source 
Implement internet gateway mocking
This commit is contained in:
Steve Pulec 2014-06-17 22:28:16 -04:00
commit aec7d8e998
6 changed files with 298 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

27
moto/ec2/exceptions.py
View File

@ -1,6 +1,7 @@
from werkzeug.exceptions import BadRequest
from jinja2 import Template
class InvalidIdError(RuntimeError):
def __init__(self, id_value):
super(InvalidIdError, self).__init__()
@ -40,10 +41,32 @@ class InvalidParameterValueError(EC2ClientError):
def __init__(self, parameter_value):
super(InvalidParameterValueError, self).__init__(
"InvalidParameterValue",
"Value ({0}) for parameter value is invalid. Invalid DHCP option value.".format(
parameter_value))
"Value {0} is invalid for parameter."
.format(parameter_value))
class InvalidInternetGatewayIDError(EC2ClientError):
def __init__(self, internet_gateway_id):
super(InvalidInternetGatewayIDError, self).__init__(
"InvalidInternetGatewayID.NotFound",
"InternetGatewayID {0} does not exist."
.format(internet_gateway_id))
class GatewayNotAttachedError(EC2ClientError):
def __init__(self, internet_gateway_id, vpc_id):
super(GatewayNotAttachedError, self).__init__(
"Gateway.NotAttached",
"InternetGatewayID {0} is not attached to a VPC {1}."
.format(internet_gateway_id, vpc_id))
class ResourceAlreadyAssociatedError(EC2ClientError):
def __init__(self, resource):
super(ResourceAlreadyAssociatedError, self).__init__(
"Resource.AlreadyAssociated",
"Resource {0} is already associated."
.format(str(resource)))
ERROR_RESPONSE = u"""<?xml version="1.0" encoding="UTF-8"?>

68
moto/ec2/models.py
View File

@ -8,15 +8,20 @@ from moto.core import BaseBackend
from .exceptions import (
InvalidIdError,
DependencyViolationError,
InvalidDHCPOptionsIdError
InvalidDHCPOptionsIdError,
InvalidInternetGatewayIDError,
GatewayNotAttachedError,
ResourceAlreadyAssociatedError,
InvalidVPCIdError
)
from .utils import (
random_ami_id,
random_dhcp_option_id,
random_eip_allocation_id,
random_eip_association_id,
random_gateway_id,
random_internet_gateway_id,
random_instance_id,
random_internet_gateway_id,
random_ip,
random_key_pair,
random_reservation_id,
@ -691,6 +696,8 @@ class VPCBackend(object):
return vpc
def get_vpc(self, vpc_id):
if vpc_id not in self.vpcs:
raise InvalidVPCIdError(vpc_id)
return self.vpcs.get(vpc_id)
def get_all_vpcs(self):
@ -838,13 +845,14 @@ class RouteBackend(object):
return route
class InternetGateway(object):
def __init__(self, gateway_id):
self.id = gateway_id
class InternetGateway(TaggedEC2Instance):
def __init__(self):
self.id = random_internet_gateway_id()
self.vpc = None
@classmethod
def create_from_cloudformation_json(cls, resource_name, cloudformation_json):
return ec2_backend.create_gateway()
return ec2_backend.create_internet_gateway()
@property
def physical_resource_id(self):
@ -853,14 +861,50 @@ class InternetGateway(object):
class InternetGatewayBackend(object):
def __init__(self):
self.gateways = {}
self.internet_gateways = {}
super(InternetGatewayBackend, self).__init__()
def create_gateway(self):
gateway_id = random_gateway_id()
gateway = InternetGateway(gateway_id)
self.gateways[gateway_id] = gateway
return gateway
def create_internet_gateway(self):
igw = InternetGateway()
self.internet_gateways[igw.id] = igw
return igw
def describe_internet_gateways(self, internet_gateway_ids=None):
igws = []
for igw_id in internet_gateway_ids or []:
if igw_id in self.internet_gateways:
igws.append(self.internet_gateways[igw_id])
else:
raise InvalidInternetGatewayIDError(igw_id)
return igws or self.internet_gateways.values()
def delete_internet_gateway(self, internet_gateway_id):
igw_ids = [internet_gateway_id]
igw = self.describe_internet_gateways(internet_gateway_ids=igw_ids)[0]
if igw.vpc:
raise DependencyViolationError(
"{0} is being utilized by {1}"
.format(internet_gateway_id, igw.vpc)
)
self.internet_gateways.pop(internet_gateway_id)
return True
def detach_internet_gateway(self, internet_gateway_id, vpc_id):
igw_ids = [internet_gateway_id]
igw = self.describe_internet_gateways(internet_gateway_ids=igw_ids)[0]
if not igw.vpc or igw.vpc.id != vpc_id:
raise GatewayNotAttachedError(internet_gateway_id, vpc_id)
igw.vpc = None
return True
def attach_internet_gateway(self, internet_gateway_id, vpc_id):
igw_ids = [internet_gateway_id]
igw = self.describe_internet_gateways(internet_gateway_ids=igw_ids)[0]
if igw.vpc:
raise ResourceAlreadyAssociatedError(igw)
vpc = self.get_vpc(vpc_id)
igw.vpc = vpc
return True
class VPCGatewayAttachment(object):

2
moto/ec2/responses/dhcp_options.py
View File

@ -20,8 +20,6 @@ class DHCPOptions(BaseResponse):
dhcp_opt = ec2_backend.describe_dhcp_options([dhcp_opt_id])[0]
vpc = ec2_backend.get_vpc(vpc_id)
if not vpc:
raise InvalidVPCIdError(vpc_id)
ec2_backend.associate_dhcp_options(dhcp_opt, vpc)

106
moto/ec2/responses/internet_gateways.py
View File

@ -1,18 +1,112 @@
from moto.core.responses import BaseResponse
from moto.ec2.models import ec2_backend
from moto.ec2.utils import sequence_from_querystring
from jinja2 import Template
class InternetGateways(BaseResponse):
def attach_internet_gateway(self):
raise NotImplementedError('InternetGateways(AmazonVPC).attach_internet_gateway is not yet implemented')
igw_id = self.querystring.get("InternetGatewayId", [None])[0]
vpc_id = self.querystring.get("VpcId", [None])[0]
ec2_backend.attach_internet_gateway(igw_id, vpc_id)
template = Template(ATTACH_INTERNET_GATEWAY_RESPONSE)
return template.render()
def create_internet_gateway(self):
raise NotImplementedError('InternetGateways(AmazonVPC).create_internet_gateway is not yet implemented')
igw = ec2_backend.create_internet_gateway()
template = Template(CREATE_INTERNET_GATEWAY_RESPONSE)
return template.render(internet_gateway=igw)
def delete_internet_gateway(self):
raise NotImplementedError('InternetGateways(AmazonVPC).delete_internet_gateway is not yet implemented')
igw_id = self.querystring.get("InternetGatewayId", [None])[0]
ec2_backend.delete_internet_gateway(igw_id)
template = Template(DELETE_INTERNET_GATEWAY_RESPONSE)
return template.render()
def describe_internet_gateways(self):
raise NotImplementedError('InternetGateways(AmazonVPC).describe_internet_gateways is not yet implemented')
if "Filter.1.Name" in self.querystring:
raise NotImplementedError(
"Filtering not supported in describe_internet_gateways.")
elif "InternetGatewayId.1" in self.querystring:
igw_ids = sequence_from_querystring(
"InternetGatewayId", self.querystring)
igws = ec2_backend.describe_internet_gateways(igw_ids)
else:
igws = ec2_backend.describe_internet_gateways()
template = Template(DESCRIBE_INTERNET_GATEWAYS_RESPONSE)
return template.render(internet_gateways=igws)
def detach_internet_gateway(self):
raise NotImplementedError('InternetGateways(AmazonVPC).detach_internet_gateway is not yet implemented')
# TODO validate no instances with EIPs in VPC before detaching
# raise else DependencyViolationError()
igw_id = self.querystring.get("InternetGatewayId", [None])[0]
vpc_id = self.querystring.get("VpcId", [None])[0]
ec2_backend.detach_internet_gateway(igw_id, vpc_id)
template = Template(DETACH_INTERNET_GATEWAY_RESPONSE)
return template.render()
ATTACH_INTERNET_GATEWAY_RESPONSE = u"""<AttachInternetGatewayResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</AttachInternetGatewayResponse>"""
CREATE_INTERNET_GATEWAY_RESPONSE = u"""<CreateInternetGatewayResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<internetGateway>
<internetGatewayId>{{ internet_gateway.id }}</internetGatewayId>
<attachmentSet/>
<tagSet>
{% for tag in internet_gateway.get_tags() %}
<item>
<resourceId>{{ tag.resource_id }}</resourceId>
<resourceType>{{ tag.resource_type }}</resourceType>
<key>{{ tag.key }}</key>
<value>{{ tag.value }}</value>
</item>
{% endfor %}
</tagSet>
</internetGateway>
</CreateInternetGatewayResponse>"""
DELETE_INTERNET_GATEWAY_RESPONSE = u"""<DeleteInternetGatewayResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</DeleteInternetGatewayResponse>"""
DESCRIBE_INTERNET_GATEWAYS_RESPONSE = u"""<DescribeInternetGatewaysResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-
15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<internetGatewaySet>
{% for igw in internet_gateways %}
<item>
<internetGatewayId>{{ igw.id }}</internetGatewayId>
{% if igw.vpc %}
<attachmentSet>
<item>
<vpcId>{{ igw.vpc.id }}</vpcId>
<state>available</state>
</item>
</attachmentSet>
{% else %}
<attachmentSet/>
{% endif %}
<tagSet>
{% for tag in igw.get_tags() %}
<item>
<resourceId>{{ tag.resource_id }}</resourceId>
<resourceType>{{ tag.resource_type }}</resourceType>
<key>{{ tag.key }}</key>
<value>{{ tag.value }}</value>
</item>
{% endfor %}
</tagSet>
</item>
{% endfor %}
</internetGatewaySet>
</DescribeInternetGatewaysResponse>"""
DETACH_INTERNET_GATEWAY_RESPONSE = u"""<DetachInternetGatewayResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
<requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
<return>true</return>
</DetachInternetGatewayResponse>"""

4
moto/ec2/utils.py
View File

@ -50,7 +50,7 @@ def random_eip_association_id():
return random_id(prefix='eipassoc')
def random_gateway_id():
def random_internet_gateway_id():
return random_id(prefix='igw')
@ -157,7 +157,7 @@ def dhcp_configuration_from_querystring(querystring, option=u'DhcpConfiguration'
def filters_from_querystring(querystring_dict):
response_values = {}
for key, value in querystring_dict.iteritems():
match = re.search("Filter.(\d).Name", key)
match = re.search(r"Filter.(\d).Name", key)
if match:
filter_index = match.groups()[0]
value_prefix = "Filter.{0}.Value".format(filter_index)

110
tests/test_ec2/test_internet_gateways.py
View File

@ -1,9 +1,115 @@
import re
import boto
from boto.exception import EC2ResponseError
import sure # noqa
from moto import mock_ec2
VPC_CIDR="10.0.0.0/16"
BAD_VPC="vpc-deadbeef"
BAD_IGW="igw-deadbeef"
@mock_ec2
def test_internet_gateways():
pass
def test_igw_create():
""" internet gateway create """
conn = boto.connect_vpc('the_key', 'the_secret')
conn.get_all_internet_gateways().should.have.length_of(0)
igw = conn.create_internet_gateway()
conn.get_all_internet_gateways().should.have.length_of(1)
igw.id.should.match(r'igw-[0-9a-f]+')
igw = conn.get_all_internet_gateways()[0]
igw.attachments.should.have.length_of(0)
@mock_ec2
def test_igw_attach():
""" internet gateway attach """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
igw = conn.get_all_internet_gateways()[0]
igw.attachments[0].vpc_id.should.be.equal(vpc.id)
@mock_ec2
def test_igw_attach_bad_vpc():
""" internet gateway fail to attach w/ bad vpc """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
conn.attach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
@mock_ec2
def test_igw_attach_twice():
""" internet gateway fail to attach twice """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc1 = conn.create_vpc(VPC_CIDR)
vpc2 = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc1.id)
conn.attach_internet_gateway.when.called_with(igw.id, vpc2.id).should.throw(EC2ResponseError)
@mock_ec2
def test_igw_detach():
""" internet gateway detach"""
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
conn.detach_internet_gateway(igw.id, vpc.id)
igw = conn.get_all_internet_gateways()[0]
igw.attachments.should.have.length_of(0)
@mock_ec2
def test_igw_detach_bad_vpc():
""" internet gateway fail to detach w/ bad vpc """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
conn.detach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
@mock_ec2
def test_igw_detach_unattached():
""" internet gateway fail to detach unattached """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
conn.detach_internet_gateway.when.called_with(igw.id, BAD_VPC).should.throw(EC2ResponseError)
@mock_ec2
def test_igw_delete():
""" internet gateway delete"""
conn = boto.connect_vpc('the_key', 'the_secret')
vpc = conn.create_vpc(VPC_CIDR)
conn.get_all_internet_gateways().should.have.length_of(0)
igw = conn.create_internet_gateway()
conn.get_all_internet_gateways().should.have.length_of(1)
conn.delete_internet_gateway(igw.id)
conn.get_all_internet_gateways().should.have.length_of(0)
@mock_ec2
def test_igw_delete_attached():
""" internet gateway fail to delete attached """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
vpc = conn.create_vpc(VPC_CIDR)
conn.attach_internet_gateway(igw.id, vpc.id)
conn.delete_internet_gateway.when.called_with(igw.id).should.throw(EC2ResponseError)
@mock_ec2
def test_igw_desribe():
""" internet gateway fetch by id """
conn = boto.connect_vpc('the_key', 'the_secret')
igw = conn.create_internet_gateway()
igw_by_search = conn.get_all_internet_gateways([igw.id])[0]
igw.id.should.equal(igw_by_search.id)
@mock_ec2
def test_igw_desribe_bad_id():
""" internet gateway fail to fetch by bad id """
conn = boto.connect_vpc('the_key', 'the_secret')
conn.get_all_internet_gateways.when.called_with([BAD_IGW]).should.throw(EC2ResponseError)