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

Security Groups: Fix for filtering support.

Browse Source
This commit is contained in:
Shawn Falkner-Horine 2014-09-10 09:42:38 -07:00
parent 11dbe5c10f
commit 3f266ebc2b
2 changed files with 30 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
moto/ec2/models.py
View File

@ -749,42 +749,42 @@ class SecurityGroup(object):
def physical_resource_id(self): def physical_resource_id(self):
return self.id return self.id
def matches_filters(self, filters):
def matches_filter(self, key, filter_value):
result = True result = True
def to_attr(filter_name): def to_attr(filter_name):
attr = None attr = None
if attr == 'group-name': if filter_name == 'group-name':
attr = 'name' attr = 'name'
elif attr == 'group-id': elif filter_name == 'group-id':
attr = 'id' attr = 'id'
elif filter_name == 'vpc-id':
attr = 'vpc_id'
else: else:
attr = filter_name.replace('-', '_') attr = filter_name.replace('-', '_')
return attr return attr
for key, value in filters.items(): if key.startswith('ip-permission'):
ret = False match = re.search(r"ip-permission.(*)", key)
ingress_attr = to_attr(match.groups()[0])
if key.startswith('ip-permission'): for ingress in self.ingress_rules:
match = re.search(r"ip-permission.(*)", key) if getattr(ingress, ingress_attr) in filters[key]:
ingress_attr = to_attr(match.groups()[0]) return True
for ingress in self.ingress_rules:
if getattr(ingress, ingress_attr) in filters[key]:
ret = True
break
else:
attr_name = to_attr(key)
ret = getattr(self, attr_name) in filters[key]
if not ret:
break
else: else:
result = False attr_name = to_attr(key)
return getattr(self, attr_name) in filter_value
return result return False
def matches_filters(self, filters):
for key, value in filters.items():
if not self.matches_filter(key, value):
return False
return True
class SecurityGroupBackend(object): class SecurityGroupBackend(object):

11
tests/test_ec2/test_security_groups.py
View File

@ -200,17 +200,24 @@ def test_authorize_group_in_vpc():
@mock_ec2 @mock_ec2
def test_get_all_security_groups(): def test_get_all_security_groups():
conn = boto.connect_ec2() conn = boto.connect_ec2()
conn.create_security_group(name='test1', description='test1', vpc_id='vpc-mjm05d27') sg1 = conn.create_security_group(name='test1', description='test1', vpc_id='vpc-mjm05d27')
conn.create_security_group(name='test2', description='test2') sg2 = conn.create_security_group(name='test2', description='test2')
resp = conn.get_all_security_groups(groupnames=['test1']) resp = conn.get_all_security_groups(groupnames=['test1'])
resp.should.have.length_of(1) resp.should.have.length_of(1)
resp[0].id.should.equal(sg1.id)
resp = conn.get_all_security_groups(filters={'vpc-id': ['vpc-mjm05d27']})
resp.should.have.length_of(1)
resp[0].id.should.equal(sg1.id)
resp = conn.get_all_security_groups(filters={'vpc_id': ['vpc-mjm05d27']}) resp = conn.get_all_security_groups(filters={'vpc_id': ['vpc-mjm05d27']})
resp.should.have.length_of(1) resp.should.have.length_of(1)
resp[0].id.should.equal(sg1.id)
resp = conn.get_all_security_groups(filters={'description': ['test1']}) resp = conn.get_all_security_groups(filters={'description': ['test1']})
resp.should.have.length_of(1) resp.should.have.length_of(1)
resp[0].id.should.equal(sg1.id)
resp = conn.get_all_security_groups() resp = conn.get_all_security_groups()
resp.should.have.length_of(2) resp.should.have.length_of(2)