--- /dev/null
+# Neutron flavors plugin
+# ----------------------
+
+FLAVORS_PLUGIN=neutron.services.flavors.flavors_plugin.FlavorsPlugin
+
+function configure_flavors {
+ _neutron_service_plugin_class_add $FLAVORS_PLUGIN
+}
LIBDIR=$DEST/neutron/devstack/lib
+source $LIBDIR/flavors
source $LIBDIR/l2_agent
source $LIBDIR/l2_agent_sriovnicswitch
source $LIBDIR/ml2
if [[ "$1" == "stack" ]]; then
case "$2" in
install)
+ if is_service_enabled q-flavors; then
+ configure_flavors
+ fi
if is_service_enabled q-qos; then
configure_qos
fi
if is_service_enabled q-sriov-agt; then
stop_l2_agent_sriov
fi
-fi
\ No newline at end of file
+fi
"update_rbac_policy": "rule:admin_or_owner",
"update_rbac_policy:target_tenant": "rule:restrict_wildcard and rule:admin_or_owner",
"get_rbac_policy": "rule:admin_or_owner",
- "delete_rbac_policy": "rule:admin_or_owner"
+ "delete_rbac_policy": "rule:admin_or_owner",
+
+ "create_flavor_service_profile": "rule:admin_only",
+ "delete_flavor_service_profile": "rule:admin_only",
+ "get_flavor_service_profile": "rule:regular_user"
}
NAME_MAX_LEN = 255
TENANT_ID_MAX_LEN = 255
DESCRIPTION_MAX_LEN = 255
+LONG_DESCRIPTION_MAX_LEN = 1024
DEVICE_ID_MAX_LEN = 255
DEVICE_OWNER_MAX_LEN = 255
# under the License.
from oslo_log import log as logging
-from oslo_serialization import jsonutils
-from oslo_utils import importutils
from oslo_utils import uuidutils
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc as sa_exc
-from neutron.common import exceptions as qexception
from neutron.db import common_db_mixin
from neutron.db import model_base
from neutron.db import models_v2
-from neutron.plugins.common import constants
-
+from neutron.db import servicetype_db as sdb
+from neutron.extensions import flavors as ext_flavors
LOG = logging.getLogger(__name__)
-# Flavor Exceptions
-class FlavorNotFound(qexception.NotFound):
- message = _("Flavor %(flavor_id)s could not be found")
-
-
-class FlavorInUse(qexception.InUse):
- message = _("Flavor %(flavor_id)s is used by some service instance")
-
-
-class ServiceProfileNotFound(qexception.NotFound):
- message = _("Service Profile %(sp_id)s could not be found")
-
-
-class ServiceProfileInUse(qexception.InUse):
- message = _("Service Profile %(sp_id)s is used by some service instance")
-
-
-class FlavorServiceProfileBindingExists(qexception.Conflict):
- message = _("Service Profile %(sp_id)s is already associated "
- "with flavor %(fl_id)s")
-
-
-class FlavorServiceProfileBindingNotFound(qexception.NotFound):
- message = _("Service Profile %(sp_id)s is not associated "
- "with flavor %(fl_id)s")
-
-
-class DummyCorePlugin(object):
- pass
-
-
-class DummyServicePlugin(object):
-
- def driver_loaded(self, driver, service_profile):
- pass
-
- def get_plugin_type(self):
- return constants.DUMMY
-
- def get_plugin_description(self):
- return "Dummy service plugin, aware of flavors"
-
-
-class DummyServiceDriver(object):
-
- @staticmethod
- def get_service_type():
- return constants.DUMMY
-
- def __init__(self, plugin):
- pass
-
-
class Flavor(model_base.BASEV2, models_v2.HasId):
name = sa.Column(sa.String(255))
description = sa.Column(sa.String(1024))
service_profile = orm.relationship(ServiceProfile)
-class FlavorManager(common_db_mixin.CommonDbMixin):
- """Class to support flavors and service profiles."""
-
- supported_extension_aliases = ["flavors"]
-
- def __init__(self, manager=None):
- # manager = None is UT usage where FlavorManager is loaded as
- # a core plugin
- self.manager = manager
-
- def get_plugin_name(self):
- return constants.FLAVORS
+class FlavorsDbMixin(common_db_mixin.CommonDbMixin):
- def get_plugin_type(self):
- return constants.FLAVORS
-
- def get_plugin_description(self):
- return "Neutron Flavors and Service Profiles manager plugin"
+ """Class to support flavors and service profiles."""
def _get_flavor(self, context, flavor_id):
try:
return self._get_by_id(context, Flavor, flavor_id)
except sa_exc.NoResultFound:
- raise FlavorNotFound(flavor_id=flavor_id)
+ raise ext_flavors.FlavorNotFound(flavor_id=flavor_id)
def _get_service_profile(self, context, sp_id):
try:
return self._get_by_id(context, ServiceProfile, sp_id)
except sa_exc.NoResultFound:
- raise ServiceProfileNotFound(sp_id=sp_id)
+ raise ext_flavors.ServiceProfileNotFound(sp_id=sp_id)
def _make_flavor_dict(self, flavor_db, fields=None):
res = {'id': flavor_db['id'],
pass
def _ensure_service_profile_not_in_use(self, context, sp_id):
- # Future TODO(enikanorov): check that there is no binding to instances
- # and no binding to flavors. Shall be addressed in future
+ """Ensures no current bindings to flavors exist."""
fl = (context.session.query(FlavorServiceProfileBinding).
filter_by(service_profile_id=sp_id).first())
if fl:
- raise ServiceProfileInUse(sp_id=sp_id)
+ raise ext_flavors.ServiceProfileInUse(sp_id=sp_id)
+
+ def _validate_driver(self, context, driver):
+ """Confirms a non-empty driver is a valid provider."""
+ service_type_manager = sdb.ServiceTypeManager.get_instance()
+ providers = service_type_manager.get_service_providers(
+ context,
+ filters={'driver': driver})
+
+ if not providers:
+ raise ext_flavors.ServiceProfileDriverNotFound(driver=driver)
def create_flavor(self, context, flavor):
fl = flavor['flavor']
self._ensure_flavor_not_in_use(context, flavor_id)
fl_db = self._get_flavor(context, flavor_id)
fl_db.update(fl)
-
return self._make_flavor_dict(fl_db)
def get_flavor(self, context, flavor_id, fields=None):
binding = bind_qry.filter_by(service_profile_id=sp['id'],
flavor_id=flavor_id).first()
if binding:
- raise FlavorServiceProfileBindingExists(
+ raise ext_flavors.FlavorServiceProfileBindingExists(
sp_id=sp['id'], fl_id=flavor_id)
binding = FlavorServiceProfileBinding(
service_profile_id=sp['id'],
flavor_id=flavor_id)
context.session.add(binding)
fl_db = self._get_flavor(context, flavor_id)
- sps = [x['service_profile_id'] for x in fl_db.service_profiles]
- return sps
+ return self._make_flavor_dict(fl_db)
def delete_flavor_service_profile(self, context,
service_profile_id, flavor_id):
filter_by(service_profile_id=service_profile_id,
flavor_id=flavor_id).first())
if not binding:
- raise FlavorServiceProfileBindingNotFound(
+ raise ext_flavors.FlavorServiceProfileBindingNotFound(
sp_id=service_profile_id, fl_id=flavor_id)
context.session.delete(binding)
filter_by(service_profile_id=service_profile_id,
flavor_id=flavor_id).first())
if not binding:
- raise FlavorServiceProfileBindingNotFound(
+ raise ext_flavors.FlavorServiceProfileBindingNotFound(
sp_id=service_profile_id, fl_id=flavor_id)
res = {'service_profile_id': service_profile_id,
'flavor_id': flavor_id}
return self._fields(res, fields)
- def _load_dummy_driver(self, driver):
- driver = DummyServiceDriver
- driver_klass = driver
- return driver_klass
-
- def _load_driver(self, profile):
- driver_klass = importutils.import_class(profile.driver)
- return driver_klass
-
def create_service_profile(self, context, service_profile):
sp = service_profile['service_profile']
- with context.session.begin(subtransactions=True):
- driver_klass = self._load_dummy_driver(sp['driver'])
- # 'get_service_type' must be a static method so it can't be changed
- svc_type = DummyServiceDriver.get_service_type()
- sp_db = ServiceProfile(id=uuidutils.generate_uuid(),
- description=sp['description'],
- driver=svc_type,
- enabled=sp['enabled'],
- metainfo=jsonutils.dumps(sp['metainfo']))
- context.session.add(sp_db)
- try:
- # driver_klass = self._load_dummy_driver(sp_db)
- # Future TODO(madhu_ak): commented for now to load dummy driver
- # until there is flavor supported driver
- # plugin = self.manager.get_service_plugins()[svc_type]
- # plugin.driver_loaded(driver_klass(plugin), sp_db)
- # svc_type = DummyServiceDriver.get_service_type()
- # plugin = self.manager.get_service_plugins()[svc_type]
- # plugin = FlavorManager(manager.NeutronManager().get_instance())
- # plugin = DummyServicePlugin.get_plugin_type(svc_type)
- plugin = DummyServicePlugin()
- plugin.driver_loaded(driver_klass(svc_type), sp_db)
- except Exception:
- # Future TODO(enikanorov): raise proper exception
- self.delete_service_profile(context, sp_db['id'])
- raise
- return self._make_service_profile_dict(sp_db)
+ if sp['driver']:
+ self._validate_driver(context, sp['driver'])
+ else:
+ if not sp['metainfo']:
+ raise ext_flavors.ServiceProfileEmpty()
- def unit_create_service_profile(self, context, service_profile):
- # Note: Triggered by unit tests pointing to dummy driver
- sp = service_profile['service_profile']
with context.session.begin(subtransactions=True):
sp_db = ServiceProfile(id=uuidutils.generate_uuid(),
description=sp['description'],
enabled=sp['enabled'],
metainfo=sp['metainfo'])
context.session.add(sp_db)
- try:
- driver_klass = self._load_driver(sp_db)
- # require get_service_type be a static method
- svc_type = driver_klass.get_service_type()
- plugin = self.manager.get_service_plugins()[svc_type]
- plugin.driver_loaded(driver_klass(plugin), sp_db)
- except Exception:
- # Future TODO(enikanorov): raise proper exception
- self.delete_service_profile(context, sp_db['id'])
- raise
+
return self._make_service_profile_dict(sp_db)
def update_service_profile(self, context,
service_profile_id, service_profile):
sp = service_profile['service_profile']
+
+ if sp.get('driver'):
+ self._validate_driver(context, sp['driver'])
+
with context.session.begin(subtransactions=True):
self._ensure_service_profile_not_in_use(context,
service_profile_id)
sorts=sorts, limit=limit,
marker_obj=marker,
page_reverse=page_reverse)
+
+ def get_flavor_next_provider(self, context, flavor_id,
+ filters=None, fields=None,
+ sorts=None, limit=None,
+ marker=None, page_reverse=False):
+ """From flavor, choose service profile and find provider for driver."""
+
+ with context.session.begin(subtransactions=True):
+ bind_qry = context.session.query(FlavorServiceProfileBinding)
+ binding = bind_qry.filter_by(flavor_id=flavor_id).first()
+ if not binding:
+ raise ext_flavors.FlavorServiceProfileBindingNotFound(
+ sp_id='', fl_id=flavor_id)
+
+ # Get the service profile from the first binding
+ # TODO(jwarendt) Should become a scheduling framework instead
+ sp_db = self._get_service_profile(context,
+ binding['service_profile_id'])
+
+ if not sp_db.enabled:
+ raise ext_flavors.ServiceProfileDisabled()
+
+ LOG.debug("Found driver %s.", sp_db.driver)
+
+ service_type_manager = sdb.ServiceTypeManager.get_instance()
+ providers = service_type_manager.get_service_providers(
+ context,
+ filters={'driver': sp_db.driver})
+
+ if not providers:
+ raise ext_flavors.ServiceProfileDriverNotFound(driver=sp_db.driver)
+
+ LOG.debug("Found providers %s.", providers)
+
+ res = {'driver': sp_db.driver,
+ 'provider': providers[0].get('name')}
+
+ return [self._fields(res, fields)]
from neutron.api.v2 import attributes as attr
from neutron.api.v2 import base
from neutron.api.v2 import resource_helper
+from neutron.common import exceptions as nexception
from neutron import manager
from neutron.plugins.common import constants
+# Flavor Exceptions
+class FlavorNotFound(nexception.NotFound):
+ message = _("Flavor %(flavor_id)s could not be found.")
+
+
+class FlavorInUse(nexception.InUse):
+ message = _("Flavor %(flavor_id)s is used by some service instance.")
+
+
+class ServiceProfileNotFound(nexception.NotFound):
+ message = _("Service Profile %(sp_id)s could not be found.")
+
+
+class ServiceProfileInUse(nexception.InUse):
+ message = _("Service Profile %(sp_id)s is used by some service instance.")
+
+
+class FlavorServiceProfileBindingExists(nexception.Conflict):
+ message = _("Service Profile %(sp_id)s is already associated "
+ "with flavor %(fl_id)s.")
+
+
+class FlavorServiceProfileBindingNotFound(nexception.NotFound):
+ message = _("Service Profile %(sp_id)s is not associated "
+ "with flavor %(fl_id)s.")
+
+
+class ServiceProfileDriverNotFound(nexception.NotFound):
+ message = _("Service Profile driver %(driver)s could not be found.")
+
+
+class ServiceProfileEmpty(nexception.InvalidInput):
+ message = _("Service Profile needs either a driver or metainfo.")
+
+
+class FlavorDisabled(nexception.ServiceUnavailable):
+ message = _("Flavor is not enabled.")
+
+
+class ServiceProfileDisabled(nexception.ServiceUnavailable):
+ message = _("Service Profile is not enabled.")
+
+
+class InvalidFlavorServiceType(nexception.InvalidInput):
+ message = _("Invalid service type %(service_type)s.")
+
+
+def _validate_flavor_service_type(validate_type, valid_values=None):
+ """Ensure requested flavor service type plugin is loaded."""
+ plugins = manager.NeutronManager.get_service_plugins()
+ if validate_type not in plugins:
+ raise InvalidFlavorServiceType(service_type=validate_type)
+
+attr.validators['type:validate_flavor_service_type'] = (
+ _validate_flavor_service_type)
+
FLAVORS = 'flavors'
SERVICE_PROFILES = 'service_profiles'
FLAVORS_PREFIX = ""
'is_visible': True,
'primary_key': True},
'name': {'allow_post': True, 'allow_put': True,
- 'validate': {'type:string': None},
+ 'validate': {'type:string': attr.NAME_MAX_LEN},
'is_visible': True, 'default': ''},
'description': {'allow_post': True, 'allow_put': True,
- 'validate': {'type:string': None},
+ 'validate': {'type:string_or_none':
+ attr.LONG_DESCRIPTION_MAX_LEN},
'is_visible': True, 'default': ''},
'service_type': {'allow_post': True, 'allow_put': False,
- 'validate': {'type:string': None},
+ 'validate':
+ {'type:validate_flavor_service_type': None},
'is_visible': True},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'validate': {'type:uuid_list': None},
'is_visible': True, 'default': []},
'enabled': {'allow_post': True, 'allow_put': True,
- 'validate': {'type:boolean': None},
+ 'convert_to': attr.convert_to_boolean_if_not_none,
'default': True,
'is_visible': True},
},
SERVICE_PROFILES: {
'id': {'allow_post': False, 'allow_put': False,
+ 'validate': {'type:uuid': None},
'is_visible': True,
'primary_key': True},
'description': {'allow_post': True, 'allow_put': True,
- 'validate': {'type:string': None},
- 'is_visible': True},
- # service_profile belong to one service type for now
- #'service_types': {'allow_post': False, 'allow_put': False,
- # 'is_visible': True},
- 'driver': {'allow_post': True, 'allow_put': False,
- 'validate': {'type:string': None},
+ 'validate': {'type:string_or_none':
+ attr.LONG_DESCRIPTION_MAX_LEN},
+ 'is_visible': True, 'default': ''},
+ 'driver': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:string':
+ attr.LONG_DESCRIPTION_MAX_LEN},
'is_visible': True,
- 'default': attr.ATTR_NOT_SPECIFIED},
+ 'default': ''},
'metainfo': {'allow_post': True, 'allow_put': True,
- 'is_visible': True},
+ 'is_visible': True,
+ 'default': ''},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'validate': {'type:string': attr.TENANT_ID_MAX_LEN},
'is_visible': True},
'enabled': {'allow_post': True, 'allow_put': True,
- 'validate': {'type:boolean': None},
+ 'convert_to': attr.convert_to_boolean_if_not_none,
'is_visible': True, 'default': True},
},
}
SUB_RESOURCE_ATTRIBUTE_MAP = {
+ 'next_providers': {
+ 'parent': {'collection_name': 'flavors',
+ 'member_name': 'flavor'},
+ 'parameters': {'provider': {'allow_post': False,
+ 'allow_put': False,
+ 'is_visible': True},
+ 'driver': {'allow_post': False,
+ 'allow_put': False,
+ 'is_visible': True},
+ 'metainfo': {'allow_post': False,
+ 'allow_put': False,
+ 'is_visible': True},
+ 'tenant_id': {'allow_post': True, 'allow_put': False,
+ 'required_by_policy': True,
+ 'validate': {'type:string':
+ attr.TENANT_ID_MAX_LEN},
+ 'is_visible': True}}
+ },
'service_profiles': {
'parent': {'collection_name': 'flavors',
'member_name': 'flavor'},
@classmethod
def get_description(cls):
- return "Service specification for advanced services"
+ return "Flavor specification for Neutron advanced services"
@classmethod
def get_updated(cls):
- return "2014-07-06T10:00:00-00:00"
+ return "2015-09-17T10:00:00-00:00"
@classmethod
def get_resources(cls):
import six
from neutron.common import utils
-from neutron.db import flavors_db
from neutron.i18n import _LI
from neutron.plugins.common import constants
LOG.info(_LI("Service %s is supported by the core plugin"),
service_type)
- def _load_flavors_manager(self):
- # pass manager instance to resolve cyclical import dependency
- self.service_plugins[constants.FLAVORS] = (
- flavors_db.FlavorManager(self))
-
def _load_service_plugins(self):
"""Loads service plugins.
"Description: %(desc)s",
{"type": plugin_inst.get_plugin_type(),
"desc": plugin_inst.get_plugin_description()})
- # do it after the loading from conf to avoid conflict with
- # configuration provided by unit tests.
- self._load_flavors_manager()
@classmethod
@utils.synchronized("manager")
--- /dev/null
+# Copyright (c) 2015, Hewlett-Packard Development Company, L.P.
+# 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 neutron.db import flavors_db
+from neutron.plugins.common import constants
+from neutron.services import service_base
+
+
+class FlavorsPlugin(service_base.ServicePluginBase,
+ flavors_db.FlavorsDbMixin):
+ """Implements Neutron Flavors Service plugin."""
+
+ supported_extension_aliases = ['flavors']
+
+ def get_plugin_type(self):
+ return constants.FLAVORS
+
+ def get_plugin_description(self):
+ return "Neutron Flavors and Service Profiles manager plugin"
# under the License.
from oslo_log import log as logging
+from tempest_lib import exceptions as lib_exc
from neutron.tests.api import base
from neutron.tests.tempest import test
if not test.is_extension_enabled('flavors', 'network'):
msg = "flavors extension not enabled."
raise cls.skipException(msg)
- service_type = "LOADBALANCER"
+
+ # Use flavors service type as know this is loaded
+ service_type = "FLAVORS"
description_flavor = "flavor is created by tempest"
name_flavor = "Best flavor created by tempest"
- cls.flavor = cls.create_flavor(name_flavor, description_flavor,
- service_type)
+
+ # The check above will pass if api_extensions=all, which does
+ # not mean flavors extension itself is present.
+ try:
+ cls.flavor = cls.create_flavor(name_flavor, description_flavor,
+ service_type)
+ except lib_exc.NotFound:
+ msg = "flavors plugin not enabled."
+ raise cls.skipException(msg)
+
description_sp = "service profile created by tempest"
- # Future TODO(madhu_ak): Right now the dummy driver is loaded. Will
- # make changes as soon I get to know the flavor supported drivers
+ # Drivers are supported as is an empty driver field. Use an
+ # empty field for now since otherwise driver is validated against the
+ # servicetype configuration which may differ in test scenarios.
driver = ""
metainfo = '{"data": "value"}'
cls.service_profile = cls.create_service_profile(
def test_create_update_delete_flavor(self):
# Creates a flavor
description = "flavor created by tempest"
- service = "LOADBALANCERS"
+ service = "FLAVORS"
name = "Best flavor created by tempest"
body = self.admin_client.create_flavor(name=name, service_type=service,
description=description)
"update_rbac_policy": "rule:admin_or_owner",
"update_rbac_policy:target_tenant": "rule:restrict_wildcard and rule:admin_or_owner",
"get_rbac_policy": "rule:admin_or_owner",
- "delete_rbac_policy": "rule:admin_or_owner"
+ "delete_rbac_policy": "rule:admin_or_owner",
+
+ "create_flavor_service_profile": "rule:admin_only",
+ "delete_flavor_service_profile": "rule:admin_only",
+ "get_flavor_service_profile": "rule:regular_user"
}
from oslo_config import cfg
from oslo_utils import uuidutils
+from webob import exc
+from neutron.api.v2 import attributes as attr
from neutron import context
from neutron.db import api as dbapi
from neutron.db import flavors_db
+from neutron.db import servicetype_db
from neutron.extensions import flavors
-from neutron import manager
from neutron.plugins.common import constants
+from neutron.services.flavors import flavors_plugin
+from neutron.services import provider_configuration as provconf
from neutron.tests import base
from neutron.tests.unit.api.v2 import test_base
from neutron.tests.unit.db import test_db_base_plugin_v2
_uuid = uuidutils.generate_uuid
_get_path = test_base._get_path
+_driver = ('neutron.tests.unit.extensions.test_flavors.'
+ 'DummyServiceDriver')
+_provider = 'dummy'
+_long_name = 'x' * (attr.NAME_MAX_LEN + 1)
+_long_description = 'x' * (attr.LONG_DESCRIPTION_MAX_LEN + 1)
+
class FlavorExtensionTestCase(extension.ExtensionTestCase):
def setUp(self):
super(FlavorExtensionTestCase, self).setUp()
self._setUpExtension(
- 'neutron.db.flavors_db.FlavorManager',
+ 'neutron.services.flavors.flavors_plugin.FlavorsPlugin',
constants.FLAVORS, flavors.RESOURCE_ATTRIBUTE_MAP,
flavors.Flavors, '', supported_extension_aliases='flavors')
def test_create_flavor(self):
tenant_id = uuidutils.generate_uuid()
+ # Use service_type FLAVORS since plugin must be loaded to validate
data = {'flavor': {'name': 'GOLD',
- 'service_type': constants.LOADBALANCER,
+ 'service_type': constants.FLAVORS,
'description': 'the best flavor',
'tenant_id': tenant_id,
'enabled': True}}
self.assertIn('flavor', res)
self.assertEqual(expected, res)
+ def test_create_flavor_invalid_service_type(self):
+ tenant_id = uuidutils.generate_uuid()
+ data = {'flavor': {'name': 'GOLD',
+ 'service_type': 'BROKEN',
+ 'description': 'the best flavor',
+ 'tenant_id': tenant_id,
+ 'enabled': True}}
+ self.api.post(_get_path('flavors', fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_create_flavor_too_long_name(self):
+ tenant_id = uuidutils.generate_uuid()
+ data = {'flavor': {'name': _long_name,
+ 'service_type': constants.FLAVORS,
+ 'description': 'the best flavor',
+ 'tenant_id': tenant_id,
+ 'enabled': True}}
+ self.api.post(_get_path('flavors', fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_create_flavor_too_long_description(self):
+ tenant_id = uuidutils.generate_uuid()
+ data = {'flavor': {'name': _long_name,
+ 'service_type': constants.FLAVORS,
+ 'description': _long_description,
+ 'tenant_id': tenant_id,
+ 'enabled': True}}
+ self.api.post(_get_path('flavors', fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_create_flavor_invalid_enabled(self):
+ tenant_id = uuidutils.generate_uuid()
+ data = {'flavor': {'name': _long_name,
+ 'service_type': constants.FLAVORS,
+ 'description': 'the best flavor',
+ 'tenant_id': tenant_id,
+ 'enabled': 'BROKEN'}}
+ self.api.post(_get_path('flavors', fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
def test_update_flavor(self):
flavor_id = 'fake_id'
data = {'flavor': {'name': 'GOLD',
self.assertIn('flavor', res)
self.assertEqual(expected, res)
+ def test_update_flavor_too_long_name(self):
+ flavor_id = 'fake_id'
+ data = {'flavor': {'name': _long_name,
+ 'description': 'the best flavor',
+ 'enabled': True}}
+ self.api.put(_get_path('flavors', id=flavor_id, fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_update_flavor_too_long_description(self):
+ flavor_id = 'fake_id'
+ data = {'flavor': {'name': 'GOLD',
+ 'description': _long_description,
+ 'enabled': True}}
+ self.api.put(_get_path('flavors', id=flavor_id, fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_update_flavor_invalid_enabled(self):
+ flavor_id = 'fake_id'
+ data = {'flavor': {'name': 'GOLD',
+ 'description': _long_description,
+ 'enabled': 'BROKEN'}}
+ self.api.put(_get_path('flavors', id=flavor_id, fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
def test_delete_flavor(self):
flavor_id = 'fake_id'
instance = self.plugin.return_value
self.assertIn('service_profile', res)
self.assertEqual(expected, res)
+ def test_create_service_profile_too_long_description(self):
+ tenant_id = uuidutils.generate_uuid()
+ expected = {'service_profile': {'description': _long_description,
+ 'driver': '',
+ 'tenant_id': tenant_id,
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ self.api.post(_get_path('service_profiles', fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_create_service_profile_too_long_driver(self):
+ tenant_id = uuidutils.generate_uuid()
+ expected = {'service_profile': {'description': 'the best sp',
+ 'driver': _long_description,
+ 'tenant_id': tenant_id,
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ self.api.post(_get_path('service_profiles', fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_create_service_profile_invalid_enabled(self):
+ tenant_id = uuidutils.generate_uuid()
+ expected = {'service_profile': {'description': 'the best sp',
+ 'driver': '',
+ 'tenant_id': tenant_id,
+ 'enabled': 'BROKEN',
+ 'metainfo': '{"data": "value"}'}}
+ self.api.post(_get_path('service_profiles', fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
def test_update_service_profile(self):
sp_id = "fake_id"
expected = {'service_profile': {'description': 'the best sp',
self.assertIn('service_profile', res)
self.assertEqual(expected, res)
+ def test_update_service_profile_too_long_description(self):
+ sp_id = "fake_id"
+ expected = {'service_profile': {'description': 'the best sp',
+ 'enabled': 'BROKEN',
+ 'metainfo': '{"data1": "value3"}'}}
+ self.api.put(_get_path('service_profiles',
+ id=sp_id, fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
+ def test_update_service_profile_invalid_enabled(self):
+ sp_id = "fake_id"
+ expected = {'service_profile': {'description': 'the best sp',
+ 'enabled': 'BROKEN',
+ 'metainfo': '{"data1": "value3"}'}}
+ self.api.put(_get_path('service_profiles',
+ id=sp_id, fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
def test_delete_service_profile(self):
sp_id = 'fake_id'
instance = self.plugin.return_value
def test_show_service_profile(self):
sp_id = 'fake_id'
expected = {'service_profile': {'id': 'id1',
- 'driver': 'entrypoint1',
+ 'driver': _driver,
'description': 'desc',
'metainfo': '{}',
'enabled': True}}
def test_get_service_profiles(self):
expected = {'service_profiles': [{'id': 'id1',
- 'driver': 'entrypoint1',
+ 'driver': _driver,
'description': 'desc',
'metainfo': '{}',
'enabled': True},
{'id': 'id2',
- 'driver': 'entrypoint2',
+ 'driver': _driver,
'description': 'desc',
'metainfo': '{}',
'enabled': True}]}
'fake_spid',
flavor_id='fl_id')
+ def test_update_association_error(self):
+ """Confirm that update is not permitted with user error."""
+ new_id = uuidutils.generate_uuid()
+ data = {'service_profile': {'id': new_id}}
+ self.api.put('/flavors/fl_id/service_profiles/%s' % 'fake_spid',
+ self.serialize(data),
+ content_type='application/%s' % self.fmt,
+ status=exc.HTTPBadRequest.code)
+
class DummyCorePlugin(object):
pass
pass
-class FlavorManagerTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
- base.PluginFixture):
+class FlavorPluginTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
+ base.PluginFixture):
def setUp(self):
- super(FlavorManagerTestCase, self).setUp()
+ super(FlavorPluginTestCase, self).setUp()
self.config_parse()
cfg.CONF.set_override(
self.useFixture(
fixtures.MonkeyPatch('neutron.manager.NeutronManager._instance'))
- self.plugin = flavors_db.FlavorManager(
- manager.NeutronManager().get_instance())
+ self.plugin = flavors_plugin.FlavorsPlugin()
self.ctx = context.get_admin_context()
+
+ providers = [DummyServiceDriver.get_service_type() +
+ ":" + _provider + ":" + _driver]
+ self.service_manager = servicetype_db.ServiceTypeManager.get_instance()
+ self.service_providers = mock.patch.object(
+ provconf.NeutronModule, 'service_providers').start()
+ self.service_providers.return_value = providers
+ for provider in providers:
+ self.service_manager.add_provider_configuration(
+ provider.split(':')[0], provconf.ProviderConfiguration())
+
dbapi.get_engine()
def _create_flavor(self, description=None):
flavor = {'flavor': {'name': 'GOLD',
- 'service_type': constants.LOADBALANCER,
+ 'service_type': constants.DUMMY,
'description': description or 'the best flavor',
'enabled': True}}
return self.plugin.create_flavor(self.ctx, flavor), flavor
res = self.ctx.session.query(flavors_db.Flavor).all()
self.assertEqual(1, len(res))
self.assertEqual('GOLD', res[0]['name'])
- self.assertEqual(constants.LOADBALANCER, res[0]['service_type'])
+ self.assertEqual(constants.DUMMY, res[0]['service_type'])
def test_update_flavor(self):
fl, flavor = self._create_flavor()
def _create_service_profile(self, description=None):
data = {'service_profile':
{'description': description or 'the best sp',
- 'driver':
- ('neutron.tests.unit.extensions.test_flavors.'
- 'DummyServiceDriver'),
+ 'driver': _driver,
'enabled': True,
'metainfo': '{"data": "value"}'}}
- sp = self.plugin.unit_create_service_profile(self.ctx,
- data)
+ sp = self.plugin.create_service_profile(self.ctx,
+ data)
return sp, data
def test_create_service_profile(self):
self.assertEqual(data['service_profile']['driver'], res['driver'])
self.assertEqual(data['service_profile']['metainfo'], res['metainfo'])
+ def test_create_service_profile_empty_driver(self):
+ data = {'service_profile':
+ {'description': 'the best sp',
+ 'driver': '',
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ sp = self.plugin.create_service_profile(self.ctx,
+ data)
+ res = (self.ctx.session.query(flavors_db.ServiceProfile).
+ filter_by(id=sp['id']).one())
+ self.assertEqual(data['service_profile']['driver'], res['driver'])
+ self.assertEqual(data['service_profile']['metainfo'], res['metainfo'])
+
+ def test_create_service_profile_invalid_driver(self):
+ data = {'service_profile':
+ {'description': 'the best sp',
+ 'driver': "Broken",
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ self.assertRaises(flavors.ServiceProfileDriverNotFound,
+ self.plugin.create_service_profile,
+ self.ctx,
+ data)
+
+ def test_create_service_profile_invalid_empty(self):
+ data = {'service_profile':
+ {'description': '',
+ 'driver': '',
+ 'enabled': True,
+ 'metainfo': ''}}
+ self.assertRaises(flavors.ServiceProfileEmpty,
+ self.plugin.create_service_profile,
+ self.ctx,
+ data)
+
def test_update_service_profile(self):
sp, data = self._create_service_profile()
data['service_profile']['metainfo'] = '{"data": "value1"}'
self.ctx,
{'service_profile': {'id': sp['id']}},
fl['id'])
- self.assertRaises(flavors_db.FlavorServiceProfileBindingExists,
+ self.assertRaises(flavors.FlavorServiceProfileBindingExists,
self.plugin.create_flavor_service_profile,
self.ctx,
{'service_profile': {'id': sp['id']}},
self.assertIsNone(binding)
self.assertRaises(
- flavors_db.FlavorServiceProfileBindingNotFound,
+ flavors.FlavorServiceProfileBindingNotFound,
self.plugin.delete_flavor_service_profile,
self.ctx, sp['id'], fl['id'])
{'service_profile': {'id': sp['id']}},
fl['id'])
self.assertRaises(
- flavors_db.ServiceProfileInUse,
+ flavors.ServiceProfileInUse,
self.plugin.delete_service_profile,
self.ctx,
sp['id'])
+
+ def test_get_flavor_next_provider_no_binding(self):
+ fl, data = self._create_flavor()
+ self.assertRaises(
+ flavors.FlavorServiceProfileBindingNotFound,
+ self.plugin.get_flavor_next_provider,
+ self.ctx,
+ fl['id'])
+
+ def test_get_flavor_next_provider_disabled(self):
+ data = {'service_profile':
+ {'description': 'the best sp',
+ 'driver': _driver,
+ 'enabled': False,
+ 'metainfo': '{"data": "value"}'}}
+ sp = self.plugin.create_service_profile(self.ctx,
+ data)
+ fl, data = self._create_flavor()
+ self.plugin.create_flavor_service_profile(
+ self.ctx,
+ {'service_profile': {'id': sp['id']}},
+ fl['id'])
+ self.assertRaises(
+ flavors.ServiceProfileDisabled,
+ self.plugin.get_flavor_next_provider,
+ self.ctx,
+ fl['id'])
+
+ def test_get_flavor_next_provider_no_driver(self):
+ data = {'service_profile':
+ {'description': 'the best sp',
+ 'driver': '',
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ sp = self.plugin.create_service_profile(self.ctx,
+ data)
+ fl, data = self._create_flavor()
+ self.plugin.create_flavor_service_profile(
+ self.ctx,
+ {'service_profile': {'id': sp['id']}},
+ fl['id'])
+ self.assertRaises(
+ flavors.ServiceProfileDriverNotFound,
+ self.plugin.get_flavor_next_provider,
+ self.ctx,
+ fl['id'])
+
+ def test_get_flavor_next_provider(self):
+ sp, data = self._create_service_profile()
+ fl, data = self._create_flavor()
+ self.plugin.create_flavor_service_profile(
+ self.ctx,
+ {'service_profile': {'id': sp['id']}},
+ fl['id'])
+ providers = self.plugin.get_flavor_next_provider(
+ self.ctx,
+ fl['id'])
+ self.assertEqual(_provider, providers[0].get('provider', None))
"MultiServiceCorePlugin")
mgr = manager.NeutronManager.get_instance()
svc_plugins = mgr.get_service_plugins()
- self.assertEqual(4, len(svc_plugins))
+ self.assertEqual(3, len(svc_plugins))
self.assertIn(constants.CORE, svc_plugins.keys())
self.assertIn(constants.LOADBALANCER, svc_plugins.keys())
self.assertIn(constants.DUMMY, svc_plugins.keys())
neutron.services.loadbalancer.plugin.LoadBalancerPlugin = neutron_lbaas.services.loadbalancer.plugin:LoadBalancerPlugin
neutron.services.vpn.plugin.VPNDriverPlugin = neutron_vpnaas.services.vpn.plugin:VPNDriverPlugin
qos = neutron.services.qos.qos_plugin:QoSPlugin
+ flavors = neutron.services.flavors.flavors_plugin:FlavorsPlugin
neutron.qos.notification_drivers =
message_queue = neutron.services.qos.notification_drivers.message_queue:RpcQosServiceNotificationDriver
neutron.ml2.type_drivers =
Code Review / openstack-build / neutron-build.git / commitdiff