x/vmware-nsx
Code Issues Proposed changes

Merge "Add update method of security group name and description"

Browse Source
This commit is contained in:
Jenkins 2013-05-21 16:13:53 +00:00 committed by Gerrit Code Review
commit 08c49053c1
3 changed files with 68 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
quantum/db/securitygroups_db.py
View File

@ -200,6 +200,15 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
with context.session.begin(subtransactions=True):
context.session.delete(sg)
def update_security_group(self, context, id, security_group):
s = security_group['security_group']
with context.session.begin(subtransactions=True):
sg = self._get_security_group(context, id)
if sg['name'] == 'default' and 'name' in s:
raise ext_sg.SecurityGroupCannotUpdateDefault()
sg.update(s)
return self._make_security_group_dict(sg)
def _make_security_group_dict(self, security_group, fields=None):
res = {'id': security_group['id'],
'name': security_group['name'],

12
quantum/extensions/securitygroup.py
View File

@ -47,6 +47,10 @@ class SecurityGroupCannotRemoveDefault(qexception.InUse):
message = _("Removing default security group not allowed.")
class SecurityGroupCannotUpdateDefault(qexception.InUse):
message = _("Updating default security group not allowed.")
class SecurityGroupDefaultAlreadyExists(qexception.InUse):
message = _("Default security group already exists.")
@ -149,10 +153,10 @@ RESOURCE_ATTRIBUTE_MAP = {
'validate': {'type:uuid': None},
'is_visible': True,
'primary_key': True},
'name': {'allow_post': True, 'allow_put': False,
'name': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': '',
'validate': {'type:name_not_default': None}},
'description': {'allow_post': True, 'allow_put': False,
'description': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': ''},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
@ -278,6 +282,10 @@ class SecurityGroupPluginBase(object):
def create_security_group(self, context, security_group):
pass
@abstractmethod
def update_security_group(self, context, id, security_group):
pass
@abstractmethod
def delete_security_group(self, context, id):
pass

49
quantum/tests/unit/test_extension_security_group.py
View File

@ -283,6 +283,55 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
'port_range_min': None}
self._assert_sg_rule_has_kvs(v6_rule, expected)
def test_update_security_group(self):
with self.security_group() as sg:
data = {'security_group': {'name': 'new_name',
'description': 'new_desc'}}
req = self.new_update_request('security-groups',
data,
sg['security_group']['id'])
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(res['security_group']['name'],
data['security_group']['name'])
self.assertEqual(res['security_group']['description'],
data['security_group']['description'])
def test_update_security_group_name_to_default_fail(self):
with self.security_group() as sg:
data = {'security_group': {'name': 'default',
'description': 'new_desc'}}
req = self.new_update_request('security-groups',
data,
sg['security_group']['id'])
req.environ['quantum.context'] = context.Context('', 'somebody')
res = req.get_response(self.ext_api)
self.assertEqual(res.status_int, 409)
def test_update_default_security_group_name_fail(self):
with self.network():
res = self.new_list_request('security-groups')
sg = self.deserialize(self.fmt, res.get_response(self.ext_api))
data = {'security_group': {'name': 'new_name',
'description': 'new_desc'}}
req = self.new_update_request('security-groups',
data,
sg['security_groups'][0]['id'])
req.environ['quantum.context'] = context.Context('', 'somebody')
res = req.get_response(self.ext_api)
self.assertEqual(res.status_int, 404)
def test_update_default_security_group_with_description(self):
with self.network():
res = self.new_list_request('security-groups')
sg = self.deserialize(self.fmt, res.get_response(self.ext_api))
data = {'security_group': {'description': 'new_desc'}}
req = self.new_update_request('security-groups',
data,
sg['security_groups'][0]['id'])
res = self.deserialize(self.fmt, req.get_response(self.ext_api))
self.assertEqual(res['security_group']['description'],
data['security_group']['description'])
def test_default_security_group(self):
with self.network():
res = self.new_list_request('security-groups')