updated SC methods to work with a group_id, which must be used if it's a group in a VPC

This commit is contained in:
Jon Haddad 2013-12-06 14:34:13 -08:00
parent 1eac424bf5
commit 48ee4b600b
3 changed files with 21 additions and 21 deletions

View File

@ -331,15 +331,17 @@ class SecurityGroupBackend(object):
def describe_security_groups(self):
return itertools.chain(*[x.values() for x in self.groups.values()])
def delete_security_group(self, name_or_group_id, vpc_id):
if name_or_group_id in self.groups[vpc_id]:
# Group Id
return self.groups[vpc_id].pop(name_or_group_id)
else:
# Group Name
group = self.get_security_group_from_name(name_or_group_id, vpc_id)
def delete_security_group(self, name=None, group_id=None):
if group_id:
# loop over all the SGs, find the right one
for vpc in self.groups.values():
if group_id in vpc:
return vpc.pop(group_id)
elif name:
# Group Name. Has to be in standard EC2, VPC needs to be identified by group_id
group = self.get_security_group_from_name(name, None)
if group:
return self.groups[vpc_id].pop(group.id)
return self.groups[None].pop(group.id)
def get_security_group_from_name(self, name, vpc_id):
for group_id, group in self.groups[vpc_id].iteritems():

View File

@ -41,11 +41,16 @@ class SecurityGroups(object):
def delete_security_group(self):
# TODO this should raise an error if there are instances in the group. See http://docs.aws.amazon.com/AWSEC2/latest/APIReference/ApiReference-query-DeleteSecurityGroup.html
name = self.querystring.get('GroupName')[0]
vpc_id = self.querystring.get("VpcId", [None])[0]
# needs vpc now
group = ec2_backend.delete_security_group(name, vpc_id)
name = self.querystring.get('GroupName')
sg_id = self.querystring.get('GroupId')
if name:
group = ec2_backend.delete_security_group(name[0])
elif sg_id:
group = ec2_backend.delete_security_group(group_id=sg_id[0])
# needs name or group now
if not group:
# There was no such group
return "There was no security group with name {0}".format(name), dict(status=404)

View File

@ -73,7 +73,7 @@ def test_deleting_security_groups():
conn.get_all_security_groups().should.have.length_of(1)
# Delete by group id
conn.delete_security_group(security_group1.id)
conn.delete_security_group(group_id=security_group1.id)
conn.get_all_security_groups().should.have.length_of(0)
@mock_ec2
@ -82,15 +82,8 @@ def test_delete_security_group_in_vpc():
vpc_id = "vpc-12345"
security_group1 = conn.create_security_group('test1', 'test1', vpc_id)
# Deleting a group that doesn't exist in the VPC should throw an error
conn.delete_security_group.when.called_with('test1').should.throw(EC2ResponseError)
# this should not throw an exception
conn.delete_security_group("test1", vpc_id=vpc_id)
# Delete by group id
# conn.delete_security_group(security_group1.id)
# conn.get_all_security_groups().should.have.length_of(0)
conn.delete_security_group(group_id=security_group1.id)
@mock_ec2