openstack/tacker
Code Issues Proposed changes

Add versioned objects for vnf packages

Browse Source 
Added new objects required for vnf packages.

Partial-Implements: blueprint tosca-csar-mgmt-driver

Co-Author: Neha Alhat <neha.alhat@nttdata.com>
Change-Id: I743c106b618aec75b4bf0877b812211296d7e816
This commit is contained in:
Niraj Singh 2019-06-03 11:34:54 +00:00 committed by nirajsingh
parent 156d9bfe47
commit 354da78e0c
15 changed files with 1729 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
tacker/common/exceptions.py
View File

@ -199,3 +199,23 @@ class DuplicateEntity(Conflict):
class ValidationError(BadRequest):
message = "%(detail)s"
class ObjectActionError(TackerException):
message = _("Object action %(action)s failed because: %(reason)s")
class VnfPackageNotFound(NotFound):
message = _("No vnf package with id %(id)s.")
class VnfDeploymentFlavourNotFound(NotFound):
message = _("No vnf deployment flavour with id %(id)s.")
class VnfSoftwareImageNotFound(NotFound):
message = _("No vnf software image with id %(id)s.")
class OrphanedObjectError(TackerException):
msg_fmt = _('Cannot call %(method)s on orphaned %(objtype)s object')

4
tacker/objects/__init__.py
View File

@ -25,3 +25,7 @@ def register_all():
# function in order for it to be registered by services that may
# need to receive it via RPC.
__import__('tacker.objects.heal_vnf_request')
__import__('tacker.objects.vnf_package')
__import__('tacker.objects.vnf_package_vnfd')
__import__('tacker.objects.vnf_deployment_flavour')
__import__('tacker.objects.vnf_software_image')

88
tacker/objects/base.py
View File

@ -12,10 +12,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import datetime
from oslo_utils import versionutils
from oslo_versionedobjects import base as ovoo_base
from tacker import objects
from tacker.objects import fields as obj_fields
def get_attrname(name):
@ -46,3 +49,88 @@ class TackerObject(ovoo_base.VersionedObject):
# from one another.
OBJ_SERIAL_NAMESPACE = 'tacker_object'
OBJ_PROJECT_NAMESPACE = 'tacker'
def tacker_obj_get_changes(self):
"""Returns a dict of changed fields with tz unaware datetimes.
Any timezone aware datetime field will be converted to UTC timezone
and returned as timezone unaware datetime.
This will allow us to pass these fields directly to a db update
method as they can't have timezone information.
"""
# Get dirtied/changed fields
changes = self.obj_get_changes()
# Look for datetime objects that contain timezone information
for k, v in changes.items():
if isinstance(v, datetime.datetime) and v.tzinfo:
# Remove timezone information and adjust the time according to
# the timezone information's offset.
changes[k] = v.replace(tzinfo=None) - v.utcoffset()
# Return modified dict
return changes
def obj_reset_changes(self, fields=None, recursive=False):
"""Reset the list of fields that have been changed.
.. note::
- This is NOT "revert to previous values"
- Specifying fields on recursive resets will only be honored at the
top level. Everything below the top will reset all.
:param fields: List of fields to reset, or "all" if None.
:param recursive: Call obj_reset_changes(recursive=True) on
any sub-objects within the list of fields
being reset.
"""
if recursive:
for field in self.obj_get_changes():
# Ignore fields not in requested set (if applicable)
if fields and field not in fields:
continue
# Skip any fields that are unset
if not self.obj_attr_is_set(field):
continue
value = getattr(self, field)
# Don't reset nulled fields
if value is None:
continue
# Reset straight Object and ListOfObjects fields
if isinstance(self.fields[field], obj_fields.ObjectField):
value.obj_reset_changes(recursive=True)
elif isinstance(self.fields[field],
obj_fields.ListOfObjectsField):
for thing in value:
thing.obj_reset_changes(recursive=True)
if fields:
self._changed_fields -= set(fields)
else:
self._changed_fields.clear()
class TackerPersistentObject(object):
"""Mixin class for Persistent objects.
This adds the fields that we use in common for most persistent objects.
"""
fields = {
'created_at': obj_fields.DateTimeField(nullable=False),
'updated_at': obj_fields.DateTimeField(nullable=True),
'deleted_at': obj_fields.DateTimeField(nullable=True),
'deleted': obj_fields.BooleanField(default=False)
}
remotable = ovoo_base.remotable
remotable_classmethod = ovoo_base.remotable_classmethod
obj_make_list = ovoo_base.obj_make_list
TackerObjectDictCompat = ovoo_base.VersionedObjectDictCompat

99
tacker/objects/fields.py
View File

@ -13,6 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import uuid
from oslo_versionedobjects import fields
@ -20,3 +22,100 @@ from oslo_versionedobjects import fields
StringField = fields.StringField
ListOfObjectsField = fields.ListOfObjectsField
ListOfStringsField = fields.ListOfStringsField
DictOfStringsField = fields.DictOfStringsField
DateTimeField = fields.DateTimeField
BooleanField = fields.BooleanField
BaseEnumField = fields.BaseEnumField
Enum = fields.Enum
ObjectField = fields.ObjectField
IntegerField = fields.IntegerField
FieldType = fields.FieldType
class BaseTackerEnum(Enum):
def __init__(self):
super(BaseTackerEnum, self).__init__(valid_values=self.__class__.ALL)
class ContainerFormat(BaseTackerEnum):
AKI = 'AKI'
AMI = 'AMI'
ARI = 'ARI'
BARE = 'BARE'
DOCKER = 'DOCKER'
OVA = 'OVA'
OVF = 'OVF'
ALL = (AKI, AMI, ARI, BARE, DOCKER, OVA, OVF)
class ContainerFormatFields(BaseEnumField):
AUTO_TYPE = ContainerFormat()
class DiskFormat(BaseTackerEnum):
AKI = 'AKI'
AMI = 'AMI'
ARI = 'ARI'
ISO = 'ISO'
QCOW2 = 'QCOW2'
RAW = 'RAW'
VDI = 'VDI'
VHD = 'VHD'
VHDX = 'VHDX'
VMDK = 'VMDK'
ALL = (AKI, AMI, ARI, ISO, QCOW2, RAW, VDI, VHD, VHDX, VMDK)
class DiskFormatFields(BaseEnumField):
AUTO_TYPE = DiskFormat()
class PackageOnboardingStateType(BaseTackerEnum):
CREATED = 'CREATED'
UPLOADING = 'UPLOADING'
PROCESSING = 'PROCESSING'
ONBOARDED = 'ONBOARDED'
ALL = (CREATED, UPLOADING, PROCESSING, ONBOARDED)
class PackageOnboardingStateTypeField(BaseEnumField):
AUTO_TYPE = PackageOnboardingStateType()
class PackageOperationalStateType(BaseTackerEnum):
ENABLED = 'ENABLED'
DISABLED = 'DISABLED'
ALL = (ENABLED, DISABLED)
class PackageOperationalStateTypeField(BaseEnumField):
AUTO_TYPE = PackageOperationalStateType()
class PackageUsageStateType(BaseTackerEnum):
IN_USE = 'IN_USE'
NOT_IN_USE = 'NOT_IN_USE'
ALL = (IN_USE, NOT_IN_USE)
class PackageUsageStateTypeField(BaseEnumField):
AUTO_TYPE = PackageUsageStateType()
class DictOfNullableField(fields.AutoTypedField):
AUTO_TYPE = fields.Dict(fields.FieldType(), nullable=True)
class UUID(fields.UUID):
def coerce(self, obj, attr, value):
uuid.UUID(str(value))
return str(value)
class UUIDField(fields.AutoTypedField):
AUTO_TYPE = UUID()

250
tacker/objects/vnf_deployment_flavour.py Normal file
View File

@ -0,0 +1,250 @@
# Copyright 2019 NTT DATA.
#
# 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 timeutils
from oslo_utils import uuidutils
from oslo_versionedobjects import base as ovoo_base
from sqlalchemy.orm import joinedload
from tacker.common import exceptions
from tacker.db import api as db_api
from tacker.db.db_sqlalchemy import api
from tacker.db.db_sqlalchemy import models
from tacker import objects
from tacker.objects import base
from tacker.objects import fields
_NO_DATA_SENTINEL = object()
VNF_DEPLOYMENT_FLAVOUR_OPTIONAL_ATTRS = ['software_images']
LOG = logging.getLogger(__name__)
@db_api.context_manager.writer
def _vnf_deployment_flavour_create(context, values):
vnf_deployment_flavour = models.VnfDeploymentFlavour()
vnf_deployment_flavour.update(values)
vnf_deployment_flavour.save(context.session)
return vnf_deployment_flavour
@db_api.context_manager.reader
def _vnf_deployment_flavour_get_by_id(context, id, columns_to_join=None):
query = api.model_query(context, models.VnfDeploymentFlavour,
read_deleted="no").filter_by(id=id)
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
result = query.first()
if not result:
raise exceptions.VnfDeploymentFlavourNotFound(id=id)
return result
@db_api.context_manager.writer
def _destroy_vnf_deployment_flavour(context, flavour_uuid):
now = timeutils.utcnow()
updated_values = {'deleted': True,
'deleted_at': now
}
software_images_query = api.model_query(
context, models.VnfSoftwareImage,
(models.VnfSoftwareImage.id,)).filter_by(flavour_uuid=flavour_uuid)
api.model_query(context, models.VnfSoftwareImageMetadata). \
filter(models.VnfSoftwareImageMetadata.image_uuid.
in_(software_images_query.subquery())).update(
updated_values, synchronize_session=False)
api.model_query(context, models.VnfSoftwareImage). \
filter_by(flavour_uuid=flavour_uuid). \
update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfDeploymentFlavour). \
filter_by(id=flavour_uuid). \
update(updated_values, synchronize_session=False)
@base.TackerObjectRegistry.register
class VnfDeploymentFlavour(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'package_uuid': fields.UUIDField(nullable=False),
'flavour_id': fields.StringField(nullable=False),
'flavour_description': fields.StringField(nullable=False),
'instantiation_levels': fields.DictOfNullableField(nullable=True),
'software_images': fields.ObjectField('VnfSoftwareImagesList'),
}
@staticmethod
def _from_db_object(context, flavour, db_flavour, expected_attrs=None):
flavour._context = context
special_cases = set(['instantiation_levels'])
fields = set(flavour.fields) - special_cases
for key in fields:
if key in VNF_DEPLOYMENT_FLAVOUR_OPTIONAL_ATTRS:
continue
if db_flavour[key]:
setattr(flavour, key, db_flavour[key])
inst_levels = db_flavour['instantiation_levels']
if inst_levels:
flavour.instantiation_levels = jsonutils.loads(inst_levels)
flavour._extra_attributes_from_db_object(flavour, db_flavour,
expected_attrs)
flavour.obj_reset_changes()
return flavour
@staticmethod
def _extra_attributes_from_db_object(flavour, db_flavour,
expected_attrs=None):
"""Method to help with migration of extra attributes to objects.
"""
if expected_attrs is None:
expected_attrs = []
if 'software_images' in expected_attrs:
flavour._load_sw_images(db_flavour.get('software_images'))
@base.remotable
def create(self):
if self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='create',
reason='already created')
updates = self.obj_get_changes()
if 'id' not in updates:
updates['id'] = uuidutils.generate_uuid()
self.id = updates['id']
if 'software_images' in updates.keys():
updates.pop('software_images')
special_key = 'instantiation_levels'
if special_key in updates.keys():
updates[special_key] = jsonutils.dumps(updates.get(special_key))
db_flavour = _vnf_deployment_flavour_create(self._context, updates)
self._from_db_object(self._context, self, db_flavour)
@base.remotable_classmethod
def get_by_id(cls, context, id, expected_attrs=None):
db_flavour = _vnf_deployment_flavour_get_by_id(
context, id, columns_to_join=expected_attrs)
return cls._from_db_object(context, cls(), db_flavour,
expected_attrs=expected_attrs)
@base.remotable
def destroy(self, context):
if not self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(
action='destroy', reason='no uuid')
_destroy_vnf_deployment_flavour(context, self.id)
def obj_load_attr(self, attrname):
if not self._context:
raise exceptions.OrphanedObjectError(
method='obj_load_attr', objtype=self.obj_name())
if 'id' not in self:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
LOG.debug("Lazy-loading '%(attr)s' on %(name)s id %(id)s",
{'attr': attrname,
'name': self.obj_name(),
'id': self.id,
})
self._obj_load_attr(attrname)
def _obj_load_attr(self, attrname):
"""Internal method for loading attributes from vnf deployment flavour.
"""
if attrname == 'software_images':
self._load_sw_images()
elif attrname in self.fields and attrname != 'id':
self._load_generic(attrname)
else:
# NOTE(nirajsingh): Raise error if non existing field is
# requested.
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
self.obj_reset_changes([attrname])
def _load_generic(self, attrname):
vnf_deployment_flavour = self.__class__.get_by_id(
self._context, id=self.id, expected_attrs=None)
if attrname not in vnf_deployment_flavour:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('loading %s requires recursion') % attrname)
for field in self.fields:
if field in vnf_deployment_flavour and field not in self:
setattr(self, field, getattr(vnf_deployment_flavour, field))
def _load_sw_images(self, db_sw_images=_NO_DATA_SENTINEL):
if db_sw_images is _NO_DATA_SENTINEL:
vnf_deployment_flavour = self.get_by_id(
self._context, self.id, expected_attrs=['software_images'])
if 'software_images' in vnf_deployment_flavour:
self.software_images = vnf_deployment_flavour.software_images
self.software_images.obj_reset_changes(recursive=True)
self.obj_reset_changes(['software_images'])
else:
self.software_images = (
objects.VnfSoftwareImagesList(objects=[]))
elif db_sw_images:
self.software_images = base.obj_make_list(
self._context, objects.VnfSoftwareImagesList(
self._context),
objects.VnfSoftwareImage, db_sw_images)
self.obj_reset_changes(['software_images'])
@base.TackerObjectRegistry.register
class VnfDeploymentFlavoursList(ovoo_base.ObjectListBase, base.TackerObject):
VERSION = '1.0'
fields = {
'objects': fields.ListOfObjectsField('VnfDeploymentFlavour')
}

390
tacker/objects/vnf_package.py Normal file
View File

@ -0,0 +1,390 @@
# Copyright 2019 NTT DATA.
#
# 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_utils import timeutils
from oslo_utils import uuidutils
from oslo_versionedobjects import base as ovoo_base
from sqlalchemy.orm import joinedload
from tacker._i18n import _
from tacker.common import exceptions
from tacker.db import api as db_api
from tacker.db.db_sqlalchemy import api
from tacker.db.db_sqlalchemy import models
from tacker import objects
from tacker.objects import base
from tacker.objects import fields
_NO_DATA_SENTINEL = object()
VNF_PACKAGE_OPTIONAL_ATTRS = ['vnf_deployment_flavours', 'vnfd']
LOG = logging.getLogger(__name__)
def _add_user_defined_data(context, package_uuid, user_data,
max_retries=10):
for attempt in range(max_retries):
with db_api.context_manager.writer.using(context):
new_entries = []
for key, value in user_data.items():
new_entries.append({"key": key,
"value": value,
"package_uuid": package_uuid})
if new_entries:
context.session.execute(
models.VnfPackageUserData.__table__.insert(None),
new_entries)
@db_api.context_manager.reader
def _vnf_package_get_by_id(context, package_uuid, columns_to_join=None):
query = api.model_query(context, models.VnfPackage,
read_deleted="no", project_only=True). \
filter_by(id=package_uuid).options(joinedload('_metadata'))
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
result = query.first()
if not result:
raise exceptions.VnfPackageNotFound(id=package_uuid)
return result
@db_api.context_manager.writer
def _vnf_package_create(context, values, user_data=None):
vnf_package = models.VnfPackage()
vnf_package.update(values)
vnf_package.save(context.session)
vnf_package._metadata = []
if user_data:
_add_user_defined_data(context, vnf_package.id, user_data)
context.session.expire(vnf_package, ['_metadata'])
vnf_package._metadata
return vnf_package
@db_api.context_manager.reader
def _vnf_package_list(context, columns_to_join=None):
query = api.model_query(context, models.VnfPackage, read_deleted="no",
project_only=True).options(joinedload('_metadata'))
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
return query.all()
@db_api.context_manager.reader
def _vnf_package_list_by_filters(context, read_deleted=None, **filters):
query = api.model_query(context, models.VnfPackage,
read_deleted=read_deleted, project_only=True)
for key, value in filters.items():
filter_obj = getattr(models.VnfPackage, key)
if key == 'deleted_at':
query = query.filter(filter_obj >= value)
else:
query = query.filter(filter_obj == value)
return query.all()
@db_api.context_manager.writer
def _vnf_package_update(context, package_uuid, values, columns_to_join=None):
vnf_package = _vnf_package_get_by_id(context, package_uuid,
columns_to_join=columns_to_join)
vnf_package.update(values)
vnf_package.save(session=context.session)
return vnf_package
@db_api.context_manager.writer
def _destroy_vnf_package(context, package_uuid):
now = timeutils.utcnow()
updated_values = {'deleted': True,
'deleted_at': now
}
flavour_query = api.model_query(
context, models.VnfDeploymentFlavour,
(models.VnfDeploymentFlavour.id, )).filter_by(
package_uuid=package_uuid)
software_images_query = api.model_query(
context, models.VnfSoftwareImage,
(models.VnfSoftwareImage.id, )).filter(
models.VnfSoftwareImage.flavour_uuid.in_(flavour_query.subquery()))
api.model_query(
context, models.VnfSoftwareImageMetadata).filter(
models.VnfSoftwareImageMetadata.image_uuid.in_(
software_images_query.subquery())).update(
updated_values, synchronize_session=False)
software_images_query.update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfPackageUserData). \
filter_by(package_uuid=package_uuid). \
update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfDeploymentFlavour). \
filter_by(package_uuid=package_uuid). \
update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfPackageVnfd). \
filter_by(package_uuid=package_uuid). \
update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfPackage).\
filter_by(id=package_uuid). \
update(updated_values, synchronize_session=False)
def _make_vnf_packages_list(context, vnf_package_list, db_vnf_package_list,
expected_attrs):
vnf_package_cls = VnfPackage
vnf_package_list.objects = []
for db_package in db_vnf_package_list:
vnf_pkg_obj = vnf_package_cls._from_db_object(
context, vnf_package_cls(context), db_package,
expected_attrs=expected_attrs)
vnf_package_list.objects.append(vnf_pkg_obj)
vnf_package_list.obj_reset_changes()
return vnf_package_list
@base.TackerObjectRegistry.register
class VnfPackage(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'onboarding_state': fields.StringField(nullable=False),
'operational_state': fields.StringField(nullable=False),
'usage_state': fields.StringField(nullable=False),
'user_data': fields.DictOfStringsField(),
'tenant_id': fields.StringField(nullable=False),
'algorithm': fields.StringField(nullable=True),
'hash': fields.StringField(nullable=True),
'location_glance_store': fields.StringField(nullable=True),
'vnf_deployment_flavours': fields.ObjectField(
'VnfDeploymentFlavoursList', nullable=True),
'vnfd': fields.ObjectField('VnfPackageVnfd', nullable=True),
}
@staticmethod
def _from_db_object(context, vnf_package, db_vnf_package,
expected_attrs=None):
if expected_attrs is None:
expected_attrs = []
vnf_package._context = context
for key in vnf_package.fields:
if key in VNF_PACKAGE_OPTIONAL_ATTRS:
continue
if key == 'user_data':
db_key = 'metadetails'
else:
db_key = key
setattr(vnf_package, key, db_vnf_package[db_key])
vnf_package._context = context
vnf_package._extra_attributes_from_db_object(
vnf_package, db_vnf_package, expected_attrs)
vnf_package.obj_reset_changes()
return vnf_package
@staticmethod
def _extra_attributes_from_db_object(vnf_package, db_vnf_package,
expected_attrs=None):
"""Method to help with migration of extra attributes to objects."""
if expected_attrs is None:
expected_attrs = []
if 'vnf_deployment_flavours' in expected_attrs:
vnf_package._load_vnf_deployment_flavours(
db_vnf_package.get('vnf_deployment_flavours'))
if 'vnfd' in expected_attrs:
vnf_package._load_vnfd(db_vnf_package.get('vnfd'))
def _load_vnf_deployment_flavours(self, db_flavours=_NO_DATA_SENTINEL):
if db_flavours is _NO_DATA_SENTINEL:
vnf_package = self.get_by_id(
self._context, self.id,
expected_attrs=['vnf_deployment_flavours'])
if 'vnf_deployment_flavours' in vnf_package:
self.vnf_deployment_flavours = \
vnf_package.vnf_deployment_flavours
self.vnf_deployment_flavours.obj_reset_changes(recursive=True)
self.obj_reset_changes(['vnf_deployment_flavours'])
else:
self.vnf_deployment_flavours = \
objects.VnfDeploymentFlavoursList(objects=[])
elif db_flavours:
self.vnf_deployment_flavours = base.obj_make_list(
self._context, objects.VnfDeploymentFlavoursList(
self._context), objects.VnfDeploymentFlavour, db_flavours)
self.obj_reset_changes(['vnf_deployment_flavours'])
def _load_vnfd(self, db_vnfd=_NO_DATA_SENTINEL):
if db_vnfd is None:
self.vnfd = None
elif db_vnfd is _NO_DATA_SENTINEL:
vnf_package = self.get_by_id(self._context, self.id,
expected_attrs=['vnfd'])
if 'vnfd' in vnf_package and vnf_package.vnfd is not None:
self.vnfd = vnf_package.vnfd
self.vnfd.obj_reset_changes(recursive=True)
self.obj_reset_changes(['vnfd'])
else:
self.vnfd = None
elif db_vnfd:
self.vnfd = objects.VnfPackageVnfd.obj_from_db_obj(
self._context, db_vnfd)
self.obj_reset_changes(['vnfd'])
def _load_generic(self, attrname):
vnf_package = self.__class__.get_by_id(self._context,
id=self.id,
expected_attrs=None)
if attrname not in vnf_package:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('loading %s requires recursion') % attrname)
for field in self.fields:
if field in vnf_package and field not in self:
setattr(self, field, getattr(vnf_package, field))
def obj_load_attr(self, attrname):
if not self._context:
raise exceptions.OrphanedObjectError(
method='obj_load_attr', objtype=self.obj_name())
if 'id' not in self:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
LOG.debug("Lazy-loading '%(attr)s' on %(name)s id %(id)s",
{'attr': attrname,
'name': self.obj_name(),
'id': self.id,
})
self._obj_load_attr(attrname)
def _obj_load_attr(self, attrname):
"""Internal method for loading attributes from vnf package."""
if attrname == 'vnf_deployment_flavours':
self._load_vnf_deployment_flavours()
elif attrname == 'vnfd':
self._load_vnfd()
elif attrname in self.fields and attrname != 'id':
self._load_generic(attrname)
else:
# NOTE(nirajsingh): Raise error if non existing field is
# requested.
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
self.obj_reset_changes([attrname])
@base.remotable
def create(self):
if self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='create',
reason=_('already created'))
updates = self.obj_get_changes()
if 'id' not in updates:
updates['id'] = uuidutils.generate_uuid()
self.id = updates['id']
for key in ['vnf_deployment_flavours']:
if key in updates.keys():
updates.pop(key)
user_data = updates.pop('user_data', None)
db_vnf_package = _vnf_package_create(self._context, updates,
user_data=user_data)
self._from_db_object(self._context, self, db_vnf_package)
@base.remotable_classmethod
def get_by_id(cls, context, id, expected_attrs=None):
db_vnf_package = _vnf_package_get_by_id(
context, id, columns_to_join=expected_attrs)
return cls._from_db_object(context, cls(), db_vnf_package,
expected_attrs=expected_attrs)
@base.remotable
def destroy(self, context):
if not self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='destroy',
reason='no uuid')
_destroy_vnf_package(context, self.id)
@base.remotable
def save(self):
updates = self.tacker_obj_get_changes()
for key in ['vnf_deployment_flavours']:
if key in updates.keys():
updates.pop(key)
db_vnf_package = _vnf_package_update(self._context,
self.id, updates)
self._from_db_object(self._context, self, db_vnf_package)
@base.TackerObjectRegistry.register
class VnfPackagesList(ovoo_base.ObjectListBase, base.TackerObject):
VERSION = '1.0'
fields = {
'objects': fields.ListOfObjectsField('VnfPackage')
}
@base.remotable_classmethod
def get_all(cls, context, expected_attrs=None):
db_vnf_packages = _vnf_package_list(context,
columns_to_join=expected_attrs)
return _make_vnf_packages_list(context, cls(), db_vnf_packages,
expected_attrs)
@base.remotable_classmethod
def get_by_filters(self, context, read_deleted=None, **filters):
return _vnf_package_list_by_filters(context,
read_deleted=read_deleted,
**filters)

81
tacker/objects/vnf_package_vnfd.py Normal file
View File

@ -0,0 +1,81 @@
# Copyright 2019 NTT DATA.
#
# 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 tacker.common import exceptions
from tacker.db import api as db_api
from tacker.db.db_sqlalchemy import models
from tacker.objects import base
from tacker.objects import fields
@db_api.context_manager.writer
def _vnf_package_vnfd_create(context, values):
vnf_package_vnfd = models.VnfPackageVnfd()
vnf_package_vnfd.update(values)
vnf_package_vnfd.save(context.session)
return vnf_package_vnfd
@base.TackerObjectRegistry.register
class VnfPackageVnfd(base.TackerObject, base.TackerObjectDictCompat,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'package_uuid': fields.UUIDField(nullable=False),
'vnfd_id': fields.UUIDField(nullable=False),
'vnf_provider': fields.StringField(nullable=False),
'vnf_product_name': fields.StringField(nullable=False),
'vnf_software_version': fields.StringField(nullable=False),
'vnfd_version': fields.StringField(nullable=False),
}
@staticmethod
def _from_db_object(context, vnf_package_vnfd, db_vnf_package_vnfd):
for key in vnf_package_vnfd.fields:
if db_vnf_package_vnfd[key]:
setattr(vnf_package_vnfd, key, db_vnf_package_vnfd[key])
vnf_package_vnfd._context = context
vnf_package_vnfd.obj_reset_changes()
return vnf_package_vnfd
@base.remotable
def create(self):
if self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='create',
reason=_('already created'))
updates = self.obj_get_changes()
if 'id' not in updates:
updates['id'] = uuidutils.generate_uuid()
self.id = updates['id']
updates = self.obj_get_changes()
db_vnf_package_vnfd = _vnf_package_vnfd_create(
self._context, updates)
self._from_db_object(self._context, self, db_vnf_package_vnfd)
@classmethod
def obj_from_db_obj(cls, context, db_obj):
return cls._from_db_object(context, cls(), db_obj)

210
tacker/objects/vnf_software_image.py Normal file
View File

@ -0,0 +1,210 @@
# Copyright 2019 NTT DATA.
#
# 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_utils import uuidutils
from oslo_versionedobjects import base as ovoo_base
from sqlalchemy.orm import joinedload
from tacker.common import exceptions
from tacker.db import api as db_api
from tacker.db.db_sqlalchemy import api
from tacker.db.db_sqlalchemy import models
from tacker.objects import base
from tacker.objects import fields
VNF_SOFTWARE_IMAGE_OPTIONAL_ATTRS = ['metadata']
LOG = logging.getLogger(__name__)
def _metadata_add_to_db(context, id, metadata, max_retries=10):
for attempt in range(max_retries):
with db_api.context_manager.writer.using(context):
new_entries = []
for key, value in metadata.items():
new_entries.append({"key": key,
"value": value,
"image_uuid": id})
if new_entries:
context.session.execute(
models.VnfSoftwareImageMetadata.__table__.insert(None),
new_entries)
return metadata
@db_api.context_manager.writer
def _vnf_sw_image_create(context, values, metadata=None):
vnf_sw_image = models.VnfSoftwareImage()
vnf_sw_image.update(values)
vnf_sw_image.save(context.session)
vnf_sw_image._metadata = []
if metadata:
_metadata_add_to_db(context, vnf_sw_image.id, metadata)
context.session.expire(vnf_sw_image, ['_metadata'])
vnf_sw_image._metadata
return vnf_sw_image
@db_api.context_manager.reader
def _vnf_sw_image_get_by_id(context, id):
query = api.model_query(context, models.VnfSoftwareImage,
read_deleted="no").filter_by(id=id).options(joinedload('_metadata'))
result = query.first()
if not result:
raise exceptions.VnfSoftwareImageNotFound(id=id)
return result
@base.TackerObjectRegistry.register
class VnfSoftwareImage(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'software_image_id': fields.StringField(nullable=False),
'flavour_uuid': fields.UUIDField(nullable=False),
'name': fields.StringField(nullable=True),
'provider': fields.StringField(nullable=True),
'version': fields.StringField(nullable=True),
'algorithm': fields.StringField(nullable=True),
'hash': fields.StringField(nullable=True),
'container_format': fields.StringField(nullable=True),
'disk_format': fields.StringField(nullable=True),
'min_disk': fields.IntegerField(),
'min_ram': fields.IntegerField(default=0),
'size': fields.IntegerField(),
'image_path': fields.StringField(),
'metadata': fields.DictOfStringsField(nullable=True)
}
@staticmethod
def _from_db_object(context, vnf_sw_image, db_sw_image,
expected_attrs=None):
vnf_sw_image._context = context
for key in vnf_sw_image.fields:
if key in VNF_SOFTWARE_IMAGE_OPTIONAL_ATTRS:
continue
else:
db_key = key
setattr(vnf_sw_image, key, db_sw_image[db_key])
vnf_sw_image._extra_attributes_from_db_object(vnf_sw_image,
db_sw_image, expected_attrs)
vnf_sw_image.obj_reset_changes()
return vnf_sw_image
@staticmethod
def _extra_attributes_from_db_object(vnf_sw_image, db_sw_image,
expected_attrs=None):
"""Method to help with migration of extra attributes to objects.
"""
if expected_attrs is None:
expected_attrs = []
if 'metadata' in expected_attrs:
setattr(vnf_sw_image, 'metadata', db_sw_image['metadetails'])
def obj_load_attr(self, attrname):
if not self._context:
raise exceptions.OrphanedObjectError(method='obj_load_attr',
objtype=self.obj_name())
if 'id' not in self:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
LOG.debug("Lazy-loading '%(attr)s' on %(name)s id %(id)s",
{'attr': attrname,
'name': self.obj_name(),
'id': self.id,
})
self._obj_load_attr(attrname)
def _obj_load_attr(self, attrname):
"""Internal method for loading attributes from vnf flavour."""
if attrname in self.fields and attrname != 'id':
self._load_generic(attrname)
else:
# NOTE(nirajsingh): Raise error if non existing field is
# requested.
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('attribute %s not lazy-loadable') % attrname)
self.obj_reset_changes([attrname])
def _load_generic(self, attrname):
software_image = self.__class__.get_by_id(self._context,
id=self.id,
expected_attrs=attrname)
if attrname not in software_image:
raise exceptions.ObjectActionError(
action='obj_load_attr',
reason=_('loading %s requires recursion') % attrname)
for field in self.fields:
if field in software_image and field not in self:
setattr(self, field, getattr(software_image, field))
@base.remotable
def create(self):
if self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='create',
reason=_('already created'))
updates = self.obj_get_changes()
if 'id' not in updates:
updates['id'] = uuidutils.generate_uuid()
self.id = updates['id']
metadata = updates.pop('metadata', None)
db_sw_image = _vnf_sw_image_create(self._context, updates,
metadata=metadata)
self._from_db_object(self._context, self, db_sw_image)
@base.remotable_classmethod
def get_by_id(cls, context, id, expected_attrs=None):
db_sw_image = _vnf_sw_image_get_by_id(context, id)
return cls._from_db_object(context, cls(), db_sw_image,
expected_attrs=expected_attrs)
@base.TackerObjectRegistry.register
class VnfSoftwareImagesList(ovoo_base.ObjectListBase, base.TackerObject):
VERSION = '1.0'
fields = {
'objects': fields.ListOfObjectsField('VnfSoftwareImage')
}

4
tacker/tests/unit/__init__.py
View File

@ -17,8 +17,12 @@ import os
from oslo_config import cfg
from tacker import objects
reldir = os.path.join(os.path.dirname(__file__), '..', '..', '..')
absdir = os.path.abspath(reldir)
cfg.CONF.state_path = absdir
cfg.CONF.use_stderr = False
objects.register_all()

15
tacker/tests/unit/base.py
View File

@ -34,6 +34,21 @@ class TestCase(base.BaseTestCase):
patcher = mock.patch(target, new)
return patcher.start()
def compare_obj(self, expected, result, subs=None, allow_missing=None):
if subs is None:
subs = {}
if allow_missing is None:
allow_missing = []
for key in expected.fields:
if key in allow_missing:
continue
obj_val = getattr(expected, key)
db_key = subs.get(key, key)
db_val = getattr(result, db_key)
self.assertEqual(db_val, obj_val)
class FixturedTestCase(TestCase):
client_fixture_class = None

79
tacker/tests/unit/objects/fakes.py Normal file
View File

@ -0,0 +1,79 @@
# Copyright (C) 2019 NTT DATA
# 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.
import copy
import datetime
import iso8601
from tacker.tests import uuidsentinel
vnf_package_data = {'algorithm': None, 'hash': None,
'location_glance_store': None,
'onboarding_state': 'CREATED',
'operational_state': 'DISABLED',
'tenant_id': uuidsentinel.tenant_id,
'usage_state': 'NOT_IN_USE',
'user_data': {'abc': 'xyz'},
'created_at': datetime.datetime(
2019, 8, 8, 0, 0, 0, tzinfo=iso8601.UTC),
}
software_image = {
'software_image_id': uuidsentinel.software_image_id,
'name': 'test', 'provider': 'test', 'version': 'test',
'algorithm': 'sha-256',
'hash': 'b9c3036539fd7a5f87a1bf38eb05fdde8b556a1'
'a7e664dbeda90ed3cd74b4f9d',
'container_format': 'test', 'disk_format': 'qcow2', 'min_disk': 1,
'min_ram': 2, 'size': 1, 'image_path': 'test',
'metadata': {'key1': 'value1'}
}
artifacts = {
'json_data': 'test data',
'type': 'tosca.artifacts.nfv.SwImage',
'algorithm': 'sha512', 'hash': uuidsentinel.hash}
fake_vnf_package_response = copy.deepcopy(vnf_package_data)
fake_vnf_package_response.pop('user_data')
fake_vnf_package_response.update({'id': uuidsentinel.package_uuid})
vnf_deployment_flavour = {'flavour_id': 'simple',
'flavour_description': 'simple flavour description',
'instantiation_levels': {
'levels': {
'instantiation_level_1': {
'description': 'Smallest size',
'scale_info': {
'worker_instance': {
'scale_level': 0
}
}
},
'instantiation_level_2': {
'description': 'Largest size',
'scale_info': {
'worker_instance': {
'scale_level': 2
}
}
}
},
'default_level': 'instantiation_level_1'
},
'created_at': datetime.datetime(
2019, 8, 8, 0, 0, 0, tzinfo=iso8601.UTC),
}

145
tacker/tests/unit/objects/test_vnf_deployment_flavour.py Normal file
View File

@ -0,0 +1,145 @@
# Copyright (c) 2019 NTT DATA
#
# 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 tacker.common import exceptions
from tacker import context
from tacker import objects
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests.unit.objects import fakes
from tacker.tests import uuidsentinel
class TestVnfDeploymentFlavour(SqlTestCase):
def setUp(self):
super(TestVnfDeploymentFlavour, self).setUp()
self.context = context.get_admin_context()
self.vnf_package = self._create_vnf_package()
self.vnf_deployment_flavour = self._create_vnf_deployment_flavour()
def _create_vnf_package(self):
vnfpkgm = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnfpkgm.create()
return vnfpkgm
def _create_vnf_deployment_flavour(self):
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour = objects.VnfDeploymentFlavour(
context=self.context, **flavour_data)
vnf_deployment_flavour.create()
return vnf_deployment_flavour
def test_create(self):
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **flavour_data)
vnf_deployment_flavour_obj.create()
self.assertTrue(vnf_deployment_flavour_obj.id)
def test_create_with_software_images(self):
software_images = objects.VnfSoftwareImage(**fakes.software_image)
fake_software_images = objects.VnfSoftwareImagesList(
objects=[software_images])
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'software_images': fake_software_images})
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **flavour_data)
vnf_deployment_flavour_obj.create()
self.assertTrue(vnf_deployment_flavour_obj.id)
def test_get_by_id(self):
vnf_deployment_flavour = objects.VnfDeploymentFlavour.get_by_id(
self.context, self.vnf_deployment_flavour.id, expected_attrs=None)
self.compare_obj(self.vnf_deployment_flavour, vnf_deployment_flavour,
allow_missing=['software_images',
'updated_at',
'deleted', 'deleted_at'])
def test_get_by_id_with_no_existing_id(self):
self.assertRaises(
exceptions.VnfDeploymentFlavourNotFound,
objects.VnfDeploymentFlavour.get_by_id, self.context,
uuidsentinel.invalid_uuid)
def test_create_with_id(self):
vnf_deployment_flavour_obj = {'id': uuidsentinel.uuid}
vnf_deployment_flavour = objects.VnfDeploymentFlavour(
context=self.context, **vnf_deployment_flavour_obj)
self.assertRaises(exceptions.ObjectActionError,
vnf_deployment_flavour.create)
@mock.patch('tacker.objects.vnf_deployment_flavour.'
'_destroy_vnf_deployment_flavour')
def test_destroy(self, mock_vnf_deployment_flavour_destroy):
self.vnf_deployment_flavour.destroy(self.context)
mock_vnf_deployment_flavour_destroy.assert_called_with(
self.context, self.vnf_deployment_flavour.id)
def test_destroy_without_id(self):
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context)
self.assertRaises(exceptions.ObjectActionError,
vnf_deployment_flavour_obj.destroy, self.context)
def test_attribute_with_valid_data(self):
data = {'id': self.vnf_deployment_flavour.id}
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **data)
vnf_deployment_flavour_obj.obj_load_attr('flavour_id')
self.assertEqual('simple', vnf_deployment_flavour_obj.flavour_id)
def test_invalid_attribute(self):
self.assertRaises(
exceptions.ObjectActionError,
self.vnf_deployment_flavour.obj_load_attr, 'invalid')
def test_obj_load_attr_without_context(self):
data = {'id': self.vnf_deployment_flavour.id}
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(**data)
self.assertRaises(exceptions.OrphanedObjectError,
vnf_deployment_flavour_obj.obj_load_attr,
'flavour_id')
def test_obj_load_attr_without_id_in_object(self):
data = {'flavour_id': self.vnf_deployment_flavour.flavour_id}
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **data)
self.assertRaises(exceptions.ObjectActionError,
vnf_deployment_flavour_obj.obj_load_attr,
'flavour_id')
def test_get_by_id_with_software_images(self):
software_image = fakes.software_image
software_image.update(
{'flavour_uuid': self.vnf_deployment_flavour.id})
vnf_soft_image_obj = objects.VnfSoftwareImage(
context=self.context, **software_image)
vnf_soft_image_obj.create()
vnf_deployment_flavour = objects.VnfDeploymentFlavour.get_by_id(
self.context, self.vnf_deployment_flavour.id,
expected_attrs=['software_images'])
self.assertEqual(1,
len(vnf_deployment_flavour.software_images.objects))
self.compare_obj(vnf_soft_image_obj,
vnf_deployment_flavour.software_images[0])

177
tacker/tests/unit/objects/test_vnf_package.py Normal file
View File

@ -0,0 +1,177 @@
# Copyright (c) 2019 NTT DATA
#
# 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 tacker.common import exceptions
from tacker import context
from tacker.db.db_sqlalchemy import models
from tacker import objects
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests.unit.objects import fakes
from tacker.tests import uuidsentinel
class TestVnfPackage(SqlTestCase):
def setUp(self):
super(TestVnfPackage, self).setUp()
self.context = context.get_admin_context()
self.vnf_package = self._create_vnf_package()
def _create_vnf_package(self):
vnfpkgm = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnfpkgm.create()
return vnfpkgm
def test_create(self):
vnfpkgm = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnfpkgm.create()
self.assertTrue(vnfpkgm.id)
self.assertEqual('CREATED', vnfpkgm.onboarding_state)
self.assertEqual('NOT_IN_USE', vnfpkgm.usage_state)
self.assertEqual('DISABLED', vnfpkgm.operational_state)
def test_create_without_user_define_data(self):
vnf_pack = fakes.vnf_package_data
vnf_pack.update({'user_data': {}})
vnfpkgm = objects.VnfPackage(context=self.context, **vnf_pack)
vnfpkgm.create()
self.assertTrue(vnfpkgm.id)
self.assertEqual('CREATED', vnfpkgm.onboarding_state)
self.assertEqual('NOT_IN_USE', vnfpkgm.usage_state)
self.assertEqual('DISABLED', vnfpkgm.operational_state)
@mock.patch.object(uuidutils, 'generate_uuid')
@mock.patch('tacker.objects.vnf_package._vnf_package_create')
def test_create_ignore_flavours(
self, mock_vnf_package_create, mock_uuid):
fake_data = fakes.fake_vnf_package_response
mock_uuid.return_value = fake_data['id']
mock_vnf_package_create.return_value = models.VnfPackage(**fake_data)
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour = objects.VnfDeploymentFlavour(**flavour_data)
fake_vnf_deployment_flavours = objects.VnfDeploymentFlavoursList(
objects=[vnf_deployment_flavour])
vnf_pack = fakes.vnf_package_data
vnf_pack.update(
{'vnf_deployment_flavours': fake_vnf_deployment_flavours})
vnfpkgm = objects.VnfPackage(context=self.context, **vnf_pack)
vnfpkgm.create()
mock_vnf_package_create.assert_called_once_with(
self.context, fake_data, user_data={'abc': 'xyz'})
def test_get_by_id(self):
vnfpkgm = objects.VnfPackage.get_by_id(self.context,
self.vnf_package.id,
expected_attrs=None)
self.compare_obj(self.vnf_package, vnfpkgm,
allow_missing=['vnf_deployment_flavours'])
def test_get_by_id_with_no_existing_id(self):
self.assertRaises(
exceptions.VnfPackageNotFound,
objects.VnfPackage.get_by_id, self.context,
uuidsentinel.invalid_uuid)
def test_create_with_id(self):
vnf_obj = {'id': uuidsentinel.uuid}
vnf_pack = objects.VnfPackage(context=self.context, **vnf_obj)
self.assertRaises(exceptions.ObjectActionError, vnf_pack.create)
def test_save(self):
self.vnf_package.onboarding_state = 'ONBOARDED'
self.vnf_package.save()
self.assertEqual('ONBOARDED', self.vnf_package.onboarding_state)
def test_save_error(self):
fake_data = {'id': uuidsentinel.uuid}
vnf_pack_obj = objects.VnfPackage(context=self.context,
**fake_data)
self.assertRaises(exceptions.VnfPackageNotFound, vnf_pack_obj.save)
@mock.patch('tacker.objects.vnf_package._destroy_vnf_package')
def test_destroy(self, mock_vnf_destroy):
self.vnf_package.destroy(self.context)
mock_vnf_destroy.assert_called_with(self.context, self.vnf_package.id)
def test_destroy_without_id(self):
vnf_obj = objects.VnfPackage(context=self.context)
self.assertRaises(exceptions.ObjectActionError, vnf_obj.destroy,
self.context)
def test_get_all(self):
result = objects.VnfPackagesList.get_all(self.context,
expected_attrs=None)
self.assertTrue(result.objects, list)
self.assertTrue(result.objects)
def test_get_by_id_with_flavours(self):
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **flavour_data)
vnf_deployment_flavour_obj.create()
vnfpkgm = objects.VnfPackage.get_by_id(
self.context, self.vnf_package.id,
expected_attrs=['vnf_deployment_flavours'])
self.assertEqual(1, len(vnfpkgm.vnf_deployment_flavours.objects))
self.compare_obj(vnf_deployment_flavour_obj,
vnfpkgm.vnf_deployment_flavours[0],
allow_missing=['software_images',
'updated_at',
'deleted', 'deleted_at'])
def test_get_by_id_without_flavours(self):
vnfpkgm = objects.VnfPackage.get_by_id(
self.context, self.vnf_package.id, expected_attrs=None)
self.assertEqual([], vnfpkgm.vnf_deployment_flavours.objects)
def test_attribute_with_valid_data(self):
data = {'id': self.vnf_package.id}
vnf_pack_obj = objects.VnfPackage(context=self.context, **data)
vnf_pack_obj.obj_load_attr('user_data')
self.assertEqual({'abc': 'xyz'}, vnf_pack_obj.user_data)
def test_invalid_attribute(self):
self.assertRaises(exceptions.ObjectActionError,
self.vnf_package.obj_load_attr, 'invalid')
def test_obj_load_attr_without_context(self):
data = {'id': self.vnf_package.id}
vnf_package_obj = objects.VnfPackage(**data)
self.assertRaises(exceptions.OrphanedObjectError,
vnf_package_obj.obj_load_attr, 'algorithm')
def test_obj_load_attr_without_id_in_object(self):
data = {'user_data': {'tests': 'test_data'}}
vnf_deployment_flavour_obj = objects.VnfDeploymentFlavour(
context=self.context, **data)
self.assertRaises(
exceptions.ObjectActionError,
vnf_deployment_flavour_obj.obj_load_attr, 'algorithm')
def test_vnf_package_list_by_filter(self):
filters = {'onboarding_state': 'CREATED'}
vnfpkgm_list = objects.VnfPackagesList.get_by_filters(
self.context, **filters)
self.assertEqual(1, len(vnfpkgm_list))

58
tacker/tests/unit/objects/test_vnf_package_vnfd.py Normal file
View File

@ -0,0 +1,58 @@
# Copyright 2019 NTT DATA.
#
# 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 tacker.common import exceptions
from tacker import context
from tacker.objects import vnf_package
from tacker.objects import vnf_package_vnfd
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests.unit.objects import fakes
from tacker.tests import uuidsentinel
class TestVnfPackageVnfd(SqlTestCase):
def setUp(self):
super(TestVnfPackageVnfd, self).setUp()
self.context = context.get_admin_context()
def test_create(self):
vnf_pack = vnf_package.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_pack.create()
vnf_pack_vnfd = {
'package_uuid': vnf_pack.id,
'vnfd_id': uuidsentinel.vnfd_id,
'vnf_provider': 'test_provider',
'vnf_product_name': 'test_product_name',
'vnf_software_version': 'test_version',
'vnfd_version': 'test_vnfd_version',
}
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
self.assertTrue(vnf_pack_vnfd_obj.id)
def test_create_with_id(self):
vnf_pack = vnf_package.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_pack.create()
vnf_pack_vnfd = {'id': uuidsentinel.id}
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
self.assertRaises(
exceptions.ObjectActionError,
vnf_pack_vnfd_obj.create)

109
tacker/tests/unit/objects/test_vnf_software_images.py Normal file
View File

@ -0,0 +1,109 @@
# Copyright (c) 2019 NTT DATA
#
# 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 tacker.common import exceptions
from tacker import context
from tacker import objects
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests.unit.objects import fakes
from tacker.tests import uuidsentinel
class TestVnfSoftwareImages(SqlTestCase):
def setUp(self):
super(TestVnfSoftwareImages, self).setUp()
self.context = context.get_admin_context()
self.vnf_package = self._create_vnf_package()
self.vnf_deployment_flavour = self._create_vnf_deployment_flavour()
self.vnf_softwate_images = self._create_vnf_softwate_images()
def _create_vnf_package(self):
vnfpkgm = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnfpkgm.create()
return vnfpkgm
def _create_vnf_deployment_flavour(self):
flavour_data = fakes.vnf_deployment_flavour
flavour_data.update({'package_uuid': self.vnf_package.id})
vnf_deployment_flavour = objects.VnfDeploymentFlavour(
context=self.context, **flavour_data)
vnf_deployment_flavour.create()
return vnf_deployment_flavour
def _create_vnf_softwate_images(self):
software_image = fakes.software_image
software_image.update(
{'flavour_uuid': self.vnf_deployment_flavour.id})
vnf_soft_image_obj = objects.VnfSoftwareImage(
context=self.context, **software_image)
vnf_soft_image_obj.create()
return vnf_soft_image_obj
def test_create(self):
software_image = fakes.software_image
software_image.update(
{'flavour_uuid': self.vnf_deployment_flavour.id})
vnf_soft_image_obj = objects.VnfSoftwareImage(
context=self.context, **software_image)
vnf_soft_image_obj.create()
self.assertTrue(vnf_soft_image_obj.id)
def test_software_image_create_with_id(self):
software_image = fakes.software_image
software_image.update({'id': uuidsentinel.id})
vnf_soft_image_obj = objects.VnfSoftwareImage(
context=self.context, **software_image)
self.assertRaises(
exceptions.ObjectActionError,
vnf_soft_image_obj.create)
def test_get_by_id(self):
vnf_software_images = objects.VnfSoftwareImage.get_by_id(
self.context, self.vnf_softwate_images.id, expected_attrs=None)
self.compare_obj(self.vnf_softwate_images, vnf_software_images)
def test_get_by_id_with_no_existing_id(self):
self.assertRaises(
exceptions.VnfSoftwareImageNotFound,
objects.VnfSoftwareImage.get_by_id, self.context,
uuidsentinel.invalid_uuid)
def test_attribute_with_valid_data(self):
data = {'id': self.vnf_softwate_images.id}
vnf_software_image_obj = objects.VnfSoftwareImage(
context=self.context, **data)
vnf_software_image_obj.obj_load_attr('name')
self.assertEqual('test', vnf_software_image_obj.name)
def test_invalid_attribute(self):
self.assertRaises(exceptions.ObjectActionError,
self.vnf_softwate_images.obj_load_attr, 'invalid')
def test_obj_load_attr_without_context(self):
data = {'id': self.vnf_softwate_images.id}
vnf_software_image_obj = objects.VnfSoftwareImage(**data)
self.assertRaises(exceptions.OrphanedObjectError,
vnf_software_image_obj.obj_load_attr, 'name')
def test_obj_load_attr_without_id_in_object(self):
data = {'name': self.vnf_softwate_images.name}
vnf_software_image_obj = objects.VnfSoftwareImage(
context=self.context, **data)
self.assertRaises(exceptions.ObjectActionError,
vnf_software_image_obj.obj_load_attr, 'name')