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

Added the ability to filter vpcs by tags.

Browse Source
This commit is contained in:
Omer Katz 2014-09-30 17:58:02 +03:00 committed by Omer Katz
parent ecb23485d0
commit 89bd78b927
2 changed files with 36 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
moto/ec2/models.py
View File

@ -84,6 +84,15 @@ class TaggedEC2Instance(object):
tags = ec2_backend.describe_tags(self.id) tags = ec2_backend.describe_tags(self.id)
return tags 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): class NetworkInterface(object):
def __init__(self, subnet, private_ip_address, device_index=0, public_ip_auto_assign=True, group_ids=None): def __init__(self, subnet, private_ip_address, device_index=0, public_ip_auto_assign=True, group_ids=None):
@ -1142,11 +1151,16 @@ class VPC(TaggedEC2Instance):
return self.dhcp_options.id 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" \ msg = "The filter '{0}' for DescribeVPCs has not been" \
" implemented in Moto yet. Feel free to open an issue at" \ " implemented in Moto yet. Feel free to open an issue at" \
" https://github.com/spulec/moto/issues".format(filter_name) " https://github.com/spulec/moto/issues".format(filter_name)
raise NotImplementedError(msg) raise NotImplementedError(msg)
return filter_value
class VPCBackend(object): class VPCBackend(object):
def __init__(self): def __init__(self):

18
tests/test_ec2/test_vpcs.py
View File

@ -111,3 +111,21 @@ def test_vpc_get_by_dhcp_options_id():
vpc_ids = map(lambda v: v.id, vpcs) vpc_ids = map(lambda v: v.id, vpcs)
vpc1.id.should.be.within(vpc_ids) vpc1.id.should.be.within(vpc_ids)
vpc2.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 = map(lambda v: v.id, vpcs)
vpc1.id.should.be.within(vpc_ids)
vpc2.id.should.be.within(vpc_ids)