x/vmware-nsx
Code Issues Proposed changes

Merge "Service Type Framework refactoring"

Browse Source
This commit is contained in:
Jenkins 2013-07-25 22:06:07 +00:00 committed by Gerrit Code Review
commit a19682e358
13 changed files with 619 additions and 819 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
etc/neutron.conf
View File

@ -290,14 +290,6 @@ notification_topics = notifications
# default driver to use for quota checks
# quota_driver = neutron.quota.ConfDriver
[default_servicetype]
# Description of the default service type (optional)
# description = "default service type"
# Enter a service definition line for each advanced service provided
# by the default service type.
# Each service definition should be in the following format:
# <service>:<plugin>[:driver]
[agent]
# Use "sudo neutron-rootwrap /etc/neutron/rootwrap.conf" to use the real
# root filter facility.
@ -365,3 +357,14 @@ signing_dir = $state_path/keystone-signing
# If set, use this value for pool_timeout with sqlalchemy
# pool_timeout = 10
[service_providers]
# Specify service providers (drivers) for advanced services like loadbalancer, VPN, Firewall.
# Must be in form:
# service_provider=<service_type>:<name>:<driver>[:default]
# List of allowed service type include LOADBALANCER, FIREWALL, VPN
# Combination of <service type> and <name> must be unique; <driver> must also be unique
# this is multiline option, example for default provider:
#service_provider=LOADBALANCER:name:lbaas_plugin_driver_path:default
# example of non-default provider:
#service_provider=FIREWALL:name2:firewall_driver_path

5
etc/policy.json
View File

@ -58,11 +58,6 @@
"create_router:external_gateway_info:enable_snat": "rule:admin_only",
"update_router:external_gateway_info:enable_snat": "rule:admin_only",
"create_service_type": "rule:admin_only",
"update_service_type": "rule:admin_only",
"delete_service_type": "rule:admin_only",
"get_service_type": "rule:regular_user",
"create_qos_queue": "rule:admin_only",
"get_qos_queue": "rule:admin_only",

80
neutron/db/migration/alembic_migrations/versions/557edfc53098_new_service_types.py Normal file
View File

@ -0,0 +1,80 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
#
# Copyright 2013 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
"""New service types framework (service providers)
Revision ID: 557edfc53098
Revises: 52c5e4a18807
Create Date: 2013-06-29 21:10:41.283358
"""
# revision identifiers, used by Alembic.
revision = '557edfc53098'
down_revision = '52c5e4a18807'
# Change to ['*'] if this migration applies to all plugins
migration_for_plugins = ['*']
from alembic import op
import sqlalchemy as sa
from neutron.db import migration
def upgrade(active_plugin=None, options=None):
if not migration.should_run(active_plugin, migration_for_plugins):
return
op.create_table(
'providerresourceassociations',
sa.Column('provider_name', sa.String(length=255), nullable=False),
sa.Column('resource_id', sa.String(length=36),
nullable=False, unique=True),
)
# dropping unused tables
op.drop_table('servicedefinitions')
op.drop_table('servicetypes')
def downgrade(active_plugin=None, options=None):
if not migration.should_run(active_plugin, migration_for_plugins):
return
op.create_table(
'servicetypes',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('tenant_id', sa.String(length=255)),
sa.Column('name', sa.String(255)),
sa.Column('description', sa.String(255)),
sa.Column('default', sa.Boolean(), nullable=False, default=False),
sa.Column('num_instances', sa.Column(sa.Integer(), default=0)),
sa.PrimaryKeyConstraint('id')
)
op.create_table(
'servicedefinitions',
sa.Column('id', sa.String(length=36), nullable=False),
sa.Column('service_class', sa.String(255)),
sa.Column('plugin', sa.String(255)),
sa.Column('driver', sa.String(255)),
sa.Column('service_type_id', sa.String(36),
sa.ForeignKey('servicetypes.id',
ondelete='CASCADE')),
sa.PrimaryKeyConstraint('id', 'service_class')
)
op.drop_table('providerresourceassociations')

1
neutron/db/routerservicetype_db.py
View File

@ -27,7 +27,6 @@ class RouterServiceTypeBinding(model_base.BASEV2):
sa.ForeignKey('routers.id', ondelete="CASCADE"),
primary_key=True)
service_type_id = sa.Column(sa.String(36),
sa.ForeignKey('servicetypes.id'),
nullable=False)

307
neutron/db/servicetype_db.py
View File

@ -17,105 +17,27 @@
# @author: Salvatore Orlando, VMware
#
from oslo.config import cfg
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc as orm_exc
from sqlalchemy.sql import expression as expr
from neutron.common import exceptions as q_exc
from neutron import context
from neutron.db import api as db
from neutron.db import model_base
from neutron.db import models_v2
from neutron.openstack.common import log as logging
from neutron.services import provider_configuration as pconf
LOG = logging.getLogger(__name__)
DEFAULT_SVCTYPE_NAME = 'default'
default_servicetype_opts = [
cfg.StrOpt('description',
default='',
help=_('Textual description for the default service type')),
cfg.MultiStrOpt('service_definition',
help=_('Defines a provider for an advanced service '
'using the format: <service>:<plugin>[:<driver>]'))
]
cfg.CONF.register_opts(default_servicetype_opts, 'default_servicetype')
def parse_service_definition_opt():
"""Parse service definition opts and returns result."""
results = []
svc_def_opt = cfg.CONF.default_servicetype.service_definition
try:
for svc_def_str in svc_def_opt:
split = svc_def_str.split(':')
svc_def = {'service_class': split[0],
'plugin': split[1]}
try:
svc_def['driver'] = split[2]
except IndexError:
# Never mind, driver is optional
LOG.debug(_("Default service type - no driver for service "
"%(service_class)s and plugin %(plugin)s"),
svc_def)
results.append(svc_def)
return results
except (TypeError, IndexError):
raise q_exc.InvalidConfigurationOption(opt_name='service_definition',
opt_value=svc_def_opt)
class NoDefaultServiceDefinition(q_exc.NeutronException):
message = _("No default service definition in configuration file. "
"Please add service definitions using the service_definition "
"variable in the [default_servicetype] section")
class ServiceTypeNotFound(q_exc.NotFound):
message = _("Service type %(service_type_id)s could not be found ")
class ServiceTypeInUse(q_exc.InUse):
message = _("There are still active instances of service type "
"'%(service_type_id)s'. Therefore it cannot be removed.")
class ServiceDefinition(model_base.BASEV2, models_v2.HasId):
service_class = sa.Column(sa.String(255), primary_key=True)
plugin = sa.Column(sa.String(255))
driver = sa.Column(sa.String(255))
service_type_id = sa.Column(sa.String(36),
sa.ForeignKey('servicetypes.id',
ondelete='CASCADE'),
primary_key=True)
class ServiceType(model_base.BASEV2, models_v2.HasId, models_v2.HasTenant):
"""Service Type Object Model."""
name = sa.Column(sa.String(255))
description = sa.Column(sa.String(255))
default = sa.Column(sa.Boolean(), nullable=False, default=False)
service_definitions = orm.relationship(ServiceDefinition,
backref='servicetypes',
lazy='joined',
cascade='all')
# Keep track of number of instances for this service type
num_instances = sa.Column(sa.Integer(), default=0)
def as_dict(self):
"""Convert a row into a dict."""
ret_dict = {}
for c in self.__table__.columns:
ret_dict[c.name] = getattr(self, c.name)
return ret_dict
class ProviderResourceAssociation(model_base.BASEV2):
provider_name = sa.Column(sa.String(255),
nullable=False, primary_key=True)
# should be manualy deleted on resource deletion
resource_id = sa.Column(sa.String(36), nullable=False, primary_key=True,
unique=True)
class ServiceTypeManager(object):
"""Manage service type objects in Neutron database."""
"""Manage service type objects in Neutron."""
_instance = None
@ -127,189 +49,42 @@ class ServiceTypeManager(object):
def __init__(self):
self._initialize_db()
ctx = context.get_admin_context()
# Init default service type from configuration file
svc_defs = cfg.CONF.default_servicetype.service_definition
if not svc_defs:
raise NoDefaultServiceDefinition()
def_service_type = {'name': DEFAULT_SVCTYPE_NAME,
'description':
cfg.CONF.default_servicetype.description,
'service_definitions':
parse_service_definition_opt(),
'default': True}
# Create or update record in database
def_svc_type_db = self._get_default_service_type(ctx)
if not def_svc_type_db:
def_svc_type_db = self._create_service_type(ctx, def_service_type)
else:
self._update_service_type(ctx,
def_svc_type_db['id'],
def_service_type,
svc_type_db=def_svc_type_db)
LOG.debug(_("Default service type record updated in Neutron database. "
"identifier is '%s'"), def_svc_type_db['id'])
self._load_conf()
def _initialize_db(self):
db.configure_db()
# Register models for service type management
# Note this might have been already done if configure_db also
# created the engine
db.register_models(models_v2.model_base.BASEV2)
def _create_service_type(self, context, service_type):
svc_defs = service_type.pop('service_definitions')
def _load_conf(self):
self.conf = pconf.ProviderConfiguration(
pconf.parse_service_provider_opt())
def get_service_providers(self, context, filters=None, fields=None):
return self.conf.get_service_providers(filters, fields)
def get_default_service_provider(self, context, service_type):
"""Return the default provider for a given service type."""
filters = {'service_type': [service_type],
'default': [True]}
providers = self.get_service_providers(context, filters=filters)
# By construction we expect at most a single item in provider
if not providers:
raise pconf.DefaultServiceProviderNotFound(
service_type=service_type
)
return providers[0]
def add_resource_association(self, context, service_type, provider_name,
resource_id):
r = self.conf.get_service_providers(
filters={'service_type': service_type, 'name': provider_name})
if not r:
raise pconf.ServiceProviderNotFound(service_type=service_type)
with context.session.begin(subtransactions=True):
svc_type_db = ServiceType(**service_type)
# and now insert provided service type definitions
for svc_def in svc_defs:
svc_type_db.service_definitions.append(
ServiceDefinition(**svc_def))
# sqlalchemy save-update on relationship is on by
# default, the following will save both the service
# type and its service definitions
context.session.add(svc_type_db)
return svc_type_db
def _update_service_type(self, context, id, service_type,
svc_type_db=None):
with context.session.begin(subtransactions=True):
if not svc_type_db:
svc_type_db = self._get_service_type(context, id)
try:
svc_defs_map = dict([(svc_def['service'], svc_def)
for svc_def in
service_type.pop('service_definitions')])
except KeyError:
# No service defs in request
svc_defs_map = {}
svc_type_db.update(service_type)
for svc_def_db in svc_type_db.service_definitions:
try:
svc_def_db.update(svc_defs_map.pop(
svc_def_db['service_class']))
except KeyError:
# too bad, the service def was not there
# then we should delete it.
context.session.delete(svc_def_db)
# Add remaining service definitions
for svc_def in svc_defs_map:
context.session.add(ServiceDefinition(**svc_def))
return svc_type_db
def _get_service_type(self, context, svc_type_id):
try:
query = context.session.query(ServiceType)
return query.filter(ServiceType.id == svc_type_id).one()
# filter is on primary key, do not catch MultipleResultsFound
except orm_exc.NoResultFound:
raise ServiceTypeNotFound(service_type_id=svc_type_id)
def _get_default_service_type(self, context):
try:
query = context.session.query(ServiceType)
return query.filter(ServiceType.default == expr.true()).one()
except orm_exc.NoResultFound:
return
except orm_exc.MultipleResultsFound:
# This should never happen. If it does, take the first instance
query2 = context.session.query(ServiceType)
results = query2.filter(ServiceType.default == expr.true()).all()
LOG.warning(_("Multiple default service type instances found."
"Will use instance '%s'"), results[0]['id'])
return results[0]
def _make_svc_type_dict(self, context, svc_type, fields=None):
def _make_svc_def_dict(svc_def_db):
svc_def = {'service_class': svc_def_db['service_class']}
svc_def.update({'plugin': svc_def_db['plugin'],
'driver': svc_def_db['driver']})
return svc_def
res = {'id': svc_type['id'],
'name': svc_type['name'],
'default': svc_type['default'],
'num_instances': svc_type['num_instances'],
'service_definitions':
[_make_svc_def_dict(svc_def) for svc_def
in svc_type['service_definitions']]}
# Field selection
if fields:
return dict(((k, v) for k, v in res.iteritems()
if k in fields))
return res
def get_service_type(self, context, id, fields=None):
"""Retrieve a service type record."""
return self._make_svc_type_dict(context,
self._get_service_type(context, id),
fields)
def get_service_types(self, context, fields=None, filters=None):
"""Retrieve a possibly filtered list of service types."""
query = context.session.query(ServiceType)
if filters:
for key, value in filters.iteritems():
column = getattr(ServiceType, key, None)
if column:
query = query.filter(column.in_(value))
return [self._make_svc_type_dict(context, svc_type, fields)
for svc_type in query]
def create_service_type(self, context, service_type):
"""Create a new service type."""
svc_type_data = service_type['service_type']
svc_type_db = self._create_service_type(context, svc_type_data)
LOG.debug(_("Created service type object:%s"), svc_type_db['id'])
return self._make_svc_type_dict(context, svc_type_db)
def update_service_type(self, context, id, service_type):
"""Update a service type."""
svc_type_data = service_type['service_type']
svc_type_db = self._update_service_type(context, id,
svc_type_data)
return self._make_svc_type_dict(context, svc_type_db)
def delete_service_type(self, context, id):
"""Delete a service type."""
# Verify that the service type is not in use.
svc_type_db = self._get_service_type(context, id)
if svc_type_db['num_instances'] > 0:
raise ServiceTypeInUse(service_type_id=svc_type_db['id'])
with context.session.begin(subtransactions=True):
context.session.delete(svc_type_db)
def increase_service_type_refcount(self, context, id):
"""Increase references count for a service type object
This method should be invoked by plugins using the service
type concept everytime an instance of an object associated
with a given service type is created.
"""
#TODO(salvatore-orlando): Devise a better solution than this
#refcount mechanisms. Perhaps adding hooks into models which
#use service types in order to enforce ref. integrity and cascade
with context.session.begin(subtransactions=True):
svc_type_db = self._get_service_type(context, id)
svc_type_db['num_instances'] = svc_type_db['num_instances'] + 1
return svc_type_db['num_instances']
def decrease_service_type_refcount(self, context, id):
"""Decrease references count for a service type object
This method should be invoked by plugins using the service
type concept everytime an instance of an object associated
with a given service type is removed
"""
#TODO(salvatore-orlando): Devise a better solution than this
#refcount mechanisms. Perhaps adding hooks into models which
#use service types in order to enforce ref. integrity and cascade
with context.session.begin(subtransactions=True):
svc_type_db = self._get_service_type(context, id)
if svc_type_db['num_instances'] == 0:
LOG.warning(_("Number of instances for service type "
"'%s' is already 0."), svc_type_db['name'])
return
svc_type_db['num_instances'] = svc_type_db['num_instances'] - 1
return svc_type_db['num_instances']
# we don't actually need service type for association.
# resource_id is unique and belongs to specific service
# which knows its type
assoc = ProviderResourceAssociation(provider_name=provider_name,
resource_id=resource_id)
context.session.add(assoc)

138
neutron/extensions/servicetype.py
View File

@ -21,149 +21,32 @@
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.api.v2 import base
from neutron import context
from neutron.db import servicetype_db
from neutron import manager
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants
LOG = logging.getLogger(__name__)
RESOURCE_NAME = "service_type"
RESOURCE_NAME = "service_provider"
COLLECTION_NAME = "%ss" % RESOURCE_NAME
SERVICE_ATTR = 'service_class'
SERVICE_ATTR = 'service_type'
PLUGIN_ATTR = 'plugin'
DRIVER_ATTR = 'driver'
EXT_ALIAS = 'service-type'
# Attribute Map for Service Type Resource
# Attribute Map for Service Provider Resource
# Allow read-only access
RESOURCE_ATTRIBUTE_MAP = {
COLLECTION_NAME: {
'id': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'name': {'allow_post': True, 'allow_put': True,
'validate': {'type:string': None},
'is_visible': True, 'default': ''},
'service_type': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'name': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'default': {'allow_post': False, 'allow_put': False,
'is_visible': True},
#TODO(salvatore-orlando): Service types should not have ownership
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'is_visible': True},
'num_instances': {'allow_post': False, 'allow_put': False,
'is_visible': True},
'service_definitions': {'allow_post': True, 'allow_put': True,
'is_visible': True, 'default': None,
'validate': {'type:service_definitions':
None}}
}
}
def set_default_svctype_id(original_id):
if not original_id:
svctype_mgr = servicetype_db.ServiceTypeManager.get_instance()
# Fetch default service type - it must exist
res = svctype_mgr.get_service_types(context.get_admin_context(),
filters={'default': [True]})
return res[0]['id']
return original_id
def _validate_servicetype_ref(data, valid_values=None):
"""Verify the service type id exists."""
svc_type_id = data
svctype_mgr = servicetype_db.ServiceTypeManager.get_instance()
try:
svctype_mgr.get_service_type(context.get_admin_context(),
svc_type_id)
except servicetype_db.ServiceTypeNotFound:
return _("The service type '%s' does not exist") % svc_type_id
def _validate_service_defs(data, valid_values=None):
"""Validate the list of service definitions."""
try:
if not data:
return _("No service type definition was provided. At least a "
"service type definition must be provided")
f_name = _validate_service_defs.__name__
for svc_def in data:
try:
# Do a copy of the original object so we can easily
# pop out stuff from it
svc_def_copy = svc_def.copy()
try:
svc_name = svc_def_copy.pop(SERVICE_ATTR)
plugin_name = svc_def_copy.pop(PLUGIN_ATTR)
except KeyError:
msg = (_("Required attributes missing in service "
"definition: %s") % svc_def)
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
# Validate 'service' attribute
if svc_name not in constants.ALLOWED_SERVICES:
msg = (_("Service name '%s' unspecified "
"or invalid") % svc_name)
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
# Validate 'plugin' attribute
if not plugin_name:
msg = (_("Plugin name not specified in "
"service definition %s") % svc_def)
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
# TODO(salvatore-orlando): This code will need to change when
# multiple plugins for each adv service will be supported
svc_plugin = manager.NeutronManager.get_service_plugins().get(
svc_name)
if not svc_plugin:
msg = _("No plugin for service '%s'") % svc_name
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
if svc_plugin.get_plugin_name() != plugin_name:
msg = _("Plugin name '%s' is not correct ") % plugin_name
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
# Validate 'driver' attribute (just check it's a string)
# FIXME(salvatore-orlando): This should be a list
# Note: using get() instead of pop() as pop raises if the
# key is not found, which might happen for the driver
driver = svc_def_copy.get(DRIVER_ATTR)
if driver:
msg = attributes._validate_string(driver,)
if msg:
return msg
del svc_def_copy[DRIVER_ATTR]
# Anything left - it should be an error
if svc_def_copy:
msg = (_("Unparseable attributes found in "
"service definition %s") % svc_def)
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
except TypeError:
LOG.exception(_("Exception while parsing service "
"definition:%s"), svc_def)
msg = (_("Was expecting a dict for service definition, found "
"the following: %s") % svc_def)
LOG.error(_("%(f_name)s: %(msg)s"),
{'f_name': f_name, 'msg': msg})
return msg
except TypeError:
return (_("%s: provided data are not iterable") %
_validate_service_defs.__name__)
attributes.validators['type:service_definitions'] = _validate_service_defs
attributes.validators['type:servicetype_ref'] = _validate_servicetype_ref
class Servicetype(extensions.ExtensionDescriptor):
@classmethod
@ -176,7 +59,7 @@ class Servicetype(extensions.ExtensionDescriptor):
@classmethod
def get_description(cls):
return _("API for retrieving and managing service types for "
return _("API for retrieving service providers for "
"Neutron advanced services")
@classmethod
@ -191,7 +74,6 @@ class Servicetype(extensions.ExtensionDescriptor):
def get_resources(cls):
"""Returns Extended Resource for service type management."""
my_plurals = [(key, key[:-1]) for key in RESOURCE_ATTRIBUTE_MAP.keys()]
my_plurals.append(('service_definitions', 'service_definition'))
attributes.PLURALS.update(dict(my_plurals))
attr_map = RESOURCE_ATTRIBUTE_MAP[COLLECTION_NAME]
collection_name = COLLECTION_NAME.replace('_', '-')
@ -206,6 +88,6 @@ class Servicetype(extensions.ExtensionDescriptor):
def get_extended_resources(self, version):
if version == "2.0":
return dict(RESOURCE_ATTRIBUTE_MAP.items())
return RESOURCE_ATTRIBUTE_MAP
else:
return {}

157
neutron/services/provider_configuration.py Normal file
View File

@ -0,0 +1,157 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 OpenStack Foundation.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo.config import cfg
from neutron.common import exceptions as n_exc
from neutron.openstack.common import log as logging
from neutron.plugins.common import constants
LOG = logging.getLogger(__name__)
serviceprovider_opts = [
cfg.MultiStrOpt('service_provider', default=[],
help=_('Defines providers for advanced services '
'using the format: '
'<service_type>:<name>:<driver>[:default]'))
]
cfg.CONF.register_opts(serviceprovider_opts, 'service_providers')
#global scope function that should be used in service APIs
def normalize_provider_name(name):
return name.lower()
def parse_service_provider_opt():
"""Parse service definition opts and returns result."""
def validate_name(name):
if len(name) > 255:
raise n_exc.Invalid("Provider name is limited by 255 characters:"
" %s" % name)
svc_providers_opt = cfg.CONF.service_providers.service_provider
res = []
for prov_def in svc_providers_opt:
split = prov_def.split(':')
try:
svc_type, name, driver = split[:3]
except ValueError:
raise n_exc.Invalid(_("Invalid service provider format"))
validate_name(name)
name = normalize_provider_name(name)
default = False
if len(split) == 4 and split[3]:
if split[3] == 'default':
default = True
else:
msg = (_("Invalid provider format. "
"Last part should be 'default' or empty: %s") %
prov_def)
LOG.error(msg)
raise n_exc.Invalid(msg)
if svc_type not in constants.ALLOWED_SERVICES:
msg = (_("Service type '%(svc_type)s' is not allowed, "
"allowed types: %(allowed)s") %
{'svc_type': svc_type,
'allowed': constants.ALLOWED_SERVICES})
LOG.error(msg)
raise n_exc.Invalid(msg)
res.append({'service_type': svc_type,
'name': name,
'driver': driver,
'default': default})
return res
class ServiceProviderNotFound(n_exc.NotFound):
message = _("Service provider could not be found "
"for service type %(service_type)s")
class DefaultServiceProviderNotFound(ServiceProviderNotFound):
message = _("Service type %(service_type)s does not have a default "
"service provider")
class ProviderConfiguration(object):
def __init__(self, prov_data):
self.providers = {}
for prov in prov_data:
self.add_provider(prov)
def _ensure_driver_unique(self, driver):
for k, v in self.providers.items():
if v['driver'] == driver:
msg = (_("Driver %s is not unique across providers") %
driver)
LOG.exception(msg)
raise n_exc.Invalid(msg)
def _ensure_default_unique(self, type, default):
if not default:
return
for k, v in self.providers.items():
if k[0] == type and v['default']:
msg = _("Multiple default providers "
"for service %s") % type
LOG.exception(msg)
raise n_exc.Invalid(msg)
def add_provider(self, provider):
self._ensure_driver_unique(provider['driver'])
self._ensure_default_unique(provider['service_type'],
provider['default'])
provider_type = (provider['service_type'], provider['name'])
if provider_type in self.providers:
msg = (_("Multiple providers specified for service "
"%s") % provider['service_type'])
LOG.exception(msg)
raise n_exc.Invalid(msg)
self.providers[provider_type] = {'driver': provider['driver'],
'default': provider['default']}
def _check_entry(self, k, v, filters):
# small helper to deal with query filters
if not filters:
return True
for index, key in enumerate(['service_type', 'name']):
if key in filters:
if k[index] not in filters[key]:
return False
for key in ['driver', 'default']:
if key in filters:
if v[key] not in filters[key]:
return False
return True
def _fields(self, resource, fields):
if fields:
return dict(((key, item) for key, item in resource.items()
if key in fields))
return resource
def get_service_providers(self, filters=None, fields=None):
res = [{'service_type': k[0],
'name': k[1],
'driver': v['driver'],
'default': v['default']}
for k, v in self.providers.items()
if self._check_entry(k, v, filters)]
return self._fields(res, fields)

4
neutron/tests/etc/neutron.conf.test
View File

@ -25,7 +25,3 @@ lock_path = $state_path/lock
[database]
connection = 'sqlite://'
[default_servicetype]
description = "default service type"
service_definition=dummy:neutron.tests.unit.dummy_plugin.NeutronDummyPlugin

1
neutron/tests/unit/dummy_plugin.py
View File

@ -45,7 +45,6 @@ RESOURCE_ATTRIBUTE_MAP = {
'service_type': {'allow_post': True,
'allow_put': False,
'validate': {'type:servicetype_ref': None},
'convert_to': servicetype.set_default_svctype_id,
'is_visible': True,
'default': None}
}

5
neutron/tests/unit/nicira/etc/neutron.conf.test
View File

@ -24,8 +24,3 @@ lock_path = $state_path/lock
[database]
connection = 'sqlite://'
[default_servicetype]
description = "default service type"
service_definition=dummy:neutron.tests.unit.dummy_plugin.NeutronDummyPlugin

183
neutron/tests/unit/test_provider_configuration.py Normal file
View File

@ -0,0 +1,183 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright 2013 VMware, Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from oslo.config import cfg
from neutron.common import exceptions as q_exc
from neutron.plugins.common import constants
from neutron.services import provider_configuration as provconf
from neutron.tests import base
class ParseServiceProviderConfigurationTestCase(base.BaseTestCase):
def test_default_service_provider_configuration(self):
providers = cfg.CONF.service_providers.service_provider
self.assertEqual(providers, [])
def test_parse_single_service_provider_opt(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path'],
'service_providers')
expected = {'service_type': constants.LOADBALANCER,
'name': 'lbaas',
'driver': 'driver_path',
'default': False}
res = provconf.parse_service_provider_opt()
self.assertEqual(len(res), 1)
self.assertEqual(res, [expected])
def test_parse_single_default_service_provider_opt(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path:default'],
'service_providers')
expected = {'service_type': constants.LOADBALANCER,
'name': 'lbaas',
'driver': 'driver_path',
'default': True}
res = provconf.parse_service_provider_opt()
self.assertEqual(len(res), 1)
self.assertEqual(res, [expected])
def test_parse_multi_service_provider_opt(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path',
constants.LOADBALANCER + ':name1:path1',
constants.LOADBALANCER +
':name2:path2:default'],
'service_providers')
expected = {'service_type': constants.LOADBALANCER,
'name': 'lbaas',
'driver': 'driver_path',
'default': False}
res = provconf.parse_service_provider_opt()
self.assertEqual(len(res), 3)
self.assertEqual(res, [expected,
{'service_type': constants.LOADBALANCER,
'name': 'name1',
'driver': 'path1',
'default': False},
{'service_type': constants.LOADBALANCER,
'name': 'name2',
'driver': 'path2',
'default': True}])
def test_parse_service_provider_opt_not_allowed_raises(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path',
'svc_type:name1:path1'],
'service_providers')
self.assertRaises(q_exc.Invalid, provconf.parse_service_provider_opt)
def test_parse_service_provider_invalid_format(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path',
'svc_type:name1:path1:def'],
'service_providers')
self.assertRaises(q_exc.Invalid, provconf.parse_service_provider_opt)
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':',
'svc_type:name1:path1:def'],
'service_providers')
self.assertRaises(q_exc.Invalid, provconf.parse_service_provider_opt)
def test_parse_service_provider_name_too_long(self):
name = 'a' * 256
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':' + name + ':driver_path',
'svc_type:name1:path1:def'],
'service_providers')
self.assertRaises(q_exc.Invalid, provconf.parse_service_provider_opt)
class ProviderConfigurationTestCase(base.BaseTestCase):
def setUp(self):
super(ProviderConfigurationTestCase, self).setUp()
def test_ensure_driver_unique(self):
pconf = provconf.ProviderConfiguration([])
pconf.providers[('svctype', 'name')] = {'driver': 'driver',
'default': True}
self.assertRaises(q_exc.Invalid,
pconf._ensure_driver_unique, 'driver')
self.assertIsNone(pconf._ensure_driver_unique('another_driver1'))
def test_ensure_default_unique(self):
pconf = provconf.ProviderConfiguration([])
pconf.providers[('svctype', 'name')] = {'driver': 'driver',
'default': True}
self.assertRaises(q_exc.Invalid,
pconf._ensure_default_unique,
'svctype', True)
self.assertIsNone(pconf._ensure_default_unique('svctype', False))
self.assertIsNone(pconf._ensure_default_unique('svctype1', True))
self.assertIsNone(pconf._ensure_default_unique('svctype1', False))
def test_add_provider(self):
pconf = provconf.ProviderConfiguration([])
prov = {'service_type': constants.LOADBALANCER,
'name': 'name',
'driver': 'path',
'default': False}
pconf.add_provider(prov)
self.assertEqual(len(pconf.providers), 1)
self.assertEqual(pconf.providers.keys(),
[(constants.LOADBALANCER, 'name')])
self.assertEqual(pconf.providers.values(),
[{'driver': 'path', 'default': False}])
def test_add_duplicate_provider(self):
pconf = provconf.ProviderConfiguration([])
prov = {'service_type': constants.LOADBALANCER,
'name': 'name',
'driver': 'path',
'default': False}
pconf.add_provider(prov)
self.assertRaises(q_exc.Invalid, pconf.add_provider, prov)
self.assertEqual(len(pconf.providers), 1)
def test_get_service_providers(self):
provs = [{'service_type': constants.LOADBALANCER,
'name': 'name',
'driver': 'path',
'default': False},
{'service_type': constants.LOADBALANCER,
'name': 'name2',
'driver': 'path2',
'default': False},
{'service_type': 'st2',
'name': 'name',
'driver': 'driver',
'default': True
},
{'service_type': 'st3',
'name': 'name2',
'driver': 'driver2',
'default': True}]
pconf = provconf.ProviderConfiguration(provs)
for prov in provs:
p = pconf.get_service_providers(
filters={'name': [prov['name']],
'service_type': prov['service_type']}
)
self.assertEqual(p, [prov])

3
neutron/tests/unit/test_routerserviceinsertion.py
View File

@ -189,8 +189,7 @@ class RouterServiceInsertionTestCase(base.BaseTestCase):
self._tenant_id = "8c70909f-b081-452d-872b-df48e6c355d1"
res = self._do_request('GET', _get_path('service-types'))
self._service_type_id = res['service_types'][0]['id']
self._service_type_id = _uuid()
self._setup_core_resources()

535
neutron/tests/unit/test_servicetype.py
View File

@ -17,25 +17,22 @@
# @author: Salvatore Orlando, VMware
#
import contextlib
import logging
import os
import tempfile
import fixtures
import mock
from oslo.config import cfg
import webob.exc as webexc
import webtest
from neutron.api import extensions
from neutron.api.v2 import attributes
from neutron.common import exceptions as q_exc
from neutron import context
from neutron.db import api as db_api
from neutron.db import servicetype_db
from neutron.db import servicetype_db as st_db
from neutron.extensions import servicetype
from neutron import manager
from neutron.plugins.common import constants
from neutron.services import provider_configuration as provconf
from neutron.tests import base
from neutron.tests.unit import dummy_plugin as dp
from neutron.tests.unit import test_api_v2
@ -52,17 +49,116 @@ _uuid = test_api_v2._uuid
_get_path = test_api_v2._get_path
class ServiceTypeManagerTestCase(base.BaseTestCase):
def setUp(self):
super(ServiceTypeManagerTestCase, self).setUp()
st_db.ServiceTypeManager._instance = None
self.manager = st_db.ServiceTypeManager.get_instance()
self.ctx = context.get_admin_context()
def test_service_provider_driver_not_unique(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver'],
'service_providers')
prov = {'service_type': constants.LOADBALANCER,
'name': 'name2',
'driver': 'driver',
'default': False}
self.manager._load_conf()
self.assertRaises(
q_exc.Invalid, self.manager.conf.add_provider, prov)
def test_get_service_providers(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path',
constants.DUMMY + ':dummy:dummy_dr'],
'service_providers')
ctx = context.get_admin_context()
provconf.parse_service_provider_opt()
self.manager._load_conf()
res = self.manager.get_service_providers(ctx)
self.assertEqual(len(res), 2)
res = self.manager.get_service_providers(
ctx,
filters=dict(service_type=[constants.DUMMY])
)
self.assertEqual(len(res), 1)
res = self.manager.get_service_providers(
ctx,
filters=dict(service_type=[constants.LOADBALANCER])
)
self.assertEqual(len(res), 1)
def test_multiple_default_providers_specified_for_service(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas1:driver_path:default',
constants.LOADBALANCER +
':lbaas2:driver_path:default'],
'service_providers')
self.assertRaises(q_exc.Invalid, self.manager._load_conf)
def test_get_default_provider(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas1:driver_path:default',
constants.DUMMY +
':lbaas2:driver_path2'],
'service_providers')
self.manager._load_conf()
# can pass None as a context
p = self.manager.get_default_service_provider(None,
constants.LOADBALANCER)
self.assertEqual(p, {'service_type': constants.LOADBALANCER,
'name': 'lbaas1',
'driver': 'driver_path',
'default': True})
self.assertRaises(
provconf.DefaultServiceProviderNotFound,
self.manager.get_default_service_provider,
None, constants.DUMMY
)
def test_add_resource_association(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas1:driver_path:default',
constants.DUMMY +
':lbaas2:driver_path2'],
'service_providers')
self.manager._load_conf()
ctx = context.get_admin_context()
self.manager.add_resource_association(ctx,
constants.LOADBALANCER,
'lbaas1', '123-123')
self.assertEqual(ctx.session.
query(st_db.ProviderResourceAssociation).count(),
1)
assoc = ctx.session.query(st_db.ProviderResourceAssociation).one()
ctx.session.delete(assoc)
def test_invalid_resource_association(self):
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas1:driver_path:default',
constants.DUMMY +
':lbaas2:driver_path2'],
'service_providers')
self.manager._load_conf()
ctx = context.get_admin_context()
self.assertRaises(provconf.ServiceProviderNotFound,
self.manager.add_resource_association,
ctx, 'BLABLA_svc', 'name', '123-123')
class TestServiceTypeExtensionManager(object):
"""Mock extensions manager."""
def get_resources(self):
# Add the resources to the global attribute map
# This is done here as the setup process won't
# initialize the main API router which extends
# the global attribute map
attributes.RESOURCE_ATTRIBUTE_MAP.update(
servicetype.RESOURCE_ATTRIBUTE_MAP)
attributes.RESOURCE_ATTRIBUTE_MAP.update(dp.RESOURCE_ATTRIBUTE_MAP)
return (servicetype.Servicetype.get_resources() +
dp.Dummy.get_resources())
@ -73,13 +169,14 @@ class TestServiceTypeExtensionManager(object):
return []
class ServiceTypeTestCaseBase(testlib_api.WebTestCase):
class ServiceTypeExtensionTestCaseBase(testlib_api.WebTestCase):
fmt = 'json'
def setUp(self):
# This is needed because otherwise a failure will occur due to
# nonexisting core_plugin
cfg.CONF.set_override('core_plugin', test_db_plugin.DB_PLUGIN_KLASS)
cfg.CONF.set_override('service_plugins',
["%s.%s" % (dp.__name__,
dp.DummyServicePlugin.__name__)])
@ -92,418 +189,58 @@ class ServiceTypeTestCaseBase(testlib_api.WebTestCase):
self.ext_mdw = test_extensions.setup_extensions_middleware(ext_mgr)
self.api = webtest.TestApp(self.ext_mdw)
self.resource_name = servicetype.RESOURCE_NAME.replace('-', '_')
super(ServiceTypeTestCaseBase, self).setUp()
super(ServiceTypeExtensionTestCaseBase, self).setUp()
class ServiceTypeExtensionTestCase(ServiceTypeTestCaseBase):
class ServiceTypeExtensionTestCase(ServiceTypeExtensionTestCaseBase):
def setUp(self):
self._patcher = mock.patch(
"%s.%s" % (servicetype_db.__name__,
servicetype_db.ServiceTypeManager.__name__),
"neutron.db.servicetype_db.ServiceTypeManager",
autospec=True)
self.addCleanup(self._patcher.stop)
self.mock_mgr = self._patcher.start()
self.mock_mgr.get_instance.return_value = self.mock_mgr.return_value
super(ServiceTypeExtensionTestCase, self).setUp()
def _test_service_type_create(self, env=None,
expected_status=webexc.HTTPCreated.code):
tenant_id = 'fake'
if env and 'neutron.context' in env:
tenant_id = env['neutron.context'].tenant_id
data = {self.resource_name:
{'name': 'test',
'tenant_id': tenant_id,
'service_definitions':
[{'service_class': constants.DUMMY,
'plugin': dp.DUMMY_PLUGIN_NAME}]}}
return_value = data[self.resource_name].copy()
svc_type_id = _uuid()
return_value['id'] = svc_type_id
def test_service_provider_list(self):
instance = self.mock_mgr.return_value
instance.create_service_type.return_value = return_value
expect_errors = expected_status >= webexc.HTTPBadRequest.code
res = self.api.post(_get_path('service-types', fmt=self.fmt),
self.serialize(data),
extra_environ=env,
expect_errors=expect_errors,
content_type='application/%s' % self.fmt)
self.assertEqual(res.status_int, expected_status)
if not expect_errors:
instance.create_service_type.assert_called_with(mock.ANY,
service_type=data)
res = self.deserialize(res)
self.assertTrue(self.resource_name in res)
svc_type = res[self.resource_name]
self.assertEqual(svc_type['id'], svc_type_id)
# NOTE(salvatore-orlando): The following two checks are
# probably not essential
self.assertEqual(svc_type['service_definitions'],
data[self.resource_name]['service_definitions'])
def _test_service_type_update(self, env=None,
expected_status=webexc.HTTPOk.code):
svc_type_name = 'updated'
data = {self.resource_name: {'name': svc_type_name}}
res = self.api.get(_get_path('service-providers', fmt=self.fmt))
svc_type_id = _uuid()
return_value = {'id': svc_type_id,
'name': svc_type_name}
instance = self.mock_mgr.return_value
expect_errors = expected_status >= webexc.HTTPBadRequest.code
instance.update_service_type.return_value = return_value
res = self.api.put(_get_path('service-types/%s' % svc_type_id,
fmt=self.fmt),
self.serialize(data))
if not expect_errors:
instance.update_service_type.assert_called_with(mock.ANY,
svc_type_id,
service_type=data)
self.assertEqual(res.status_int, webexc.HTTPOk.code)
res = self.deserialize(res)
self.assertTrue(self.resource_name in res)
svc_type = res[self.resource_name]
self.assertEqual(svc_type['id'], svc_type_id)
self.assertEqual(svc_type['name'],
data[self.resource_name]['name'])
def test_service_type_create(self):
self._test_service_type_create()
def test_service_type_update(self):
self._test_service_type_update()
def test_service_type_delete(self):
svctype_id = _uuid()
instance = self.mock_mgr.return_value
res = self.api.delete(_get_path('service-types/%s' % svctype_id,
fmt=self.fmt))
instance.delete_service_type.assert_called_with(mock.ANY,
svctype_id)
self.assertEqual(res.status_int, webexc.HTTPNoContent.code)
def test_service_type_get(self):
svctype_id = _uuid()
return_value = {self.resource_name: {'name': 'test',
'service_definitions': [],
'id': svctype_id}}
instance = self.mock_mgr.return_value
instance.get_service_type.return_value = return_value
res = self.api.get(_get_path('service-types/%s' % svctype_id,
fmt=self.fmt))
instance.get_service_type.assert_called_with(mock.ANY,
svctype_id,
fields=mock.ANY)
instance.get_service_providers.assert_called_with(mock.ANY,
filters={},
fields=[])
self.assertEqual(res.status_int, webexc.HTTPOk.code)
def test_service_type_list(self):
svctype_id = _uuid()
return_value = [{self.resource_name: {'name': 'test',
'service_definitions': [],
'id': svctype_id}}]
instance = self.mock_mgr.return_value
instance.get_service_types.return_value = return_value
res = self.api.get(_get_path('service-types',
fmt=self.fmt))
instance.get_service_types.assert_called_with(mock.ANY,
fields=mock.ANY,
filters=mock.ANY)
self.assertEqual(res.status_int, webexc.HTTPOk.code)
def test_create_service_type_nonadminctx_returns_403(self):
tenant_id = _uuid()
env = {'neutron.context': context.Context('', tenant_id,
is_admin=False)}
self._test_service_type_create(
env=env, expected_status=webexc.HTTPForbidden.code)
def test_create_service_type_adminctx_returns_200(self):
env = {'neutron.context': context.Context('', '', is_admin=True)}
self._test_service_type_create(env=env)
def test_update_service_type_nonadminctx_returns_403(self):
tenant_id = _uuid()
env = {'neutron.context': context.Context('', tenant_id,
is_admin=False)}
self._test_service_type_update(
env=env, expected_status=webexc.HTTPForbidden.code)
def test_update_service_type_adminctx_returns_200(self):
env = {'neutron.context': context.Context('', '', is_admin=True)}
self._test_service_type_update(env=env)
class ServiceTypeExtensionTestCaseXML(ServiceTypeExtensionTestCase):
fmt = 'xml'
class ServiceTypeManagerTestCase(ServiceTypeTestCaseBase):
class ServiceTypeManagerExtTestCase(ServiceTypeExtensionTestCaseBase):
"""Tests ServiceTypemanager as a public API."""
def setUp(self):
# Blank out service type manager instance
servicetype_db.ServiceTypeManager._instance = None
plugin_name = "%s.%s" % (dp.__name__, dp.DummyServicePlugin.__name__)
cfg.CONF.set_override('service_definition', ['dummy:%s' % plugin_name],
group='default_servicetype')
st_db.ServiceTypeManager._instance = None
cfg.CONF.set_override('service_provider',
[constants.LOADBALANCER +
':lbaas:driver_path',
constants.DUMMY + ':dummy:dummy_dr'],
'service_providers')
self.addCleanup(db_api.clear_db)
super(ServiceTypeManagerTestCase, self).setUp()
super(ServiceTypeManagerExtTestCase, self).setUp()
@contextlib.contextmanager
def service_type(self, name='svc_type',
default=True,
service_defs=None,
do_delete=True):
if not service_defs:
service_defs = [{'service_class': constants.DUMMY,
'plugin': dp.DUMMY_PLUGIN_NAME}]
res = self._create_service_type(name, service_defs)
if res.status_int >= 400:
raise webexc.HTTPClientError(code=res.status_int)
svc_type = self.deserialize(res)
yield svc_type
def _list_service_providers(self):
return self.api.get(_get_path('service-providers', fmt=self.fmt))
if do_delete:
# The do_delete parameter allows you to control whether the
# created network is immediately deleted again. Therefore, this
# function is also usable in tests, which require the creation
# of many networks.
self._delete_service_type(svc_type[self.resource_name]['id'])
def _list_service_types(self):
return self.api.get(_get_path('service-types', fmt=self.fmt))
def _show_service_type(self, svctype_id, expect_errors=False):
return self.api.get(_get_path('service-types/%s' % str(svctype_id),
fmt=self.fmt),
expect_errors=expect_errors)
def _create_service_type(self, name, service_defs,
default=None, expect_errors=False):
data = {self.resource_name:
{'name': name,
'service_definitions': service_defs}
}
if default:
data[self.resource_name]['default'] = default
if 'tenant_id' not in data[self.resource_name]:
data[self.resource_name]['tenant_id'] = 'fake'
return self.api.post(_get_path('service-types', fmt=self.fmt),
self.serialize(data),
expect_errors=expect_errors,
content_type='application/%s' % self.fmt)
def _create_dummy(self, dummyname='dummyobject'):
data = {'dummy': {'name': dummyname,
'tenant_id': 'fake'}}
dummy_res = self.api.post(_get_path('dummys', fmt=self.fmt),
self.serialize(data),
content_type='application/%s' % self.fmt)
dummy_res = self.deserialize(dummy_res)
return dummy_res['dummy']
def _update_service_type(self, svc_type_id, name, service_defs,
default=None, expect_errors=False):
data = {self.resource_name:
{'name': name}}
if service_defs is not None:
data[self.resource_name]['service_definitions'] = service_defs
# set this attribute only if True
if default:
data[self.resource_name]['default'] = default
return self.api.put(
_get_path('service-types/%s' % str(svc_type_id), fmt=self.fmt),
self.serialize(data),
expect_errors=expect_errors)
def _delete_service_type(self, svctype_id, expect_errors=False):
return self.api.delete(_get_path('service-types/%s' % str(svctype_id),
fmt=self.fmt),
expect_errors=expect_errors)
def _validate_service_type(self, res, name, service_defs,
svc_type_id=None):
res = self.deserialize(res)
self.assertTrue(self.resource_name in res)
svc_type = res[self.resource_name]
if svc_type_id:
self.assertEqual(svc_type['id'], svc_type_id)
if name:
self.assertEqual(svc_type['name'], name)
if service_defs:
target_defs = []
# unspecified drivers will value None in response
for svc_def in service_defs:
new_svc_def = svc_def.copy()
new_svc_def['driver'] = svc_def.get('driver')
target_defs.append(new_svc_def)
self.assertEqual(svc_type['service_definitions'],
target_defs)
self.assertEqual(svc_type['default'], False)
def _test_service_type_create(self, name='test',
service_defs=DEFAULT_SERVICE_DEFS,
default=None,
expected_status=webexc.HTTPCreated.code):
expect_errors = expected_status >= webexc.HTTPBadRequest.code
res = self._create_service_type(name, service_defs,
default, expect_errors)
self.assertEqual(res.status_int, expected_status)
if not expect_errors:
self.assertEqual(res.status_int, webexc.HTTPCreated.code)
self._validate_service_type(res, name, service_defs)
def _test_service_type_update(self, svc_type_id, name='test-updated',
default=None, service_defs=None,
expected_status=webexc.HTTPOk.code):
expect_errors = expected_status >= webexc.HTTPBadRequest.code
res = self._update_service_type(svc_type_id, name, service_defs,
default, expect_errors)
if not expect_errors:
self.assertEqual(res.status_int, webexc.HTTPOk.code)
self._validate_service_type(res, name, service_defs, svc_type_id)
def test_service_type_create(self):
self._test_service_type_create()
def test_create_service_type_default_returns_400(self):
self._test_service_type_create(
default=True, expected_status=webexc.HTTPBadRequest.code)
def test_create_service_type_no_svcdef_returns_400(self):
self._test_service_type_create(
service_defs=None,
expected_status=webexc.HTTPBadRequest.code)
def test_service_type_update_name(self):
with self.service_type() as svc_type:
self._test_service_type_update(svc_type[self.resource_name]['id'])
def test_service_type_update_set_default_returns_400(self):
with self.service_type() as svc_type:
self._test_service_type_update(
svc_type[self.resource_name]['id'], default=True,
expected_status=webexc.HTTPBadRequest.code)
def test_service_type_update_clear_svc_defs_returns_400(self):
with self.service_type() as svc_type:
self._test_service_type_update(
svc_type[self.resource_name]['id'], service_defs=[],
expected_status=webexc.HTTPBadRequest.code)
def test_service_type_update_svc_defs(self):
with self.service_type() as svc_type:
svc_defs = [{'service': constants.DUMMY,
'plugin': 'foobar'}]
self._test_service_type_update(
svc_type[self.resource_name]['id'], service_defs=svc_defs,
expected_status=webexc.HTTPBadRequest.code)
def test_list_service_types(self):
with contextlib.nested(self.service_type('st1'),
self.service_type('st2')):
res = self._list_service_types()
self.assertEqual(res.status_int, webexc.HTTPOk.code)
data = self.deserialize(res)
self.assertTrue('service_types' in data)
# it must be 3 because we have the default service type too!
self.assertEqual(len(data['service_types']), 3)
def test_get_default_service_type(self):
res = self._list_service_types()
def test_list_service_providers(self):
res = self._list_service_providers()
self.assertEqual(res.status_int, webexc.HTTPOk.code)
data = self.deserialize(res)
self.assertTrue('service_types' in data)
self.assertEqual(len(data['service_types']), 1)
def_svc_type = data['service_types'][0]
self.assertEqual(def_svc_type['default'], True)
def test_get_service_type(self):
with self.service_type() as svc_type:
svc_type_data = svc_type[self.resource_name]
res = self._show_service_type(svc_type_data['id'])
self.assertEqual(res.status_int, webexc.HTTPOk.code)
self._validate_service_type(res, svc_type_data['name'],
svc_type_data['service_definitions'],
svc_type_data['id'])
def test_delete_service_type_in_use_returns_409(self):
with self.service_type() as svc_type:
svc_type_data = svc_type[self.resource_name]
mgr = servicetype_db.ServiceTypeManager.get_instance()
ctx = context.Context('', '', is_admin=True)
mgr.increase_service_type_refcount(ctx, svc_type_data['id'])
res = self._delete_service_type(svc_type_data['id'], True)
self.assertEqual(res.status_int, webexc.HTTPConflict.code)
mgr.decrease_service_type_refcount(ctx, svc_type_data['id'])
def test_create_dummy_increases_service_type_refcount(self):
dummy = self._create_dummy()
svc_type_res = self._show_service_type(dummy['service_type'])
svc_type_res = self.deserialize(svc_type_res)
svc_type = svc_type_res[self.resource_name]
self.assertEqual(svc_type['num_instances'], 1)
def test_delete_dummy_decreases_service_type_refcount(self):
dummy = self._create_dummy()
svc_type_res = self._show_service_type(dummy['service_type'])
svc_type_res = self.deserialize(svc_type_res)
svc_type = svc_type_res[self.resource_name]
self.assertEqual(svc_type['num_instances'], 1)
self.api.delete(_get_path('dummys/%s' % str(dummy['id']),
fmt=self.fmt))
svc_type_res = self._show_service_type(dummy['service_type'])
svc_type_res = self.deserialize(svc_type_res)
svc_type = svc_type_res[self.resource_name]
self.assertEqual(svc_type['num_instances'], 0)
self.assertTrue('service_providers' in data)
self.assertEqual(len(data['service_providers']), 2)
class ServiceTypeManagerTestCaseXML(ServiceTypeManagerTestCase):
class ServiceTypeManagerExtTestCaseXML(ServiceTypeManagerExtTestCase):
fmt = 'xml'
class CompatServiceTypeConfigTestCase(base.BaseTestCase):
def setUp(self):
super(CompatServiceTypeConfigTestCase, self).setUp()
self.useFixture(fixtures.NestedTempfile())
self.conf = cfg.ConfigOpts()
self.conf.register_opts(servicetype_db.default_servicetype_opts,
'default_servicetype')
def _write_neutron_conf(self, contents):
(fd, path) = tempfile.mkstemp(prefix='neutron-', suffix='.conf')
try:
os.write(fd, contents)
finally:
os.close(fd)
return path
def _test_default_servicetype_section(self, section_name):
path = self._write_neutron_conf(
'[%(section_name)s]\n'
'description = test service type\n'
'service_definition=test:testing.NeutronTestPlugin\n' %
{'section_name': section_name})
self.conf(['--config-file', path])
self.assertEqual(self.conf.default_servicetype.description,
'test service type')
self.assertEqual(self.conf.default_servicetype.service_definition,
['test:testing.NeutronTestPlugin'])
def test_default_servicetype_lowercase(self):
self._test_default_servicetype_section('default_servicetype')
def test_default_servicetype_uppercase(self):
self._test_default_servicetype_section('DEFAULT_SERVICETYPE')