This patch introduces API and DB plugin for flavor framework.
API adds Flavors and Service Profiles which are resources
available only for admins to operate.
This framework then should be leveraged by advanced services.
Included tempest API tests in neutron tree
Implements: blueprint neutron-flavor-framework
Change-Id: I99ba0ce520ae3d8696eca5c994777c7d5ba3d4b1
Co-Authored-By: Doug Wiegley <dougw@a10networks.com>
Co-Authored-By: Madhusudhan Kandadai <madhusudhan.kandadai@hp.com>
"get_service_provider": "rule:regular_user",
"get_lsn": "rule:admin_only",
- "create_lsn": "rule:admin_only"
+ "create_lsn": "rule:admin_only",
+
+ "create_flavor": "rule:admin_only",
+ "update_flavor": "rule:admin_only",
+ "delete_flavor": "rule:admin_only",
+ "get_flavors": "rule:regular_user",
+ "get_flavor": "rule:regular_user",
+ "create_service_profile": "rule:admin_only",
+ "update_service_profile": "rule:admin_only",
+ "delete_service_profile": "rule:admin_only",
+ "get_service_profiles": "rule:admin_only",
+ "get_service_profile": "rule:admin_only"
}
action,
item[self._resource],
pluralized=self._collection)
+ if 'tenant_id' not in item[self._resource]:
+ # no tenant_id - no quota check
+ continue
try:
tenant_id = item[self._resource]['tenant_id']
count = quota.QUOTAS.count(request.context, self._resource,
return result
@staticmethod
- def _populate_tenant_id(context, res_dict, is_create):
-
+ def _populate_tenant_id(context, res_dict, attr_info, is_create):
if (('tenant_id' in res_dict and
res_dict['tenant_id'] != context.tenant_id and
not context.is_admin)):
if is_create and 'tenant_id' not in res_dict:
if context.tenant_id:
res_dict['tenant_id'] = context.tenant_id
- else:
+ elif 'tenant_id' in attr_info:
msg = _("Running without keystone AuthN requires "
- " that tenant_id is specified")
+ "that tenant_id is specified")
raise webob.exc.HTTPBadRequest(msg)
@staticmethod
msg = _("Unable to find '%s' in request body") % resource
raise webob.exc.HTTPBadRequest(msg)
- Controller._populate_tenant_id(context, res_dict, is_create)
+ Controller._populate_tenant_id(context, res_dict, attr_info, is_create)
Controller._verify_attributes(res_dict, attr_info)
if is_create: # POST
--- /dev/null
+# 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_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
+
+
+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))
+ enabled = sa.Column(sa.Boolean, nullable=False, default=True,
+ server_default=sa.sql.true())
+ # Make it True for multi-type flavors
+ service_type = sa.Column(sa.String(36), nullable=True)
+ service_profiles = orm.relationship("FlavorServiceProfileBinding",
+ cascade="all, delete-orphan")
+
+
+class ServiceProfile(model_base.BASEV2, models_v2.HasId):
+ description = sa.Column(sa.String(1024))
+ driver = sa.Column(sa.String(1024), nullable=False)
+ enabled = sa.Column(sa.Boolean, nullable=False, default=True,
+ server_default=sa.sql.true())
+ metainfo = sa.Column(sa.String(4096))
+ flavors = orm.relationship("FlavorServiceProfileBinding")
+
+
+class FlavorServiceProfileBinding(model_base.BASEV2):
+ flavor_id = sa.Column(sa.String(36),
+ sa.ForeignKey("flavors.id",
+ ondelete="CASCADE"),
+ nullable=False, primary_key=True)
+ flavor = orm.relationship(Flavor)
+ service_profile_id = sa.Column(sa.String(36),
+ sa.ForeignKey("serviceprofiles.id",
+ ondelete="CASCADE"),
+ nullable=False, primary_key=True)
+ 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
+
+ def get_plugin_type(self):
+ return constants.FLAVORS
+
+ def get_plugin_description(self):
+ return "Neutron Flavors and Service Profiles manager plugin"
+
+ 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)
+
+ 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)
+
+ def _make_flavor_dict(self, flavor_db, fields=None):
+ res = {'id': flavor_db['id'],
+ 'name': flavor_db['name'],
+ 'description': flavor_db['description'],
+ 'enabled': flavor_db['enabled'],
+ 'service_profiles': []}
+ if flavor_db.service_profiles:
+ res['service_profiles'] = [sp['service_profile_id']
+ for sp in flavor_db.service_profiles]
+ return self._fields(res, fields)
+
+ def _make_service_profile_dict(self, sp_db, fields=None):
+ res = {'id': sp_db['id'],
+ 'description': sp_db['description'],
+ 'driver': sp_db['driver'],
+ 'enabled': sp_db['enabled'],
+ 'metainfo': sp_db['metainfo']}
+ if sp_db.flavors:
+ res['flavors'] = [fl['flavor_id']
+ for fl in sp_db.flavors]
+ return self._fields(res, fields)
+
+ def _ensure_flavor_not_in_use(self, context, flavor_id):
+ """Checks that flavor is not associated with service instance."""
+ # Future TODO(enikanorov): check that there is no binding to
+ # instances. Shall address in future upon getting the right
+ # flavor supported driver
+ 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
+ fl = (context.session.query(FlavorServiceProfileBinding).
+ filter_by(service_profile_id=sp_id).first())
+ if fl:
+ raise ServiceProfileInUse(sp_id=sp_id)
+
+ def create_flavor(self, context, flavor):
+ fl = flavor['flavor']
+ with context.session.begin(subtransactions=True):
+ fl_db = Flavor(id=uuidutils.generate_uuid(),
+ name=fl['name'],
+ description=fl['description'],
+ enabled=fl['enabled'])
+ context.session.add(fl_db)
+ return self._make_flavor_dict(fl_db)
+
+ def update_flavor(self, context, flavor_id, flavor):
+ fl = flavor['flavor']
+ with context.session.begin(subtransactions=True):
+ 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):
+ fl = self._get_flavor(context, flavor_id)
+ return self._make_flavor_dict(fl, fields)
+
+ def delete_flavor(self, context, flavor_id):
+ with context.session.begin(subtransactions=True):
+ self._ensure_flavor_not_in_use(context, flavor_id)
+ fl_db = self._get_flavor(context, flavor_id)
+ context.session.delete(fl_db)
+
+ def get_flavors(self, context, filters=None, fields=None,
+ sorts=None, limit=None, marker=None, page_reverse=False):
+ return self._get_collection(context, Flavor, self._make_flavor_dict,
+ filters=filters, fields=fields,
+ sorts=sorts, limit=limit,
+ marker_obj=marker,
+ page_reverse=page_reverse)
+
+ def create_flavor_service_profile(self, context,
+ service_profile, flavor_id):
+ sp = service_profile['service_profile']
+ with context.session.begin(subtransactions=True):
+ bind_qry = context.session.query(FlavorServiceProfileBinding)
+ binding = bind_qry.filter_by(service_profile_id=sp['id'],
+ flavor_id=flavor_id).first()
+ if binding:
+ raise 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
+
+ def delete_flavor_service_profile(self, context,
+ service_profile_id, flavor_id):
+ with context.session.begin(subtransactions=True):
+ binding = (context.session.query(FlavorServiceProfileBinding).
+ filter_by(service_profile_id=service_profile_id,
+ flavor_id=flavor_id).first())
+ if not binding:
+ raise FlavorServiceProfileBindingNotFound(
+ sp_id=service_profile_id, fl_id=flavor_id)
+ context.session.delete(binding)
+
+ def get_flavor_service_profile(self, context,
+ service_profile_id, flavor_id, fields=None):
+ with context.session.begin(subtransactions=True):
+ binding = (context.session.query(FlavorServiceProfileBinding).
+ filter_by(service_profile_id=service_profile_id,
+ flavor_id=flavor_id).first())
+ if not binding:
+ raise 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 cant 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)
+
+ 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'],
+ driver=sp['driver'],
+ 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']
+ with context.session.begin(subtransactions=True):
+ self._ensure_service_profile_not_in_use(context,
+ service_profile_id)
+ sp_db = self._get_service_profile(context, service_profile_id)
+ sp_db.update(sp)
+ return self._make_service_profile_dict(sp_db)
+
+ def get_service_profile(self, context, sp_id, fields=None):
+ sp_db = self._get_service_profile(context, sp_id)
+ return self._make_service_profile_dict(sp_db, fields)
+
+ def delete_service_profile(self, context, sp_id):
+ with context.session.begin(subtransactions=True):
+ self._ensure_service_profile_not_in_use(context, sp_id)
+ sp_db = self._get_service_profile(context, sp_id)
+ context.session.delete(sp_db)
+
+ def get_service_profiles(self, context, filters=None, fields=None,
+ sorts=None, limit=None, marker=None,
+ page_reverse=False):
+ return self._get_collection(context, ServiceProfile,
+ self._make_service_profile_dict,
+ filters=filters, fields=fields,
+ sorts=sorts, limit=limit,
+ marker_obj=marker,
+ page_reverse=page_reverse)
30018084ec99
-52c5312f6baf
+313373c0ffee
kilo
--- /dev/null
+# Copyright 2014-2015 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.
+#
+
+"""Flavor framework
+
+Revision ID: 313373c0ffee
+Revises: 52c5312f6baf
+
+Create Date: 2014-07-17 03:00:00.00
+"""
+# revision identifiers, used by Alembic.
+revision = '313373c0ffee'
+down_revision = '52c5312f6baf'
+
+from alembic import op
+import sqlalchemy as sa
+
+
+def upgrade():
+ op.create_table(
+ 'flavors',
+ sa.Column('id', sa.String(36)),
+ sa.Column('name', sa.String(255)),
+ sa.Column('description', sa.String(1024)),
+ sa.Column('enabled', sa.Boolean, nullable=False,
+ server_default=sa.sql.true()),
+ sa.Column('service_type', sa.String(36), nullable=True),
+ sa.PrimaryKeyConstraint('id')
+ )
+
+ op.create_table(
+ 'serviceprofiles',
+ sa.Column('id', sa.String(36)),
+ sa.Column('description', sa.String(1024)),
+ sa.Column('driver', sa.String(1024), nullable=False),
+ sa.Column('enabled', sa.Boolean, nullable=False,
+ server_default=sa.sql.true()),
+ sa.Column('metainfo', sa.String(4096)),
+ sa.PrimaryKeyConstraint('id')
+ )
+
+ op.create_table(
+ 'flavorserviceprofilebindings',
+ sa.Column('service_profile_id', sa.String(36), nullable=False),
+ sa.Column('flavor_id', sa.String(36), nullable=False),
+ sa.ForeignKeyConstraint(['service_profile_id'],
+ ['serviceprofiles.id']),
+ sa.ForeignKeyConstraint(['flavor_id'], ['flavors.id']),
+ sa.PrimaryKeyConstraint('service_profile_id', 'flavor_id')
+ )
from neutron.db import external_net_db # noqa
from neutron.db import extradhcpopt_db # noqa
from neutron.db import extraroute_db # noqa
+from neutron.db import flavors_db # noqa
from neutron.db import l3_agentschedulers_db # noqa
from neutron.db import l3_attrs_db # noqa
from neutron.db import l3_db # noqa
--- /dev/null
+# 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.api import extensions
+from neutron.api.v2 import attributes as attr
+from neutron.api.v2 import base
+from neutron.api.v2 import resource_helper
+from neutron import manager
+from neutron.plugins.common import constants
+
+
+FLAVORS = 'flavors'
+SERVICE_PROFILES = 'service_profiles'
+FLAVORS_PREFIX = ""
+
+RESOURCE_ATTRIBUTE_MAP = {
+ FLAVORS: {
+ 'id': {'allow_post': False, 'allow_put': False,
+ 'validate': {'type:uuid': None},
+ 'is_visible': True,
+ 'primary_key': True},
+ 'name': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:string': None},
+ 'is_visible': True, 'default': ''},
+ 'description': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:string': None},
+ 'is_visible': True, 'default': ''},
+ 'service_type': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:string': None},
+ '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': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:uuid_list': None},
+ 'is_visible': True, 'default': []},
+ 'enabled': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:boolean': None},
+ 'default': True,
+ 'is_visible': True},
+ },
+ SERVICE_PROFILES: {
+ 'id': {'allow_post': False, 'allow_put': False,
+ '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},
+ 'is_visible': True,
+ 'default': attr.ATTR_NOT_SPECIFIED},
+ 'metainfo': {'allow_post': True, 'allow_put': True,
+ '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},
+ 'enabled': {'allow_post': True, 'allow_put': True,
+ 'validate': {'type:boolean': None},
+ 'is_visible': True, 'default': True},
+ },
+}
+
+
+SUB_RESOURCE_ATTRIBUTE_MAP = {
+ 'service_profiles': {
+ 'parent': {'collection_name': 'flavors',
+ 'member_name': 'flavor'},
+ 'parameters': {'id': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:uuid': None},
+ 'is_visible': True}}
+ }
+}
+
+
+class Flavors(extensions.ExtensionDescriptor):
+
+ @classmethod
+ def get_name(cls):
+ return "Neutron Service Flavors"
+
+ @classmethod
+ def get_alias(cls):
+ return "flavors"
+
+ @classmethod
+ def get_description(cls):
+ return "Service specification for advanced services"
+
+ @classmethod
+ def get_updated(cls):
+ return "2014-07-06T10:00:00-00:00"
+
+ @classmethod
+ def get_resources(cls):
+ """Returns Ext Resources."""
+ plural_mappings = resource_helper.build_plural_mappings(
+ {}, RESOURCE_ATTRIBUTE_MAP)
+ attr.PLURALS.update(plural_mappings)
+ resources = resource_helper.build_resource_info(
+ plural_mappings,
+ RESOURCE_ATTRIBUTE_MAP,
+ constants.FLAVORS)
+ plugin = manager.NeutronManager.get_service_plugins()[
+ constants.FLAVORS]
+ for collection_name in SUB_RESOURCE_ATTRIBUTE_MAP:
+ # Special handling needed for sub-resources with 'y' ending
+ # (e.g. proxies -> proxy)
+ resource_name = collection_name[:-1]
+ parent = SUB_RESOURCE_ATTRIBUTE_MAP[collection_name].get('parent')
+ params = SUB_RESOURCE_ATTRIBUTE_MAP[collection_name].get(
+ 'parameters')
+
+ controller = base.create_resource(collection_name, resource_name,
+ plugin, params,
+ allow_bulk=True,
+ parent=parent)
+
+ resource = extensions.ResourceExtension(
+ collection_name,
+ controller, parent,
+ path_prefix=FLAVORS_PREFIX,
+ attr_map=params)
+ resources.append(resource)
+
+ return resources
+
+ def update_attributes_map(self, attributes):
+ super(Flavors, self).update_attributes_map(
+ attributes, extension_attrs_map=RESOURCE_ATTRIBUTE_MAP)
+
+ def get_extended_resources(self, version):
+ if version == "2.0":
+ return RESOURCE_ATTRIBUTE_MAP
+ else:
+ return {}
import six
from neutron.common import utils
+from neutron.db import flavors_db
from neutron.i18n import _LE, _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")
VPN = "VPN"
METERING = "METERING"
L3_ROUTER_NAT = "L3_ROUTER_NAT"
+FLAVORS = "FLAVORS"
# Maps extension alias to service type
EXT_TO_SERVICE_MAPPING = {
'fwaas': FIREWALL,
'vpnaas': VPN,
'metering': METERING,
- 'router': L3_ROUTER_NAT
+ 'router': L3_ROUTER_NAT,
+ 'flavors': FLAVORS
}
# Service operation status constants
cls.ikepolicies = []
cls.floating_ips = []
cls.metering_labels = []
+ cls.service_profiles = []
+ cls.flavors = []
cls.metering_label_rules = []
cls.fw_rules = []
cls.fw_policies = []
cls._try_delete_resource(
cls.admin_client.delete_metering_label,
metering_label['id'])
+ # Clean up flavors
+ for flavor in cls.flavors:
+ cls._try_delete_resource(
+ cls.admin_client.delete_flavor,
+ flavor['id'])
+ # Clean up service profiles
+ for service_profile in cls.service_profiles:
+ cls._try_delete_resource(
+ cls.admin_client.delete_service_profile,
+ service_profile['id'])
# Clean up ports
for port in cls.ports:
cls._try_delete_resource(cls.client.delete_port,
metering_label_rule = body['metering_label_rule']
cls.metering_label_rules.append(metering_label_rule)
return metering_label_rule
+
+ @classmethod
+ def create_flavor(cls, name, description, service_type):
+ """Wrapper utility that returns a test flavor."""
+ body = cls.admin_client.create_flavor(
+ description=description, service_type=service_type,
+ name=name)
+ flavor = body['flavor']
+ cls.flavors.append(flavor)
+ return flavor
+
+ @classmethod
+ def create_service_profile(cls, description, metainfo, driver):
+ """Wrapper utility that returns a test service profile."""
+ body = cls.admin_client.create_service_profile(
+ driver=driver, metainfo=metainfo, description=description)
+ service_profile = body['service_profile']
+ cls.service_profiles.append(service_profile)
+ return service_profile
--- /dev/null
+# Copyright 2015 Hewlett-Packard Development Company, L.P.
+#
+# 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_log import log as logging
+
+from neutron.tests.api import base
+from neutron.tests.tempest import test
+
+
+LOG = logging.getLogger(__name__)
+
+
+class TestFlavorsJson(base.BaseAdminNetworkTest):
+
+ """
+ Tests the following operations in the Neutron API using the REST client for
+ Neutron:
+
+ List, Show, Create, Update, Delete Flavors
+ List, Show, Create, Update, Delete service profiles
+ """
+
+ @classmethod
+ def resource_setup(cls):
+ super(TestFlavorsJson, cls).resource_setup()
+ if not test.is_extension_enabled('flavors', 'network'):
+ msg = "flavors extension not enabled."
+ raise cls.skipException(msg)
+ service_type = "LOADBALANCER"
+ 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)
+ 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
+ driver = ""
+ metainfo = '{"data": "value"}'
+ cls.service_profile = cls.create_service_profile(
+ description=description_sp, metainfo=metainfo, driver=driver)
+
+ def _delete_service_profile(self, service_profile_id):
+ # Deletes a service profile and verifies if it is deleted or not
+ self.admin_client.delete_service_profile(service_profile_id)
+ # Asserting that service profile is not found in list after deletion
+ labels = self.admin_client.list_service_profiles(id=service_profile_id)
+ self.assertEqual(len(labels['service_profiles']), 0)
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('ec8e15ff-95d0-433b-b8a6-b466bddb1e50')
+ def test_create_update_delete_service_profile(self):
+ # Creates a service profile
+ description = "service_profile created by tempest"
+ driver = ""
+ metainfo = '{"data": "value"}'
+ body = self.admin_client.create_service_profile(
+ description=description, driver=driver, metainfo=metainfo)
+ service_profile = body['service_profile']
+ # Updates a service profile
+ self.admin_client.update_service_profile(service_profile['id'],
+ enabled=False)
+ self.assertTrue(service_profile['enabled'])
+ # Deletes a service profile
+ self.addCleanup(self._delete_service_profile,
+ service_profile['id'])
+ # Assert whether created service profiles are found in service profile
+ # lists or fail if created service profiles are not found in service
+ # profiles list
+ labels = (self.admin_client.list_service_profiles(
+ id=service_profile['id']))
+ self.assertEqual(len(labels['service_profiles']), 1)
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('ec8e15ff-95d0-433b-b8a6-b466bddb1e50')
+ def test_create_update_delete_flavor(self):
+ # Creates a flavor
+ description = "flavor created by tempest"
+ service = "LOADBALANCERS"
+ name = "Best flavor created by tempest"
+ body = self.admin_client.create_flavor(name=name, service_type=service,
+ description=description)
+ flavor = body['flavor']
+ # Updates a flavor
+ self.admin_client.update_flavor(flavor['id'], enabled=False)
+ self.assertTrue(flavor['enabled'])
+ # Deletes a flavor
+ self.addCleanup(self._delete_flavor, flavor['id'])
+ # Assert whether created flavors are found in flavor lists or fail
+ # if created flavors are not found in flavors list
+ labels = (self.admin_client.list_flavors(id=flavor['id']))
+ self.assertEqual(len(labels['flavors']), 1)
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('30abb445-0eea-472e-bd02-8649f54a5968')
+ def test_show_service_profile(self):
+ # Verifies the details of a service profile
+ body = self.admin_client.show_service_profile(
+ self.service_profile['id'])
+ service_profile = body['service_profile']
+ self.assertEqual(self.service_profile['id'], service_profile['id'])
+ self.assertEqual(self.service_profile['description'],
+ service_profile['description'])
+ self.assertEqual(self.service_profile['metainfo'],
+ service_profile['metainfo'])
+ self.assertEqual(True, service_profile['enabled'])
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('30abb445-0eea-472e-bd02-8649f54a5968')
+ def test_show_flavor(self):
+ # Verifies the details of a flavor
+ body = self.admin_client.show_flavor(self.flavor['id'])
+ flavor = body['flavor']
+ self.assertEqual(self.flavor['id'], flavor['id'])
+ self.assertEqual(self.flavor['description'], flavor['description'])
+ self.assertEqual(self.flavor['name'], flavor['name'])
+ self.assertEqual(True, flavor['enabled'])
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('e2fb2f8c-45bf-429a-9f17-171c70444612')
+ def test_list_flavors(self):
+ # Verify flavor lists
+ body = self.admin_client.list_flavors(id=33)
+ flavors = body['flavors']
+ self.assertEqual(0, len(flavors))
+
+ @test.attr(type='smoke')
+ @test.idempotent_id('e2fb2f8c-45bf-429a-9f17-171c70444612')
+ def test_list_service_profiles(self):
+ # Verify service profiles lists
+ body = self.admin_client.list_service_profiles(id=33)
+ service_profiles = body['service_profiles']
+ self.assertEqual(0, len(service_profiles))
+
+ def _delete_flavor(self, flavor_id):
+ # Deletes a flavor and verifies if it is deleted or not
+ self.admin_client.delete_flavor(flavor_id)
+ # Asserting that the flavor is not found in list after deletion
+ labels = self.admin_client.list_flavors(id=flavor_id)
+ self.assertEqual(len(labels['flavors']), 0)
+
+
+class TestFlavorsIpV6TestJSON(TestFlavorsJson):
+ _ip_version = 6
"get_service_provider": "rule:regular_user",
"get_lsn": "rule:admin_only",
- "create_lsn": "rule:admin_only"
+ "create_lsn": "rule:admin_only",
+
+ "create_flavor": "rule:admin_only",
+ "update_flavor": "rule:admin_only",
+ "delete_flavor": "rule:admin_only",
+ "get_flavors": "rule:regular_user",
+ "get_flavor": "rule:regular_user",
+ "create_service_profile": "rule:admin_only",
+ "update_service_profile": "rule:admin_only",
+ "delete_service_profile": "rule:admin_only",
+ "get_service_profiles": "rule:admin_only",
+ "get_service_profile": "rule:admin_only"
}
# The following list represents resource names that do not require
# changing underscore to a hyphen
hyphen_exceptions = ["health_monitors", "firewall_rules",
- "firewall_policies"]
+ "firewall_policies", "service_profiles"]
# the following map is used to construct proper URI
# for the given neutron resource
service_resource_prefix_map = {
--- /dev/null
+#
+# 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.
+#
+
+
+import copy
+import fixtures
+import mock
+
+from oslo_config import cfg
+from oslo_utils import uuidutils
+
+from neutron import context
+from neutron.db import api as dbapi
+from neutron.db import flavors_db
+from neutron.extensions import flavors
+from neutron import manager
+from neutron.plugins.common import constants
+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
+from neutron.tests.unit.extensions import base as extension
+
+_uuid = uuidutils.generate_uuid
+_get_path = test_base._get_path
+
+
+class FlavorExtensionTestCase(extension.ExtensionTestCase):
+
+ def setUp(self):
+ super(FlavorExtensionTestCase, self).setUp()
+ self._setUpExtension(
+ 'neutron.db.flavors_db.FlavorManager',
+ constants.FLAVORS, flavors.RESOURCE_ATTRIBUTE_MAP,
+ flavors.Flavors, '', supported_extension_aliases='flavors')
+
+ def test_create_flavor(self):
+ tenant_id = uuidutils.generate_uuid()
+ data = {'flavor': {'name': 'GOLD',
+ 'service_type': constants.LOADBALANCER,
+ 'description': 'the best flavor',
+ 'tenant_id': tenant_id,
+ 'enabled': True}}
+
+ expected = copy.deepcopy(data)
+ expected['flavor']['service_profiles'] = []
+
+ instance = self.plugin.return_value
+ instance.create_flavor.return_value = expected['flavor']
+ res = self.api.post(_get_path('flavors', fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt)
+
+ instance.create_flavor.assert_called_with(mock.ANY,
+ flavor=expected)
+ res = self.deserialize(res)
+ self.assertIn('flavor', res)
+ self.assertEqual(expected, res)
+
+ def test_update_flavor(self):
+ flavor_id = 'fake_id'
+ data = {'flavor': {'name': 'GOLD',
+ 'description': 'the best flavor',
+ 'enabled': True}}
+ expected = copy.copy(data)
+ expected['flavor']['service_profiles'] = []
+
+ instance = self.plugin.return_value
+ instance.update_flavor.return_value = expected['flavor']
+ res = self.api.put(_get_path('flavors', id=flavor_id, fmt=self.fmt),
+ self.serialize(data),
+ content_type='application/%s' % self.fmt)
+
+ instance.update_flavor.assert_called_with(mock.ANY,
+ flavor_id,
+ flavor=expected)
+ res = self.deserialize(res)
+ self.assertIn('flavor', res)
+ self.assertEqual(expected, res)
+
+ def test_delete_flavor(self):
+ flavor_id = 'fake_id'
+ instance = self.plugin.return_value
+ self.api.delete(_get_path('flavors', id=flavor_id, fmt=self.fmt),
+ content_type='application/%s' % self.fmt)
+
+ instance.delete_flavor.assert_called_with(mock.ANY,
+ flavor_id)
+
+ def test_show_flavor(self):
+ flavor_id = 'fake_id'
+ expected = {'flavor': {'id': flavor_id,
+ 'name': 'GOLD',
+ 'description': 'the best flavor',
+ 'enabled': True,
+ 'service_profiles': ['profile-1']}}
+ instance = self.plugin.return_value
+ instance.get_flavor.return_value = expected['flavor']
+ res = self.api.get(_get_path('flavors', id=flavor_id, fmt=self.fmt))
+ instance.get_flavor.assert_called_with(mock.ANY,
+ flavor_id,
+ fields=mock.ANY)
+ res = self.deserialize(res)
+ self.assertEqual(expected, res)
+
+ def test_get_flavors(self):
+ data = {'flavors': [{'id': 'id1',
+ 'name': 'GOLD',
+ 'description': 'the best flavor',
+ 'enabled': True,
+ 'service_profiles': ['profile-1']},
+ {'id': 'id2',
+ 'name': 'GOLD',
+ 'description': 'the best flavor',
+ 'enabled': True,
+ 'service_profiles': ['profile-2', 'profile-1']}]}
+ instance = self.plugin.return_value
+ instance.get_flavors.return_value = data['flavors']
+ res = self.api.get(_get_path('flavors', fmt=self.fmt))
+ instance.get_flavors.assert_called_with(mock.ANY,
+ fields=mock.ANY,
+ filters=mock.ANY)
+ res = self.deserialize(res)
+ self.assertEqual(data, res)
+
+ def test_create_service_profile(self):
+ tenant_id = uuidutils.generate_uuid()
+ expected = {'service_profile': {'description': 'the best sp',
+ 'driver': '',
+ 'tenant_id': tenant_id,
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+
+ instance = self.plugin.return_value
+ instance.create_service_profile.return_value = (
+ expected['service_profile'])
+ res = self.api.post(_get_path('service_profiles', fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt)
+ instance.create_service_profile.assert_called_with(
+ mock.ANY,
+ service_profile=expected)
+ res = self.deserialize(res)
+ self.assertIn('service_profile', res)
+ self.assertEqual(expected, res)
+
+ def test_update_service_profile(self):
+ sp_id = "fake_id"
+ expected = {'service_profile': {'description': 'the best sp',
+ 'enabled': False,
+ 'metainfo': '{"data1": "value3"}'}}
+
+ instance = self.plugin.return_value
+ instance.update_service_profile.return_value = (
+ expected['service_profile'])
+ res = self.api.put(_get_path('service_profiles',
+ id=sp_id, fmt=self.fmt),
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt)
+
+ instance.update_service_profile.assert_called_with(
+ mock.ANY,
+ sp_id,
+ service_profile=expected)
+ res = self.deserialize(res)
+ self.assertIn('service_profile', res)
+ self.assertEqual(expected, res)
+
+ def test_delete_service_profile(self):
+ sp_id = 'fake_id'
+ instance = self.plugin.return_value
+ self.api.delete(_get_path('service_profiles', id=sp_id, fmt=self.fmt),
+ content_type='application/%s' % self.fmt)
+ instance.delete_service_profile.assert_called_with(mock.ANY,
+ sp_id)
+
+ def test_show_service_profile(self):
+ sp_id = 'fake_id'
+ expected = {'service_profile': {'id': 'id1',
+ 'driver': 'entrypoint1',
+ 'description': 'desc',
+ 'metainfo': '{}',
+ 'enabled': True}}
+ instance = self.plugin.return_value
+ instance.get_service_profile.return_value = (
+ expected['service_profile'])
+ res = self.api.get(_get_path('service_profiles',
+ id=sp_id, fmt=self.fmt))
+ instance.get_service_profile.assert_called_with(mock.ANY,
+ sp_id,
+ fields=mock.ANY)
+ res = self.deserialize(res)
+ self.assertEqual(expected, res)
+
+ def test_get_service_profiles(self):
+ expected = {'service_profiles': [{'id': 'id1',
+ 'driver': 'entrypoint1',
+ 'description': 'desc',
+ 'metainfo': '{}',
+ 'enabled': True},
+ {'id': 'id2',
+ 'driver': 'entrypoint2',
+ 'description': 'desc',
+ 'metainfo': '{}',
+ 'enabled': True}]}
+ instance = self.plugin.return_value
+ instance.get_service_profiles.return_value = (
+ expected['service_profiles'])
+ res = self.api.get(_get_path('service_profiles', fmt=self.fmt))
+ instance.get_service_profiles.assert_called_with(mock.ANY,
+ fields=mock.ANY,
+ filters=mock.ANY)
+ res = self.deserialize(res)
+ self.assertEqual(expected, res)
+
+ def test_associate_service_profile_with_flavor(self):
+ expected = {'service_profile': {'id': _uuid()}}
+ instance = self.plugin.return_value
+ instance.create_flavor_service_profile.return_value = (
+ expected['service_profile'])
+ res = self.api.post('/flavors/fl_id/service_profiles',
+ self.serialize(expected),
+ content_type='application/%s' % self.fmt)
+ instance.create_flavor_service_profile.assert_called_with(
+ mock.ANY, service_profile=expected, flavor_id='fl_id')
+ res = self.deserialize(res)
+ self.assertEqual(expected, res)
+
+ def test_disassociate_service_profile_with_flavor(self):
+ instance = self.plugin.return_value
+ instance.delete_flavor_service_profile.return_value = None
+ self.api.delete('/flavors/fl_id/service_profiles/%s' % 'fake_spid',
+ content_type='application/%s' % self.fmt)
+ instance.delete_flavor_service_profile.assert_called_with(
+ mock.ANY,
+ 'fake_spid',
+ flavor_id='fl_id')
+
+
+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 FlavorManagerTestCase(test_db_base_plugin_v2.NeutronDbPluginV2TestCase,
+ base.PluginFixture):
+ def setUp(self):
+ super(FlavorManagerTestCase, self).setUp()
+
+ self.config_parse()
+ cfg.CONF.set_override(
+ 'core_plugin',
+ 'neutron.tests.unit.extensions.test_flavors.DummyCorePlugin')
+ cfg.CONF.set_override(
+ 'service_plugins',
+ ['neutron.tests.unit.extensions.test_flavors.DummyServicePlugin'])
+
+ self.useFixture(
+ fixtures.MonkeyPatch('neutron.manager.NeutronManager._instance'))
+
+ self.plugin = flavors_db.FlavorManager(
+ manager.NeutronManager().get_instance())
+ self.ctx = context.get_admin_context()
+ dbapi.get_engine()
+
+ def _create_flavor(self, description=None):
+ flavor = {'flavor': {'name': 'GOLD',
+ 'service_type': constants.LOADBALANCER,
+ 'description': description or 'the best flavor',
+ 'enabled': True}}
+ return self.plugin.create_flavor(self.ctx, flavor), flavor
+
+ def test_create_flavor(self):
+ self._create_flavor()
+ res = self.ctx.session.query(flavors_db.Flavor).all()
+ self.assertEqual(1, len(res))
+ self.assertEqual('GOLD', res[0]['name'])
+
+ def test_update_flavor(self):
+ fl, flavor = self._create_flavor()
+ flavor = {'flavor': {'name': 'Silver',
+ 'enabled': False}}
+ self.plugin.update_flavor(self.ctx, fl['id'], flavor)
+ res = (self.ctx.session.query(flavors_db.Flavor).
+ filter_by(id=fl['id']).one())
+ self.assertEqual('Silver', res['name'])
+ self.assertFalse(res['enabled'])
+
+ def test_delete_flavor(self):
+ fl, data = self._create_flavor()
+ self.plugin.delete_flavor(self.ctx, fl['id'])
+ res = (self.ctx.session.query(flavors_db.Flavor).all())
+ self.assertFalse(res)
+
+ def test_show_flavor(self):
+ fl, data = self._create_flavor()
+ show_fl = self.plugin.get_flavor(self.ctx, fl['id'])
+ self.assertEqual(fl, show_fl)
+
+ def test_get_flavors(self):
+ fl, flavor = self._create_flavor()
+ flavor['flavor']['name'] = 'SILVER'
+ self.plugin.create_flavor(self.ctx, flavor)
+ show_fl = self.plugin.get_flavors(self.ctx)
+ self.assertEqual(2, len(show_fl))
+
+ def _create_service_profile(self, description=None):
+ data = {'service_profile':
+ {'description': description or 'the best sp',
+ 'driver':
+ ('neutron.tests.unit.extensions.test_flavors.'
+ 'DummyServiceDriver'),
+ 'enabled': True,
+ 'metainfo': '{"data": "value"}'}}
+ sp = self.plugin.unit_create_service_profile(self.ctx,
+ data)
+ return sp, data
+
+ def test_create_service_profile(self):
+ sp, data = self._create_service_profile()
+ 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_update_service_profile(self):
+ sp, data = self._create_service_profile()
+ data['service_profile']['metainfo'] = '{"data": "value1"}'
+ sp = self.plugin.update_service_profile(self.ctx, sp['id'],
+ data)
+ res = (self.ctx.session.query(flavors_db.ServiceProfile).
+ filter_by(id=sp['id']).one())
+ self.assertEqual(data['service_profile']['metainfo'], res['metainfo'])
+
+ def test_delete_service_profile(self):
+ sp, data = self._create_service_profile()
+ self.plugin.delete_service_profile(self.ctx, sp['id'])
+ res = self.ctx.session.query(flavors_db.ServiceProfile).all()
+ self.assertFalse(res)
+
+ def test_show_service_profile(self):
+ sp, data = self._create_service_profile()
+ sp_show = self.plugin.get_service_profile(self.ctx, sp['id'])
+ self.assertEqual(sp, sp_show)
+
+ def test_get_service_profiles(self):
+ self._create_service_profile()
+ self._create_service_profile(description='another sp')
+ self.assertEqual(2, len(self.plugin.get_service_profiles(self.ctx)))
+
+ def test_associate_service_profile_with_flavor(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'])
+ binding = (
+ self.ctx.session.query(flavors_db.FlavorServiceProfileBinding).
+ first())
+ self.assertEqual(fl['id'], binding['flavor_id'])
+ self.assertEqual(sp['id'], binding['service_profile_id'])
+
+ res = self.plugin.get_flavor(self.ctx, fl['id'])
+ self.assertEqual(1, len(res['service_profiles']))
+ self.assertEqual(sp['id'], res['service_profiles'][0])
+
+ res = self.plugin.get_service_profile(self.ctx, sp['id'])
+ self.assertEqual(1, len(res['flavors']))
+ self.assertEqual(fl['id'], res['flavors'][0])
+
+ def test_autodelete_flavor_associations(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'])
+ self.plugin.delete_flavor(self.ctx, fl['id'])
+ binding = (
+ self.ctx.session.query(flavors_db.FlavorServiceProfileBinding).
+ first())
+ self.assertIsNone(binding)
+
+ def test_associate_service_profile_with_flavor_exists(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'])
+ self.assertRaises(flavors_db.FlavorServiceProfileBindingExists,
+ self.plugin.create_flavor_service_profile,
+ self.ctx,
+ {'service_profile': {'id': sp['id']}},
+ fl['id'])
+
+ def test_disassociate_service_profile_with_flavor(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'])
+ self.plugin.delete_flavor_service_profile(
+ self.ctx, sp['id'], fl['id'])
+ binding = (
+ self.ctx.session.query(flavors_db.FlavorServiceProfileBinding).
+ first())
+ self.assertIsNone(binding)
+
+ self.assertRaises(
+ flavors_db.FlavorServiceProfileBindingNotFound,
+ self.plugin.delete_flavor_service_profile,
+ self.ctx, sp['id'], fl['id'])
+
+ def test_delete_service_profile_in_use(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'])
+ self.assertRaises(
+ flavors_db.ServiceProfileInUse,
+ self.plugin.delete_service_profile,
+ self.ctx,
+ sp['id'])
"MultiServiceCorePlugin")
mgr = manager.NeutronManager.get_instance()
svc_plugins = mgr.get_service_plugins()
- self.assertEqual(3, len(svc_plugins))
+ self.assertEqual(4, len(svc_plugins))
self.assertIn(constants.CORE, svc_plugins.keys())
self.assertIn(constants.LOADBALANCER, svc_plugins.keys())
self.assertIn(constants.DUMMY, svc_plugins.keys())
Code Review / openstack-build / neutron-build.git / commitdiff