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

Merge pull request #218 from thedrow/feature/vpc-filters

Browse Source 
Added the cider block, dhcp options id and tag filter to the VPC backend
This commit is contained in:
Steve Pulec 2014-10-01 08:04:12 -04:00
commit 27ef345b64
4 changed files with 128 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
moto/ec2/models.py
View File

@ -84,6 +84,15 @@ class TaggedEC2Instance(object):
tags = ec2_backend.describe_tags(self.id)
return tags
def get_filter_value(self, filter_name):
tags = self.get_tags()
if filter_name.startswith('tag:'):
tagname = filter_name.split('tag:')[1]
for tag in tags:
if tag['key'] == tagname:
return tag['value']
class NetworkInterface(object):
def __init__(self, subnet, private_ip_address, device_index=0, public_ip_auto_assign=True, group_ids=None):
@ -1139,6 +1148,25 @@ class VPC(TaggedEC2Instance):
def physical_resource_id(self):
return self.id
def get_filter_value(self, filter_name):
if filter_name == 'cidr':
return self.cidr_block
elif filter_name == 'dhcp-options-id':
if not self.dhcp_options:
return None
return self.dhcp_options.id
filter_value = super(VPC, self).get_filter_value(filter_name)
if not filter_value:
msg = "The filter '{0}' for DescribeVPCs has not been" \
" implemented in Moto yet. Feel free to open an issue at" \
" https://github.com/spulec/moto/issues".format(filter_name)
raise NotImplementedError(msg)
return filter_value
class VPCBackend(object):
def __init__(self):
@ -1164,8 +1192,17 @@ class VPCBackend(object):
raise InvalidVPCIdError(vpc_id)
return self.vpcs.get(vpc_id)
def get_all_vpcs(self):
return self.vpcs.values()
def get_all_vpcs(self, vpc_ids=None, filters=None):
if vpc_ids:
vpcs = [vpc for vpc in self.vpcs.values() if vpc.id in vpc_ids]
else:
vpcs = self.vpcs.values()
if filters:
for (_filter, _filter_value) in filters.items():
vpcs = [ vpc for vpc in vpcs if vpc.get_filter_value(_filter) in _filter_value ]
return vpcs
def delete_vpc(self, vpc_id):
# Delete route table if only main route table remains.

5
moto/ec2/responses/vpcs.py
View File

@ -3,6 +3,7 @@ from jinja2 import Template
from moto.core.responses import BaseResponse
from moto.ec2.models import ec2_backend
from moto.ec2.utils import filters_from_querystring, vpc_ids_from_querystring
class VPCs(BaseResponse):
@ -19,7 +20,9 @@ class VPCs(BaseResponse):
return template.render(vpc=vpc)
def describe_vpcs(self):
vpcs = ec2_backend.get_all_vpcs()
vpc_ids = vpc_ids_from_querystring(self.querystring)
filters = filters_from_querystring(self.querystring)
vpcs = ec2_backend.get_all_vpcs(vpc_ids=vpc_ids, filters=filters)
template = Template(DESCRIBE_VPCS_RESPONSE)
return template.render(vpcs=vpcs)

22
moto/ec2/utils.py
View File

@ -84,6 +84,7 @@ def random_public_ip():
return '54.214.{0}.{1}'.format(random.choice(range(255)),
random.choice(range(255)))
def random_ip():
return "127.{0}.{1}.{2}".format(
random.randint(0, 255),
@ -97,7 +98,7 @@ def generate_route_id(route_table_id, cidr_block):
def split_route_id(route_id):
values = string.split(route_id,'~')
values = string.split(route_id, '~')
return values[0], values[1]
@ -125,6 +126,14 @@ def route_table_ids_from_querystring(querystring_dict):
return route_table_ids
def vpc_ids_from_querystring(querystring_dict):
vpc_ids = []
for key, value in querystring_dict.items():
if 'VpcId' in key:
vpc_ids.append(value[0])
return vpc_ids
def sequence_from_querystring(parameter, querystring_dict):
parameter_values = []
for key, value in querystring_dict.items():
@ -201,13 +210,14 @@ def filters_from_querystring(querystring_dict):
if match:
filter_index = match.groups()[0]
value_prefix = "Filter.{0}.Value".format(filter_index)
filter_values = [filter_value[0] for filter_key, filter_value in querystring_dict.items() if filter_key.startswith(value_prefix)]
filter_values = [filter_value[0] for filter_key, filter_value in querystring_dict.items() if
filter_key.startswith(value_prefix)]
response_values[value[0]] = filter_values
return response_values
def dict_from_querystring(parameter, querystring_dict):
use_dict={}
use_dict = {}
for key, value in querystring_dict.items():
match = re.search(r"{0}.(\d).(\w+)".format(parameter), key)
if match:
@ -216,7 +226,7 @@ def dict_from_querystring(parameter, querystring_dict):
if not use_dict.get(use_dict_index):
use_dict[use_dict_index] = {}
use_dict[use_dict_index][use_dict_element_property]=value[0]
use_dict[use_dict_index][use_dict_element_property] = value[0]
return use_dict
@ -249,7 +259,9 @@ def passes_filter_dict(instance, filter_dict):
if tag_value not in filter_values:
return False
else:
raise NotImplementedError("Filter dicts have not been implemented in Moto for '%s' yet. Feel free to open an issue at https://github.com/spulec/moto/issues", filter_name)
raise NotImplementedError(
"Filter dicts have not been implemented in Moto for '%s' yet. Feel free to open an issue at https://github.com/spulec/moto/issues",
filter_name)
return True

69
tests/test_ec2/test_vpcs.py
View File

@ -1,6 +1,6 @@
from __future__ import unicode_literals
# Ensure 'assert_raises' context manager support for Python 2.6
import tests.backport_assert_raises
#import tests.backport_assert_raises
from nose.tools import assert_raises
import boto
@ -9,6 +9,9 @@ import sure # noqa
from moto import mock_ec2
SAMPLE_DOMAIN_NAME = u'example.com'
SAMPLE_NAME_SERVERS = [u'10.0.0.6', u'10.0.0.7']
@mock_ec2
def test_vpcs():
@ -62,3 +65,67 @@ def test_vpc_tagging():
vpc = conn.get_all_vpcs()[0]
vpc.tags.should.have.length_of(1)
vpc.tags["a key"].should.equal("some value")
@mock_ec2
def test_vpc_get_by_id():
conn = boto.connect_vpc()
vpc1 = conn.create_vpc("10.0.0.0/16")
vpc2 = conn.create_vpc("10.0.0.0/16")
vpc3 = conn.create_vpc("10.0.0.0/16")
vpcs = conn.get_all_vpcs(vpc_ids=[vpc1.id, vpc2.id])
vpcs.should.have.length_of(2)
vpc_ids = tuple(map(lambda v: v.id, vpcs))
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)
@mock_ec2
def test_vpc_get_by_cidr_block():
conn = boto.connect_vpc()
vpc1 = conn.create_vpc("10.0.0.0/16")
vpc2 = conn.create_vpc("10.0.0.0/16")
vpc3 = conn.create_vpc("10.0.0.0/24")
vpcs = conn.get_all_vpcs(filters={'cidr': '10.0.0.0/16'})
vpcs.should.have.length_of(2)
vpc_ids = tuple(map(lambda v: v.id, vpcs))
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)
@mock_ec2
def test_vpc_get_by_dhcp_options_id():
conn = boto.connect_vpc()
dhcp_options = conn.create_dhcp_options(SAMPLE_DOMAIN_NAME, SAMPLE_NAME_SERVERS)
vpc1 = conn.create_vpc("10.0.0.0/16")
vpc2 = conn.create_vpc("10.0.0.0/16")
vpc3 = conn.create_vpc("10.0.0.0/24")
conn.associate_dhcp_options(dhcp_options.id, vpc1.id)
conn.associate_dhcp_options(dhcp_options.id, vpc2.id)
vpcs = conn.get_all_vpcs(filters={'dhcp-options-id': dhcp_options.id})
vpcs.should.have.length_of(2)
vpc_ids = tuple(map(lambda v: v.id, vpcs))
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)
@mock_ec2
def test_vpc_get_by_tag():
conn = boto.connect_vpc()
vpc1 = conn.create_vpc("10.0.0.0/16")
vpc2 = conn.create_vpc("10.0.0.0/16")
vpc3 = conn.create_vpc("10.0.0.0/24")
vpc1.add_tag('Name', 'TestVPC')
vpc2.add_tag('Name', 'TestVPC')
vpc3.add_tag('Name', 'TestVPC2')
vpcs = conn.get_all_vpcs(filters={'tag:Name': 'TestVPC'})
vpcs.should.have.length_of(2)
vpc_ids = tuple(map(lambda v: v.id, vpcs))
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)