openstack/tacker
Code Issues Proposed changes

Add version objects and db apis for vnf lcm

Browse Source 
Added versioned objects and db apis for vnf lifecycle
management.

Co-authored-By: Ajay Parja <ajay.parja@nttdata.com>
Co-authored-By: Nitin Uikey <nitin.uikey@nttdata.com>
Co-authored-By: Niraj <Niraj.Singh@nttdata.com>

Change-Id: Ife7971a395c60fcb71dffb4427105bceb508dbc6
Blueprint: support-etsi-nfv-specs
This commit is contained in:
tpatil 2020-01-01 14:13:34 +00:00 committed by Shubham
parent 30fcfcc0a8
commit aa3508fb6f
18 changed files with 2670 additions and 11 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
tacker/common/exceptions.py
View File

@ -211,6 +211,14 @@ class VnfPackageVnfdIdDuplicate(TackerException):
message = _("Vnf package with vnfd id %(vnfd_id)s already exists.")
class VnfInstanceNotFound(NotFound):
message = _("No vnf instance with id %(id)s.")
class VnfResourceNotFound(NotFound):
message = _("No vnf resource with id %(id)s.")
class VnfDeploymentFlavourNotFound(NotFound):
message = _("No vnf deployment flavour with id %(id)s.")
@ -219,6 +227,10 @@ class VnfSoftwareImageNotFound(NotFound):
message = _("No vnf software image with id %(id)s.")
class VnfInstantiatedInfoNotFound(NotFound):
message = _("No vnf instantiated info for vnf id %(vnf_instance_id)s.")
class OrphanedObjectError(TackerException):
msg_fmt = _('Cannot call %(method)s on orphaned %(objtype)s object')

4
tacker/objects/__init__.py
View File

@ -29,3 +29,7 @@ def register_all():
__import__('tacker.objects.vnf_package_vnfd')
__import__('tacker.objects.vnf_deployment_flavour')
__import__('tacker.objects.vnf_software_image')
__import__('tacker.objects.vnf_instance')
__import__('tacker.objects.vnf_instantiated_info')
__import__('tacker.objects.vim_connection')
__import__('tacker.objects.vnf_resources')

14
tacker/objects/base.py
View File

@ -12,6 +12,7 @@
# License for the specific language governing permissions and limitations
# under the License.
import contextlib
import datetime
import oslo_messaging as messaging
@ -52,6 +53,10 @@ class TackerObject(ovoo_base.VersionedObject):
OBJ_SERIAL_NAMESPACE = 'tacker_object'
OBJ_PROJECT_NAMESPACE = 'tacker'
def __init__(self, context=None, **kwargs):
super(TackerObject, self).__init__(context, **kwargs)
self.obj_set_defaults()
def tacker_obj_get_changes(self):
"""Returns a dict of changed fields with tz unaware datetimes.
@ -118,6 +123,15 @@ class TackerObject(ovoo_base.VersionedObject):
else:
self._changed_fields.clear()
@contextlib.contextmanager
def obj_alternate_context(self, context):
original_context = self._context
self._context = context
try:
yield
finally:
self._context = original_context
class TackerObjectSerializer(messaging.NoOpSerializer):
"""A TackerObject-aware Serializer.

47
tacker/objects/fields.py
View File

@ -23,6 +23,7 @@ StringField = fields.StringField
ListOfObjectsField = fields.ListOfObjectsField
ListOfStringsField = fields.ListOfStringsField
DictOfStringsField = fields.DictOfStringsField
DictOfNullableStringsField = fields.DictOfNullableStringsField
DateTimeField = fields.DateTimeField
BooleanField = fields.BooleanField
BaseEnumField = fields.BaseEnumField
@ -119,3 +120,49 @@ class UUID(fields.UUID):
class UUIDField(fields.AutoTypedField):
AUTO_TYPE = UUID()
class VnfInstanceState(BaseTackerEnum):
INSTANTIATED = 'INSTANTIATED'
NOT_INSTANTIATED = 'NOT_INSTANTIATED'
ALL = (INSTANTIATED, NOT_INSTANTIATED)
class VnfInstanceStateField(BaseEnumField):
AUTO_TYPE = VnfInstanceState()
class VnfInstanceTaskState(BaseTackerEnum):
INSTANTIATING = 'INSTANTIATING'
HEALING = 'HEALING'
TERMINATING = 'TERMINATING'
ERROR = 'ERROR'
ALL = (INSTANTIATING, HEALING, TERMINATING, ERROR)
class VnfInstanceTaskStateField(BaseEnumField):
AUTO_TYPE = VnfInstanceTaskState()
class VnfOperationalStateType(BaseTackerEnum):
STARTED = 'STARTED'
STOPPED = 'STOPPED'
ALL = (STARTED, STOPPED)
class VnfOperationalStateTypeField(BaseEnumField):
AUTO_TYPE = VnfOperationalStateType()
class IpAddressType(BaseTackerEnum):
IPV4 = 'IPV4'
IPV6 = 'IPV6'
ALL = (IPV4, IPV6)
class IpAddressTypeField(BaseEnumField):
AUTO_TYPE = IpAddressType()

58
tacker/objects/vim_connection.py Normal file
View File

@ -0,0 +1,58 @@
# Copyright (C) 2020 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.
from tacker.objects import base
from tacker.objects import fields
@base.TackerObjectRegistry.register
class VimConnectionInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'vim_id': fields.StringField(nullable=True, default=None),
'vim_type': fields.StringField(nullable=False),
'access_info': fields.DictOfNullableStringsField(nullable=True,
default={}),
}
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
vim_id = data_dict.get('vim_id')
vim_type = data_dict.get('vim_type')
access_info = data_dict.get('access_info', {})
obj = cls(id=id, vim_id=vim_id, vim_type=vim_type,
access_info=access_info)
return obj
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
vim_connection_info = super(
VimConnectionInfo, cls).obj_from_primitive(
primitive, context)
else:
vim_connection_info = VimConnectionInfo._from_dict(primitive)
return vim_connection_info
def to_dict(self):
return {'id': self.id,
'vim_id': self.vim_id,
'vim_type': self.vim_type,
'access_info': self.access_info}

284
tacker/objects/vnf_instance.py Normal file
View File

@ -0,0 +1,284 @@
# Copyright (C) 2020 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.
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
LOG = logging.getLogger(__name__)
@db_api.context_manager.reader
def _vnf_instance_get_by_id(context, vnf_instance_id, columns_to_join=None):
query = api.model_query(context, models.VnfInstance,
read_deleted="no", project_only=True). \
filter_by(id=vnf_instance_id)
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
result = query.first()
if not result:
raise exceptions.VnfInstanceNotFound(id=vnf_instance_id)
return result
@db_api.context_manager.writer
def _vnf_instance_create(context, values):
vnf_instance = models.VnfInstance()
vnf_instance.update(values)
vnf_instance.save(context.session)
return _vnf_instance_get_by_id(context, vnf_instance.id,
columns_to_join=["instantiated_vnf_info"])
@db_api.context_manager.writer
def _vnf_instance_update(context, vnf_instance_id, values,
columns_to_join=None):
vnf_instance = _vnf_instance_get_by_id(context, vnf_instance_id,
columns_to_join=columns_to_join)
vnf_instance.update(values)
vnf_instance.save(session=context.session)
return vnf_instance
@db_api.context_manager.writer
def _destroy_vnf_instance(context, uuid):
now = timeutils.utcnow()
updated_values = {'deleted': True,
'deleted_at': now
}
api.model_query(context, models.VnfInstantiatedInfo). \
filter_by(vnf_instance_id=uuid). \
update(updated_values, synchronize_session=False)
api.model_query(context, models.VnfInstance).\
filter_by(id=uuid). \
update(updated_values, synchronize_session=False)
@db_api.context_manager.reader
def _vnf_instance_list(context, columns_to_join=None):
query = api.model_query(context, models.VnfInstance, read_deleted="no",
project_only=True)
if columns_to_join:
for column in columns_to_join:
query = query.options(joinedload(column))
return query.all()
def _make_vnf_instance_list(context, vnf_instance_list, db_vnf_instance_list,
expected_attrs):
vnf_instance_cls = VnfInstance
vnf_instance_list.objects = []
for db_vnf_instance in db_vnf_instance_list:
vnf_instance_obj = vnf_instance_cls._from_db_object(
context, vnf_instance_cls(context), db_vnf_instance,
expected_attrs=expected_attrs)
vnf_instance_list.objects.append(vnf_instance_obj)
vnf_instance_list.obj_reset_changes()
return vnf_instance_list
@base.TackerObjectRegistry.register
class VnfInstance(base.TackerObject, base.TackerPersistentObject,
base.TackerObjectDictCompat):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'vnf_instance_name': fields.StringField(nullable=True),
'vnf_instance_description': fields.StringField(nullable=True),
'instantiation_state': fields.VnfInstanceStateField(nullable=False,
default=fields.VnfInstanceState.NOT_INSTANTIATED),
'task_state': fields.StringField(nullable=True, default=None),
'vnfd_id': fields.StringField(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),
'vim_connection_info': fields.ListOfObjectsField(
'VimConnectionInfo', nullable=True, default=[]),
'tenant_id': fields.StringField(nullable=False),
'instantiated_vnf_info': fields.ObjectField('InstantiatedVnfInfo',
nullable=True, default=None)
}
def __init__(self, context=None, **kwargs):
super(VnfInstance, self).__init__(context, **kwargs)
self.obj_set_defaults()
@staticmethod
def _from_db_object(context, vnf_instance, db_vnf_instance,
expected_attrs=None):
special_fields = ["instantiated_vnf_info", "vim_connection_info"]
for key in vnf_instance.fields:
if key in special_fields:
continue
setattr(vnf_instance, key, db_vnf_instance[key])
VnfInstance._load_instantiated_vnf_info_from_db_object(context,
vnf_instance, db_vnf_instance)
vim_connection_info = db_vnf_instance['vim_connection_info']
vim_connection_list = [objects.VimConnectionInfo.obj_from_primitive(
vim_info, context) for vim_info in vim_connection_info]
vnf_instance.vim_connection_info = vim_connection_list
vnf_instance._context = context
vnf_instance.obj_reset_changes()
return vnf_instance
@staticmethod
def _load_instantiated_vnf_info_from_db_object(context, vnf_instance,
db_vnf_instance):
if db_vnf_instance['instantiated_vnf_info']:
inst_vnf_info = \
objects.InstantiatedVnfInfo.obj_from_db_obj(context,
db_vnf_instance['instantiated_vnf_info'])
vnf_instance.instantiated_vnf_info = inst_vnf_info
@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']
db_vnf_instance = _vnf_instance_create(self._context, updates)
expected_attrs = ["instantiated_vnf_info"]
self._from_db_object(self._context, self, db_vnf_instance,
expected_attrs=expected_attrs)
@base.remotable
def save(self):
context = self._context
updates = {}
changes = self.obj_what_changed()
for field in self.fields:
if (self.obj_attr_is_set(field) and
isinstance(self.fields[field], fields.ObjectField)):
try:
getattr(self, '_save_%s' % field)(context)
except AttributeError:
LOG.exception('No save handler for %s', field)
elif (self.obj_attr_is_set(field) and
isinstance(self.fields[field], fields.ListOfObjectsField)):
field_list = getattr(self, field)
updates[field] = [obj.obj_to_primitive() for obj in field_list]
elif field in changes:
updates[field] = self[field]
expected_attrs = ["instantiated_vnf_info"]
db_vnf_instance = _vnf_instance_update(self._context,
self.id, updates,
columns_to_join=expected_attrs)
self._from_db_object(self._context, self, db_vnf_instance)
def _save_instantiated_vnf_info(self, context):
if self.instantiated_vnf_info:
with self.instantiated_vnf_info.obj_alternate_context(context):
self.instantiated_vnf_info.save()
@base.remotable
def destroy(self, context):
if not self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='destroy',
reason='no uuid')
_destroy_vnf_instance(context, self.id)
def to_dict(self):
data = {'id': self.id,
'vnf_instance_name': self.vnf_instance_name,
'vnf_instance_description': self.vnf_instance_description,
'instantiation_state': self.instantiation_state,
'vnfd_id': self.vnfd_id,
'vnf_provider': self.vnf_provider,
'vnf_product_name': self.vnf_product_name,
'vnf_software_version': self.vnf_software_version,
'vnfd_version': self.vnfd_version}
if (self.instantiation_state == fields.VnfInstanceState.INSTANTIATED
and self.instantiated_vnf_info):
data.update({'instantiated_vnf_info':
self.instantiated_vnf_info.to_dict()})
vim_connection_info_list = []
for vim_connection_info in self.vim_connection_info:
vim_connection_info_list.append(vim_connection_info.to_dict())
data.update({'vim_connection_info': vim_connection_info_list})
return data
@base.remotable_classmethod
def get_by_id(cls, context, id):
expected_attrs = ["instantiated_vnf_info"]
db_vnf_instance = _vnf_instance_get_by_id(
context, id, columns_to_join=expected_attrs)
return cls._from_db_object(context, cls(), db_vnf_instance,
expected_attrs=expected_attrs)
@base.TackerObjectRegistry.register
class VnfInstanceList(ovoo_base.ObjectListBase, base.TackerObject):
VERSION = '1.0'
fields = {
'objects': fields.ListOfObjectsField('VnfInstance')
}
@base.remotable_classmethod
def get_all(cls, context, expected_attrs=None):
expected_attrs = ["instantiated_vnf_info"]
db_vnf_instances = _vnf_instance_list(context,
columns_to_join=expected_attrs)
return _make_vnf_instance_list(context, cls(), db_vnf_instances,
expected_attrs)

974
tacker/objects/vnf_instantiated_info.py Normal file
View File

@ -0,0 +1,974 @@
# Copyright (C) 2020 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.
from oslo_log import log as logging
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
LOG = logging.getLogger(__name__)
@db_api.context_manager.writer
def _instantiate_vnf_info_update(context, vnf_instance_id, values):
vnf_info = api.model_query(context, models.VnfInstantiatedInfo). \
filter_by(vnf_instance_id=vnf_instance_id).first()
needs_create = False
if vnf_info and vnf_info['deleted']:
raise exceptions.VnfInstantiatedInfoNotFound(
vnf_instance_id=vnf_instance_id)
elif not vnf_info:
values['vnf_instance_id'] = vnf_instance_id
vnf_info = models.VnfInstantiatedInfo(**values)
needs_create = True
if needs_create:
vnf_info.save(session=context.session)
else:
vnf_info.update(values)
vnf_info.save(session=context.session)
return vnf_info
@base.TackerObjectRegistry.register
class InstantiatedVnfInfo(base.TackerObject, base.TackerObjectDictCompat,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'flavour_id': fields.StringField(nullable=False),
'vnf_instance_id': fields.UUIDField(nullable=False),
'ext_cp_info': fields.ListOfObjectsField(
'VnfExtCpInfo', nullable=False),
'ext_virtual_link_info': fields.ListOfObjectsField(
'ExtVirtualLinkInfo', nullable=True, default=[]),
'ext_managed_virtual_link_info': fields.ListOfObjectsField(
'ExtManagedVirtualLinkInfo', nullable=True, default=[]),
'vnfc_resource_info': fields.ListOfObjectsField(
'VnfcResourceInfo', nullable=True, default=[]),
'vnf_virtual_link_resource_info': fields.ListOfObjectsField(
'VnfVirtualLinkResourceInfo', nullable=True, default=[]),
'virtual_storage_resource_info': fields.ListOfObjectsField(
'VirtualStorageResourceInfo', nullable=True, default=[]),
'vnf_state': fields.VnfOperationalStateTypeField(nullable=False,
default=fields.VnfOperationalStateType.STOPPED),
'instance_id': fields.StringField(nullable=True, default=None),
'instantiation_level_id': fields.StringField(nullable=True,
default=None),
'additional_params': fields.DictOfStringsField(nullable=True,
default={})
}
@staticmethod
def _from_db_object(context, inst_vnf_info, db_inst_vnf_info):
special_fields = ['ext_cp_info', 'ext_virtual_link_info',
'ext_managed_virtual_link_info',
'vnfc_resource_info',
'vnf_virtual_link_resource_info',
'virtual_storage_resource_info']
for key in inst_vnf_info.fields:
if key in special_fields:
continue
setattr(inst_vnf_info, key, db_inst_vnf_info.get(key))
ext_cp_info = db_inst_vnf_info['ext_cp_info']
ext_cp_info_list = [VnfExtCpInfo.obj_from_primitive(ext_cp, context)
for ext_cp in ext_cp_info]
inst_vnf_info.ext_cp_info = ext_cp_info_list
vnfc_resource_info = db_inst_vnf_info['vnfc_resource_info']
vnfc_resource_info_list = [VnfcResourceInfo.obj_from_primitive(
vnfc_resource, context) for vnfc_resource in vnfc_resource_info]
inst_vnf_info.vnfc_resource_info = vnfc_resource_info_list
storage_res_info = db_inst_vnf_info['virtual_storage_resource_info']
storage_res_info_list = [VirtualStorageResourceInfo.
obj_from_primitive(storage_resource, context)
for storage_resource in storage_res_info]
inst_vnf_info.virtual_storage_resource_info = storage_res_info_list
ext_virtual_link_info = db_inst_vnf_info['ext_virtual_link_info']
ext_vl_info_list = [ExtVirtualLinkInfo.obj_from_primitive(
ext_vl_info, context) for ext_vl_info in ext_virtual_link_info]
inst_vnf_info.ext_virtual_link_info = ext_vl_info_list
ext_mng_vl_info = db_inst_vnf_info['ext_managed_virtual_link_info']
ext_managed_vl_info_list = [ExtManagedVirtualLinkInfo.
obj_from_primitive(ext_managed_vl_info, context) for
ext_managed_vl_info in ext_mng_vl_info]
inst_vnf_info.ext_managed_virtual_link_info = ext_managed_vl_info_list
vnf_vl_resource_info = db_inst_vnf_info[
'vnf_virtual_link_resource_info']
vnf_vl_info_list = [VnfVirtualLinkResourceInfo.
obj_from_primitive(vnf_vl_info, context) for vnf_vl_info in
vnf_vl_resource_info]
inst_vnf_info.vnf_virtual_link_resource_info = vnf_vl_info_list
inst_vnf_info._context = context
inst_vnf_info.obj_reset_changes()
return inst_vnf_info
@base.remotable
def save(self):
updates = {}
changes = self.obj_what_changed()
for field in self.fields:
if (self.obj_attr_is_set(field) and
isinstance(self.fields[field], fields.ListOfObjectsField)):
field_list = getattr(self, field)
updates[field] = [obj.obj_to_primitive() for obj in field_list]
elif field in changes:
updates[field] = self[field]
vnf_info = _instantiate_vnf_info_update(self._context,
self.vnf_instance_id,
updates)
self._from_db_object(self._context, self, vnf_info)
self.obj_reset_changes()
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
instantiate_vnf_info = super(
InstantiatedVnfInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'ext_cp_info' in primitive.keys():
obj_data = [VnfExtCpInfo.obj_from_primitive(
vnf_ext_cp, context) for vnf_ext_cp in primitive.get(
'ext_cp_info', [])]
primitive.update({'ext_cp_info': obj_data})
if 'ext_virtual_link_info' in primitive.keys():
obj_data = [ExtVirtualLinkInfo.obj_from_primitive(
ext_virtual_link,
context) for ext_virtual_link in primitive.get(
'ext_virtual_link_info', [])]
primitive.update({'ext_virtual_link_info': obj_data})
if 'ext_managed_virtual_link_info' in primitive.keys():
obj_data = [ExtManagedVirtualLinkInfo.obj_from_primitive(
ext_managed_v_link,
context) for ext_managed_v_link in primitive.get(
'ext_managed_virtual_link_info', [])]
primitive.update({'ext_managed_virtual_link_info': obj_data})
if 'vnfc_resource_info' in primitive.keys():
obj_data = [VnfcResourceInfo.obj_from_primitive(
vnf_resource_info,
context) for vnf_resource_info in primitive.get(
'vnfc_resource_info', [])]
primitive.update({'vnfc_resource_info': obj_data})
if 'vnf_virtual_link_resource_info' in primitive.keys():
obj_data = [VnfVirtualLinkResourceInfo.obj_from_primitive(
vnf_v_link_resource,
context) for vnf_v_link_resource in primitive.get(
'vnf_virtual_link_resource_info', [])]
primitive.update({'vnf_virtual_link_resource_info': obj_data})
if 'virtual_storage_resource_info' in primitive.keys():
obj_data = [VirtualStorageResourceInfo.obj_from_primitive(
virtual_storage_info,
context) for virtual_storage_info in primitive.get(
'virtual_storage_resource_info', [])]
primitive.update({'virtual_storage_resource_info': obj_data})
instantiate_vnf_info = \
InstantiatedVnfInfo._from_dict(primitive)
return instantiate_vnf_info
@classmethod
def obj_from_db_obj(cls, context, db_obj):
return cls._from_db_object(context, cls(), db_obj)
@classmethod
def _from_dict(cls, data_dict):
flavour_id = data_dict.get('flavour_id')
ext_cp_info = data_dict.get('ext_cp_info', [])
ext_virtual_link_info = data_dict.get('ext_virtual_link_info', [])
ext_managed_virtual_link_info = data_dict.get(
'ext_managed_virtual_link_info', [])
vnfc_resource_info = data_dict.get('vnfc_resource_info', [])
vnf_virtual_link_resource_info = data_dict.get(
'vnf_virtual_link_resource_info', [])
virtual_storage_resource_info = data_dict.get(
'virtual_storage_resource_info', [])
vnf_state = data_dict.get('vnf_state')
instantiation_level_id = data_dict.get('instantiation_level_id')
additional_params = data_dict.get('additional_params', {})
obj = cls(flavour_id=flavour_id, ext_cp_info=ext_cp_info,
ext_virtual_link_info=ext_virtual_link_info,
ext_managed_virtual_link_info=ext_managed_virtual_link_info,
vnfc_resource_info=vnfc_resource_info,
vnf_virtual_link_resource_info=vnf_virtual_link_resource_info,
virtual_storage_resource_info=virtual_storage_resource_info,
vnf_state=vnf_state,
instantiation_level_id=instantiation_level_id,
additional_params=additional_params)
return obj
def to_dict(self):
data = {'flavour_id': self.flavour_id,
'vnf_state': self.vnf_state}
ext_cp_info_list = []
for ext_cp_info in self.ext_cp_info:
ext_cp_info_list.append(ext_cp_info.to_dict())
data.update({'ext_cp_info': ext_cp_info_list})
if self.ext_virtual_link_info:
exp_virt_link_info_list = []
for exp_virt_link_info in self.ext_virtual_link_info:
exp_virt_link_info_list.append(exp_virt_link_info.to_dict())
data.update({'ext_virtual_link_info': exp_virt_link_info_list})
if self.ext_managed_virtual_link_info:
ext_managed_virt_info_list = []
for exp_managed_virt_link_info in \
self.ext_managed_virtual_link_info:
info = exp_managed_virt_link_info.to_dict()
ext_managed_virt_info_list.append(info)
data.update({'ext_managed_virtual_link_info':
ext_managed_virt_info_list})
if self.vnfc_resource_info:
vnfc_resource_info_list = []
for vnfc_resource_info in self.vnfc_resource_info:
vnfc_resource_info_list.append(vnfc_resource_info.to_dict())
data.update({'vnfc_resource_info': vnfc_resource_info_list})
if self.vnf_virtual_link_resource_info:
virt_link_info = []
for vnf_virtual_link_resource_info in \
self.vnf_virtual_link_resource_info:
info = vnf_virtual_link_resource_info.to_dict()
virt_link_info.append(info)
data.update({'vnf_virtual_link_resource_info': virt_link_info})
if self.virtual_storage_resource_info:
virtual_storage_resource_info_list = []
for virtual_storage_resource_info in \
self.virtual_storage_resource_info:
info = virtual_storage_resource_info.to_dict()
virtual_storage_resource_info_list.append(info)
data.update({'virtual_storage_resource_info':
virtual_storage_resource_info_list})
data.update({'additional_params':
self.additional_params})
return data
@base.TackerObjectRegistry.register
class VnfExtCpInfo(base.TackerObject, base.TackerObjectDictCompat,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'cpd_id': fields.StringField(nullable=False),
'cp_protocol_info': fields.ListOfObjectsField(
'CpProtocolInfo', nullable=False, default=[]),
'ext_link_port_id': fields.StringField(nullable=True, default=None),
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_ext_cp_info = super(
VnfExtCpInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'cp_protocol_info' in primitive.keys():
obj_data = [CpProtocolInfo.obj_from_primitive(
ext_cp, context) for ext_cp in primitive.get(
'cp_protocol_info', [])]
primitive.update({'cp_protocol_info': obj_data})
obj_ext_cp_info = VnfExtCpInfo._from_dict(primitive)
return obj_ext_cp_info
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
cpd_id = data_dict.get('cpd_id')
cp_protocol_info = data_dict.get('cp_protocol_info', [])
ext_link_port_id = data_dict.get('ext_link_port_id')
obj = cls(id=id, cpd_id=cpd_id,
cp_protocol_info=cp_protocol_info,
ext_link_port_id=ext_link_port_id)
return obj
def to_dict(self):
data = {'id': self.id,
'cpd_id': self.cpd_id,
'ext_link_port_id': self.ext_link_port_id}
cp_protocol_info_list = []
for cp_protocol_info in self.cp_protocol_info:
cp_protocol_info_list.append(cp_protocol_info.to_dict())
data.update({'cp_protocol_info': cp_protocol_info_list})
return data
@base.TackerObjectRegistry.register
class CpProtocolInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'layer_protocol': fields.StringField(nullable=False),
'ip_over_ethernet': fields.ObjectField(
'IpOverEthernetAddressInfo', nullable=True, default=None),
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_cp_protocol = super(CpProtocolInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'ip_over_ethernet' in primitive.keys():
obj_data = IpOverEthernetAddressInfo.obj_from_primitive(
primitive.get('ip_over_ethernet', {}), context)
primitive.update({'ip_over_ethernet': obj_data})
obj_cp_protocol = CpProtocolInfo._from_dict(primitive)
return obj_cp_protocol
@classmethod
def _from_dict(cls, data_dict):
layer_protocol = data_dict.get('layer_protocol')
ip_over_ethernet = data_dict.get('ip_over_ethernet')
obj = cls(layer_protocol=layer_protocol,
ip_over_ethernet=ip_over_ethernet)
return obj
def to_dict(self):
data = {'layer_protocol': self.layer_protocol}
if self.ip_over_ethernet:
data.update({'ip_over_ethernet': self.ip_over_ethernet.to_dict()})
return data
@base.TackerObjectRegistry.register
class IpOverEthernetAddressInfo(base.TackerObject,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'mac_address': fields.StringField(nullable=True, default=None),
'ip_addresses': fields.ListOfObjectsField('IpAddress', nullable=True,
default=[]),
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
ip_over_ethernet = super(
IpOverEthernetAddressInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'ip_addresses' in primitive.keys():
obj_data = [IpAddress._from_dict(
ip_address) for ip_address in primitive.get(
'ip_addresses', [])]
primitive.update({'ip_addresses': obj_data})
ip_over_ethernet = IpOverEthernetAddressInfo._from_dict(primitive)
return ip_over_ethernet
@classmethod
def _from_dict(cls, data_dict):
mac_address = data_dict.get('mac_address')
ip_addresses = data_dict.get('ip_addresses', [])
obj = cls(mac_address=mac_address, ip_addresses=ip_addresses)
return obj
def to_dict(self):
data = {'mac_address': self.mac_address}
if self.ip_addresses:
ip_addresses_list = []
for ip_addresses in self.ip_addresses:
ip_addresses_list.append(ip_addresses.to_dict())
data.update({'ip_addresses': ip_addresses_list})
return data
@base.TackerObjectRegistry.register
class IpAddress(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'type': fields.IpAddressTypeField(nullable=False),
'subnet_id': fields.StringField(nullable=True, default=None),
'is_dynamic': fields.BooleanField(nullable=True, default=False),
'addresses': fields.ListOfStringsField(nullable=True, default=[]),
}
@classmethod
def _from_dict(cls, data_dict):
type = data_dict.get('type', fields.IpAddressType.IPV4)
subnet_id = data_dict.get('subnet_id')
is_dynamic = data_dict.get('is_dynamic', False)
addresses = data_dict.get('addresses', [])
obj = cls(type=type, subnet_id=subnet_id, is_dynamic=is_dynamic,
addresses=addresses)
return obj
def to_dict(self):
return {'type': self.type,
'subnet_id': self.subnet_id,
'is_dynamic': self.is_dynamic,
'addresses': self.addresses}
@base.TackerObjectRegistry.register
class ExtVirtualLinkInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'resource_handle': fields.ObjectField(
'ResourceHandle', nullable=False),
'ext_link_ports': fields.ListOfObjectsField(
'ExtLinkPortInfo', nullable=True, default=[]),
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_ext_virt_link = super(
ExtVirtualLinkInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'resource_handle' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('resource_handle'))
primitive.update({'resource_handle': obj_data})
if 'ext_link_ports' in primitive.keys():
obj_data = [ExtLinkPortInfo.obj_from_primitive(
ext_link_port_info, context)
for ext_link_port_info in primitive.get(
'ext_link_ports', [])]
primitive.update({'ext_link_ports': obj_data})
obj_ext_virt_link = ExtVirtualLinkInfo._from_dict(primitive)
return obj_ext_virt_link
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id', '')
resource_handle = data_dict.get('resource_handle')
ext_link_ports = data_dict.get('ext_link_ports', [])
obj = cls(id=id, resource_handle=resource_handle,
ext_link_ports=ext_link_ports)
return obj
def to_dict(self):
data = {'id': self.id,
'resource_handle': self.resource_handle.to_dict()}
if self.ext_link_ports:
ext_link_ports = []
for ext_link_port in self.ext_link_ports:
ext_link_ports.append(ext_link_port.to_dict())
data.update({'ext_link_ports': ext_link_ports})
return data
@base.TackerObjectRegistry.register
class ExtLinkPortInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'resource_handle': fields.ObjectField(
'ResourceHandle', nullable=False),
'cp_instance_id': fields.StringField(nullable=True, default=None)
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_link_port_info = super(
ExtLinkPortInfo, cls).obj_from_primitive(primitive, context)
else:
if 'resource_handle' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('resource_handle'))
primitive.update({'resource_handle': obj_data})
obj_link_port_info = ExtLinkPortInfo._from_dict(primitive)
return obj_link_port_info
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
resource_handle = data_dict.get('resource_handle')
cp_instance_id = data_dict.get('cp_instance_id')
obj = cls(id=id, resource_handle=resource_handle,
cp_instance_id=cp_instance_id)
return obj
def to_dict(self):
return {'id': self.id,
'resource_handle': self.resource_handle.to_dict(),
'cp_instance_id': self.cp_instance_id}
@base.TackerObjectRegistry.register
class ExtManagedVirtualLinkInfo(base.TackerObject,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'vnf_virtual_link_desc_id': fields.StringField(nullable=False),
'network_resource': fields.ObjectField(
'ResourceHandle', nullable=False),
'vnf_link_ports': fields.ListOfObjectsField(
'VnfLinkPortInfo', nullable=True, default=[]),
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_ext_managed_virt_link = super(
ExtManagedVirtualLinkInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'network_resource' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('network_resource'))
primitive.update({'network_resource': obj_data})
if 'vnf_link_ports' in primitive.keys():
obj_data = [VnfLinkPortInfo.obj_from_primitive(
vnf_link_port, context)
for vnf_link_port in primitive.get(
'vnf_link_ports', [])]
primitive.update({'vnf_link_ports': obj_data})
obj_ext_managed_virt_link = ExtManagedVirtualLinkInfo._from_dict(
primitive)
return obj_ext_managed_virt_link
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
vnf_virtual_link_desc_id = data_dict.get('vnf_virtual_link_desc_id')
network_resource = data_dict.get('network_resource')
vnf_link_ports = data_dict.get('vnf_link_ports', [])
obj = cls(id=id, vnf_virtual_link_desc_id=vnf_virtual_link_desc_id,
network_resource=network_resource,
vnf_link_ports=vnf_link_ports)
return obj
def to_dict(self):
data = {'id': self.id,
'vnf_virtual_link_desc_id': self.vnf_virtual_link_desc_id,
'network_resource': self.network_resource.to_dict()}
if self.vnf_link_ports:
vnf_link_ports = []
for vnf_link_port in self.vnf_link_ports:
vnf_link_ports.append(vnf_link_port.to_dict())
data.update({'vnf_link_ports': vnf_link_ports})
return data
@base.TackerObjectRegistry.register
class VnfLinkPortInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'resource_handle': fields.ObjectField(
'ResourceHandle', nullable=False),
'cp_instance_id': fields.StringField(nullable=True, default=None)
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
vnf_port_link_info = super(
VnfLinkPortInfo, cls).obj_from_primitive(primitive, context)
else:
if 'resource_handle' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('resource_handle'))
primitive.update({'resource_handle': obj_data})
vnf_port_link_info = VnfLinkPortInfo._from_dict(primitive)
return vnf_port_link_info
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
resource_handle = data_dict.get('resource_handle')
cp_instance_id = data_dict.get('cp_instance_id')
obj = cls(id=id, resource_handle=resource_handle,
cp_instance_id=cp_instance_id)
return obj
def to_dict(self):
return {'id': self.id,
'resource_handle': self.resource_handle.to_dict(),
'cp_instance_id': self.cp_instance_id}
@base.TackerObjectRegistry.register
class VnfcResourceInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'vdu_id': fields.StringField(nullable=False),
'compute_resource': fields.ObjectField(
'ResourceHandle', nullable=False),
'storage_resource_ids': fields.ListOfStringsField(nullable=True,
default=[]),
'vnfc_cp_info': fields.ListOfObjectsField(
'VnfcCpInfo', nullable=True, default=[]),
'metadata': fields.DictOfStringsField(nullable=True, default={})
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
vnfc_resource_info = super(
VnfcResourceInfo, cls).obj_from_primitive(primitive, context)
else:
if 'compute_resource' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('compute_resource'))
primitive.update({'compute_resource': obj_data})
if 'vnfc_cp_info' in primitive.keys():
obj_data = [VnfcCpInfo.obj_from_primitive(
vnfc_cp_info, context)
for vnfc_cp_info in primitive.get(
'vnfc_cp_info', [])]
primitive.update({'vnfc_cp_info': obj_data})
vnfc_resource_info = VnfcResourceInfo._from_dict(primitive)
return vnfc_resource_info
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
vdu_id = data_dict.get('vdu_id')
compute_resource = data_dict.get('compute_resource')
storage_resource_ids = data_dict.get('storage_resource_ids', [])
vnfc_cp_info = data_dict.get('vnfc_cp_info', [])
metadata = data_dict.get('metadata', {})
obj = cls(id=id, vdu_id=vdu_id,
compute_resource=compute_resource,
storage_resource_ids=storage_resource_ids,
vnfc_cp_info=vnfc_cp_info, metadata=metadata)
return obj
def to_dict(self):
data = {'id': self.id,
'vdu_id': self.vdu_id,
'compute_resource': self.compute_resource.to_dict(),
'storage_resource_ids': self.storage_resource_ids}
if self.vnfc_cp_info:
vnfc_cp_info_list = []
for vnfc_cp_info in self.vnfc_cp_info:
vnfc_cp_info_list.append(vnfc_cp_info.to_dict())
data.update({'vnfc_cp_info': vnfc_cp_info_list})
return data
@base.TackerObjectRegistry.register
class VnfcCpInfo(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'cpd_id': fields.StringField(nullable=False),
'vnf_ext_cp_id': fields.StringField(nullable=True, default=None),
'cp_protocol_info': fields.ListOfObjectsField(
'CpProtocolInfo', nullable=True, default=[]),
'vnf_link_port_id': fields.StringField(nullable=True, default=None)
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_vnfc_cp_info = super(VnfcCpInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'cp_protocol_info' in primitive.keys():
obj_data = [CpProtocolInfo.obj_from_primitive(
ext_cp, context) for ext_cp in primitive.get(
'cp_protocol_info', [])]
primitive.update({'cp_protocol_info': obj_data})
obj_vnfc_cp_info = VnfcCpInfo._from_dict(primitive)
return obj_vnfc_cp_info
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
cpd_id = data_dict.get('cpd_id')
vnf_ext_cp_id = data_dict.get('vnf_ext_cp_id')
cp_protocol_info = data_dict.get('cp_protocol_info', [])
vnf_link_port_id = data_dict.get('vnf_link_port_id')
obj = cls(id=id, cpd_id=cpd_id,
vnf_ext_cp_id=vnf_ext_cp_id,
cp_protocol_info=cp_protocol_info,
vnf_link_port_id=vnf_link_port_id)
return obj
def to_dict(self):
data = {'id': self.id,
'cpd_id': self.cpd_id,
'vnf_ext_cp_id': self.vnf_ext_cp_id,
'vnf_link_port_id': self.vnf_link_port_id}
if self.cp_protocol_info:
cp_protocol_info_list = []
for cp_protocol_info in self.cp_protocol_info:
cp_protocol_info_list.append(cp_protocol_info.to_dict())
data.update({'cp_protocol_info': cp_protocol_info_list})
return data
@base.TackerObjectRegistry.register
class VnfVirtualLinkResourceInfo(base.TackerObject,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'vnf_virtual_link_desc_id': fields.StringField(nullable=False),
'network_resource': fields.ObjectField(
'ResourceHandle', nullable=False),
'vnf_link_ports': fields.ListOfObjectsField(
'VnfLinkPortInfo', nullable=True, default=[])
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_vnf_virtual_link = super(
VnfVirtualLinkResourceInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'network_resource' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('network_resource'))
primitive.update({'network_resource': obj_data})
if 'vnf_link_ports' in primitive.keys():
obj_data = [VnfLinkPortInfo.obj_from_primitive(
vnf_link_port, context)
for vnf_link_port in primitive.get(
'vnf_link_ports', [])]
primitive.update({'vnf_link_ports': obj_data})
obj_vnf_virtual_link = VnfVirtualLinkResourceInfo._from_dict(
primitive)
return obj_vnf_virtual_link
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
vnf_virtual_link_desc_id = data_dict.get(
'vnf_virtual_link_desc_id')
network_resource = data_dict.get('network_resource')
vnf_link_ports = data_dict.get('vnf_link_ports', [])
obj = cls(id=id, vnf_virtual_link_desc_id=vnf_virtual_link_desc_id,
network_resource=network_resource,
vnf_link_ports=vnf_link_ports)
return obj
def to_dict(self):
data = {'id': self.id,
'vnf_virtual_link_desc_id': self.vnf_virtual_link_desc_id,
'network_resource': self.network_resource.to_dict()}
if self.vnf_link_ports:
vnf_link_ports = []
for vnf_link_port in self.vnf_link_ports:
vnf_link_ports.append(vnf_link_port.to_dict())
data['vnf_link_ports'] = vnf_link_ports
return data
@base.TackerObjectRegistry.register
class VirtualStorageResourceInfo(base.TackerObject,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.StringField(nullable=False),
'virtual_storage_desc_id': fields.StringField(nullable=False),
'storage_resource': fields.ObjectField(
'ResourceHandle', nullable=False)
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
obj_virtual_storage = super(
VirtualStorageResourceInfo, cls).obj_from_primitive(
primitive, context)
else:
if 'storage_resource' in primitive.keys():
obj_data = ResourceHandle._from_dict(
primitive.get('storage_resource'))
primitive.update({'storage_resource': obj_data})
obj_virtual_storage = VirtualStorageResourceInfo._from_dict(
primitive)
return obj_virtual_storage
@classmethod
def _from_dict(cls, data_dict):
id = data_dict.get('id')
virtual_storage_desc_id = data_dict.get('virtual_storage_desc_id')
storage_resource = data_dict.get('storage_resource')
obj = cls(id=id, virtual_storage_desc_id=virtual_storage_desc_id,
storage_resource=storage_resource)
return obj
def to_dict(self):
return {'id': self.id,
'virtual_storage_desc_id': self.virtual_storage_desc_id,
'storage_resource': self.storage_resource.to_dict()}
@base.TackerObjectRegistry.register
class ResourceHandle(base.TackerObject,
base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'resource_id': fields.StringField(nullable=False, default=""),
'vim_level_resource_type': fields.StringField(nullable=True,
default=None)
}
@classmethod
def obj_from_primitive(cls, primitive, context):
if 'tacker_object.name' in primitive:
resource_handle = super(
ResourceHandle, cls).obj_from_primitive(
primitive, context)
else:
resource_handle = ResourceHandle._from_dict(primitive)
return resource_handle
@classmethod
def _from_dict(cls, data_dict):
resource_id = data_dict.get('resource_id', "")
vim_level_resource_type = data_dict.get('vim_level_resource_type')
obj = cls(resource_id=resource_id,
vim_level_resource_type=vim_level_resource_type)
return obj
def to_dict(self):
return {'resource_id': self.resource_id,
'vim_level_resource_type': self.vim_level_resource_type}

27
tacker/objects/vnf_package_vnfd.py
View File

@ -16,7 +16,9 @@ from oslo_db import exception as db_exc
from oslo_utils import uuidutils
from tacker.common import exceptions
import tacker.context
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
@ -37,6 +39,26 @@ def _vnf_package_vnfd_create(context, values):
return vnf_package_vnfd
@db_api.context_manager.reader
def _vnf_package_vnfd_get_by_id(context, vnfd_id):
query = api.model_query(context, models.VnfPackageVnfd,
read_deleted="no", project_only=False). \
filter_by(vnfd_id=vnfd_id).\
join((models.VnfPackage, models.VnfPackage.id ==
models.VnfPackageVnfd.package_uuid))
if tacker.context.is_user_context(context):
query = query.filter(models.VnfPackage.tenant_id == context.project_id)
result = query.first()
if not result:
raise exceptions.VnfPackageVnfdNotFound(id=vnfd_id)
return result
@base.TackerObjectRegistry.register
class VnfPackageVnfd(base.TackerObject, base.TackerObjectDictCompat,
base.TackerPersistentObject):
@ -85,3 +107,8 @@ class VnfPackageVnfd(base.TackerObject, base.TackerObjectDictCompat,
@classmethod
def obj_from_db_obj(cls, context, db_obj):
return cls._from_db_object(context, cls(), db_obj)
@base.remotable_classmethod
def get_by_id(cls, context, id):
db_vnf_package_vnfd = _vnf_package_vnfd_get_by_id(context, id)
return cls._from_db_object(context, cls(), db_vnf_package_vnfd)

176
tacker/objects/vnf_resources.py Normal file
View File

@ -0,0 +1,176 @@
# Copyright (C) 2020 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.
from oslo_utils import timeutils
from oslo_utils import uuidutils
from oslo_versionedobjects import base as ovoo_base
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.objects import base
from tacker.objects import fields
@db_api.context_manager.writer
def _vnf_resource_create(context, values):
vnf_resource = models.VnfResource()
vnf_resource.update(values)
vnf_resource.save(context.session)
return vnf_resource
@db_api.context_manager.reader
def _vnf_resource_get_by_id(context, id):
query = api.model_query(context, models.VnfResource,
read_deleted="no", project_only=True). \
filter_by(id=id)
result = query.first()
if not result:
raise exceptions.VnfResourceNotFound(id=id)
return result
@db_api.context_manager.writer
def _vnf_resource_update(context, id, values):
vnf_resource = _vnf_resource_get_by_id(context, id)
vnf_resource.update(values)
vnf_resource.save(session=context.session)
return vnf_resource
@db_api.context_manager.writer
def _destroy_vnf_resource(context, id):
now = timeutils.utcnow()
updated_values = {'deleted': True,
'deleted_at': now
}
api.model_query(context, models.VnfResource).\
filter_by(id=id). \
update(updated_values, synchronize_session=False)
@db_api.context_manager.reader
def _vnf_resource_list(context, vnf_instance_id):
query = api.model_query(context, models.VnfResource, read_deleted="no",
project_only=True).\
filter_by(vnf_instance_id=vnf_instance_id)
return query.all()
def _make_vnf_resources_list(context, vnf_resource_list, db_vnf_resource_list):
vnf_resource_cls = VnfResource
vnf_resource_list.objects = []
for db_vnf_resource in db_vnf_resource_list:
vnf_resource_obj = vnf_resource_cls._from_db_object(
context, vnf_resource_cls(context), db_vnf_resource)
vnf_resource_list.objects.append(vnf_resource_obj)
vnf_resource_list.obj_reset_changes()
return vnf_resource_list
@base.TackerObjectRegistry.register
class VnfResource(base.TackerObject, base.TackerPersistentObject):
# Version 1.0: Initial version
VERSION = '1.0'
fields = {
'id': fields.UUIDField(nullable=False),
'vnf_instance_id': fields.StringField(nullable=False),
'resource_name': fields.StringField(nullable=True),
'resource_type': fields.StringField(nullable=False),
'resource_identifier': fields.StringField(nullable=False),
'resource_status': fields.StringField(nullable=True, default='status')
}
def __init__(self, context=None, **kwargs):
super(VnfResource, self).__init__(context, **kwargs)
self.obj_set_defaults()
@staticmethod
def _from_db_object(context, vnf_resource, db_vnf_resource):
for key in vnf_resource.fields:
if db_vnf_resource[key]:
setattr(vnf_resource, key, db_vnf_resource[key])
vnf_resource._context = context
vnf_resource.obj_reset_changes()
return vnf_resource
@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']
db_vnf_resource = _vnf_resource_create(self._context, updates)
self._from_db_object(self._context, self, db_vnf_resource)
@base.remotable
def save(self):
updates = self.tacker_obj_get_changes()
db_vnf_resource = _vnf_resource_update(self._context,
self.id, updates)
self._from_db_object(self._context, self, db_vnf_resource)
@base.remotable
def destroy(self, context):
if not self.obj_attr_is_set('id'):
raise exceptions.ObjectActionError(action='destroy',
reason='no uuid')
_destroy_vnf_resource(context, self.id)
@base.remotable_classmethod
def get_by_id(cls, context, id):
db_vnf_package = _vnf_resource_get_by_id(context, id)
return cls._from_db_object(context, cls(), db_vnf_package)
@base.TackerObjectRegistry.register
class VnfResourceList(ovoo_base.ObjectListBase, base.TackerObject):
VERSION = '1.0'
fields = {
'objects': fields.ListOfObjectsField('VnfResource')
}
@base.remotable_classmethod
def get_by_vnf_instance_id(cls, context, vnf_instance_id):
db_vnf_resources = _vnf_resource_list(context, vnf_instance_id)
return _make_vnf_resources_list(context, cls(), db_vnf_resources)

82
tacker/tests/unit/db/test_vnf_instance.py Normal file
View File

@ -0,0 +1,82 @@
# Copyright (C) 2020 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.
from tacker.common import exceptions
from tacker import context
from tacker import objects
from tacker.objects import vnf_instance
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 TestVnfLcm(SqlTestCase):
def setUp(self):
super(TestVnfLcm, self).setUp()
self.context = context.get_admin_context()
self.vnf_instance = self._create_vnf_instance()
def _create_vnf_instance(self):
vnf_package_vnfd = self._create_and_upload_vnf_package()
vnf_instance_data = fakes.get_vnf_instance_data(
vnf_package_vnfd.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
return vnf_instance
def _create_and_upload_vnf_package(self):
vnf_package = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_package.create()
vnf_pack_vnfd = fakes.get_vnf_package_vnfd_data(
vnf_package.id, uuidsentinel.vnfd_id)
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
vnf_package.vnf_package = "ONBOARDED"
vnf_package.save()
return vnf_pack_vnfd_obj
def test_vnf_instance_get_by_id(self):
result = vnf_instance._vnf_instance_get_by_id(
self.context, self.vnf_instance.id, columns_to_join=None)
self.assertEqual(self.vnf_instance.id, result.id)
def test_vnf_instance_create(self):
result = vnf_instance._vnf_instance_create(
self.context, fakes.fake_vnf_instance_model_dict())
self.assertTrue(result.id)
def test_vnf_instance_update(self):
update = {"vnf_instance_name": 'updated_instance'}
result = vnf_instance._vnf_instance_update(
self.context, self.vnf_instance.id, update)
self.assertEqual('updated_instance', result.vnf_instance_name)
def test_destroy_vnf_instance(self):
vnf_instance._destroy_vnf_instance(self.context,
self.vnf_instance.id)
self.assertRaises(
exceptions.VnfInstanceNotFound,
objects.VnfInstance.get_by_id, self.context,
self.vnf_instance.id)

51
tacker/tests/unit/db/test_vnf_package_vnfd.py Normal file
View File

@ -0,0 +1,51 @@
# 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 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_vnf_package_vnfd_create(self):
vnf_pack = vnf_package.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_pack.create()
vnf_pack_vnfd_data = fakes.vnf_pack_vnfd_data(vnf_pack.id)
vnf_pack_vnfd_data.update({'id': uuidsentinel.id})
result = vnf_package_vnfd._vnf_package_vnfd_create(
self.context, vnf_pack_vnfd_data)
self.assertTrue(result.id)
self.assertEqual('test_provider', result.vnf_provider)
def test_vnf_package_vnfd_get_by_id(self):
vnf_pack = vnf_package.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_pack.create()
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **fakes.vnf_pack_vnfd_data(vnf_pack.id))
vnf_pack_vnfd_obj.create()
result = vnf_package_vnfd._vnf_package_vnfd_get_by_id(
self.context, vnf_pack_vnfd_obj.vnfd_id)
self.assertEqual(vnf_pack_vnfd_obj.id, result.id)
self.assertEqual(vnf_pack_vnfd_obj.vnf_provider, result.vnf_provider)

107
tacker/tests/unit/db/test_vnf_resource.py Normal file
View File

@ -0,0 +1,107 @@
# Copyright (C) 2020 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.
from tacker.common import exceptions
from tacker import context
from tacker.db.db_sqlalchemy.models import VnfResource
from tacker import objects
from tacker.objects import vnf_package_vnfd
from tacker.objects import vnf_resources
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests.unit.objects import fakes
from tacker.tests import uuidsentinel
class TestVnfResource(SqlTestCase):
def setUp(self):
super(TestVnfResource, self).setUp()
self.context = context.get_admin_context()
self.vnf_instance = self._create_vnf_instance()
self.vnf_resource = self._create_vnf_resource()
def _create_vnf_instance(self):
vnf_package = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_package.create()
vnf_pack_vnfd = fakes.get_vnf_package_vnfd_data(
vnf_package.id, uuidsentinel.vnfd_id)
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
vnf_package.vnf_package = "ONBOARDED"
vnf_package.save()
vnf_instance_data = fakes.get_vnf_instance_data(
vnf_pack_vnfd_obj.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
return vnf_instance
def _create_vnf_resource(self):
vnf_resource = vnf_resources.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
return vnf_resource
def test_vnf_resource_create(self):
resource_data = fakes.fake_vnf_resource_data(
self.vnf_instance.id)
resource_data.update({'id': uuidsentinel.id})
result = vnf_resources._vnf_resource_create(
self.context, resource_data)
self.assertTrue(result.id)
self.assertEqual('test', result.resource_name)
def test_vnf_resource_get_by_id(self):
result = vnf_resources._vnf_resource_get_by_id(
self.context, self.vnf_resource.id)
self.assertEqual(self.vnf_resource.id, result.id)
def test_vnf_resource_update(self):
update = {'resource_name': 'fake'}
result = vnf_resources._vnf_resource_update(
self.context, self.vnf_resource.id, update)
self.assertEqual('fake', result.resource_name)
def test_destroy_vnf_resource(self):
vnf_resources._destroy_vnf_resource(
self.context, self.vnf_resource.id)
self.assertRaises(
exceptions.VnfResourceNotFound,
vnf_resources.VnfResource.get_by_id, self.context,
self.vnf_resource.id)
def test_vnf_resource_list(self):
result = vnf_resources._vnf_resource_list(
self.context, self.vnf_instance.id)
self.assertTrue(result[0].id)
self.assertIsInstance(result[0], VnfResource)
def test_make_vnf_resources_list(self):
vnf_resource_db = vnf_resources._vnf_resource_list(
self.context, self.vnf_instance.id)
vnf_resource_list = vnf_resources._make_vnf_resources_list(
self.context, vnf_resources.VnfResourceList(), vnf_resource_db)
self.assertIsInstance(vnf_resource_list,
vnf_resources.VnfResourceList)
self.assertTrue(vnf_resource_list.objects[0].id)

241
tacker/tests/unit/objects/fakes.py
View File

@ -17,9 +17,10 @@ import copy
import datetime
import iso8601
from tacker.db.db_sqlalchemy import models
from tacker.tests import constants
from tacker.tests import uuidsentinel
vnf_package_data = {'algorithm': None, 'hash': None,
'location_glance_store': None,
'onboarding_state': 'CREATED',
@ -29,6 +30,7 @@ vnf_package_data = {'algorithm': None, 'hash': None,
'user_data': {'abc': 'xyz'},
'created_at': datetime.datetime(
2019, 8, 8, 0, 0, 0, tzinfo=iso8601.UTC),
'deleted': False
}
software_image = {
@ -77,3 +79,240 @@ vnf_deployment_flavour = {'flavour_id': 'simple',
'created_at': datetime.datetime(
2019, 8, 8, 0, 0, 0, tzinfo=iso8601.UTC),
}
def get_vnf_package_vnfd_data(vnf_package_id, vnfd_id):
return {
'package_uuid': vnf_package_id,
'vnfd_id': vnfd_id,
'vnf_provider': 'test vnf provider',
'vnf_product_name': 'Sample VNF',
'vnf_software_version': '1.0',
'vnfd_version': '1.0',
}
def get_vnf_instance_data(vnfd_id):
return {
"vnf_software_version": "1.0",
"vnf_product_name": "Sample VNF",
"vnf_instance_name": 'Sample VNF Instance',
"vnf_instance_description": 'Sample vnf_instance_description',
"instantiation_state": "NOT_INSTANTIATED",
"vnf_provider": "test vnf provider",
"vnfd_id": vnfd_id,
"vnfd_version": "1.0",
"tenant_id": uuidsentinel.tenant_id
}
def get_vnf_instance_data_with_id(vnfd_id):
return {
"id": uuidsentinel.tenant_id,
"vnf_software_version": "1.0",
"vnf_product_name": "Sample VNF",
"vnf_instance_name": 'Sample VNF Instance',
"vnf_instance_description": 'Sample vnf_instance_description',
"instantiation_state": "NOT_INSTANTIATED",
"vnf_provider": "test vnf provider",
"vnfd_id": vnfd_id,
"vnfd_version": "1.0",
"tenant_id": uuidsentinel.tenant_id
}
def fake_vnf_instance_model_dict(**updates):
vnf_instance = {
'deleted': False,
'deleted_at': None,
'updated_at': None,
'created_at': datetime.datetime(1900, 1, 1, 1, 1, 1,
tzinfo=iso8601.UTC),
'vnf_product_name': 'Sample VNF',
'vnf_instance_name': 'Sample VNF',
'vnf_instance_description': None,
'vnf_provider': 'test vnf provider',
'vnf_software_version': '1.0',
'vnfd_id': uuidsentinel.vnfd_id,
'vnfd_version': '1.0',
'instantiation_state': 'NOT_INSTANTIATED',
'vim_connection_info': [],
'tenant_id': '33f8dbdae36142eebf214c1869eb4e4c',
'id': constants.UUID,
}
if updates:
vnf_instance.update(updates)
return vnf_instance
def fake_vnf_resource_data(instance_id):
return {
'vnf_instance_id': instance_id,
'resource_name': "test",
'resource_type': "image",
'resource_identifier': uuidsentinel.image_id,
'resource_status': "status"
}
def vnf_pack_vnfd_data(vnf_pack_id):
return {
'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',
}
ip_address = [{
'type': 'IPV4',
'is_dynamic': True
}]
ip_over_ethernet_address_info = {
'mac_address': 'fake_mac',
'ip_addresses': ip_address,
}
cp_protocol_info = {
'layer_protocol': 'IP_OVER_ETHERNET',
'ip_over_ethernet': ip_over_ethernet_address_info,
}
vnf_external_cp_info = {
'id': uuidsentinel.external_cp_id,
'cpd_id': uuidsentinel.cpd_id,
'ext_link_port_id': uuidsentinel.ext_link_port_id
}
resource_handle_info = {
'resource_id': uuidsentinel.resource_id,
'vim_level_resource_type': 'TEST'
}
ext_link_port_info = {
'id': uuidsentinel.ext_link_port_id,
'resource_handle': resource_handle_info,
'cp_instance_id': uuidsentinel.cp_instance_id,
}
ext_virtual_link_info = {
'id': uuidsentinel.virtual_link_id,
'resource_handle': resource_handle_info,
'ext_link_ports': [ext_link_port_info],
}
vnf_link_ports = {
'id': uuidsentinel.vnf_link_ports_id,
'resource_handle': resource_handle_info,
'cp_instance_id': uuidsentinel.cp_instance_id
}
ext_managed_virtual_link_info = {
'id': uuidsentinel.ext_managed_virtual_link_id,
'vnf_virtual_link_desc_id': uuidsentinel.vnf_virtual_link_desc_id,
'network_resource': resource_handle_info,
'vnf_link_ports': [vnf_link_ports],
}
vnfc_cp_info = {
'id': uuidsentinel.cp_info,
'cpd_id': uuidsentinel.cpd_id,
'vnf_ext_cp_id': uuidsentinel.vnf_ext_cp_id,
'cp_protocol_info': [cp_protocol_info],
'vnf_link_port_id': uuidsentinel.vnf_link_port_id,
}
vnfc_resource_info = {
'id': uuidsentinel.resource_info_id,
'vdu_id': uuidsentinel.vdu_id,
'compute_resource': resource_handle_info,
'storage_resource_ids': [uuidsentinel.id1, uuidsentinel.id2],
'reservation_id': uuidsentinel.reservation_id,
'vnfc_cp_info': [vnfc_cp_info],
'metadata': {'key': 'value'}
}
ip_address_info = {
'type': 'IPV4',
'subnet_id': uuidsentinel.subnet_id,
'is_dynamic': False,
'addresses': ['10.10.1', '10.10.2'],
}
vnf_virtual_link_resource_info = {
'id': uuidsentinel.virtual_link_resource_id,
'vnf_virtual_link_desc_id': uuidsentinel.vnf_virtual_link_desc_id,
'network_resource': resource_handle_info,
'vnf_link_ports': vnf_link_ports,
}
virtual_storage_resource_info = {
'id': uuidsentinel.virtual_storage_resource_id,
'virtual_storage_desc_id': uuidsentinel.virtual_storage_desc_id,
'storage_resource': resource_handle_info,
}
vnf_ext_cp_info = {
'id': uuidsentinel.id,
'cpd_id': 'CP1',
'cp_protocol_info': [cp_protocol_info]
}
def get_instantiated_vnf_info():
instantiated_vnf_info = {
'flavour_id': uuidsentinel.flavour_id,
'vnf_state': 'STARTED',
'instance_id': uuidsentinel.instance_id
}
return instantiated_vnf_info
instantiated_vnf_info = {
'ext_cp_info': [vnf_ext_cp_info],
'flavour_id': uuidsentinel.flavour_id,
'vnf_state': 'STARTED',
'vnf_instance_id': uuidsentinel.vnf_instance_id
}
def vnf_resource_model_object(vnf_resource):
resource_dict = {
'id': vnf_resource.id,
'vnf_instance_id': vnf_resource.vnf_instance_id,
'resource_name': vnf_resource.resource_name,
'resource_type': vnf_resource.resource_type,
'resource_identifier': vnf_resource.resource_identifier,
'resource_status': vnf_resource.resource_status
}
vnf_resource_db_obj = models.VnfResource()
vnf_resource_db_obj.update(resource_dict)
return vnf_resource_db_obj
def vnf_instance_model_object(vnf_instance):
instance_dict = {
'id': vnf_instance.id,
'vnf_instance_name': vnf_instance.vnf_instance_name,
'vnf_instance_description': vnf_instance.vnf_instance_description,
'instantiation_state': vnf_instance.instantiation_state,
'task_state': vnf_instance.task_state,
'vnfd_id': vnf_instance.vnfd_id,
'vnf_provider': vnf_instance.vnf_provider,
'vnf_product_name': vnf_instance.vnf_product_name,
'vnf_software_version': vnf_instance.vnf_software_version,
'vnfd_version': vnf_instance.vnfd_version,
'vim_connection_info': vnf_instance.vim_connection_info,
'tenant_id': vnf_instance.tenant_id,
'created_at': vnf_instance.created_at
}
vnf_instance_db_obj = models.VnfInstance()
vnf_instance_db_obj.update(instance_dict)
return vnf_instance_db_obj

38
tacker/tests/unit/objects/test_vim_connection.py Normal file
View File

@ -0,0 +1,38 @@
# Copyright (C) 2020 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.
from tacker import context
from tacker import objects
from tacker.tests.unit.db.base import SqlTestCase
from tacker.tests import uuidsentinel
class TestVnfResource(SqlTestCase):
def setUp(self):
super(TestVnfResource, self).setUp()
self.context = context.get_admin_context()
def test_obj_from_primitive_and_object_to_dict(self):
vim_connection_dict = {'id': uuidsentinel.uuid,
'vim_id': uuidsentinel.uuid,
'vim_type': 'openstack'}
result = objects.VimConnectionInfo.obj_from_primitive(
vim_connection_dict, self.context)
self.assertEqual(True, isinstance(result, objects.VimConnectionInfo))
self.assertEqual('openstack', result.vim_type)
vim_connection_dict = result.to_dict()
self.assertEqual(True, isinstance(vim_connection_dict, dict))
self.assertEqual('openstack', vim_connection_dict['vim_type'])

146
tacker/tests/unit/objects/test_vnf_instance.py Normal file
View File

@ -0,0 +1,146 @@
# Copyright (C) 2020 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 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
from tacker.db import api as sqlalchemy_api
get_engine = sqlalchemy_api.get_engine
class TestVnfInstance(SqlTestCase):
def setUp(self):
super(TestVnfInstance, self).setUp()
self.context = context.get_admin_context()
self.vnf_package = self._create_and_upload_vnf_package()
self.engine = get_engine()
self.conn = self.engine.connect()
def _create_and_upload_vnf_package(self):
vnf_package = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_package.create()
vnf_pack_vnfd = fakes.get_vnf_package_vnfd_data(
vnf_package.id, uuidsentinel.vnfd_id)
vnf_pack_vnfd_obj = objects.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
vnf_package.vnf_package = "ONBOARDED"
vnf_package.save()
return vnf_pack_vnfd_obj
def test_create(self):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
self.assertTrue(vnf_instance.id)
self.assertEqual('NOT_INSTANTIATED', vnf_instance.instantiation_state)
self.assertEqual(self.vnf_package.vnfd_id,
vnf_instance.vnfd_id)
self.assertEqual('test vnf provider', vnf_instance.vnf_provider)
self.assertEqual('Sample VNF', vnf_instance.vnf_product_name)
self.assertEqual('1.0', vnf_instance.vnf_software_version)
self.assertEqual('1.0', vnf_instance.vnfd_version)
self.assertEqual(vnf_instance_data.get('tenant_id'),
vnf_instance.tenant_id)
def test_create_failure_with_id(self):
vnf_instance_data = fakes.get_vnf_instance_data_with_id(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
self.assertRaises(exceptions.ObjectActionError, vnf_instance.create)
def test_get_by_id(self):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
vnf_instance_by_id = objects.VnfInstance.get_by_id(
self.context, vnf_instance.id)
self.compare_obj(vnf_instance, vnf_instance_by_id,
allow_missing=['instantiated_vnf_info',
'vim_connection_info'])
def test_get_by_id_non_existing_vnf_instance(self):
self.assertRaises(
exceptions.VnfInstanceNotFound,
objects.VnfInstance.get_by_id, self.context,
uuidsentinel.invalid_uuid)
@mock.patch('tacker.objects.vnf_instance._vnf_instance_update')
def test_save(self, mock_update_vnf):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
mock_update_vnf.return_value = \
fakes.vnf_instance_model_object(vnf_instance)
vnf_instance.vnf_instance_name = 'fake-name'
vnf_instance.save()
mock_update_vnf.assert_called_with(
self.context, vnf_instance.id, {
'vnf_instance_name': 'fake-name',
'vim_connection_info': []},
columns_to_join=['instantiated_vnf_info'])
def test_save_error(self):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.id = uuidsentinel.id
self.assertRaises(exceptions.VnfInstanceNotFound, vnf_instance.save)
def test_get_all(self):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
result = objects.VnfInstanceList.get_all(self.context,
expected_attrs=None)
self.assertTrue(result.objects, list)
self.assertTrue(result.objects)
@mock.patch('tacker.objects.vnf_instance._destroy_vnf_instance')
def test_destroy(self, mock_vnf_destroy):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
vnf_instance.destroy(self.context)
mock_vnf_destroy.assert_called_with(self.context, vnf_instance.id)
def test_destroy_failure_without_id(self):
vnf_instance_obj = objects.VnfInstance(context=self.context)
self.assertRaises(exceptions.ObjectActionError,
vnf_instance_obj.destroy, self.context)

278
tacker/tests/unit/objects/test_vnf_instantiated_info.py Normal file
View File

@ -0,0 +1,278 @@
# Copyright (c) 2020 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 copy
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 TestInstantiatedVnfInfo(SqlTestCase):
def setUp(self):
super(TestInstantiatedVnfInfo, self).setUp()
self.context = context.get_admin_context()
self.vnf_package = self._create_and_upload_vnf_package()
self.vnf_instance = self._create_vnf_instance()
self.resource_handle_info = self._create_resource_handle()
self.ext_link_port_info = self._create_ext_link_port_info()
self.ext_virtual_link_info = self._create_ext_virtual_link_info()
self.vnf_link_ports_info = self._create_vnf_link_ports()
self.ip_addresses_info = self._create_ip_addresses_info()
self.ip_over_ethernet = self._create_ip_over_ethernet_info()
self.cp_protocol_info = self._create_cp_protocol_info()
self.vnf_external_cp_info = self._create_vnf_external_cp_info()
self.vnfc_cp_info = self._create_vnfc_cp_info()
self.vnfc_resource_info = self._create_vnfc_resource_info()
self.virtual_link_resource_info = \
self._create_virtual_link_resource_info()
self.virtual_storage_resource_info = \
self._create_virtual_storage_resource_info()
self.ext_managed_virtual_link_info = \
self._create_ext_managed_virtual_link_info()
def _create_and_upload_vnf_package(self):
vnf_package = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_package.create()
vnf_pack_vnfd = fakes.get_vnf_package_vnfd_data(
vnf_package.id, uuidsentinel.vnfd_id)
vnf_pack_vnfd_obj = objects.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
self.vnf_package_vnfd = vnf_pack_vnfd_obj
vnf_package.vnf_package = "ONBOARDED"
vnf_package.save()
return vnf_package
def _create_vnf_instance(self):
vnf_instance_data = fakes.get_vnf_instance_data(
self.vnf_package_vnfd.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
return vnf_instance
def _create_vnf_external_cp_info(self):
vnf_external_cp_data = copy.deepcopy(fakes.vnf_external_cp_info)
vnf_external_cp_data.update(
{'cp_protocol_info': [self.cp_protocol_info]})
vnf_external_cp_info = objects.VnfExtCpInfo(
context=self.context, **vnf_external_cp_data)
return vnf_external_cp_info
def _create_resource_handle(self):
resource_handle_data = copy.deepcopy(fakes.resource_handle_info)
resource_handle_info = objects.ResourceHandle(
context=self.context, **resource_handle_data)
return resource_handle_info
def _create_ext_link_port_info(self):
ext_link_port_info = copy.deepcopy(fakes.ext_link_port_info)
ext_link_port_info.update(
{'resource_handle': self.resource_handle_info})
ext_link_port_info = objects.ExtLinkPortInfo(
context=self.context, **ext_link_port_info)
return ext_link_port_info
def _create_ext_virtual_link_info(self):
ext_virtual_link_info = copy.deepcopy(fakes.ext_virtual_link_info)
ext_virtual_link_info.update(
{'resource_handle_info': self.resource_handle_info,
'ext_link_ports': self.ext_link_port_info})
ext_virtual_link_info = objects.VnfExtCpInfo(
context=self.context, **ext_virtual_link_info)
return ext_virtual_link_info
def _create_vnf_link_ports(self):
vnf_link_ports_info = copy.deepcopy(fakes.vnf_link_ports)
vnf_link_ports_info.update(
{'resource_handle': self.resource_handle_info})
vnf_link_ports_info = objects.VnfLinkPortInfo(
context=self.context, **vnf_link_ports_info)
return vnf_link_ports_info
def _create_ext_managed_virtual_link_info(self):
ext_managed_virtual_link_info = copy.deepcopy(
fakes.ext_managed_virtual_link_info)
ext_managed_virtual_link_info.update(
{'network_resource': self.resource_handle_info,
'vnf_link_ports': [self.vnf_link_ports_info]})
ext_managed_virtual_link_info = objects.ExtManagedVirtualLinkInfo(
context=self.context, **ext_managed_virtual_link_info)
return ext_managed_virtual_link_info
def _create_ip_addresses_info(self):
ip_address_info = copy.deepcopy(fakes.ip_address_info)
ip_address_info = objects.IpAddress(
context=self.context, **ip_address_info)
return ip_address_info
def _create_ip_over_ethernet_info(self):
ip_over_ethernet_onfo = copy.deepcopy(
fakes.ip_over_ethernet_address_info)
ip_over_ethernet_onfo.update(
{'ip_addresses': [self.ip_addresses_info]})
ip_over_ethernet_onfo = objects.IpOverEthernetAddressInfo(
context=self.context, **ip_over_ethernet_onfo)
return ip_over_ethernet_onfo
def _create_cp_protocol_info(self):
cp_protocol_info = copy.deepcopy(fakes.cp_protocol_info)
cp_protocol_info.update(
{'ip_over_ethernet': self.ip_over_ethernet})
cp_protocol_info = objects.CpProtocolInfo(
context=self.context, **cp_protocol_info)
return cp_protocol_info
def _create_vnfc_cp_info(self):
vnfc_cp_info = copy.deepcopy(fakes.vnfc_cp_info)
vnfc_cp_info.update(
{'cp_protocol_info': [self.cp_protocol_info]})
vnfc_cp_info = objects.VnfcCpInfo(
context=self.context, **vnfc_cp_info)
return vnfc_cp_info
def _create_vnfc_resource_info(self):
vnfc_resource_info = copy.deepcopy(fakes.vnfc_resource_info)
vnfc_resource_info.update(
{'compute_resource': self.resource_handle_info,
'vnf_link_ports': [self.vnf_link_ports_info],
'vnfc_cp_info': [self.vnfc_cp_info]})
vnfc_resource_info = objects.VnfcResourceInfo(
context=self.context, **vnfc_resource_info)
return vnfc_resource_info
def _create_virtual_link_resource_info(self):
vnf_virtual_link_resource_info = copy.deepcopy(
fakes.vnf_virtual_link_resource_info)
vnf_virtual_link_resource_info.update(
{'network_resource': self.resource_handle_info,
'vnf_link_ports': [self.vnf_link_ports_info]})
vnf_virtual_link_resource_info = objects.VnfVirtualLinkResourceInfo(
context=self.context, **vnf_virtual_link_resource_info)
return vnf_virtual_link_resource_info
def _create_virtual_storage_resource_info(self):
virtual_storage_resource_info = copy.deepcopy(
fakes.virtual_storage_resource_info)
virtual_storage_resource_info.update(
{'storage_resource': self.resource_handle_info})
virtual_storage_resource_info = objects.VirtualStorageResourceInfo(
context=self.context, **virtual_storage_resource_info)
return virtual_storage_resource_info
def test_save(self):
instantiated_vnf_info = copy.deepcopy(
fakes.get_instantiated_vnf_info())
instantiated_vnf_info.update(
{'ext_cp_info': [self.vnf_external_cp_info],
'vnf_instance_id': self.vnf_instance.id,
'ext_link_port_info': self.ext_link_port_info,
'ext_managed_virtual_link_info': [
self.ext_managed_virtual_link_info],
'vnfc_resource_info': [self.vnfc_resource_info],
'vnf_virtual_link_resource_info': [
self.virtual_link_resource_info],
'virtual_storage_resource_info': [
self.virtual_storage_resource_info]})
instantiated_vnf_info = objects.InstantiatedVnfInfo(
context=self.context, **instantiated_vnf_info)
instantiated_vnf_info.save()
self.assertIsNotNone(instantiated_vnf_info.created_at)
def test_resource_handle_obj_from_primitive_and_object_to_dict(self):
resource_handle = copy.deepcopy(fakes.resource_handle_info)
result = objects.ResourceHandle.obj_from_primitive(
resource_handle, self.context)
self.assertTrue(isinstance(result, objects.ResourceHandle))
self.assertEqual('TEST', result.vim_level_resource_type)
resource_handle_dict = result.to_dict()
self.assertTrue(isinstance(resource_handle_dict, dict))
self.assertEqual(
'TEST', resource_handle_dict['vim_level_resource_type'])
def test_virt_strg_res_info_obj_from_primitive_and_obj_to_dict(self):
virtual_storage_resource_info = copy.deepcopy(
fakes.virtual_storage_resource_info)
result = objects.VirtualStorageResourceInfo.obj_from_primitive(
virtual_storage_resource_info, self.context)
self.assertTrue(isinstance(result,
objects.VirtualStorageResourceInfo))
virt_strg_res_info_dict = result.to_dict()
self.assertTrue(isinstance(virt_strg_res_info_dict, dict))
def test_vnfc_cp_info_obj_from_primitive_and_obj_to_dict(self):
vnfc_cp_info = copy.deepcopy(fakes.vnfc_cp_info)
result = objects.VnfcCpInfo.obj_from_primitive(
vnfc_cp_info, self.context)
self.assertTrue(isinstance(result, objects.VnfcCpInfo))
vnfc_cp_info = result.to_dict()
self.assertTrue(isinstance(vnfc_cp_info, dict))
def test_vnfc_resource_info_obj_from_primitive_and_obj_to_dict(self):
vnfc_resource_info = copy.deepcopy(fakes.vnfc_resource_info)
result = objects.VnfcResourceInfo.obj_from_primitive(
vnfc_resource_info, self.context)
self.assertTrue(isinstance(result, objects.VnfcResourceInfo))
self.assertEqual({'key': 'value'}, result.metadata)
vnfc_resource_info = result.to_dict()
self.assertTrue(isinstance(vnfc_resource_info, dict))
def test_ext_mng_virt_link_obj_from_primitive_and_obj_to_dict(self):
ext_managed_virtual_link_info = copy.deepcopy(
fakes.ext_managed_virtual_link_info)
result = objects.ExtManagedVirtualLinkInfo.obj_from_primitive(
ext_managed_virtual_link_info, self.context)
self.assertTrue(isinstance(result, objects.ExtManagedVirtualLinkInfo))
ext_mng_virt_link = result.to_dict()
self.assertTrue(isinstance(ext_mng_virt_link, dict))
def test_ext_link_port_info_obj_from_primitive_and_obj_to_dict(self):
ext_link_port_info_data = copy.deepcopy(fakes.ext_link_port_info)
result = objects.ExtLinkPortInfo.obj_from_primitive(
ext_link_port_info_data, self.context)
self.assertTrue(isinstance(result, objects.ExtLinkPortInfo))
ext_link_port_info = result.to_dict()
self.assertTrue(isinstance(ext_link_port_info, dict))
def test_ext_virt_link_info_obj_from_primitive_and_obj_to_dict(self):
ext_virtual_link_info = copy.deepcopy(fakes.ext_virtual_link_info)
result = objects.ExtVirtualLinkInfo.obj_from_primitive(
ext_virtual_link_info, self.context)
self.assertTrue(isinstance(result, objects.ExtVirtualLinkInfo))
ext_virt_link_info = result.to_dict()
self.assertTrue(isinstance(ext_virt_link_info, dict))
def test_vnf_ext_cp_info_obj_from_primitive_and_obj_to_dict(self):
vnf_ext_cp_info = copy.deepcopy(fakes.vnf_ext_cp_info)
result = objects.VnfExtCpInfo.obj_from_primitive(
vnf_ext_cp_info, self.context)
self.assertTrue(isinstance(result, objects.VnfExtCpInfo))
ext_virt_link_info = result.to_dict()
self.assertTrue(isinstance(ext_virt_link_info, dict))
def test_instantiated_info_obj_from_primitive_and_obj_to_dict(self):
instantiated_vnf_info = copy.deepcopy(fakes.instantiated_vnf_info)
result = objects.InstantiatedVnfInfo.obj_from_primitive(
instantiated_vnf_info, self.context)
self.assertTrue(isinstance(result, objects.InstantiatedVnfInfo))
instantiated_vnf_info_dict = result.to_dict()
self.assertTrue(isinstance(instantiated_vnf_info_dict, dict))

24
tacker/tests/unit/objects/test_vnf_package_vnfd.py
View File

@ -31,17 +31,8 @@ class TestVnfPackageVnfd(SqlTestCase):
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)
context=self.context, **fakes.vnf_pack_vnfd_data(vnf_pack.id))
vnf_pack_vnfd_obj.create()
self.assertTrue(vnf_pack_vnfd_obj.id)
@ -56,3 +47,16 @@ class TestVnfPackageVnfd(SqlTestCase):
self.assertRaises(
exceptions.ObjectActionError,
vnf_pack_vnfd_obj.create)
def test_get_by_id(self):
vnf_pack = vnf_package.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_pack.create()
vnf_pack_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd(
context=self.context, **fakes.vnf_pack_vnfd_data(vnf_pack.id))
vnf_pack_vnfd_obj.create()
vnf_package_vnfd_obj = vnf_package_vnfd.VnfPackageVnfd()
result = vnf_package_vnfd_obj.get_by_id(
self.context, vnf_pack_vnfd_obj.vnfd_id)
self.assertEqual('test_provider', result.vnf_provider)
self.assertEqual('test_version', result.vnf_software_version)

118
tacker/tests/unit/objects/test_vnf_resource.py Normal file
View File

@ -0,0 +1,118 @@
# Copyright (C) 2020 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 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 TestVnfResource(SqlTestCase):
def setUp(self):
super(TestVnfResource, self).setUp()
self.context = context.get_admin_context()
self.vnf_instance = self._create_vnf_instance()
def _create_vnf_instance(self):
vnf_package = objects.VnfPackage(context=self.context,
**fakes.vnf_package_data)
vnf_package.create()
vnf_pack_vnfd = fakes.get_vnf_package_vnfd_data(
vnf_package.id, uuidsentinel.vnfd_id)
vnf_pack_vnfd_obj = objects.VnfPackageVnfd(
context=self.context, **vnf_pack_vnfd)
vnf_pack_vnfd_obj.create()
vnf_package.vnf_package = "ONBOARDED"
vnf_package.save()
vnf_instance_data = fakes.get_vnf_instance_data(
vnf_pack_vnfd_obj.vnfd_id)
vnf_instance = objects.VnfInstance(context=self.context,
**vnf_instance_data)
vnf_instance.create()
return vnf_instance
def test_create(self):
vnf_resource = objects.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
self.assertTrue(vnf_resource.id)
def test_create_with_id(self):
vnf_resource_data = fakes.fake_vnf_resource_data(self.vnf_instance.id)
vnf_resource_data.update({'id': uuidsentinel.uuid})
vnf_resource_obj = objects.VnfResource(
context=self.context, **vnf_resource_data)
self.assertRaises(exceptions.ObjectActionError,
vnf_resource_obj.create)
@mock.patch('tacker.objects.vnf_resources._vnf_resource_update')
def test_save(self, mock_resource_update):
vnf_resource = objects.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
mock_resource_update.return_value = \
fakes.vnf_resource_model_object(vnf_resource)
vnf_resource.resource_name = 'fake'
vnf_resource.save()
mock_resource_update.assert_called_with(
self.context, vnf_resource.id, {'resource_name': 'fake'})
def test_save_error(self):
vnf_resource = objects.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
vnf_resource.destroy(self.context)
vnf_resource.resource_name = 'fake'
self.assertRaises(exceptions.VnfResourceNotFound, vnf_resource.save)
@mock.patch('tacker.objects.vnf_resources._destroy_vnf_resource')
def test_destroy(self, mock_vnf_destroy):
vnf_resource = objects.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
vnf_resource.destroy(self.context)
mock_vnf_destroy.assert_called_with(self.context, vnf_resource.id)
def test_destroy_failure_without_id(self):
vnf_resource_obj = objects.VnfResource(context=self.context)
self.assertRaises(exceptions.ObjectActionError,
vnf_resource_obj.destroy, self.context)
def test_get_by_vnf_instance_id(self):
vnf_resource = objects.VnfResource(
context=self.context,
**fakes.fake_vnf_resource_data(self.vnf_instance.id))
vnf_resource.create()
vnf_resource_list = objects.VnfResourceList()
result = vnf_resource_list.get_by_vnf_instance_id(
self.context, self.vnf_instance.id)
self.assertIsInstance(result.objects, list)
self.assertTrue(result.objects)