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'],
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.")
'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,
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
'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')