x/python-ganttclient
Code Issues Proposed changes

Alright, first hole poked all the way through. We can now create security groups and read them back.

Browse Source
This commit is contained in:
Soren Hansen
2010-09-09 12:35:46 +02:00
parent 0b7b343d37
commit 0da3cec279
3 changed files with 47 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
nova/auth/manager.py
View File

@@ -640,11 +640,17 @@ class AuthManager(object):
with self.driver() as drv:
user_dict = drv.create_user(name, access, secret, admin)
if user_dict:
db.security_group_create(context={},
values={ 'name' : 'default',
'description' : 'default',
'user_id' : name })
return User(**user_dict)
def delete_user(self, user):
"""Deletes a user"""
with self.driver() as drv:
for security_group in db.security_group_get_by_user(context = {}, user_id=user.id):
db.security_group_destroy({}, security_group.id)
drv.delete_user(User.safe_id(user))
def generate_key_pair(self, user, key_name):

14
nova/endpoint/cloud.py
View File

@@ -212,10 +212,12 @@ class CloudController(object):
return True
@rbac.allow('all')
def describe_security_groups(self, context, group_names, **kwargs):
groups = {'securityGroupSet': []}
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) ] }
# Stubbed for now to unblock other things.
return groups
@rbac.allow('netadmin')
@@ -223,7 +225,11 @@ class CloudController(object):
return True
@rbac.allow('netadmin')
def create_security_group(self, context, group_name, **kwargs):
def create_security_group(self, context, group_name, group_description):
db.security_group_create(context,
values = { 'user_id' : context.user.id,
'name': group_name,
'description': group_description })
return True
@rbac.allow('netadmin')

34
nova/tests/api_unittest.py
View File

@@ -185,6 +185,9 @@ class ApiEc2TestCase(test.BaseTestCase):
self.host = '127.0.0.1'
self.app = api.APIServerApplication({'Cloud': self.cloud})
def expect_http(self, host=None, is_secure=False):
"""Returns a new EC2 connection"""
self.ec2 = boto.connect_ec2(
aws_access_key_id='fake',
aws_secret_access_key='fake',
@@ -194,9 +197,6 @@ class ApiEc2TestCase(test.BaseTestCase):
path='/services/Cloud')
self.mox.StubOutWithMock(self.ec2, 'new_http_connection')
def expect_http(self, host=None, is_secure=False):
"""Returns a new EC2 connection"""
http = FakeHttplibConnection(
self.app, '%s:%d' % (self.host, FLAGS.cc_port), False)
# pylint: disable-msg=E1103
@@ -231,3 +231,31 @@ class ApiEc2TestCase(test.BaseTestCase):
self.assertEquals(len(results), 1)
self.manager.delete_project(project)
self.manager.delete_user(user)
def test_get_all_security_groups(self):
"""Test that operations on security groups stick"""
self.expect_http()
self.mox.ReplayAll()
security_group_name = "".join(random.choice("sdiuisudfsdcnpaqwertasd") \
for x in range(random.randint(4, 8)))
user = self.manager.create_user('fake', 'fake', 'fake', admin=True)
project = self.manager.create_project('fake', 'fake', 'fake')
rv = self.ec2.get_all_security_groups()
self.assertEquals(len(rv), 1)
self.assertEquals(rv[0].name, 'default')
self.expect_http()
self.mox.ReplayAll()
self.ec2.create_security_group(security_group_name, 'test group')
self.expect_http()
self.mox.ReplayAll()
rv = self.ec2.get_all_security_groups()
self.assertEquals(len(rv), 2)
self.assertTrue(security_group_name in [group.name for group in rv])
self.manager.delete_project(project)
self.manager.delete_user(user)