x/vmware-nsx
Code Issues Proposed changes

Merge "Remove external_id and security group proxy code"

Browse Source
This commit is contained in:
Jenkins 2013-02-24 18:11:39 +00:00 committed by Gerrit Code Review
commit 3ee4d6d9d9
8 changed files with 32 additions and 364 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
etc/quantum.conf
View File

@ -217,10 +217,6 @@ notification_topics = notifications
# Each service definition should be in the following format:
# <service>:<plugin>[:driver]
[SECURITYGROUP]
# If set to true this allows quantum to receive proxied security group calls from nova
# proxy_mode = False
[AGENT]
# Use "sudo quantum-rootwrap /etc/quantum/rootwrap.conf" to use the real
# root filter facility.

5
quantum/db/migration/alembic_migrations/versions/3cb5d900c5de_security_groups.py
View File

@ -54,15 +54,12 @@ def upgrade(active_plugin=None, options=None):
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('name', sa.String(length=255), nullable=True),
sa.Column('description', sa.String(length=255), nullable=True),
sa.Column('external_id', sa.Integer(), nullable=True),
sa.PrimaryKeyConstraint('id'),
sa.UniqueConstraint('external_id')
sa.PrimaryKeyConstraint('id')
)
op.create_table(
'securitygrouprules',
sa.Column('tenant_id', sa.String(length=255), nullable=True),
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('external_id', sa.Integer(), nullable=True),
sa.Column('security_group_id', sa.String(length=36), nullable=False),
sa.Column('source_group_id', sa.String(length=36), nullable=True),
sa.Column('direction',

79
quantum/db/securitygroups_db.py
View File

@ -16,7 +16,6 @@
#
# @author: Aaron Rosen, Nicira, Inc
from oslo.config import cfg
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc
@ -33,7 +32,6 @@ class SecurityGroup(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
"""Represents a v2 quantum security group."""
name = sa.Column(sa.String(255))
description = sa.Column(sa.String(255))
external_id = sa.Column(sa.Integer, unique=True)
class SecurityGroupPortBinding(model_base.BASEV2):
@ -50,7 +48,6 @@ class SecurityGroupPortBinding(model_base.BASEV2):
class SecurityGroupRule(model_base.BASEV2, models_v2.HasId,
models_v2.HasTenant):
"""Represents a v2 quantum security group rule."""
external_id = sa.Column(sa.Integer)
security_group_id = sa.Column(sa.String(36),
sa.ForeignKey("securitygroups.id",
ondelete="CASCADE"),
@ -93,37 +90,17 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
a given tenant if it does not exist.
"""
s = security_group['security_group']
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
if (cfg.CONF.SECURITYGROUP.proxy_mode and not s.get('external_id')):
raise ext_sg.SecurityGroupProxyMode()
if not cfg.CONF.SECURITYGROUP.proxy_mode and s.get('external_id'):
raise ext_sg.SecurityGroupNotProxyMode()
tenant_id = self._get_tenant_id_for_create(context, s)
# if in proxy mode a default security group will be created by source
if not default_sg and not cfg.CONF.SECURITYGROUP.proxy_mode:
self._ensure_default_security_group(context, tenant_id,
security_group)
if s.get('external_id'):
try:
# Check if security group already exists
sg = self.get_security_group(context, s.get('external_id'))
if sg:
raise ext_sg.SecurityGroupAlreadyExists(
name=sg.get('name', ''),
external_id=s.get('external_id'))
except ext_sg.SecurityGroupNotFound:
pass
if not default_sg:
self._ensure_default_security_group(context, tenant_id)
with context.session.begin(subtransactions=True):
security_group_db = SecurityGroup(id=s.get('id') or (
uuidutils.generate_uuid()),
description=s['description'],
tenant_id=tenant_id,
name=s['name'],
external_id=s.get('external_id'))
name=s['name'])
context.session.add(security_group_db)
if s.get('name') == 'default':
for ethertype in ext_sg.sg_supported_ethertypes:
@ -178,19 +155,13 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
def _get_security_group(self, context, id):
try:
query = self._model_query(context, SecurityGroup)
if uuidutils.is_uuid_like(id):
sg = query.filter(SecurityGroup.id == id).one()
else:
sg = query.filter(SecurityGroup.external_id == id).one()
except exc.NoResultFound:
raise ext_sg.SecurityGroupNotFound(id=id)
return sg
def delete_security_group(self, context, id):
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
filters = {'security_group_id': [id]}
ports = self._get_port_security_group_bindings(context, filters)
if ports:
@ -208,8 +179,6 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
'name': security_group['name'],
'tenant_id': security_group['tenant_id'],
'description': security_group['description']}
if security_group.get('external_id'):
res['external_id'] = security_group['external_id']
res['security_group_rules'] = [self._make_security_group_rule_dict(r)
for r in security_group.rules]
return self._fields(res, fields)
@ -265,7 +234,6 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
id=uuidutils.generate_uuid(), tenant_id=tenant_id,
security_group_id=rule['security_group_id'],
direction=rule['direction'],
external_id=rule.get('external_id'),
source_group_id=rule.get('source_group_id'),
ethertype=rule['ethertype'],
protocol=rule['protocol'],
@ -286,23 +254,12 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
group, source_group_id/security_group_id belong to the same tenant,
and rules are valid.
"""
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
new_rules = set()
tenant_ids = set()
for rules in security_group_rule['security_group_rules']:
rule = rules.get('security_group_rule')
new_rules.add(rule['security_group_id'])
if (cfg.CONF.SECURITYGROUP.proxy_mode and
not rule.get('external_id')):
raise ext_sg.SecurityGroupProxyMode()
if (not cfg.CONF.SECURITYGROUP.proxy_mode and
rule.get('external_id')):
raise ext_sg.SecurityGroupNotProxyMode()
# Check that port_range's are valid
if (rule['port_range_min'] is None and
rule['port_range_max'] is None):
@ -347,8 +304,7 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
'port_range_min': security_group_rule['port_range_min'],
'port_range_max': security_group_rule['port_range_max'],
'source_ip_prefix': security_group_rule['source_ip_prefix'],
'source_group_id': security_group_rule['source_group_id'],
'external_id': security_group_rule['external_id']}
'source_group_id': security_group_rule['source_group_id']}
return self._fields(res, fields)
@ -360,7 +316,7 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
include_if_present = ['protocol', 'port_range_max', 'port_range_min',
'ethertype', 'source_ip_prefix',
'source_group_id', 'external_id']
'source_group_id']
for key in include_if_present:
value = sgr.get(key)
if value:
@ -405,19 +361,13 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
def _get_security_group_rule(self, context, id):
try:
if uuidutils.is_uuid_like(id):
query = self._model_query(context, SecurityGroupRule)
sgr = query.filter(SecurityGroupRule.id == id).one()
else:
query = self._model_query(context, SecurityGroupRule)
sgr = query.filter(SecurityGroupRule.external_id == id).one()
except exc.NoResultFound:
raise ext_sg.SecurityGroupRuleNotFound(id=id)
return sgr
def delete_security_group_rule(self, context, id):
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
with context.session.begin(subtransactions=True):
rule = self._get_security_group_rule(context, id)
context.session.delete(rule)
@ -442,25 +392,17 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
self._create_port_security_group_binding(context, port_id,
security_group_id)
def _ensure_default_security_group(self, context, tenant_id,
security_group=None):
def _ensure_default_security_group(self, context, tenant_id):
"""Create a default security group if one doesn't exist.
:returns: the default security group id.
"""
# if in proxy mode a default security group will be created by source
if not security_group and cfg.CONF.SECURITYGROUP.proxy_mode:
return
filters = {'name': ['default'], 'tenant_id': [tenant_id]}
default_group = self.get_security_groups(context, filters)
if not default_group:
security_group = {'security_group': {'name': 'default',
'tenant_id': tenant_id,
'description': 'default'}}
if security_group:
security_group['security_group']['external_id'] = (
security_group['security_group'].get('external_id'))
ret = self.create_security_group(context, security_group, True)
return ret['id']
else:
@ -477,11 +419,8 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
if p.get('device_owner') and p['device_owner'].startswith('network:'):
return
valid_groups = self.get_security_groups(
context, fields=['external_id', 'id'])
valid_groups = self.get_security_groups(context, fields=['id'])
valid_group_map = dict((g['id'], g['id']) for g in valid_groups)
valid_group_map.update((g['external_id'], g['id'])
for g in valid_groups if g.get('external_id'))
try:
return set([valid_group_map[sg_id]
for sg_id in p.get(ext_sg.SECURITYGROUPS, [])])
@ -489,10 +428,6 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
raise ext_sg.SecurityGroupNotFound(id=str(e))
def _ensure_default_security_group_on_port(self, context, port):
# return if proxy_mode is enabled since nova will handle adding
# the port to the default security group.
if cfg.CONF.SECURITYGROUP.proxy_mode:
return
# we don't apply security groups for dhcp, router
if (port['port'].get('device_owner') and
port['port']['device_owner'].startswith('network:')):

59
quantum/extensions/securitygroup.py
View File

@ -30,11 +30,6 @@ from quantum import quota
# Security group Exceptions
class SecurityGroupAlreadyExists(qexception.InUse):
# This can only happen if the external_id database is cleared
message = _("Security group %(name)s id %(external_id)s already exists")
class SecurityGroupInvalidPortRange(qexception.InvalidInput):
message = _("For TCP/UDP protocols, port_range_min must be "
"<= port_range_max")
@ -100,22 +95,6 @@ class SecurityGroupRuleExists(qexception.InUse):
message = _("Security group rule already exists. Group id is %(id)s.")
class SecurityGroupProxyMode(qexception.InUse):
message = _("Did not recieve external id and in proxy mode")
class SecurityGroupNotProxyMode(qexception.InUse):
message = _("Recieve external id and not in proxy mode")
class SecurityGroupProxyModeNotAdmin(qexception.NotAuthorized):
message = _("In Proxy Mode and not from admin")
class SecurityGroupInvalidExternalID(qexception.InvalidInput):
message = _("external_id wrong type %(data)s")
def convert_protocol_to_case_insensitive(value):
if value is None:
return value
@ -147,36 +126,22 @@ def convert_validate_port_value(port):
raise SecurityGroupInvalidPortValue(port=port)
def convert_to_uuid_or_int_list(value_list):
def convert_to_uuid_list_or_none(value_list):
if value_list is None:
return
try:
return [sg_id if uuidutils.is_uuid_like(sg_id) else int(sg_id)
for sg_id in value_list]
except (ValueError, TypeError):
for sg_id in value_list:
if not uuidutils.is_uuid_like(sg_id):
msg = _("'%s' is not an integer or uuid") % sg_id
raise qexception.InvalidInput(error_message=msg)
return value_list
def _validate_name_not_default(data, valid_values=None):
if not cfg.CONF.SECURITYGROUP.proxy_mode and data == "default":
if data == "default":
raise SecurityGroupDefaultAlreadyExists()
def _validate_external_id_and_mode(external_id, valid_values=None):
if not cfg.CONF.SECURITYGROUP.proxy_mode and not external_id:
return
elif not cfg.CONF.SECURITYGROUP.proxy_mode and external_id:
raise SecurityGroupNotProxyMode()
try:
int(external_id)
except (ValueError, TypeError):
raise SecurityGroupInvalidExternalID(data=external_id)
if cfg.CONF.SECURITYGROUP.proxy_mode and not external_id:
raise SecurityGroupProxyMode()
attr.validators['type:name_not_default'] = _validate_name_not_default
attr.validators['type:external_id_and_mode'] = _validate_external_id_and_mode
sg_supported_protocols = [None, 'tcp', 'udp', 'icmp']
sg_supported_ethertypes = ['IPv4', 'IPv6']
@ -193,9 +158,6 @@ RESOURCE_ATTRIBUTE_MAP = {
'validate': {'type:name_not_default': None}},
'description': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': ''},
'external_id': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': None,
'validate': {'type:external_id_and_mode': None}},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'is_visible': True},
@ -207,10 +169,6 @@ RESOURCE_ATTRIBUTE_MAP = {
'validate': {'type:uuid': None},
'is_visible': True,
'primary_key': True},
# external_id can be used to be backwards compatible with nova
'external_id': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'default': None,
'validate': {'type:external_id_and_mode': None}},
'security_group_id': {'allow_post': True, 'allow_put': False,
'is_visible': True, 'required_by_policy': True},
'source_group_id': {'allow_post': True, 'allow_put': False,
@ -246,7 +204,7 @@ EXTENDED_ATTRIBUTES_2_0 = {
'ports': {SECURITYGROUPS: {'allow_post': True,
'allow_put': True,
'is_visible': True,
'convert_to': convert_to_uuid_or_int_list,
'convert_to': convert_to_uuid_list_or_none,
'default': attr.ATTR_NOT_SPECIFIED}}}
security_group_quota_opts = [
cfg.IntOpt('quota_security_group',
@ -260,11 +218,6 @@ security_group_quota_opts = [
]
cfg.CONF.register_opts(security_group_quota_opts, 'QUOTAS')
security_group_opts = [
cfg.StrOpt('proxy_mode', default=False)
]
cfg.CONF.register_opts(security_group_opts, 'SECURITYGROUP')
class Securitygroup(extensions.ExtensionDescriptor):
""" Security group extension"""

29
quantum/plugins/nicira/nicira_nvp_plugin/QuantumPlugin.py
View File

@ -2005,25 +2005,11 @@ class NvpPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
group and we don't need to check if one exists.
"""
s = security_group.get('security_group')
if cfg.CONF.SECURITYGROUP.proxy_mode:
if not context.is_admin:
raise ext_sg.SecurityGroupProxyModeNotAdmin()
elif not s.get('external_id'):
raise ext_sg.SecurityGroupProxyMode()
elif s.get('external_id'):
raise ext_sg.SecurityGroupNotProxyMode()
tenant_id = self._get_tenant_id_for_create(context, s)
if not default_sg and not cfg.CONF.SECURITYGROUP.proxy_mode:
self._ensure_default_security_group(context, tenant_id,
security_group)
if s.get('external_id'):
filters = {'external_id': [s.get('external_id')]}
security_groups = super(NvpPluginV2, self).get_security_groups(
context, filters=filters)
if security_groups:
raise ext_sg.SecurityGroupAlreadyExists(
name=s.get('name', ''), external_id=s.get('external_id'))
if not default_sg:
self._ensure_default_security_group(context, tenant_id)
nvp_secgroup = nvplib.create_security_profile(self.default_cluster,
tenant_id, s)
security_group['security_group']['id'] = nvp_secgroup['uuid']
@ -2034,9 +2020,6 @@ class NvpPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
"""Delete a security group
:param security_group_id: security group rule to remove.
"""
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
with context.session.begin(subtransactions=True):
security_group = super(NvpPluginV2, self).get_security_group(
context, security_group_id)
@ -2074,8 +2057,7 @@ class NvpPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
security_group_id = self._validate_security_group_rules(
context, security_group_rule)
# Check to make sure security group exists and retrieve
# security_group['id'] needed incase it only has an external_id
# Check to make sure security group exists
security_group = super(NvpPluginV2, self).get_security_group(
context, security_group_id)
@ -2098,9 +2080,6 @@ class NvpPluginV2(db_base_plugin_v2.QuantumDbPluginV2,
""" Delete a security group rule
:param sgrid: security group id to remove.
"""
if (cfg.CONF.SECURITYGROUP.proxy_mode and not context.is_admin):
raise ext_sg.SecurityGroupProxyModeNotAdmin()
with context.session.begin(subtransactions=True):
# determine security profile id
security_group_rule = (

3
quantum/plugins/nicira/nicira_nvp_plugin/common/securitygroups.py
View File

@ -63,8 +63,7 @@ class NVPSecurityGroups(object):
def _get_security_group_rules_nvp_format(self, context, security_group_id,
with_id=False):
"""Query quantum db for security group rules. If external_id is
provided the external_id will also be returned.
"""Query quantum db for security group rules.
"""
fields = ['source_ip_prefix', 'source_group_id', 'protocol',
'port_range_min', 'port_range_max', 'protocol', 'ethertype']

19
quantum/plugins/nicira/nicira_nvp_plugin/nvplib.py
View File

@ -1024,31 +1024,12 @@ def set_tenant_id_tag(tenant_id, taglist=None):
return new_taglist
def set_ext_security_profile_id_tag(external_id, taglist=None):
"""Convenience function to add spid tag to taglist.
:param external_id: the security_profile id from nova
:param taglist: the taglist to append to (or None).
:returns: a new taglist that includes the old taglist with the new
spid tag set."""
new_taglist = []
if taglist:
new_taglist = [x for x in taglist if x['scope'] !=
EXT_SECURITY_PROFILE_ID_SCOPE]
if external_id:
new_taglist.append(dict(scope=EXT_SECURITY_PROFILE_ID_SCOPE,
tag=str(external_id)))
return new_taglist
# -----------------------------------------------------------------------------
# Security Group API Calls
# -----------------------------------------------------------------------------
def create_security_profile(cluster, tenant_id, security_profile):
path = "/ws.v1/security-profile"
tags = set_tenant_id_tag(tenant_id)
tags = set_ext_security_profile_id_tag(
security_profile.get('external_id'), tags)
# Allow all dhcp responses in
dhcp = {'logical_port_egress_rules': [{'ethertype': 'IPv4',
'protocol': 17,

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

@ -52,15 +52,12 @@ class SecurityGroupTestExtensionManager(object):
class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
def _create_security_group(self, fmt, name, description, external_id=None,
**kwargs):
def _create_security_group(self, fmt, name, description, **kwargs):
data = {'security_group': {'name': name,
'tenant_id': kwargs.get('tenant_id',
'test_tenant'),
'description': description}}
if external_id:
data['security_group']['external_id'] = external_id
security_group_req = self.new_create_request('security-groups', data,
fmt)
if (kwargs.get('set_context') and 'tenant_id' in kwargs):
@ -72,7 +69,7 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
def _build_security_group_rule(self, security_group_id, direction,
protocol, port_range_min, port_range_max,
source_ip_prefix=None, source_group_id=None,
external_id=None, tenant_id='test_tenant',
tenant_id='test_tenant',
ethertype='IPv4'):
data = {'security_group_rule': {'security_group_id': security_group_id,
@ -83,9 +80,6 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
'port_range_max': port_range_max,
'tenant_id': tenant_id,
'ethertype': ethertype}}
if external_id:
data['security_group_rule']['external_id'] = external_id
if source_ip_prefix:
data['security_group_rule']['source_ip_prefix'] = source_ip_prefix
@ -105,10 +99,8 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
context.Context('', kwargs['tenant_id']))
return security_group_rule_req.get_response(self.ext_api)
def _make_security_group(self, fmt, name, description, external_id=None,
**kwargs):
res = self._create_security_group(fmt, name, description,
external_id, **kwargs)
def _make_security_group(self, fmt, name, description, **kwargs):
res = self._create_security_group(fmt, name, description, **kwargs)
if res.status_int >= 400:
raise webob.exc.HTTPClientError(code=res.status_int)
return self.deserialize(fmt, res)
@ -121,11 +113,10 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
@contextlib.contextmanager
def security_group(self, name='webservers', description='webservers',
external_id=None, fmt=None, no_delete=False):
fmt=None, no_delete=False):
if not fmt:
fmt = self.fmt
security_group = self._make_security_group(fmt, name, description,
external_id)
security_group = self._make_security_group(fmt, name, description)
try:
yield security_group
finally:
@ -139,8 +130,7 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
direction='ingress', protocol='tcp',
port_range_min='22', port_range_max='22',
source_ip_prefix=None, source_group_id=None,
external_id=None, fmt=None, no_delete=False,
ethertype='IPv4'):
fmt=None, no_delete=False, ethertype='IPv4'):
if not fmt:
fmt = self.fmt
rule = self._build_security_group_rule(security_group_id,
@ -149,7 +139,6 @@ class SecurityGroupsTestCase(test_db_plugin.QuantumDbPluginV2TestCase):
port_range_max,
source_ip_prefix,
source_group_id,
external_id,
ethertype=ethertype)
security_group_rule = self._make_security_group_rule(self.fmt, rule)
try:
@ -243,45 +232,12 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
for k, v, in keys:
self.assertEqual(security_group['security_group'][k], v)
def test_create_security_group_external_id(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
name = 'webservers'
description = 'my webservers'
external_id = 10
keys = [('name', name,), ('description', description),
('external_id', external_id)]
with self.security_group(name, description, external_id) as sg:
for k, v, in keys:
self.assertEqual(sg['security_group'][k], v)
def test_default_security_group(self):
with self.network():
res = self.new_list_request('security-groups')
groups = self.deserialize(self.fmt, res.get_response(self.ext_api))
self.assertEqual(len(groups['security_groups']), 1)
def test_create_security_group_proxy_mode_not_admin(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
res = self._create_security_group(self.fmt, 'webservers',
'webservers', '1',
tenant_id='bad_tenant',
set_context=True)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 403)
def test_create_security_group_no_external_id_proxy_mode(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
res = self._create_security_group(self.fmt, 'webservers',
'webservers')
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 400)
def test_create_security_group_no_external_id_not_proxy_mode(self):
res = self._create_security_group(self.fmt, 'webservers',
'webservers', '1')
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 409)
def test_create_default_security_group_fail(self):
name = 'default'
description = 'my webservers'
@ -289,17 +245,6 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 409)
def test_create_security_group_duplicate_external_id(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
name = 'webservers'
description = 'my webservers'
external_id = 1
with self.security_group(name, description, external_id):
res = self._create_security_group(self.fmt, name, description,
external_id)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 409)
def test_list_security_groups(self):
with contextlib.nested(self.security_group(name='sg1',
description='sg'),
@ -554,58 +499,6 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 404)
def test_create_security_group_rule_exteral_id_proxy_mode(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
with self.security_group(external_id=1) as sg:
rule = {'security_group_rule':
{'security_group_id': sg['security_group']['id'],
'direction': 'ingress',
'protocol': 'tcp',
'port_range_min': '22',
'port_range_max': '22',
'external_id': '1',
'tenant_id': 'test_tenant',
'source_group_id': sg['security_group']['id']}}
res = self._create_security_group_rule(self.fmt, rule)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 201)
def test_create_security_group_rule_exteral_id_not_proxy_mode(self):
with self.security_group() as sg:
rule = {'security_group_rule':
{'security_group_id': sg['security_group']['id'],
'direction': 'ingress',
'protocol': 'tcp',
'port_range_min': '22',
'port_range_max': '22',
'external_id': 1,
'tenant_id': 'test_tenant',
'source_group_id': sg['security_group']['id']}}
res = self._create_security_group_rule(self.fmt, rule)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 409)
def test_create_security_group_rule_not_admin(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
with self.security_group(external_id='1') as sg:
rule = {'security_group_rule':
{'security_group_id': sg['security_group']['id'],
'direction': 'ingress',
'protocol': 'tcp',
'port_range_min': '22',
'port_range_max': '22',
'tenant_id': 'bad_tenant',
'external_id': 1,
'source_group_id': sg['security_group']['id']}}
res = self._create_security_group_rule(self.fmt, rule,
tenant_id='bad_tenant',
set_context=True)
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 403)
def test_create_security_group_rule_bad_tenant_source_group_id(self):
with self.security_group() as sg:
res = self._create_security_group(self.fmt, 'webservers',
@ -1080,60 +973,7 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 400)
def test_validate_port_external_id_quantum_id(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
with self.network() as n:
with self.subnet(n):
sg1 = (self.deserialize(self.fmt,
self._create_security_group(self.fmt,
'foo', 'bar', '1')))
sg2 = (self.deserialize(self.fmt,
self._create_security_group(self.fmt,
'foo', 'bar', '2')))
res = self._create_port(
self.fmt, n['network']['id'],
security_groups=[sg1['security_group']['id']])
port = self.deserialize(self.fmt, res)
# This request updates the port sending the quantum security
# group id in and a nova security group id.
data = {'port': {'fixed_ips': port['port']['fixed_ips'],
'name': port['port']['name'],
ext_sg.SECURITYGROUPS:
[sg1['security_group']['external_id'],
sg2['security_group']['id']]}}
req = self.new_update_request('ports', data,
port['port']['id'])
res = self.deserialize(self.fmt, req.get_response(self.api))
self.assertEquals(len(res['port'][ext_sg.SECURITYGROUPS]), 2)
for sg_id in res['port'][ext_sg.SECURITYGROUPS]:
# only security group id's should be
# returned and not external_ids
self.assertEquals(len(sg_id), 36)
self._delete('ports', port['port']['id'])
def test_validate_port_external_id_string_or_int(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
with self.network() as n:
with self.subnet(n):
string_id = '1'
int_id = 2
self.deserialize(
self.fmt, self._create_security_group(self.fmt,
'foo', 'bar',
string_id))
self.deserialize(
self.fmt, self._create_security_group(self.fmt,
'foo', 'bar',
int_id))
res = self._create_port(
self.fmt, n['network']['id'],
security_groups=[string_id, int_id])
port = self.deserialize(self.fmt, res)
self._delete('ports', port['port']['id'])
def test_create_port_with_non_uuid_or_int(self):
def test_create_port_with_non_uuid(self):
with self.network() as n:
with self.subnet(n):
res = self._create_port(self.fmt, n['network']['id'],
@ -1142,18 +982,6 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 400)
def test_validate_port_external_id_fail(self):
cfg.CONF.set_override('proxy_mode', True, 'SECURITYGROUP')
with self.network() as n:
with self.subnet(n):
bad_id = 1
res = self._create_port(
self.fmt, n['network']['id'],
security_groups=[bad_id])
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, 404)
class TestSecurityGroupsXML(TestSecurityGroups):
fmt = 'xml'