Add flavor, flavor_profile table and their APIs

This patch adds flavor and flavor_profile tables.
It also implements flavors and flavorprofiles apis.

Partially-Implements: Blueprint octavia-lbaas-flavors
Co-Authored-By: Michael Johnson <johnsomor@gmail.com>

Change-Id: I99a673438458757d0acdaa46dd8ee041edb3be9c
This commit is contained in:
Pradeep Kumar Singh 2017-07-24 06:24:01 +00:00 committed by Michael Johnson
parent 8b4a01c5bb
commit 637009ecd0
29 changed files with 2177 additions and 15 deletions

View File

@ -231,14 +231,14 @@ class NoopManager(object):
LOG.debug('Provider %s no-op, get_supported_flavor_metadata',
self.__class__.__name__)
return {'amp_image_tag': 'The glance image tag to use for this load '
'balancer.'}
return {"amp_image_tag": "The glance image tag to use for this load "
"balancer."}
def validate_flavor(self, flavor_metadata):
LOG.debug('Provider %s no-op, validate_flavor metadata: %s',
self.__class__.__name__, flavor_metadata)
flavor_hash = hash(frozenset(flavor_metadata.items()))
flavor_hash = hash(frozenset(flavor_metadata))
self.driverconfig[flavor_hash] = (flavor_metadata, 'validate_flavor')

View File

@ -79,9 +79,13 @@ class RootController(rest.RestController):
'2018-07-31T00:00:00Z', host_url)
self._add_a_version(versions, 'v2.3', 'v2', 'SUPPORTED',
'2018-12-18T00:00:00Z', host_url)
self._add_a_version(versions, 'v2.4', 'v2', 'CURRENT',
# amp statistics
self._add_a_version(versions, 'v2.4', 'v2', 'SUPPORTED',
'2018-12-19T00:00:00Z', host_url)
# Tags
self._add_a_version(versions, 'v2.5', 'v2', 'CURRENT',
self._add_a_version(versions, 'v2.5', 'v2', 'SUPPORTED',
'2019-01-21T00:00:00Z', host_url)
# Flavors
self._add_a_version(versions, 'v2.6', 'v2', 'CURRENT',
'2019-01-25T00:00:00Z', host_url)
return {'versions': versions}

View File

@ -17,6 +17,8 @@ from wsmeext import pecan as wsme_pecan
from octavia.api.v2.controllers import amphora
from octavia.api.v2.controllers import base
from octavia.api.v2.controllers import flavor_profiles
from octavia.api.v2.controllers import flavors
from octavia.api.v2.controllers import health_monitor
from octavia.api.v2.controllers import l7policy
from octavia.api.v2.controllers import listener
@ -43,6 +45,8 @@ class BaseV2Controller(base.BaseController):
self.healthmonitors = health_monitor.HealthMonitorController()
self.quotas = quotas.QuotasController()
self.providers = provider.ProviderController()
self.flavors = flavors.FlavorsController()
self.flavorprofiles = flavor_profiles.FlavorProfileController()
@wsme_pecan.wsexpose(wtypes.text)
def get(self):

View File

@ -99,6 +99,16 @@ class BaseController(rest.RestController):
data_models.HealthMonitor, id,
show_deleted=show_deleted)
def _get_db_flavor(self, session, id):
"""Get a flavor from the database."""
return self._get_db_obj(session, self.repositories.flavor,
data_models.Flavor, id)
def _get_db_flavor_profile(self, session, id):
"""Get a flavor profile from the database."""
return self._get_db_obj(session, self.repositories.flavor_profile,
data_models.FlavorProfile, id)
def _get_db_l7policy(self, session, id, show_deleted=True):
"""Get a L7 Policy from the database."""
return self._get_db_obj(session, self.repositories.l7policy,

View File

@ -0,0 +1,191 @@
# Copyright 2014 Rackspace
# Copyright 2016 Blue Box, an IBM Company
#
# 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_db import exception as odb_exceptions
from oslo_log import log as logging
from oslo_serialization import jsonutils
from oslo_utils import excutils
from oslo_utils import uuidutils
import pecan
from sqlalchemy.orm import exc as sa_exception
from wsme import types as wtypes
from wsmeext import pecan as wsme_pecan
from octavia.api.drivers import driver_factory
from octavia.api.drivers import utils as driver_utils
from octavia.api.v2.controllers import base
from octavia.api.v2.types import flavor_profile as profile_types
from octavia.common import constants
from octavia.common import exceptions
from octavia.db import api as db_api
LOG = logging.getLogger(__name__)
class FlavorProfileController(base.BaseController):
RBAC_TYPE = constants.RBAC_FLAVOR_PROFILE
def __init__(self):
super(FlavorProfileController, self).__init__()
@wsme_pecan.wsexpose(profile_types.FlavorProfileRootResponse, wtypes.text,
[wtypes.text], ignore_extra_args=True)
def get_one(self, id, fields=None):
"""Gets a flavor profile's detail."""
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_GET_ONE)
db_flavor_profile = self._get_db_flavor_profile(context.session, id)
result = self._convert_db_to_type(db_flavor_profile,
profile_types.FlavorProfileResponse)
if fields is not None:
result = self._filter_fields([result], fields)[0]
return profile_types.FlavorProfileRootResponse(flavorprofile=result)
@wsme_pecan.wsexpose(profile_types.FlavorProfilesRootResponse,
[wtypes.text], ignore_extra_args=True)
def get_all(self, fields=None):
"""Lists all flavor profiles."""
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_GET_ALL)
db_flavor_profiles, links = self.repositories.flavor_profile.get_all(
context.session,
pagination_helper=pcontext.get(constants.PAGINATION_HELPER))
result = self._convert_db_to_type(
db_flavor_profiles, [profile_types.FlavorProfileResponse])
if fields is not None:
result = self._filter_fields(result, fields)
return profile_types.FlavorProfilesRootResponse(
flavorprofiles=result, flavorprofile_links=links)
@wsme_pecan.wsexpose(profile_types.FlavorProfileRootResponse,
body=profile_types.FlavorProfileRootPOST,
status_code=201)
def post(self, flavor_profile_):
"""Creates a flavor Profile."""
flavorprofile = flavor_profile_.flavorprofile
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_POST)
# Do a basic JSON validation on the metadata
try:
flavor_data_dict = jsonutils.loads(flavorprofile.flavor_data)
except Exception:
raise exceptions.InvalidOption(
value=flavorprofile.flavor_data,
option=constants.FLAVOR_DATA)
# Validate that the provider driver supports the metadata
driver = driver_factory.get_driver(flavorprofile.provider_name)
driver_utils.call_provider(driver.name, driver.validate_flavor,
flavor_data_dict)
lock_session = db_api.get_session(autocommit=False)
try:
flavorprofile_dict = flavorprofile.to_dict(render_unsets=True)
flavorprofile_dict['id'] = uuidutils.generate_uuid()
db_flavor_profile = self.repositories.flavor_profile.create(
lock_session, **flavorprofile_dict)
lock_session.commit()
except odb_exceptions.DBDuplicateEntry:
lock_session.rollback()
raise exceptions.IDAlreadyExists()
except Exception:
with excutils.save_and_reraise_exception():
lock_session.rollback()
result = self._convert_db_to_type(
db_flavor_profile, profile_types.FlavorProfileResponse)
return profile_types.FlavorProfileRootResponse(flavorprofile=result)
@wsme_pecan.wsexpose(profile_types.FlavorProfileRootResponse,
wtypes.text, status_code=200,
body=profile_types.FlavorProfileRootPUT)
def put(self, id, flavor_profile_):
"""Updates a flavor Profile."""
flavorprofile = flavor_profile_.flavorprofile
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_PUT)
# Don't allow changes to the flavor_data or provider_name if it
# is in use.
if (not isinstance(flavorprofile.flavor_data, wtypes.UnsetType) or
not isinstance(flavorprofile.provider_name, wtypes.UnsetType)):
if self.repositories.flavor.count(context.session,
flavor_profile_id=id) > 0:
raise exceptions.ObjectInUse(object='Flavor profile', id=id)
if not isinstance(flavorprofile.flavor_data, wtypes.UnsetType):
# Do a basic JSON validation on the metadata
try:
flavor_data_dict = jsonutils.loads(flavorprofile.flavor_data)
except Exception:
raise exceptions.InvalidOption(
value=flavorprofile.flavor_data,
option=constants.FLAVOR_DATA)
if isinstance(flavorprofile.provider_name, wtypes.UnsetType):
db_flavor_profile = self._get_db_flavor_profile(
context.session, id)
provider_driver = db_flavor_profile.provider_name
else:
provider_driver = flavorprofile.provider_name
# Validate that the provider driver supports the metadata
driver = driver_factory.get_driver(provider_driver)
driver_utils.call_provider(driver.name, driver.validate_flavor,
flavor_data_dict)
lock_session = db_api.get_session(autocommit=False)
try:
flavorprofile_dict = flavorprofile.to_dict(render_unsets=False)
if flavorprofile_dict:
db_flavor_profile = self.repositories.flavor_profile.update(
lock_session, id, **flavorprofile_dict)
lock_session.commit()
except Exception:
with excutils.save_and_reraise_exception():
lock_session.rollback()
# Force SQL alchemy to query the DB, otherwise we get inconsistent
# results
context.session.expire_all()
db_flavor_profile = self._get_db_flavor_profile(context.session, id)
result = self._convert_db_to_type(
db_flavor_profile, profile_types.FlavorProfileResponse)
return profile_types.FlavorProfileRootResponse(flavorprofile=result)
@wsme_pecan.wsexpose(None, wtypes.text, status_code=204)
def delete(self, flavor_profile_id):
"""Deletes a Flavor Profile"""
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_DELETE)
# Don't allow it to be deleted if it is in use by a flavor
if self.repositories.flavor.count(
context.session, flavor_profile_id=flavor_profile_id) > 0:
raise exceptions.ObjectInUse(object='Flavor profile',
id=flavor_profile_id)
try:
self.repositories.flavor_profile.delete(context.session,
id=flavor_profile_id)
except sa_exception.NoResultFound:
raise exceptions.NotFound(resource='Flavor profile',
id=flavor_profile_id)

View File

@ -0,0 +1,144 @@
# Copyright 2014 Rackspace
# Copyright 2016 Blue Box, an IBM Company
#
# 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_db import exception as odb_exceptions
from oslo_log import log as logging
from oslo_utils import excutils
from oslo_utils import uuidutils
import pecan
from sqlalchemy.orm import exc as sa_exception
from wsme import types as wtypes
from wsmeext import pecan as wsme_pecan
from octavia.api.v2.controllers import base
from octavia.api.v2.types import flavors as flavor_types
from octavia.common import constants
from octavia.common import exceptions
from octavia.db import api as db_api
LOG = logging.getLogger(__name__)
class FlavorsController(base.BaseController):
RBAC_TYPE = constants.RBAC_FLAVOR
def __init__(self):
super(FlavorsController, self).__init__()
@wsme_pecan.wsexpose(flavor_types.FlavorRootResponse, wtypes.text,
[wtypes.text], ignore_extra_args=True)
def get_one(self, id, fields=None):
"""Gets a flavor's detail."""
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_GET_ONE)
db_flavor = self._get_db_flavor(context.session, id)
result = self._convert_db_to_type(db_flavor,
flavor_types.FlavorResponse)
if fields is not None:
result = self._filter_fields([result], fields)[0]
return flavor_types.FlavorRootResponse(flavor=result)
@wsme_pecan.wsexpose(flavor_types.FlavorsRootResponse,
[wtypes.text], ignore_extra_args=True)
def get_all(self, fields=None):
"""Lists all flavors."""
pcontext = pecan.request.context
context = pcontext.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_GET_ALL)
db_flavors, links = self.repositories.flavor.get_all(
context.session,
pagination_helper=pcontext.get(constants.PAGINATION_HELPER))
result = self._convert_db_to_type(
db_flavors, [flavor_types.FlavorResponse])
if fields is not None:
result = self._filter_fields(result, fields)
return flavor_types.FlavorsRootResponse(
flavors=result, flavors_links=links)
@wsme_pecan.wsexpose(flavor_types.FlavorRootResponse,
body=flavor_types.FlavorRootPOST, status_code=201)
def post(self, flavor_):
"""Creates a flavor."""
flavor = flavor_.flavor
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_POST)
# TODO(johnsom) Validate the flavor profile ID
lock_session = db_api.get_session(autocommit=False)
try:
flavor_dict = flavor.to_dict(render_unsets=True)
flavor_dict['id'] = uuidutils.generate_uuid()
db_flavor = self.repositories.flavor.create(lock_session,
**flavor_dict)
lock_session.commit()
except odb_exceptions.DBDuplicateEntry:
lock_session.rollback()
raise exceptions.RecordAlreadyExists(field='flavor',
name=flavor.name)
except Exception:
with excutils.save_and_reraise_exception():
lock_session.rollback()
result = self._convert_db_to_type(db_flavor,
flavor_types.FlavorResponse)
return flavor_types.FlavorRootResponse(flavor=result)
@wsme_pecan.wsexpose(flavor_types.FlavorRootResponse,
wtypes.text, status_code=200,
body=flavor_types.FlavorRootPUT)
def put(self, id, flavor_):
flavor = flavor_.flavor
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_PUT)
lock_session = db_api.get_session(autocommit=False)
try:
flavor_dict = flavor.to_dict(render_unsets=False)
if flavor_dict:
db_flavor = self.repositories.flavor.update(lock_session, id,
**flavor_dict)
lock_session.commit()
except Exception:
with excutils.save_and_reraise_exception():
lock_session.rollback()
# Force SQL alchemy to query the DB, otherwise we get inconsistent
# results
context.session.expire_all()
db_flavor = self._get_db_flavor(context.session, id)
result = self._convert_db_to_type(db_flavor,
flavor_types.FlavorResponse)
return flavor_types.FlavorRootResponse(flavor=result)
@wsme_pecan.wsexpose(None, wtypes.text, status_code=204)
def delete(self, flavor_id):
"""Deletes a Flavor"""
context = pecan.request.context.get('octavia_context')
self._auth_validate_action(context, context.project_id,
constants.RBAC_DELETE)
try:
self.repositories.flavor.delete(context.session, id=flavor_id)
# Handle when load balancers still reference this flavor
except odb_exceptions.DBReferenceError:
raise exceptions.ObjectInUse(object='Flavor', id=flavor_id)
except sa_exception.NoResultFound:
raise exceptions.NotFound(resource='Flavor',
id=flavor_id)

View File

@ -0,0 +1,69 @@
# Copyright 2014 Rackspace
#
# 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 wsme import types as wtypes
from octavia.api.common import types
class BaseFlavorProfileType(types.BaseType):
_type_to_model_map = {}
_child_map = {}
class FlavorProfileResponse(BaseFlavorProfileType):
"""Defines which attributes are to be shown on any response."""
id = wtypes.wsattr(wtypes.UuidType())
name = wtypes.wsattr(wtypes.StringType())
provider_name = wtypes.wsattr(wtypes.StringType())
flavor_data = wtypes.wsattr(wtypes.StringType())
@classmethod
def from_data_model(cls, data_model, children=False):
flavorprofile = super(FlavorProfileResponse, cls).from_data_model(
data_model, children=children)
return flavorprofile
class FlavorProfileRootResponse(types.BaseType):
flavorprofile = wtypes.wsattr(FlavorProfileResponse)
class FlavorProfilesRootResponse(types.BaseType):
flavorprofiles = wtypes.wsattr([FlavorProfileResponse])
flavorprofile_links = wtypes.wsattr([types.PageType])
class FlavorProfilePOST(BaseFlavorProfileType):
"""Defines mandatory and optional attributes of a POST request."""
name = wtypes.wsattr(wtypes.StringType(max_length=255), mandatory=True)
provider_name = wtypes.wsattr(wtypes.StringType(max_length=255),
mandatory=True)
flavor_data = wtypes.wsattr(wtypes.StringType(max_length=4096),
mandatory=True)
class FlavorProfileRootPOST(types.BaseType):
flavorprofile = wtypes.wsattr(FlavorProfilePOST)
class FlavorProfilePUT(BaseFlavorProfileType):
"""Defines the attributes of a PUT request."""
name = wtypes.wsattr(wtypes.StringType(max_length=255))
provider_name = wtypes.wsattr(wtypes.StringType(max_length=255))
flavor_data = wtypes.wsattr(wtypes.StringType(max_length=4096))
class FlavorProfileRootPUT(types.BaseType):
flavorprofile = wtypes.wsattr(FlavorProfilePUT)

View File

@ -0,0 +1,69 @@
# Copyright 2014 Rackspace
#
# 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 wsme import types as wtypes
from octavia.api.common import types
class BaseFlavorType(types.BaseType):
_type_to_model_map = {}
_child_map = {}
class FlavorResponse(BaseFlavorType):
"""Defines which attributes are to be shown on any response."""
id = wtypes.wsattr(wtypes.UuidType())
name = wtypes.wsattr(wtypes.StringType())
description = wtypes.wsattr(wtypes.StringType())
enabled = wtypes.wsattr(bool)
flavor_profile_id = wtypes.wsattr(wtypes.StringType())
@classmethod
def from_data_model(cls, data_model, children=False):
flavor = super(FlavorResponse, cls).from_data_model(
data_model, children=children)
return flavor
class FlavorRootResponse(types.BaseType):
flavor = wtypes.wsattr(FlavorResponse)
class FlavorsRootResponse(types.BaseType):
flavors = wtypes.wsattr([FlavorResponse])
flavors_links = wtypes.wsattr([types.PageType])
class FlavorPOST(BaseFlavorType):
"""Defines mandatory and optional attributes of a POST request."""
name = wtypes.wsattr(wtypes.StringType(max_length=255), mandatory=True)
description = wtypes.wsattr(wtypes.StringType(max_length=255))
enabled = wtypes.wsattr(bool, default=True)
flavor_profile_id = wtypes.wsattr(wtypes.UuidType(), mandatory=True)
class FlavorRootPOST(types.BaseType):
flavor = wtypes.wsattr(FlavorPOST)
class FlavorPUT(BaseFlavorType):
"""Defines the attributes of a PUT request."""
name = wtypes.wsattr(wtypes.StringType(max_length=255))
description = wtypes.wsattr(wtypes.StringType(max_length=255))
enabled = wtypes.wsattr(bool)
class FlavorRootPUT(types.BaseType):
flavor = wtypes.wsattr(FlavorPUT)

View File

@ -534,6 +534,8 @@ RBAC_L7RULE = '{}:l7rule:'.format(LOADBALANCER_API)
RBAC_QUOTA = '{}:quota:'.format(LOADBALANCER_API)
RBAC_AMPHORA = '{}:amphora:'.format(LOADBALANCER_API)
RBAC_PROVIDER = '{}:provider:'.format(LOADBALANCER_API)
RBAC_FLAVOR = '{}:flavor:'.format(LOADBALANCER_API)
RBAC_FLAVOR_PROFILE = '{}:flavor-profile:'.format(LOADBALANCER_API)
RBAC_POST = 'post'
RBAC_PUT = 'put'
RBAC_PUT_FAILOVER = 'put_failover'
@ -562,3 +564,5 @@ AMP_NETNS_SVC_PREFIX = 'amphora-netns'
# Amphora Feature Compatibility
HTTP_REUSE = 'has_http_reuse'
FLAVOR_DATA = 'flavor_data'

View File

@ -728,3 +728,25 @@ class Quotas(BaseDataModel):
self.in_use_load_balancer = in_use_load_balancer
self.in_use_member = in_use_member
self.in_use_pool = in_use_pool
class Flavor(BaseDataModel):
def __init__(self, id=None, name=None,
description=None, enabled=None,
flavor_profile_id=None):
self.id = id
self.name = name
self.description = description
self.enabled = enabled
self.flavor_profile_id = flavor_profile_id
class FlavorProfile(BaseDataModel):
def __init__(self, id=None, name=None, provider_name=None,
flavor_data=None):
self.id = id
self.name = name
self.provider_name = provider_name
self.flavor_data = flavor_data

View File

@ -216,6 +216,11 @@ class IDAlreadyExists(APIException):
code = 409
class RecordAlreadyExists(APIException):
msg = _('A %(field)s of %(name)s already exists.')
code = 409
class NoReadyAmphoraeException(OctaviaException):
message = _('There are not any READY amphora available.')
@ -367,3 +372,8 @@ class ProviderUnsupportedOptionError(APIException):
class InputFileError(OctaviaException):
message = _('Error with file %(file_name)s. Reason: %(reason)s')
class ObjectInUse(APIException):
msg = _("%(object)s %(id)s is in use and cannot be modified.")
code = 409

View File

@ -31,7 +31,7 @@ class OctaviaBase(models.ModelBase):
# objects.
if obj.__class__.__name__ in ['Member', 'Pool', 'LoadBalancer',
'Listener', 'Amphora', 'L7Policy',
'L7Rule']:
'L7Rule', 'Flavor', 'FlavorProfile']:
return obj.__class__.__name__ + obj.id
elif obj.__class__.__name__ in ['SessionPersistence', 'HealthMonitor']:
return obj.__class__.__name__ + obj.pool_id

View File

@ -0,0 +1,53 @@
# Copyright 2017 Walmart Stores Inc.
#
# 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.
"""add flavor and flavor_profile table
Revision ID: b9c703669314
Revises: 4f65b4f91c39
Create Date: 2018-01-02 16:05:29.745457
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision = 'b9c703669314'
down_revision = '4f65b4f91c39'
def upgrade():
op.create_table(
u'flavor_profile',
sa.Column(u'id', sa.String(36), nullable=False),
sa.Column(u'name', sa.String(255), nullable=False),
sa.Column(u'provider_name', sa.String(255), nullable=False),
sa.Column(u'flavor_data', sa.String(4096), nullable=False),
sa.PrimaryKeyConstraint(u'id'))
op.create_table(
u'flavor',
sa.Column(u'id', sa.String(36), nullable=False),
sa.Column(u'name', sa.String(255), nullable=False),
sa.Column(u'description', sa.String(255), nullable=True),
sa.Column(u'enabled', sa.Boolean(), nullable=False),
sa.Column(u'flavor_profile_id', sa.String(36), nullable=False),
sa.ForeignKeyConstraint([u'flavor_profile_id'],
[u'flavor_profile.id'],
name=u'fk_flavor_flavor_profile_id'),
sa.PrimaryKeyConstraint(u'id'),
sa.UniqueConstraint(u'name',
name=u'uq_flavor_name'),)

View File

@ -1,5 +1,6 @@
# Copyright 2014 Rackspace
# Copyright 2016 Blue Box, an IBM Company
# Copyright 2017 Walmart Stores Inc.
#
# 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
@ -13,6 +14,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_db.sqlalchemy import models
import sqlalchemy as sa
from sqlalchemy.ext import orderinglist
@ -21,6 +23,8 @@ from sqlalchemy.orm import validates
from sqlalchemy.sql import func
from octavia.api.v2.types import amphora
from octavia.api.v2.types import flavor_profile
from octavia.api.v2.types import flavors
from octavia.api.v2.types import health_monitor
from octavia.api.v2.types import l7policy
from octavia.api.v2.types import l7rule
@ -717,3 +721,40 @@ class Quotas(base_models.BASE):
in_use_load_balancer = sa.Column(sa.Integer(), nullable=True)
in_use_member = sa.Column(sa.Integer(), nullable=True)
in_use_pool = sa.Column(sa.Integer(), nullable=True)
class FlavorProfile(base_models.BASE, base_models.IdMixin,
base_models.NameMixin):
__data_model__ = data_models.FlavorProfile
__tablename__ = "flavor_profile"
__v2_wsme__ = flavor_profile.FlavorProfileResponse
provider_name = sa.Column(sa.String(255), nullable=False)
flavor_data = sa.Column(sa.String(4096), nullable=False)
class Flavor(base_models.BASE,
base_models.IdMixin,
base_models.NameMixin):
__data_model__ = data_models.Flavor
__tablename__ = "flavor"
__v2_wsme__ = flavors.FlavorResponse
__table_args__ = (
sa.UniqueConstraint('name',
name='uq_flavor_name'),
)
description = sa.Column(sa.String(255), nullable=True)
enabled = sa.Column(sa.Boolean(), nullable=False)
flavor_profile_id = sa.Column(
sa.String(36),
sa.ForeignKey("flavor_profile.id",
name="fk_flavor_flavor_profile_id"),
nullable=False)

View File

@ -187,6 +187,8 @@ class Repositories(object):
self.amp_build_slots = AmphoraBuildSlotsRepository()
self.amp_build_req = AmphoraBuildReqRepository()
self.quotas = QuotasRepository()
self.flavor = FlavorRepository()
self.flavor_profile = FlavorProfileRepository()
def create_load_balancer_and_vip(self, session, lb_dict, vip_dict):
"""Inserts load balancer and vip entities into the database.
@ -1768,3 +1770,11 @@ class QuotasRepository(BaseRepository):
quotas.member = None
quotas.pool = None
session.flush()
class FlavorRepository(BaseRepository):
model_class = models.Flavor
class FlavorProfileRepository(BaseRepository):
model_class = models.FlavorProfile

View File

@ -15,6 +15,8 @@ import itertools
from octavia.policies import amphora
from octavia.policies import base
from octavia.policies import flavor
from octavia.policies import flavor_profile
from octavia.policies import healthmonitor
from octavia.policies import l7policy
from octavia.policies import l7rule
@ -29,6 +31,8 @@ from octavia.policies import quota
def list_rules():
return itertools.chain(
base.list_rules(),
flavor.list_rules(),
flavor_profile.list_rules(),
healthmonitor.list_rules(),
l7policy.list_rules(),
l7rule.list_rules(),

View File

@ -0,0 +1,61 @@
# Copyright 2017 Walmart Stores Inc..
# 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_policy import policy
from octavia.common import constants
rules = [
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR,
action=constants.RBAC_GET_ALL),
constants.RULE_API_READ,
"List Flavors",
[{'method': 'GET', 'path': '/v2.0/lbaas/flavors'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR,
action=constants.RBAC_POST),
constants.RULE_API_ADMIN,
"Create a Flavor",
[{'method': 'POST', 'path': '/v2.0/lbaas/flavors'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR,
action=constants.RBAC_PUT),
constants.RULE_API_ADMIN,
"Update a Flavor",
[{'method': 'PUT', 'path': '/v2.0/lbaas/flavors/{flavor_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR,
action=constants.RBAC_GET_ONE),
constants.RULE_API_READ,
"Show Flavor details",
[{'method': 'GET',
'path': '/v2.0/lbaas/flavors/{flavor_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR,
action=constants.RBAC_DELETE),
constants.RULE_API_ADMIN,
"Remove a flavor",
[{'method': 'DELETE',
'path': '/v2.0/lbaas/flavors/{flavor_id}'}]
),
]
def list_rules():
return rules

View File

@ -0,0 +1,62 @@
# Copyright 2017 Walmart Stores Inc..
# 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_policy import policy
from octavia.common import constants
rules = [
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR_PROFILE,
action=constants.RBAC_GET_ALL),
constants.RULE_API_ADMIN,
"List Flavors",
[{'method': 'GET', 'path': '/v2.0/lbaas/flavorprofiles'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR_PROFILE,
action=constants.RBAC_POST),
constants.RULE_API_ADMIN,
"Create a Flavor",
[{'method': 'POST', 'path': '/v2.0/lbaas/flavorprofiles'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR_PROFILE,
action=constants.RBAC_PUT),
constants.RULE_API_ADMIN,
"Update a Flavor",
[{'method': 'PUT',
'path': '/v2.0/lbaas/flavorprofiles/{flavor_profile_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR_PROFILE,
action=constants.RBAC_GET_ONE),
constants.RULE_API_ADMIN,
"Show Flavor details",
[{'method': 'GET',
'path': '/v2.0/lbaas/flavorprofiles/{flavor_profile_id}'}]
),
policy.DocumentedRuleDefault(
'{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_FLAVOR_PROFILE,
action=constants.RBAC_DELETE),
constants.RULE_API_ADMIN,
"Remove a flavor",
[{'method': 'DELETE',
'path': '/v2.0/lbaas/flavorprofiles/{flavor_profile_id}'}]
),
]
def list_rules():
return rules

View File

@ -46,13 +46,15 @@ class TestRootController(base_db_test.OctaviaDBTestBase):
versions = self._get_versions_with_config(
api_v1_enabled=True, api_v2_enabled=True)
version_ids = tuple(v.get('id') for v in versions)
self.assertEqual(7, len(version_ids))
self.assertEqual(8, len(version_ids))
self.assertIn('v1', version_ids)
self.assertIn('v2.0', version_ids)
self.assertIn('v2.1', version_ids)
self.assertIn('v2.2', version_ids)
self.assertIn('v2.3', version_ids)
self.assertIn('v2.4', version_ids)
self.assertIn('v2.5', version_ids)
self.assertIn('v2.6', version_ids)
# Each version should have a 'self' 'href' to the API version URL
# [{u'rel': u'self', u'href': u'http://localhost/v2'}]
@ -72,12 +74,14 @@ class TestRootController(base_db_test.OctaviaDBTestBase):
def test_api_v1_disabled(self):
versions = self._get_versions_with_config(
api_v1_enabled=False, api_v2_enabled=True)
self.assertEqual(6, len(versions))
self.assertEqual(7, len(versions))
self.assertEqual('v2.0', versions[0].get('id'))
self.assertEqual('v2.1', versions[1].get('id'))
self.assertEqual('v2.2', versions[2].get('id'))
self.assertEqual('v2.3', versions[3].get('id'))
self.assertEqual('v2.4', versions[4].get('id'))
self.assertEqual('v2.5', versions[5].get('id'))
self.assertEqual('v2.6', versions[6].get('id'))
def test_api_v2_disabled(self):
versions = self._get_versions_with_config(

View File

@ -32,6 +32,14 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
BASE_PATH = '/v2'
BASE_PATH_v2_0 = '/v2.0'
# /lbaas/flavors
FLAVORS_PATH = '/flavors'
FLAVOR_PATH = FLAVORS_PATH + '/{flavor_id}'
# /lbaas/flavorprofiles
FPS_PATH = '/flavorprofiles'
FP_PATH = FPS_PATH + '/{fp_id}'
# /lbaas/loadbalancers
LBS_PATH = '/lbaas/loadbalancers'
LB_PATH = LBS_PATH + '/{lb_id}'
@ -89,6 +97,7 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
enabled_provider_drivers={
'amphora': 'Amp driver.',
'noop_driver': 'NoOp driver.',
'noop_driver-alt': 'NoOp driver alt alisas.',
'octavia': 'Octavia driver.'})
self.lb_repo = repositories.LoadBalancerRepository()
self.listener_repo = repositories.ListenerRepository()
@ -99,6 +108,8 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
self.l7rule_repo = repositories.L7RuleRepository()
self.health_monitor_repo = repositories.HealthMonitorRepository()
self.amphora_repo = repositories.AmphoraRepository()
self.flavor_repo = repositories.FlavorRepository()
self.flavor_profile_repo = repositories.FlavorProfileRepository()
patcher2 = mock.patch('octavia.certificates.manager.barbican.'
'BarbicanCertManager')
self.cert_manager_mock = patcher2.start()
@ -183,6 +194,21 @@ class BaseAPITest(base_db_test.OctaviaDBTestBase):
expect_errors=expect_errors)
return response
def create_flavor(self, name, description, flavor_profile_id, enabled):
req_dict = {'name': name, 'description': description,
'flavor_profile_id': flavor_profile_id,
'enabled': enabled}
body = {'flavor': req_dict}
response = self.post(self.FLAVORS_PATH, body)
return response.json.get('flavor')
def create_flavor_profile(self, name, privider_name, flavor_data):
req_dict = {'name': name, 'provider_name': privider_name,
constants.FLAVOR_DATA: flavor_data}
body = {'flavorprofile': req_dict}
response = self.post(self.FPS_PATH, body)
return response.json.get('flavorprofile')
def create_load_balancer(self, vip_subnet_id,
**optionals):
req_dict = {'vip_subnet_id': vip_subnet_id,

View File

@ -0,0 +1,530 @@
# Copyright 2017 Walmart Stores Inc.
#
# 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 mock
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from oslo_db import exception as odb_exceptions
from oslo_utils import uuidutils
from octavia.common import constants
import octavia.common.context
from octavia.tests.functional.api.v2 import base
class TestFlavorProfiles(base.BaseAPITest):
root_tag = 'flavorprofile'
root_tag_list = 'flavorprofiles'
root_tag_links = 'flavorprofile_links'
def _assert_request_matches_response(self, req, resp, **optionals):
self.assertTrue(uuidutils.is_uuid_like(resp.get('id')))
self.assertEqual(req.get('name'), resp.get('name'))
self.assertEqual(req.get('provider_name'),
resp.get('provider_name'))
self.assertEqual(req.get(constants.FLAVOR_DATA),
resp.get(constants.FLAVOR_DATA))
def test_empty_list(self):
response = self.get(self.FPS_PATH)
api_list = response.json.get(self.root_tag_list)
self.assertEqual([], api_list)
def test_create(self):
fp_json = {'name': 'test1', 'provider_name': 'noop_driver',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body)
api_fp = response.json.get(self.root_tag)
self._assert_request_matches_response(fp_json, api_fp)
def test_create_with_missing_name(self):
fp_json = {'provider_name': 'pr1', constants.FLAVOR_DATA: '{"x": "y"}'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute name. Value: "
"'None'. Mandatory field missing.")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_with_missing_provider(self):
fp_json = {'name': 'xyz', constants.FLAVOR_DATA: '{"x": "y"}'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute provider_name. "
"Value: 'None'. Mandatory field missing.")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_with_missing_flavor_data(self):
fp_json = {'name': 'xyz', 'provider_name': 'pr1'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute flavor_data. "
"Value: 'None'. Mandatory field missing.")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_with_empty_flavor_data(self):
fp_json = {'name': 'test1', 'provider_name': 'noop_driver',
constants.FLAVOR_DATA: '{}'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body)
api_fp = response.json.get(self.root_tag)
self._assert_request_matches_response(fp_json, api_fp)
def test_create_with_long_name(self):
fp_json = {'name': 'n' * 256, 'provider_name': 'test1',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(fp_json)
self.post(self.FPS_PATH, body, status=400)
def test_create_with_long_provider(self):
fp_json = {'name': 'name1', 'provider_name': 'n' * 256,
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(fp_json)
self.post(self.FPS_PATH, body, status=400)
def test_create_with_long_flavor_data(self):
fp_json = {'name': 'name1', 'provider_name': 'amp',
constants.FLAVOR_DATA: 'n' * 4097}
body = self._build_body(fp_json)
self.post(self.FPS_PATH, body, status=400)
def test_create_authorized(self):
fp_json = {'name': 'test1', 'provider_name': 'noop_driver',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(fp_json)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.post(self.FPS_PATH, body)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
api_fp = response.json.get(self.root_tag)
self._assert_request_matches_response(fp_json, api_fp)
def test_create_not_authorized(self):
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
fp_json = {'name': 'name',
'provider_name': 'xyz', constants.FLAVOR_DATA: '{"x": "y"}'}
body = self._build_body(fp_json)
response = self.post(self.FPS_PATH, body, status=403)
api_fp = response.json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_fp)
def test_create_db_failure(self):
fp_json = {'name': 'test1', 'provider_name': 'noop_driver',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(fp_json)
with mock.patch("octavia.db.repositories.FlavorProfileRepository."
"create") as mock_create:
mock_create.side_effect = Exception
self.post(self.FPS_PATH, body, status=500)
mock_create.side_effect = odb_exceptions.DBDuplicateEntry
self.post(self.FPS_PATH, body, status=409)
def test_create_with_invalid_json(self):
fp_json = {'name': 'test1', 'provider_name': 'noop_driver',
constants.FLAVOR_DATA: '{hello: "world"}'}
body = self._build_body(fp_json)
self.post(self.FPS_PATH, body, status=400)
def test_get(self):
fp = self.create_flavor_profile('name', 'noop_driver',
'{"x": "y"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
response = self.get(
self.FP_PATH.format(
fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('name', response.get('name'))
self.assertEqual(fp.get('id'), response.get('id'))
def test_get_one_fields_filter(self):
fp = self.create_flavor_profile('name', 'noop_driver',
'{"x": "y"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id')), params={
'fields': ['id', 'provider_name']}).json.get(self.root_tag)
self.assertEqual(fp.get('id'), response.get('id'))
self.assertIn(u'id', response)
self.assertIn(u'provider_name', response)
self.assertNotIn(u'name', response)
self.assertNotIn(constants.FLAVOR_DATA, response)
def test_get_authorized(self):
fp = self.create_flavor_profile('name', 'noop_driver',
'{"x": "y"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(
self.FP_PATH.format(
fp_id=fp.get('id'))).json.get(self.root_tag)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual('name', response.get('name'))
self.assertEqual(fp.get('id'), response.get('id'))
def test_get_not_authorized(self):
fp = self.create_flavor_profile('name', 'noop_driver',
'{"x": "y"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
self.get(self.FP_PATH.format(fp_id=fp.get('id')), status=403)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
def test_get_all(self):
fp1 = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
ref_fp_1 = {u'flavor_data': u'{"image": "ubuntu"}',
u'id': fp1.get('id'), u'name': u'test1',
u'provider_name': u'noop_driver'}
self.assertTrue(uuidutils.is_uuid_like(fp1.get('id')))
fp2 = self.create_flavor_profile('test2', 'noop_driver-alt',
'{"image": "ubuntu"}')
ref_fp_2 = {u'flavor_data': u'{"image": "ubuntu"}',
u'id': fp2.get('id'), u'name': u'test2',
u'provider_name': u'noop_driver-alt'}
self.assertTrue(uuidutils.is_uuid_like(fp2.get('id')))
response = self.get(self.FPS_PATH)
api_list = response.json.get(self.root_tag_list)
self.assertEqual(2, len(api_list))
self.assertIn(ref_fp_1, api_list)
self.assertIn(ref_fp_2, api_list)
def test_get_all_fields_filter(self):
fp1 = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp1.get('id')))
fp2 = self.create_flavor_profile('test2', 'noop_driver-alt',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp2.get('id')))
response = self.get(self.FPS_PATH, params={
'fields': ['id', 'name']})
api_list = response.json.get(self.root_tag_list)
self.assertEqual(2, len(api_list))
for profile in api_list:
self.assertIn(u'id', profile)
self.assertIn(u'name', profile)
self.assertNotIn(u'provider_name', profile)
self.assertNotIn(constants.FLAVOR_DATA, profile)
def test_get_all_authorized(self):
fp1 = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp1.get('id')))
fp2 = self.create_flavor_profile('test2', 'noop_driver-alt',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp2.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(self.FPS_PATH)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
api_list = response.json.get(self.root_tag_list)
self.assertEqual(2, len(api_list))
def test_get_all_not_authorized(self):
fp1 = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp1.get('id')))
fp2 = self.create_flavor_profile('test2', 'noop_driver-alt',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp2.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
self.get(self.FPS_PATH, status=403)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
def test_update(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
update_data = {'name': 'the_profile',
'provider_name': 'noop_driver-alt',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(update_data)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('the_profile', response.get('name'))
self.assertEqual('noop_driver-alt', response.get('provider_name'))
self.assertEqual('{"hello": "world"}',
response.get(constants.FLAVOR_DATA))
def test_update_none(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
body = self._build_body({})
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test_profile', response.get('name'))
self.assertEqual('noop_driver', response.get('provider_name'))
self.assertEqual('{"x": "y"}',
response.get(constants.FLAVOR_DATA))
def test_update_no_flavor_data(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
update_data = {'name': 'the_profile',
'provider_name': 'noop_driver-alt'}
body = self._build_body(update_data)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('the_profile', response.get('name'))
self.assertEqual('noop_driver-alt', response.get('provider_name'))
self.assertEqual('{"x": "y"}', response.get(constants.FLAVOR_DATA))
def test_update_authorized(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
update_data = {'name': 'the_profile',
'provider_name': 'noop_driver-alt',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(update_data)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')),
body)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('the_profile', response.get('name'))
self.assertEqual('noop_driver-alt', response.get('provider_name'))
self.assertEqual('{"hello": "world"}',
response.get(constants.FLAVOR_DATA))
def test_update_not_authorized(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
update_data = {'name': 'the_profile', 'provider_name': 'amp',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(update_data)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')),
body, status=403)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test_profile', response.get('name'))
self.assertEqual('noop_driver', response.get('provider_name'))
self.assertEqual('{"x": "y"}',
response.get(constants.FLAVOR_DATA))
def test_update_in_use(self):
fp = self.create_flavor_profile('test_profile', 'noop_driver',
'{"x": "y"}')
self.create_flavor('name1', 'description', fp.get('id'), True)
# Test updating provider while in use is not allowed
update_data = {'name': 'the_profile',
'provider_name': 'noop_driver-alt'}
body = self._build_body(update_data)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body,
status=409)
err_msg = ("Flavor profile {} is in use and cannot be "
"modified.".format(fp.get('id')))
self.assertEqual(err_msg, response.json.get('faultstring'))
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test_profile', response.get('name'))
self.assertEqual('noop_driver', response.get('provider_name'))
self.assertEqual('{"x": "y"}', response.get(constants.FLAVOR_DATA))
# Test updating flavor data while in use is not allowed
update_data = {'name': 'the_profile',
constants.FLAVOR_DATA: '{"hello": "world"}'}
body = self._build_body(update_data)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body,
status=409)
err_msg = ("Flavor profile {} is in use and cannot be "
"modified.".format(fp.get('id')))
self.assertEqual(err_msg, response.json.get('faultstring'))
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test_profile', response.get('name'))
self.assertEqual('noop_driver', response.get('provider_name'))
self.assertEqual('{"x": "y"}', response.get(constants.FLAVOR_DATA))
# Test that you can still update the name when in use
update_data = {'name': 'the_profile'}
body = self._build_body(update_data)
response = self.put(self.FP_PATH.format(fp_id=fp.get('id')), body)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('the_profile', response.get('name'))
self.assertEqual('noop_driver', response.get('provider_name'))
self.assertEqual('{"x": "y"}', response.get(constants.FLAVOR_DATA))
def test_delete(self):
fp = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
self.delete(self.FP_PATH.format(fp_id=fp.get('id')))
response = self.get(self.FP_PATH.format(
fp_id=fp.get('id')), status=404)
err_msg = "Flavor Profile %s not found." % fp.get('id')
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_delete_authorized(self):
fp = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
self.delete(self.FP_PATH.format(fp_id=fp.get('id')))
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
response = self.get(self.FP_PATH.format(
fp_id=fp.get('id')), status=404)
err_msg = "Flavor Profile %s not found." % fp.get('id')
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_delete_not_authorized(self):
fp = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.assertTrue(uuidutils.is_uuid_like(fp.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.delete(self.FP_PATH.format(
fp_id=fp.get('id')), status=403)
api_fp = response.json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_fp)
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test1', response.get('name'))
def test_delete_in_use(self):
fp = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
self.create_flavor('name1', 'description', fp.get('id'), True)
response = self.delete(self.FP_PATH.format(fp_id=fp.get('id')),
status=409)
err_msg = ("Flavor profile {} is in use and cannot be "
"modified.".format(fp.get('id')))
self.assertEqual(err_msg, response.json.get('faultstring'))
response = self.get(
self.FP_PATH.format(fp_id=fp.get('id'))).json.get(self.root_tag)
self.assertEqual('test1', response.get('name'))

View File

@ -0,0 +1,541 @@
# Copyright 2017 Walmart Stores Inc.
#
# 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 mock
from oslo_utils import uuidutils
from oslo_config import cfg
from oslo_config import fixture as oslo_fixture
from octavia.common import constants
import octavia.common.context
from octavia.tests.functional.api.v2 import base
class TestFlavors(base.BaseAPITest):
root_tag = 'flavor'
root_tag_list = 'flavors'
root_tag_links = 'flavors_links'
def setUp(self):
super(TestFlavors, self).setUp()
self.fp = self.create_flavor_profile('test1', 'noop_driver',
'{"image": "ubuntu"}')
def _assert_request_matches_response(self, req, resp, **optionals):
self.assertTrue(uuidutils.is_uuid_like(resp.get('id')))
req_description = req.get('description')
self.assertEqual(req.get('name'), resp.get('name'))
if not req_description:
self.assertEqual('', resp.get('description'))
else:
self.assertEqual(req.get('description'), resp.get('description'))
self.assertEqual(req.get('flavor_profile_id'),
resp.get('flavor_profile_id'))
self.assertEqual(req.get('enabled', True),
resp.get('enabled'))
def test_empty_list(self):
response = self.get(self.FLAVORS_PATH)
api_list = response.json.get(self.root_tag_list)
self.assertEqual([], api_list)
def test_create(self):
flavor_json = {'name': 'test1',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
self._assert_request_matches_response(flavor_json, api_flavor)
def test_create_with_missing_name(self):
flavor_json = {'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute name. Value: "
"'None'. Mandatory field missing.")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_with_long_name(self):
flavor_json = {'name': 'n' * 256,
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
self.post(self.FLAVORS_PATH, body, status=400)
def test_create_with_long_description(self):
flavor_json = {'name': 'test-flavor',
'description': 'n' * 256,
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
self.post(self.FLAVORS_PATH, body, status=400)
def test_create_with_missing_flavor_profile(self):
flavor_json = {'name': 'xyz'}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute flavor_profile_id. "
"Value: 'None'. Mandatory field missing.")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_with_bad_flavor_profile(self):
flavor_json = {'name': 'xyz', 'flavor_profile_id': 'bogus'}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body, status=400)
err_msg = ("Invalid input for field/attribute flavor_profile_id. "
"Value: 'bogus'. Value should be UUID format")
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_duplicate_names(self):
flavor1 = self.create_flavor('name', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor1.get('id')))
flavor_json = {'name': 'name',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body, status=409)
err_msg = "A flavor of name already exists."
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_create_authorized(self):
flavor_json = {'name': 'test1',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self._assert_request_matches_response(flavor_json, api_flavor)
def test_create_not_authorized(self):
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
flavor_json = {'name': 'name',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body, status=403)
api_flavor = response.json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_flavor)
def test_create_db_failure(self):
flavor_json = {'name': 'test1',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
with mock.patch("octavia.db.repositories.FlavorRepository."
"create") as mock_create:
mock_create.side_effect = Exception
self.post(self.FLAVORS_PATH, body, status=500)
def test_get(self):
flavor = self.create_flavor('name', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
response = self.get(
self.FLAVOR_PATH.format(
flavor_id=flavor.get('id'))).json.get(self.root_tag)
self.assertEqual('name', response.get('name'))
self.assertEqual('description', response.get('description'))
self.assertEqual(flavor.get('id'), response.get('id'))
self.assertEqual(self.fp.get('id'), response.get('flavor_profile_id'))
self.assertTrue(response.get('enabled'))
def test_get_one_fields_filter(self):
flavor = self.create_flavor('name', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
response = self.get(
self.FLAVOR_PATH.format(flavor_id=flavor.get('id')), params={
'fields': ['id', 'flavor_profile_id']}).json.get(self.root_tag)
self.assertEqual(flavor.get('id'), response.get('id'))
self.assertEqual(self.fp.get('id'), response.get('flavor_profile_id'))
self.assertIn(u'id', response)
self.assertIn(u'flavor_profile_id', response)
self.assertNotIn(u'name', response)
self.assertNotIn(u'description', response)
self.assertNotIn(u'enabled', response)
def test_get_authorized(self):
flavor = self.create_flavor('name', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.get(
self.FLAVOR_PATH.format(
flavor_id=flavor.get('id'))).json.get(self.root_tag)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual('name', response.get('name'))
self.assertEqual('description', response.get('description'))
self.assertEqual(flavor.get('id'), response.get('id'))
self.assertEqual(self.fp.get('id'), response.get('flavor_profile_id'))
self.assertTrue(response.get('enabled'))
def test_get_not_authorized(self):
flavor = self.create_flavor('name', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor.get('id')), status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response)
def test_get_all(self):
ref_flavor_1 = {
u'description': u'description', u'enabled': True,
u'flavor_profile_id': u'd21bf20d-c323-4004-bf67-f90591ceced9',
u'id': u'172ccb10-a3b7-4c73-aee8-bdb77fb51ed5',
u'name': u'name1'}
flavor1 = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor1.get('id')))
ref_flavor_1 = {
u'description': u'description', u'enabled': True,
u'flavor_profile_id': self.fp.get('id'),
u'id': flavor1.get('id'),
u'name': u'name1'}
flavor2 = self.create_flavor('name2', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor2.get('id')))
ref_flavor_2 = {
u'description': u'description', u'enabled': True,
u'flavor_profile_id': self.fp.get('id'),
u'id': flavor2.get('id'),
u'name': u'name2'}
response = self.get(self.FLAVORS_PATH)
api_list = response.json.get(self.root_tag_list)
self.assertEqual(2, len(api_list))
self.assertIn(ref_flavor_1, api_list)
self.assertIn(ref_flavor_2, api_list)
def test_get_all_fields_filter(self):
flavor1 = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor1.get('id')))
flavor2 = self.create_flavor('name2', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor2.get('id')))
response = self.get(self.FLAVORS_PATH, params={
'fields': ['id', 'name']})
api_list = response.json.get(self.root_tag_list)
self.assertEqual(2, len(api_list))
for flavor in api_list:
self.assertIn(u'id', flavor)
self.assertIn(u'name', flavor)
self.assertNotIn(u'flavor_profile_id', flavor)
self.assertNotIn(u'description', flavor)
self.assertNotIn(u'enabled', flavor)
def test_get_all_authorized(self):
flavor1 = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor1.get('id')))
flavor2 = self.create_flavor('name2', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor2.get('id')))
response = self.get(self.FLAVORS_PATH)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': False,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
api_list = response.json.get(self.root_tag_list)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(2, len(api_list))
def test_get_all_not_authorized(self):
flavor1 = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor1.get('id')))
flavor2 = self.create_flavor('name2', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor2.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.get(self.FLAVORS_PATH, status=403).json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, response)
def test_update(self):
flavor_json = {'name': 'Fancy_Flavor',
'description': 'A great flavor. Pick me!',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
flavor_id = api_flavor.get('id')
flavor_json = {'name': 'Better_Flavor',
'description': 'An even better flavor. Pick me!',
'enabled': False}
body = self._build_body(flavor_json)
response = self.put(self.FLAVOR_PATH.format(flavor_id=flavor_id), body)
updated_flavor = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor_id)).json.get(self.root_tag)
self.assertEqual('Better_Flavor', updated_flavor.get('name'))
self.assertEqual('An even better flavor. Pick me!',
updated_flavor.get('description'))
self.assertEqual(flavor_id, updated_flavor.get('id'))
self.assertEqual(self.fp.get('id'),
updated_flavor.get('flavor_profile_id'))
self.assertFalse(updated_flavor.get('enabled'))
def test_update_none(self):
flavor_json = {'name': 'Fancy_Flavor',
'description': 'A great flavor. Pick me!',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
flavor_id = api_flavor.get('id')
flavor_json = {}
body = self._build_body(flavor_json)
response = self.put(self.FLAVOR_PATH.format(flavor_id=flavor_id), body)
updated_flavor = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor_id)).json.get(self.root_tag)
self.assertEqual('Fancy_Flavor', updated_flavor.get('name'))
self.assertEqual('A great flavor. Pick me!',
updated_flavor.get('description'))
self.assertEqual(flavor_id, updated_flavor.get('id'))
self.assertEqual(self.fp.get('id'),
updated_flavor.get('flavor_profile_id'))
self.assertTrue(updated_flavor.get('enabled'))
def test_update_flavor_profile_id(self):
flavor_json = {'name': 'Fancy_Flavor',
'description': 'A great flavor. Pick me!',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
flavor_id = api_flavor.get('id')
flavor_json = {'flavor_profile_id': uuidutils.generate_uuid()}
body = self._build_body(flavor_json)
response = self.put(self.FLAVOR_PATH.format(flavor_id=flavor_id),
body, status=400)
updated_flavor = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor_id)).json.get(self.root_tag)
self.assertEqual(self.fp.get('id'),
updated_flavor.get('flavor_profile_id'))
def test_update_authorized(self):
flavor_json = {'name': 'Fancy_Flavor',
'description': 'A great flavor. Pick me!',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
flavor_id = api_flavor.get('id')
flavor_json = {'name': 'Better_Flavor',
'description': 'An even better flavor. Pick me!',
'enabled': False}
body = self._build_body(flavor_json)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
response = self.put(self.FLAVOR_PATH.format(
flavor_id=flavor_id), body)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
updated_flavor = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor_id)).json.get(self.root_tag)
self.assertEqual('Better_Flavor', updated_flavor.get('name'))
self.assertEqual('An even better flavor. Pick me!',
updated_flavor.get('description'))
self.assertEqual(flavor_id, updated_flavor.get('id'))
self.assertEqual(self.fp.get('id'),
updated_flavor.get('flavor_profile_id'))
self.assertFalse(updated_flavor.get('enabled'))
def test_update_not_authorized(self):
flavor_json = {'name': 'Fancy_Flavor',
'description': 'A great flavor. Pick me!',
'flavor_profile_id': self.fp.get('id')}
body = self._build_body(flavor_json)
response = self.post(self.FLAVORS_PATH, body)
api_flavor = response.json.get(self.root_tag)
flavor_id = api_flavor.get('id')
flavor_json = {'name': 'Better_Flavor',
'description': 'An even better flavor. Pick me!',
'enabled': False}
body = self._build_body(flavor_json)
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.put(self.FLAVOR_PATH.format(flavor_id=flavor_id),
body, status=403)
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
updated_flavor = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor_id)).json.get(self.root_tag)
self.assertEqual('Fancy_Flavor', updated_flavor.get('name'))
self.assertEqual('A great flavor. Pick me!',
updated_flavor.get('description'))
self.assertEqual(flavor_id, updated_flavor.get('id'))
self.assertEqual(self.fp.get('id'),
updated_flavor.get('flavor_profile_id'))
self.assertTrue(updated_flavor.get('enabled'))
def test_delete(self):
flavor = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
self.delete(self.FLAVOR_PATH.format(flavor_id=flavor.get('id')))
response = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor.get('id')), status=404)
err_msg = "Flavor %s not found." % flavor.get('id')
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_delete_authorized(self):
flavor = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
project_id = uuidutils.generate_uuid()
with mock.patch.object(octavia.common.context.Context, 'project_id',
project_id):
override_credentials = {
'service_user_id': None,
'user_domain_id': None,
'is_admin_project': True,
'service_project_domain_id': None,
'service_project_id': None,
'roles': ['load-balancer_member'],
'user_id': None,
'is_admin': True,
'service_user_domain_id': None,
'project_domain_id': None,
'service_roles': [],
'project_id': project_id}
with mock.patch(
"oslo_context.context.RequestContext.to_policy_values",
return_value=override_credentials):
self.delete(
self.FLAVOR_PATH.format(flavor_id=flavor.get('id')))
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
response = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor.get('id')), status=404)
err_msg = "Flavor %s not found." % flavor.get('id')
self.assertEqual(err_msg, response.json.get('faultstring'))
def test_delete_not_authorized(self):
flavor = self.create_flavor('name1', 'description', self.fp.get('id'),
True)
self.assertTrue(uuidutils.is_uuid_like(flavor.get('id')))
self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
response = self.delete(self.FLAVOR_PATH.format(
flavor_id=flavor.get('id')), status=403)
api_flavor = response.json
self.conf.config(group='api_settings', auth_strategy=auth_strategy)
self.assertEqual(self.NOT_AUTHORIZED_BODY, api_flavor)
response = self.get(self.FLAVOR_PATH.format(
flavor_id=flavor.get('id'))).json.get(self.root_tag)
self.assertEqual('name1', response.get('name'))

View File

@ -28,7 +28,7 @@ class TestProvider(base.BaseAPITest):
amphora_dict = {u'description': u'Amp driver.', u'name': u'amphora'}
noop_dict = {u'description': u'NoOp driver.', u'name': u'noop_driver'}
providers = self.get(self.PROVIDERS_PATH).json.get(self.root_tag_list)
self.assertEqual(3, len(providers))
self.assertEqual(4, len(providers))
self.assertTrue(octavia_dict in providers)
self.assertTrue(amphora_dict in providers)
self.assertTrue(noop_dict in providers)
@ -39,7 +39,7 @@ class TestProvider(base.BaseAPITest):
noop_dict = {u'name': u'noop_driver'}
providers = self.get(self.PROVIDERS_PATH, params={'fields': ['name']})
providers_list = providers.json.get(self.root_tag_list)
self.assertEqual(3, len(providers_list))
self.assertEqual(4, len(providers_list))
self.assertTrue(octavia_dict in providers_list)
self.assertTrue(amphora_dict in providers_list)
self.assertTrue(noop_dict in providers_list)

View File

@ -37,6 +37,23 @@ class ModelTestMixin(object):
session.add(model)
return model
def create_flavor_profile(self, session, **overrides):
kwargs = {'id': self.FAKE_UUID_1,
'name': 'fake_profile',
'provider_name': 'fake_provider',
'flavor_data': "{'glance_image': 'ubuntu-16.04.03'}"}
kwargs.update(overrides)
return self._insert(session, models.FlavorProfile, kwargs)
def create_flavor(self, session, profile, **overrides):
kwargs = {'id': self.FAKE_UUID_1,
'name': 'fake_flavor',
'flavor_profile_id': profile,
'description': 'fake flavor',
'enabled': True}
kwargs.update(overrides)
return self._insert(session, models.Flavor, kwargs)
def associate_amphora(self, load_balancer, amphora):
load_balancer.amphorae.append(amphora)
@ -1715,3 +1732,45 @@ class TestDataModelManipulations(base.OctaviaDBTestBase, ModelTestMixin):
self.assertEqual(l7p.redirect_pool, new_pool)
self.assertIn(new_pool, listener.pools)
self.assertIn(listener, new_pool.listeners)
class FlavorModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def setUp(self):
super(FlavorModelTest, self).setUp()
self.profile = self.create_flavor_profile(self.session)
def test_create(self):
flavor = self.create_flavor(self.session, self.profile.id)
self.assertIsNotNone(flavor.id)
def test_delete(self):
flavor = self.create_flavor(self.session, self.profile.id)
self.assertIsNotNone(flavor.id)
id = flavor.id
with self.session.begin():
self.session.delete(flavor)
self.session.flush()
new_flavor = self.session.query(
models.Flavor).filter_by(id=id).first()
self.assertIsNone(new_flavor)
class FlavorProfileModelTest(base.OctaviaDBTestBase, ModelTestMixin):
def test_create(self):
fp = self.create_flavor_profile(self.session)
self.assertIsNotNone(fp.id)
def test_delete(self):
fp = self.create_flavor_profile(self.session)
self.assertIsNotNone(fp.id)
id = fp.id
with self.session.begin():
self.session.delete(fp)
self.session.flush()
new_fp = self.session.query(
models.FlavorProfile).filter_by(id=id).first()
self.assertIsNone(new_fp)

View File

@ -62,6 +62,8 @@ class BaseRepositoryTest(base.OctaviaDBTestBase):
self.l7policy_repo = repo.L7PolicyRepository()
self.l7rule_repo = repo.L7RuleRepository()
self.quota_repo = repo.QuotasRepository()
self.flavor_repo = repo.FlavorRepository()
self.flavor_profile_repo = repo.FlavorProfileRepository()
def test_get_all_return_value(self):
pool_list, _ = self.pool_repo.get_all(self.session,
@ -76,6 +78,12 @@ class BaseRepositoryTest(base.OctaviaDBTestBase):
member_list, _ = self.member_repo.get_all(self.session,
project_id=self.FAKE_UUID_2)
self.assertIsInstance(member_list, list)
fp_list, _ = self.flavor_profile_repo.get_all(
self.session, id=self.FAKE_UUID_2)
self.assertIsInstance(fp_list, list)
flavor_list, _ = self.flavor_repo.get_all(
self.session, id=self.FAKE_UUID_2)
self.assertIsInstance(flavor_list, list)
class AllRepositoriesTest(base.OctaviaDBTestBase):
@ -109,7 +117,8 @@ class AllRepositoriesTest(base.OctaviaDBTestBase):
'session_persistence', 'pool', 'member', 'listener',
'listener_stats', 'amphora', 'sni',
'amphorahealth', 'vrrpgroup', 'l7rule', 'l7policy',
'amp_build_slots', 'amp_build_req', 'quotas')
'amp_build_slots', 'amp_build_req', 'quotas',
'flavor', 'flavor_profile')
for repo_attr in repo_attr_names:
single_repo = getattr(self.repos, repo_attr, None)
message = ("Class Repositories should have %s instance"
@ -4218,3 +4227,84 @@ class TestQuotasRepository(BaseRepositoryTest):
self.assertRaises(exceptions.NotFound,
self.quota_repo.delete,
self.session, 'bogus')
class FlavorProfileRepositoryTest(BaseRepositoryTest):
def create_flavor_profile(self, fp_id):
fp = self.flavor_profile_repo.create(
self.session, id=fp_id, name="fp1", provider_name='pr1',
flavor_data="{'image': 'unbuntu'}")
return fp
def test_get(self):
fp = self.create_flavor_profile(fp_id=self.FAKE_UUID_1)
new_fp = self.flavor_profile_repo.get(self.session, id=fp.id)
self.assertIsInstance(new_fp, models.FlavorProfile)
self.assertEqual(fp, new_fp)
def test_get_all(self):
fp1 = self.create_flavor_profile(fp_id=self.FAKE_UUID_1)
fp2 = self.create_flavor_profile(fp_id=self.FAKE_UUID_2)
fp_list, _ = self.flavor_profile_repo.get_all(self.session)
self.assertIsInstance(fp_list, list)
self.assertEqual(2, len(fp_list))
self.assertEqual(fp1, fp_list[0])
self.assertEqual(fp2, fp_list[1])
def test_create(self):
fp = self.create_flavor_profile(fp_id=self.FAKE_UUID_1)
self.assertIsInstance(fp, models.FlavorProfile)
self.assertEqual(self.FAKE_UUID_1, fp.id)
self.assertEqual("fp1", fp.name)
def test_delete(self):
fp = self.create_flavor_profile(fp_id=self.FAKE_UUID_1)
self.flavor_profile_repo.delete(self.session, id=fp.id)
self.assertIsNone(self.flavor_profile_repo.get(
self.session, id=fp.id))
class FlavorRepositoryTest(BaseRepositoryTest):
def create_flavor_profile(self):
fp = self.flavor_profile_repo.create(
self.session, id=uuidutils.generate_uuid(),
name="fp1", provider_name='pr1',
flavor_data="{'image': 'unbuntu'}")
return fp
def create_flavor(self, flavor_id, name):
fp = self.create_flavor_profile()
flavor = self.flavor_repo.create(
self.session, id=flavor_id, name=name,
flavor_profile_id=fp.id, description='test',
enabled=True)
return flavor
def test_get(self):
flavor = self.create_flavor(flavor_id=self.FAKE_UUID_2, name='flavor')
new_flavor = self.flavor_repo.get(self.session, id=flavor.id)
self.assertIsInstance(new_flavor, models.Flavor)
self.assertEqual(flavor, new_flavor)
def test_get_all(self):
fl1 = self.create_flavor(flavor_id=self.FAKE_UUID_2, name='flavor1')
fl2 = self.create_flavor(flavor_id=self.FAKE_UUID_3, name='flavor2')
fl_list, _ = self.flavor_repo.get_all(self.session)
self.assertIsInstance(fl_list, list)
self.assertEqual(2, len(fl_list))
self.assertEqual(fl1, fl_list[0])
self.assertEqual(fl2, fl_list[1])
def test_create(self):
fl = self.create_flavor(flavor_id=self.FAKE_UUID_2, name='fl1')
self.assertIsInstance(fl, models.Flavor)
self.assertEqual(self.FAKE_UUID_2, fl.id)
self.assertEqual("fl1", fl.name)
def test_delete(self):
fl = self.create_flavor(flavor_id=self.FAKE_UUID_2, name='fl1')
self.flavor_repo.delete(self.session, id=fl.id)
self.assertIsNone(self.flavor_repo.get(
self.session, id=fl.id))

View File

@ -141,9 +141,8 @@ class TestNoopProviderDriver(base.TestCase):
vip_port_id=self.vip_port_id,
vip_subnet_id=self.vip_subnet_id)
self.ref_flavor_metadata = {
'amp_image_tag': 'The glance image tag to use for this load '
'balancer.'}
self.ref_flavor_metadata = {"amp_image_tag": "The glance image tag "
"to use for this load balancer."}
def test_create_vip_port(self):
vip_dict = self.driver.create_vip_port(self.loadbalancer_id,
@ -302,6 +301,6 @@ class TestNoopProviderDriver(base.TestCase):
def test_validate_flavor(self):
self.driver.validate_flavor(self.ref_flavor_metadata)
flavor_hash = hash(frozenset(self.ref_flavor_metadata.items()))
flavor_hash = hash(frozenset(self.ref_flavor_metadata))
self.assertEqual((self.ref_flavor_metadata, 'validate_flavor'),
self.driver.driver.driverconfig[flavor_hash])

View File

@ -0,0 +1,69 @@
# Copyright 2017 Walmart Stores Inc.
#
# 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 wsme import exc
from wsme.rest import json as wsme_json
from octavia.api.v2.types import flavor_profile as fp_type
from octavia.common import constants
from octavia.tests.unit.api.common import base
class TestFlavorProfile(object):
_type = None
def test_flavor_profile(self):
body = {"name": "test_name", "provider_name": "test1",
constants.FLAVOR_DATA: '{"hello": "world"}'}
flavor = wsme_json.fromjson(self._type, body)
self.assertEqual(flavor.name, body["name"])
def test_invalid_name(self):
body = {"name": 0}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_name_length(self):
body = {"name": "x" * 256}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_provider_name_length(self):
body = {"name": "x" * 250,
"provider_name": "X" * 256}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson,
self._type, body)
def test_name_mandatory(self):
body = {"provider_name": "test1",
constants.FLAVOR_DATA: '{"hello": "world"}'}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_provider_name_mandatory(self):
body = {"name": "test_name",
constants.FLAVOR_DATA: '{"hello": "world"}'}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_meta_mandatory(self):
body = {"name": "test_name", "provider_name": "test1"}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
class TestFlavorProfilePOST(base.BaseTypesTest, TestFlavorProfile):
_type = fp_type.FlavorProfilePOST

View File

@ -0,0 +1,85 @@
# Copyright 2017 Walmart Stores Inc.
#
# 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_utils import uuidutils
from wsme import exc
from wsme.rest import json as wsme_json
from octavia.api.v2.types import flavors as flavor_type
from octavia.tests.unit.api.common import base
class TestFlavor(object):
_type = None
def test_flavor(self):
body = {"name": "test_name", "description": "test_description",
"flavor_profile_id": uuidutils.generate_uuid()}
flavor = wsme_json.fromjson(self._type, body)
self.assertTrue(flavor.enabled)
def test_invalid_name(self):
body = {"name": 0, "flavor_profile_id": uuidutils.generate_uuid()}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_name_length(self):
body = {"name": "x" * 256,
"flavor_profile_id": uuidutils.generate_uuid()}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_invalid_description(self):
body = {"flavor_profile_id": uuidutils.generate_uuid(),
"description": 0, "name": "test"}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_description_length(self):
body = {"name": "x" * 250,
"flavor_profile_id": uuidutils.generate_uuid(),
"description": "0" * 256}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_invalid_enabled(self):
body = {"name": "test_name",
"flavor_profile_id": uuidutils.generate_uuid(),
"enabled": "notvalid"}
self.assertRaises(ValueError, wsme_json.fromjson, self._type,
body)
def test_name_mandatory(self):
body = {"description": "xyz",
"flavor_profile_id": uuidutils.generate_uuid(),
"enabled": True}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
def test_flavor_profile_id_mandatory(self):
body = {"name": "test_name"}
self.assertRaises(exc.InvalidInput, wsme_json.fromjson, self._type,
body)
class TestFlavorPOST(base.BaseTypesTest, TestFlavor):
_type = flavor_type.FlavorPOST
def test_non_uuid_project_id(self):
body = {"name": "test_name", "description": "test_description",
"flavor_profile_id": uuidutils.generate_uuid()}
lb = wsme_json.fromjson(self._type, body)
self.assertEqual(lb.flavor_profile_id, body['flavor_profile_id'])

View File

@ -54,6 +54,7 @@ console_scripts =
octavia-status = octavia.cmd.status:main
octavia.api.drivers =
noop_driver = octavia.api.drivers.noop_driver.driver:NoopProviderDriver
noop_driver-alt = octavia.api.drivers.noop_driver.driver:NoopProviderDriver
amphora = octavia.api.drivers.amphora_driver.driver:AmphoraProviderDriver
# octavia is an alias for backward compatibility
octavia = octavia.api.drivers.amphora_driver.driver:AmphoraProviderDriver