x/python-ganttclient
Code Issues Proposed changes

Create and delete security groups works.

Browse Source 
Adding and revoking rules works.
DescribeSecurityGroups returns the groups and rules.
So, the API seems to be done. Yay.
This commit is contained in:
Soren Hansen
2010-09-10 11:47:06 +02:00
parent 1e4c80f5f5
commit 4a8fcffa67
3 changed files with 88 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
nova/endpoint/api.py
View File

@@ -135,6 +135,7 @@ class APIRequest(object):
response = xml.toxml()
xml.unlink()
# print response
_log.debug(response)
return response

41
nova/endpoint/cloud.py
View File

@@ -213,14 +213,41 @@ class CloudController(object):
@rbac.allow('all')
def describe_security_groups(self, context, **kwargs):
groups = {'securityGroupSet':
[{ 'groupDescription': group.description,
'groupName' : group.name,
'ownerId': context.user.id } for group in \
db.security_group_get_by_user(context,
context.user.id) ] }
groups = []
for group in db.security_group_get_by_user(context, context.user.id):
group_dict = {}
group_dict['groupDescription'] = group.description
group_dict['groupName'] = group.name
group_dict['ownerId'] = context.user.id
group_dict['ipPermissions'] = []
for rule in group.rules:
rule_dict = {}
rule_dict['ipProtocol'] = rule.protocol
rule_dict['fromPort'] = rule.from_port
rule_dict['toPort'] = rule.to_port
rule_dict['groups'] = []
rule_dict['ipRanges'] = []
if rule.group_id:
foreign_group = db.security_group_get_by_id({}, rule.group_id)
rule_dict['groups'] += [ { 'groupName': foreign_group.name,
'userId': foreign_group.user_id } ]
else:
rule_dict['ipRanges'] += [ { 'cidrIp': rule.cidr } ]
group_dict['ipPermissions'] += [ rule_dict ]
groups += [ group_dict ]
return groups
return {'securityGroupInfo': groups }
#
# [{ 'groupDescription': group.description,
# 'groupName' : group.name,
# 'ownerId': context.user.id,
# 'ipPermissions' : [
# { 'ipProtocol' : rule.protocol,
# 'fromPort' : rule.from_port,
# 'toPort' : rule.to_port,
# 'ipRanges' : [ { 'cidrIp' : rule.cidr } ] } for rule in group.rules ] } for group in \
#
# return groups
@rbac.allow('netadmin')
def revoke_security_group_ingress(self, context, group_name,

58
nova/tests/api_unittest.py
View File

@@ -293,19 +293,43 @@ class ApiEc2TestCase(test.BaseTestCase):
self.mox.ReplayAll()
group.connection = self.ec2
group.authorize('tcp', 80, 80, '0.0.0.0/0')
group.authorize('tcp', 80, 81, '0.0.0.0/0')
self.expect_http()
self.mox.ReplayAll()
rv = self.ec2.get_all_security_groups()
# I don't bother checkng that we actually find it here,
# because the create/delete unit test further up should
# be good enough for that.
for group in rv:
if group.name == security_group_name:
self.assertEquals(len(group.rules), 1)
self.assertEquals(int(group.rules[0].from_port), 80)
self.assertEquals(int(group.rules[0].to_port), 81)
self.assertEquals(len(group.rules[0].grants), 1)
self.assertEquals(str(group.rules[0].grants[0]), '0.0.0.0/0')
self.expect_http()
self.mox.ReplayAll()
group.connection = self.ec2
group.revoke('tcp', 80, 80, '0.0.0.0/0')
group.revoke('tcp', 80, 81, '0.0.0.0/0')
self.expect_http()
self.mox.ReplayAll()
self.ec2.delete_security_group(security_group_name)
self.expect_http()
self.mox.ReplayAll()
group.connection = self.ec2
rv = self.ec2.get_all_security_groups()
self.assertEqual(len(rv), 1)
self.assertEqual(rv[0].name, 'default')
self.manager.delete_project(project)
self.manager.delete_user(user)
@@ -323,13 +347,16 @@ class ApiEc2TestCase(test.BaseTestCase):
security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \
for x in range(random.randint(4, 8)))
other_security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \
for x in range(random.randint(4, 8)))
group = self.ec2.create_security_group(security_group_name, 'test group')
self.expect_http()
self.mox.ReplayAll()
other_group = self.ec2.create_security_group('appserver', 'The application tier')
other_group = self.ec2.create_security_group(other_security_group_name,
'some other group')
self.expect_http()
self.mox.ReplayAll()
@@ -339,9 +366,30 @@ class ApiEc2TestCase(test.BaseTestCase):
self.expect_http()
self.mox.ReplayAll()
group.connection = self.ec2
group.revoke(src_group=other_group)
rv = self.ec2.get_all_security_groups()
# I don't bother checkng that we actually find it here,
# because the create/delete unit test further up should
# be good enough for that.
for group in rv:
if group.name == security_group_name:
self.assertEquals(len(group.rules), 1)
self.assertEquals(len(group.rules[0].grants), 1)
self.assertEquals(str(group.rules[0].grants[0]),
'%s-%s' % (other_security_group_name, 'fake'))
self.expect_http()
self.mox.ReplayAll()
rv = self.ec2.get_all_security_groups()
for group in rv:
if group.name == security_group_name:
self.expect_http()
self.mox.ReplayAll()
group.connection = self.ec2
group.revoke(src_group=other_group)
self.expect_http()
self.mox.ReplayAll()