openstack/tacker
Code Issues Proposed changes

Merge "Support LCM operation with user data"

Browse Source
This commit is contained in:
Zuul 2020-04-24 04:25:58 +00:00 committed by Gerrit Code Review
commit ffee7ed0dc
45 changed files with 2317 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
tacker/extensions/vnfm.py
View File

@ -108,6 +108,10 @@ class VNFNotFound(exceptions.NotFound):
message = _('VNF %(vnf_id)s could not be found')
class LCMUserDataFailed(exceptions.TackerException):
message = _('LCM user data %(reason)s')
class ParamYAMLNotWellFormed(exceptions.InvalidInput):
message = _("Parameter YAML not well formed - %(error_msg_details)s")

94
tacker/tests/etc/samples/etsi/nfv/user_data_common/Definitions/sample_lcm_with_user_data_df_simple.yaml Normal file
View File

@ -0,0 +1,94 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: >
Template for test _generate_hot_from_tosca().
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
- sample_lcm_with_user_data_types.yaml
topology_template:
inputs:
descriptor_id:
type: string
descriptor_version:
type: string
provider:
type: string
product_name:
type: string
software_version:
type: string
vnfm_info:
type: list
entry_schema:
type: string
flavour_id:
type: string
flavour_description:
type: string
substitution_mappings:
node_type: ntt.nslab.VNF
properties:
flavour_id: simple
requirements:
virtual_link_external: [ CP1, virtual_link ]
node_templates:
VNF:
type: ntt.nslab.VNF
properties:
flavour_description: A simple flavour
interfaces:
Vnflcm:
instantiate: []
instantiate_start: []
instantiate_end: []
terminate: []
terminate_start: []
terminate_end: []
modify_information: []
modify_information_start: []
modify_information_end: []
VDU1:
type: tosca.nodes.nfv.Vdu.Compute
properties:
name: VDU1
description: VDU1 compute node
vdu_profile:
min_number_of_instances: 1
max_number_of_instances: 1
sw_image_data:
name: Software of VDU1
version: '0.4.0'
checksum:
algorithm: sha-512
hash: 6513f21e44aa3da349f248188a44bc304a3653a04122d8fb4535423c8e1d14cd6a153f735bb0982e2161b5b5186106570c17a9e58b64dd39390617cd5a350f78
container_format: bare
disk_format: qcow2
min_disk: 1 GiB
size: 1 GiB
artifacts:
sw_image:
type: tosca.artifacts.nfv.SwImage
file: ../Files/images/cirros-0.4.0-x86_64-disk.img
capabilities:
virtual_compute:
properties:
virtual_memory:
virtual_mem_size: 512 MiB
virtual_cpu:
num_virtual_cpu: 1
virtual_local_storage:
- size_of_storage: 1 GiB
CP1:
type: tosca.nodes.nfv.VduCp
properties:
layer_protocols: [ ipv4 ]
order: 0
requirements:
- virtual_binding: VDU1

31
tacker/tests/etc/samples/etsi/nfv/user_data_common/Definitions/sample_lcm_with_user_data_top.vnfd.yaml Normal file
View File

@ -0,0 +1,31 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: Sample VNFD for LCM with user data
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
- sample_lcm_with_user_data_types.yaml
- sample_lcm_with_user_data_df_simple.yaml
topology_template:
inputs:
selected_flavour:
type: string
description: VNF deployment flavour selected by the consumer. It is provided in the API
node_templates:
VNF:
type: ntt.nslab.VNF
properties:
flavour_id: { get_input: selected_flavour }
descriptor_id: eeeeeeee-ebca-4fa7-95ed-4840d70a1111
provider: NTT NS lab
product_name: Sample VNF for LCM with user data
software_version: '1.0'
descriptor_version: '1.0'
vnfm_info:
- Tacker
requirements:
#- virtual_link_external # mapped in lower-level templates
#- virtual_link_internal # mapped in lower-level templates

53
tacker/tests/etc/samples/etsi/nfv/user_data_common/Definitions/sample_lcm_with_user_data_types.yaml Normal file
View File

@ -0,0 +1,53 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: VNF type definition
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
node_types:
ntt.nslab.VNF:
derived_from: tosca.nodes.nfv.VNF
properties:
descriptor_id:
type: string
constraints: [ valid_values: [ eeeeeeee-ebca-4fa7-95ed-4840d70a1111 ] ]
default: eeeeeeee-ebca-4fa7-95ed-4840d70a1111
descriptor_version:
type: string
constraints: [ valid_values: [ '1.0' ] ]
default: '1.0'
provider:
type: string
constraints: [ valid_values: [ 'NTT NS lab' ] ]
default: 'NTT NS lab'
product_name:
type: string
constraints: [ valid_values: [ 'Sample VNF for LCM with user data' ] ]
default: 'Sample VNF for LCM with user data'
software_version:
type: string
constraints: [ valid_values: [ '1.0' ] ]
default: '1.0'
vnfm_info:
type: list
entry_schema:
type: string
constraints: [ valid_values: [ Tacker ] ]
default: [ Tacker ]
flavour_id:
type: string
constraints: [ valid_values: [ simple ] ]
default: simple
flavour_description:
type: string
default: "falvour"
requirements:
- virtual_link_external:
capability: tosca.capabilities.nfv.VirtualLinkable
- virtual_link_internal:
capability: tosca.capabilities.nfv.VirtualLinkable
interfaces:
Vnflcm:
type: tosca.interfaces.nfv.Vnflcm

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_basehot_invalid/BaseHOT/sample_lcm_with_user_data_invalid_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
# flavor:
# get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_basehot_invalid/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_basehot_invalid/UserData/__init__.py Normal file
View File

39
tacker/tests/etc/samples/etsi/nfv/user_data_sample_basehot_invalid/UserData/lcm_user_data.py Normal file
View File

@ -0,0 +1,39 @@
#
# 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 tacker.vnfm.lcm_user_data.utils as UserDataUtil
from tacker.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
# Create HOT input parameter using util functions.
initial_param_dict = UserDataUtil.create_initial_param_dict(
base_hot_dict)
vdu_flavor_dict = UserDataUtil.create_vdu_flavor_dict(vnfd_dict)
vdu_image_dict = UserDataUtil.create_vdu_image_dict(grant_info)
cpd_vl_dict = UserDataUtil.create_cpd_vl_dict(
base_hot_dict, inst_req_info)
final_param_dict = UserDataUtil.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
return final_param_dict

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_non_dict/UserData/__init__.py Normal file
View File

26
tacker/tests/etc/samples/etsi/nfv/user_data_sample_non_dict/UserData/lcm_user_data_non_dict.py Normal file
View File

@ -0,0 +1,26 @@
#
# 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.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
# return non dict value
return 'DUMMY'

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_normal/BaseHOT/sample_lcm_with_user_data_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_normal/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_normal/UserData/__init__.py Normal file
View File

39
tacker/tests/etc/samples/etsi/nfv/user_data_sample_normal/UserData/lcm_user_data.py Normal file
View File

@ -0,0 +1,39 @@
#
# 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 tacker.vnfm.lcm_user_data.utils as UserDataUtil
from tacker.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
# Create HOT input parameter using util functions.
initial_param_dict = UserDataUtil.create_initial_param_dict(
base_hot_dict)
vdu_flavor_dict = UserDataUtil.create_vdu_flavor_dict(vnfd_dict)
vdu_image_dict = UserDataUtil.create_vdu_image_dict(grant_info)
cpd_vl_dict = UserDataUtil.create_cpd_vl_dict(
base_hot_dict, inst_req_info)
final_param_dict = UserDataUtil.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
return final_param_dict

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_hot_param/BaseHOT/sample_lcm_with_user_data_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_hot_param/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_hot_param/UserData/__init__.py Normal file
View File

39
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_hot_param/UserData/lcm_user_data_invalid_hot_param.py Normal file
View File

@ -0,0 +1,39 @@
#
# 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 tacker.vnfm.lcm_user_data.utils as UserDataUtil
from tacker.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
# Create HOT input parameter using util functions.
initial_param_dict = UserDataUtil.create_initial_param_dict(
base_hot_dict)
# vdu_flavor_dict = UserDataUtil.create_vdu_flavor_dict(vnfd_dict)
# vdu_image_dict = UserDataUtil.create_vdu_image_dict(grant_info)
# cpd_vl_dict = UserDataUtil.create_cpd_vl_dict(
# base_hot_dict, inst_req_info)
#
# final_param_dict = UserDataUtil.create_final_param_dict(
# initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
return initial_param_dict

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_script/BaseHOT/sample_lcm_with_user_data_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_script/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_script/UserData/__init__.py Normal file
View File

28
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_invalid_script/UserData/lcm_user_data_invalid_script.py Normal file
View File

@ -0,0 +1,28 @@
#
# 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.extensions import vnfm
from tacker.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
error_reason = _(
"invalid user data script.")
raise vnfm.LCMUserDataFailed(reason=error_reason)

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_none/BaseHOT/sample_lcm_with_user_data_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_none/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_none/UserData/__init__.py Normal file
View File

32
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_timeout/BaseHOT/sample_lcm_with_user_data_hot.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

7
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_timeout/TOSCA-Metadata/TOSCA.meta Normal file
View File

@ -0,0 +1,7 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/sample_lcm_with_user_data_top.vnfd.yaml
Name: Files/images/cirros-0.4.0-x86_64-disk.img
Content-type: application/x-iso9066-image

0
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_timeout/UserData/__init__.py Normal file
View File

51
tacker/tests/etc/samples/etsi/nfv/user_data_sample_userdata_timeout/UserData/lcm_user_data_sleeping.py Normal file
View File

@ -0,0 +1,51 @@
#
# 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 time
from oslo_log import log as logging
import tacker.vnfm.lcm_user_data.utils as UserDataUtil
from tacker.vnfm.lcm_user_data.abstract_user_data import AbstractUserData
from tacker.vnfm.lcm_user_data.constants import USER_DATA_TIMEOUT
LOG = logging.getLogger(__name__)
class SampleUserData(AbstractUserData):
@staticmethod
def instantiate(base_hot_dict=None,
vnfd_dict=None,
inst_req_info=None,
grant_info=None):
# Sleep more than timeout.
LOG.debug('Sleep start.')
time.sleep(USER_DATA_TIMEOUT + 60)
LOG.debug('Sleep end.')
# Create HOT input parameter using util functions.
initial_param_dict = UserDataUtil.create_initial_param_dict(
base_hot_dict)
vdu_flavor_dict = UserDataUtil.create_vdu_flavor_dict(vnfd_dict)
vdu_image_dict = UserDataUtil.create_vdu_image_dict(grant_info)
cpd_vl_dict = UserDataUtil.create_cpd_vl_dict(
base_hot_dict, inst_req_info)
final_param_dict = UserDataUtil.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
return final_param_dict

32
tacker/tests/etc/samples/hot_lcm_user_data.yaml Normal file
View File

@ -0,0 +1,32 @@
heat_template_version: 2013-05-23
description: 'Template for test _generate_hot_from_tosca().'
parameters:
nfv:
type: json
resources:
VDU1:
type: OS::Nova::Server
properties:
flavor:
get_resource: VDU1_flavor
name: VDU1
image: { get_param: [ nfv, VDU, VDU1, image ] }
networks:
- port:
get_resource: CP1
CP1:
type: OS::Neutron::Port
properties:
network: { get_param: [ nfv, CP, CP1, network ] }
VDU1_flavor:
type: OS::Nova::Flavor
properties:
ram: { get_param: [ nfv, VDU, VDU1, flavor, ram ] }
vcpus: { get_param: [ nfv, VDU, VDU1, flavor, vcpus ] }
disk: { get_param: [ nfv, VDU, VDU1, flavor, disk ] }
outputs: {}

34
tacker/tests/etc/samples/lcm_instantiate_request/instantiate_vnf_request_lcm_userdata.json Normal file
View File

@ -0,0 +1,34 @@
{
"flavourId": "simple",
"instantiationLevelId": "instantiation_level_1",
"extVirtualLinks": [
{
"id": "ext-vl-uuid-VL1",
"resourceId": "neutron-network-uuid_VL1",
"extCps": [
{
"cpdId": "CP1",
"cpConfig": [
{
"cpProtocolData": [
{
"layerProtocol": "IP_OVER_ETHERNET"
}
]
}
]
}
]
}
],
"vimConnectionInfo": [
{
"id": "vim-uuid",
"vimType": "openstack"
}
],
"additionalParams":{
"lcm-operation-user-data":"./UserData/lcm_user_data.py",
"lcm-operation-user-data-class":"SampleUserData"
}
}

94
tacker/tests/etc/samples/vnfd_lcm_user_data.yaml Normal file
View File

@ -0,0 +1,94 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: >
Template for test _generate_hot_from_tosca().
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
- sample_lcm_with_user_data_types.yaml
topology_template:
inputs:
descriptor_id:
type: string
descriptor_version:
type: string
provider:
type: string
product_name:
type: string
software_version:
type: string
vnfm_info:
type: list
entry_schema:
type: string
flavour_id:
type: string
flavour_description:
type: string
substitution_mappings:
node_type: ntt.nslab.VNF
properties:
flavour_id: simple
requirements:
virtual_link_external: [ CP1, virtual_link ]
node_templates:
VNF:
type: ntt.nslab.VNF
properties:
flavour_description: A simple flavour
interfaces:
Vnflcm:
instantiate: []
instantiate_start: []
instantiate_end: []
terminate: []
terminate_start: []
terminate_end: []
modify_information: []
modify_information_start: []
modify_information_end: []
VDU1:
type: tosca.nodes.nfv.Vdu.Compute
properties:
name: VDU1
description: VDU1 compute node
vdu_profile:
min_number_of_instances: 1
max_number_of_instances: 1
sw_image_data:
name: Software of VDU1
version: '0.4.0'
checksum:
algorithm: sha-256
hash: cf83e1357eefb8bdf1542850d66d8007d620e4050b5715dc83f4a921d36ce9ce47d0d13c5d85f2b0ff8318d2877eec2f63b931bd47417a81a538327af927da3e
container_format: bare
disk_format: qcow2
min_disk: 1 GiB
size: 1 GiB
artifacts:
sw_image:
type: tosca.artifacts.nfv.SwImage
file: ../Files/images/cirros-0.4.0-x86_64-disk.img
capabilities:
virtual_compute:
properties:
virtual_memory:
virtual_mem_size: 512 MiB
virtual_cpu:
num_virtual_cpu: 1
virtual_local_storage:
- size_of_storage: 1 GiB
CP1:
type: tosca.nodes.nfv.VduCp
properties:
layer_protocols: [ ipv4 ]
order: 0
requirements:
- virtual_binding: VDU1

501
tacker/tests/functional/vnflcm/test_vnf_instance_with_user_data.py Normal file
View File

@ -0,0 +1,501 @@
#
# 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 os
import tempfile
import time
import yaml
import zipfile
from oslo_serialization import jsonutils
from oslo_utils import uuidutils
from tacker.objects import fields
from tacker.tests.functional import base
from tacker.tests import utils
VNF_PACKAGE_UPLOAD_TIMEOUT = 60
VNF_INSTANTIATE_TIMEOUT = 60
VNF_TERMINATE_TIMEOUT = 60
VNF_INSTANTIATE_ERROR_WAIT = 80
VNF_DELETE_COMPLETION_WAIT = 60
def _get_external_virtual_links(net0_id):
return [
{
"id": "net0",
"resourceId": net0_id,
"extCps": [{
"cpdId": "CP1",
"cpConfig": [{
"cpProtocolData": [{
"layerProtocol": "IP_OVER_ETHERNET",
}]
}]
}]
}
]
def _create_csar_with_unique_vnfd_id(csar_dir):
unique_id = uuidutils.generate_uuid()
tempfd, tempname = tempfile.mkstemp(suffix=".zip",
dir=os.path.dirname(csar_dir))
os.close(tempfd)
common_dir = os.path.join(csar_dir, "../common/")
ud_common_dir = os.path.join(csar_dir, "../user_data_common/")
zcsar = zipfile.ZipFile(tempname, 'w')
target_dir_list = [csar_dir, common_dir, ud_common_dir]
_write_zipfile(zcsar, unique_id, target_dir_list)
zcsar.close()
return tempname, unique_id
def _write_zipfile(zcsar, unique_id, target_dir_list):
for target_dir in target_dir_list:
for (dpath, _, fnames) in os.walk(target_dir):
if not fnames:
continue
for fname in fnames:
src_file = os.path.join(dpath, fname)
dst_file = os.path.relpath(
os.path.join(dpath, fname), target_dir)
if fname.endswith('.yaml') or fname.endswith('.yml'):
with open(src_file, 'rb') as yfile:
data = yaml.safe_load(yfile)
utils._update_unique_id_in_yaml(data, unique_id)
zcsar.writestr(dst_file, yaml.dump(
data, default_flow_style=False,
allow_unicode=True))
else:
zcsar.write(src_file, dst_file)
def _create_and_upload_vnf_package(tacker_client, csar_package_name,
user_defined_data):
# create vnf package
body = jsonutils.dumps({"userDefinedData": user_defined_data})
resp, vnf_package = tacker_client.do_request(
'/vnfpkgm/v1/vnf_packages', "POST", body=body)
# upload vnf package
csar_package_path = \
"../../etc/samples/etsi/nfv/%s" % csar_package_name
file_path = os.path.abspath(os.path.join(os.path.dirname(__file__),
csar_package_path))
# Generating unique vnfd id. This is required when multiple workers
# are running concurrently. The call below creates a new temporary
# CSAR with unique vnfd id.
file_path, uniqueid = _create_csar_with_unique_vnfd_id(file_path)
with open(file_path, 'rb') as file_object:
resp, resp_body = tacker_client.do_request(
'/vnfpkgm/v1/vnf_packages/{id}/package_content'.format(
id=vnf_package['id']),
"PUT", body=file_object, content_type='application/zip')
# wait for onboard
timeout = VNF_PACKAGE_UPLOAD_TIMEOUT
start_time = int(time.time())
show_url = os.path.join('/vnfpkgm/v1/vnf_packages', vnf_package['id'])
vnfd_id = None
while True:
resp, body = tacker_client.do_request(show_url, "GET")
if body['onboardingState'] == "ONBOARDED":
vnfd_id = body['vnfdId']
break
if ((int(time.time()) - start_time) > timeout):
raise Exception("Failed to onboard vnf package")
time.sleep(1)
# remove temporarily created CSAR file
os.remove(file_path)
return vnf_package['id'], vnfd_id
class VnfLcmWithUserDataTest(base.BaseTackerTest):
def setUp(self):
super(VnfLcmWithUserDataTest, self).setUp()
self.tacker_client = base.BaseTackerTest.tacker_http_client()
self.base_url = "/vnflcm/v1/vnf_instances"
vim_list = self.client.list_vims()
self.vim = self.get_vim(vim_list, 'VIM0')
if not self.vim:
assert False, "vim_list is Empty: Default VIM is missing"
neutron_client = self.neutronclient()
net = neutron_client.list_networks()
networks = {}
for network in net['networks']:
networks[network['name']] = network['id']
net0_id = networks.get('net0')
if not net0_id:
self.fail("net0 network is not available")
self.ext_vl = _get_external_virtual_links(net0_id)
def _create_instantiate_vnf_request_body(self, flavour_id,
instantiation_level_id=None, vim_id=None, ext_vl=None,
add_params=None):
request_body = {"flavourId": flavour_id}
if instantiation_level_id:
request_body["instantiationLevelId"] = instantiation_level_id
if ext_vl:
request_body["extVirtualLinks"] = ext_vl
if vim_id:
request_body["vimConnectionInfo"] = [
{"id": uuidutils.generate_uuid(),
"vimId": vim_id,
"vimType": "ETSINFV.OPENSTACK_KEYSTONE.v_2"}]
if add_params:
request_body["additionalParams"] = add_params
return request_body
def _create_vnf_instance(self, vnfd_id, vnf_instance_name=None,
vnf_instance_description=None):
request_body = {'vnfdId': vnfd_id}
if vnf_instance_name:
request_body['vnfInstanceName'] = vnf_instance_name
if vnf_instance_description:
request_body['vnfInstanceDescription'] = vnf_instance_description
resp, response_body = self.http_client.do_request(
self.base_url, "POST", body=jsonutils.dumps(request_body))
return resp, response_body
def _delete_vnf_instance(self, id):
url = os.path.join(self.base_url, id)
resp, body = self.http_client.do_request(url, "DELETE")
self.assertEqual(204, resp.status_code)
# verify vnf instance is deleted
url = os.path.join(self.base_url, id)
resp, body = self.http_client.do_request(url, "GET")
self.assertEqual(404, resp.status_code)
def _show_vnf_instance(self, id, expected_result=None):
show_url = os.path.join(self.base_url, id)
resp, vnf_instance = self.http_client.do_request(show_url, "GET")
self.assertEqual(200, resp.status_code)
if expected_result:
self.assertDictSupersetOf(expected_result, vnf_instance)
return vnf_instance
def _vnf_instance_wait(self, id,
instantiation_state=fields.VnfInstanceState.INSTANTIATED,
timeout=VNF_INSTANTIATE_TIMEOUT):
show_url = os.path.join(self.base_url, id)
start_time = int(time.time())
while True:
resp, body = self.http_client.do_request(show_url, "GET")
if body['instantiationState'] == instantiation_state:
break
if ((int(time.time()) - start_time) > timeout):
error = ("Vnf instance %(id)s status is %(current)s, "
"expected status should be %(expected)s")
self.fail(error % {"id": id,
"current": body['instantiationState'],
"expected": instantiation_state})
time.sleep(5)
def _vnf_instance_wait_until_fail_detected(self, id,
instantiation_state=fields.VnfInstanceState.NOT_INSTANTIATED,
timeout=VNF_INSTANTIATE_ERROR_WAIT):
show_url = os.path.join(self.base_url, id)
time.sleep(VNF_INSTANTIATE_ERROR_WAIT)
resp, body = self.http_client.do_request(show_url, "GET")
if body['instantiationState'] != instantiation_state:
error = ("Vnf instance %(id)s status is %(current)s, "
"expected status should be %(expected)s")
self.fail(error % {"id": id,
"current": body['instantiationState'],
"expected": instantiation_state})
def _instantiate_vnf_instance(self, id, request_body):
url = os.path.join(self.base_url, id, "instantiate")
resp, body = self.http_client.do_request(url, "POST",
body=jsonutils.dumps(request_body))
self.assertEqual(202, resp.status_code)
self._vnf_instance_wait(id)
def _instantiate_vnf_instance_fail(self, id, request_body):
url = os.path.join(self.base_url, id, "instantiate")
resp, body = self.http_client.do_request(url, "POST",
body=jsonutils.dumps(request_body))
self.assertEqual(202, resp.status_code)
# Confirm that the state doesn't change from NOT_INSTANTIATED.
self._vnf_instance_wait_until_fail_detected(id)
def _terminate_vnf_instance(self, id, request_body):
url = os.path.join(self.base_url, id, "terminate")
resp, body = self.http_client.do_request(url, "POST",
body=jsonutils.dumps(request_body))
self.assertEqual(202, resp.status_code)
timeout = request_body.get('gracefulTerminationTimeout')
start_time = int(time.time())
self._vnf_instance_wait(id,
instantiation_state=fields.VnfInstanceState.NOT_INSTANTIATED,
timeout=VNF_TERMINATE_TIMEOUT)
# If gracefulTerminationTimeout is set, check whether vnf
# instantiation_state is set to NOT_INSTANTIATED after
# gracefulTerminationTimeout seconds.
if timeout and int(time.time()) - start_time < timeout:
self.fail("Vnf is terminated before graceful termination"
"timeout period")
# wait for status completion
time.sleep(VNF_DELETE_COMPLETION_WAIT)
def _delete_vnf_package(self, vnf_package_id):
url = '/vnfpkgm/v1/vnf_packages/%s' % vnf_package_id
# Update vnf package before delete
req_body = jsonutils.dumps({"operationalState": "DISABLED"})
self.tacker_client.do_request(url, "PATCH", body=req_body)
# Delete vnf package before delete
self.tacker_client.do_request(url, "DELETE")
def test_instantiate_vnf_normal(self):
# Create vnf package
sample_name = "user_data_sample_normal"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_normal"
add_params = {
"lcm-operation-user-data": "./UserData/lcm_user_data.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance(vnf_instance['id'], request_body)
vnf_instance = self._show_vnf_instance(vnf_instance['id'])
vdu_count = len(vnf_instance['instantiatedVnfInfo']
['vnfcResourceInfo'])
self.assertEqual(1, vdu_count)
# Terminate vnf forcefully
terminate_req_body = {
"terminationType": fields.VnfInstanceTerminationType.FORCEFUL
}
self._terminate_vnf_instance(vnf_instance['id'], terminate_req_body)
def test_instantiate_vnf_basehot_invalid(self):
# Create vnf package
sample_name = "user_data_sample_basehot_invalid"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_basehot_invalid"
add_params = {
"lcm-operation-user-data": "./UserData/lcm_user_data.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance_fail(vnf_instance['id'], request_body)
def test_instantiate_vnf_userdata_timeout(self):
# Create vnf package
sample_name = "user_data_sample_userdata_timeout"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_timeout"
add_params = {
"lcm-operation-user-data": "./UserData/lcm_user_data_sleeping.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance_fail(vnf_instance['id'], request_body)
def test_instantiate_vnf_userdata_invalid_hot_param(self):
# Create vnf package
sample_name = "user_data_sample_userdata_invalid_hot_param"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_timeout"
add_params = {
"lcm-operation-user-data": "./UserData/"
"lcm_user_data_invalid_hot_param.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance_fail(vnf_instance['id'], request_body)
def test_instantiate_vnf_userdata_none(self):
# Create vnf package
sample_name = "user_data_sample_userdata_none"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_timeout"
add_params = {
"lcm-operation-user-data": "./UserData/lcm_user_data.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance_fail(vnf_instance['id'], request_body)
def test_instantiate_vnf_userdata_invalid_script(self):
# Create vnf package
sample_name = "user_data_sample_userdata_invalid_script"
vnf_package_id, vnfd_id = _create_and_upload_vnf_package(
self.tacker_client, sample_name, {"key": sample_name})
# Reserve deleting vnf package
self.addCleanup(self._delete_vnf_package, vnf_package_id)
# Settings
vnf_instance_name = "vnf_with_user_data-%s" % \
uuidutils.generate_uuid()
vnf_instance_description = "vnf_with_user_data_timeout"
add_params = {
"lcm-operation-user-data": "./UserData/"
"lcm_user_data_invalid_script.py",
"lcm-operation-user-data-class": "SampleUserData"}
# Create vnf instance
resp, vnf_instance = self._create_vnf_instance(vnfd_id,
vnf_instance_name=vnf_instance_name,
vnf_instance_description=vnf_instance_description)
self.assertIsNotNone(vnf_instance['id'])
self.assertEqual(201, resp.status_code)
# Reserve deleting vnf instance
self.addCleanup(self._delete_vnf_instance, vnf_instance['id'])
request_body = self._create_instantiate_vnf_request_body("simple",
vim_id=self.vim['id'], ext_vl=self.ext_vl, add_params=add_params)
self._instantiate_vnf_instance_fail(vnf_instance['id'], request_body)

0
tacker/tests/unit/vnflcm/vnflcm_driver/__init__.py Normal file
View File

438
tacker/tests/unit/vnfm/infra_drivers/openstack/test_openstack_driver.py
View File

@ -14,16 +14,20 @@
# under the License.
import ddt
import importlib
import json
import mock
import os
import requests
import tempfile
import yaml
from tacker.common import exceptions
from tacker import context
from tacker.extensions import vnfm
from tacker import objects
from tacker.tests.common import helpers
from tacker.tests import constants
from tacker.tests.unit import base
from tacker.tests.unit.db import utils
from tacker.tests.unit.vnfm.infra_drivers.openstack.fixture_data import client
@ -46,6 +50,8 @@ class TestOpenStack(base.FixturedTestCase):
self.glance_url = client.GLANCE_URL
self.instance_uuid = uuidsentinel.instance_id
self.stack_id = uuidsentinel.stack_id
self.auth_attr = None
self.plugin = None
self.json_headers = {'content-type': 'application/json',
'location': 'http://heat-api/stacks/'
+ self.instance_uuid + '/myStack/60f83b5e'}
@ -111,6 +117,438 @@ class TestOpenStack(base.FixturedTestCase):
else:
self.requests_mock.register_uri('PATCH', url)
def _json_load(self, input_file):
json_file = os.path.abspath(os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/lcm_instantiate_request/",
str(input_file)))
with open(json_file) as f:
json_dict = json.load(f)
return json_dict
def _read_file(self):
yaml_file = os.path.abspath(os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/",
"hot_lcm_user_data.yaml"))
with open(yaml_file, 'r') as f:
yaml_file_dict = yaml.safe_load(f)
return yaml_file_dict
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_normal(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
vnf_resource = type('', (), {})
vnf_resource.resource_identifier = constants.INVALID_UUID
grant_info_test = {'vdu_name': {vnf_resource}}
self.openstack.create(self.plugin, self.context, vnf,
self.auth_attr, inst_req_info=inst_req_info_test,
vnf_package_path=vnf_package_path_test,
base_hot_dict=base_hot_dict_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_heat_stack(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict = None
vnf_package_path = None
self.openstack.create(self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict, vnf_package_path)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_userdata_none(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# delete lcm-operation-user-data from additionalParams
del test_json['additionalParams']['lcm-operation-user-data']
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_userdataclass_none(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# delete lcm-operation-user-data-class from additionalParams
del test_json['additionalParams']['lcm-operation-user-data-class']
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_userdata_null(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# set null to setlcm-operation-user-data from additionalParams
test_json['additionalParams']['lcm-operation-user-data'] = ''
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_userdataclass_null(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# set null to setlcm-operation-user-data-class from additionalParams
test_json['additionalParams']['lcm-operation-user-data-class'] = ''
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_import_module_exception(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
with mock.patch.object(importlib, 'import_module') as mock_importlib:
mock_importlib.side_effect = Exception('Test Exception')
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_getattr_none(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
with mock.patch.object(importlib, 'import_module') as mock_importlib:
mock_importlib.return_value = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_missing_file(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_none"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
vnf_resource = type('', (), {})
vnf_resource.resource_identifier = constants.INVALID_UUID
grant_info_test = {'vdu_name': {vnf_resource}}
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, inst_req_info=inst_req_info_test,
vnf_package_path=vnf_package_path_test,
base_hot_dict=base_hot_dict_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_return_none_dict(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_non_dict"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
test_json['additionalParams']['lcm-operation-user-data'] = \
'UserData/lcm_user_data_non_dict.py'
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
vnf_resource = type('', (), {})
vnf_resource.resource_identifier = constants.INVALID_UUID
grant_info_test = {'vdu_name': {vnf_resource}}
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, inst_req_info=inst_req_info_test,
vnf_package_path=vnf_package_path_test,
base_hot_dict=base_hot_dict_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_none_base_hot_dict(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
inst_req_info_test.additional_params = test_json['additionalParams']
base_hot_dict_test = None
vnf_package_path_test = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_invalid_user_data(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# set dummy to setlcm-operation-user-data-class from additionalParams
test_json['additionalParams']['lcm-operation-user-data-class'] = \
'DummyUserData'
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_invalid_user_data_class(self,
mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# set dummy to setlcm-operation-user-data-class from additionalParams
test_json['additionalParams']['lcm-operation-user-data-class'] = \
'DummyUserData'
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
grant_info_test = None
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict_test,
vnf_package_path_test,
inst_req_info=inst_req_info_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_lcm_user_data_and_user_data_class_no_value(self,
mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# set null to setlcm-operation-user-data and
# lcm-operation-user-data-class from additionalParams
test_json['additionalParams']['lcm-operation-user-data'] = ''
test_json['additionalParams']['lcm-operation-user-data-class'] = ''
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = test_json['extVirtualLinks']
vnf_resource = type('', (), {})
vnf_resource.resource_identifier = constants.INVALID_UUID
grant_info_test = {'vdu_name': {vnf_resource}}
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, inst_req_info=inst_req_info_test,
vnf_package_path=vnf_package_path_test,
base_hot_dict=base_hot_dict_test,
grant_info=grant_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_lcm_user_data_and_user_data_class_none(self,
mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_normal"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
# delete lcm-operation-user-data and
# lcm-operation-user-data-class from additionalParams
del test_json['additionalParams']['lcm-operation-user-data']
del test_json['additionalParams']['lcm-operation-user-data-class']
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = test_json['extVirtualLinks']
base_hot_dict = None
self.assertRaises(BaseException,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict,
vnf_package_path=vnf_package_path_test,
inst_req_info=inst_req_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_no_additionalparams(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_normal"))
vnf['placement_attr'] = {'region_name': 'dummy_region'}
inst_req_info_test = type('', (), {})
inst_req_info_test.additional_params = None
inst_req_info_test.ext_virtual_links = None
base_hot_dict = None
self.assertRaises(BaseException,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, base_hot_dict,
vnf_package_path=vnf_package_path_test,
inst_req_info=inst_req_info_test)
@mock.patch('tacker.common.clients.OpenstackClients')
def test_create_instance_exception(self, mock_OpenstackClients_heat):
vnf = utils.get_dummy_vnf(instance_id=self.instance_uuid)
vnf['placement_attr'] = {'region_name': 'dummy_region'}
base_hot_dict_test = self._read_file()
vnf_package_path_test = os.path.abspath(
os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/etsi/nfv",
"user_data_sample_userdata_invalid_script"))
inst_req_info_test = type('', (), {})
test_json = self._json_load(
'instantiate_vnf_request_lcm_userdata.json')
test_json['additionalParams']['lcm-operation-user-data'] = \
'UserData/lcm_user_data_invalid_script.py'
inst_req_info_test.additional_params = test_json['additionalParams']
inst_req_info_test.ext_virtual_links = None
vnf_resource = type('', (), {})
vnf_resource.resource_identifier = constants.INVALID_UUID
grant_info_test = {'vdu_name': {vnf_resource}}
self.assertRaises(vnfm.LCMUserDataFailed,
self.openstack.create,
self.plugin, self.context, vnf,
self.auth_attr, inst_req_info=inst_req_info_test,
vnf_package_path=vnf_package_path_test,
base_hot_dict=base_hot_dict_test,
grant_info=grant_info_test)
def test_create_wait(self):
self._response_in_wait_until_stack_ready(["CREATE_IN_PROGRESS",
"CREATE_COMPLETE"])

0
tacker/tests/unit/vnfm/lcm_user_data/__init__.py Normal file
View File

0
tacker/tests/unit/vnfm/lcm_user_data/utils/__init__.py Normal file
View File

192
tacker/tests/unit/vnfm/lcm_user_data/utils/test_utils.py Normal file
View File

@ -0,0 +1,192 @@
#
# 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 os
import testtools
import yaml
from tacker.objects import instantiate_vnf_req
from tacker.tests import constants
from tacker.vnfm.lcm_user_data import utils
default_initial_param_dict = {
'nfv': {
'VDU': {
},
'CP': {
}
}
}
example_initial_param_dict = {
'nfv': {
'VDU': {
'VDU1': {},
},
'CP': {
'CP1': {},
}
}
}
class TestUtils(testtools.TestCase):
def _read_file(self, input_file):
yaml_file = os.path.abspath(os.path.join(os.path.dirname(__file__),
"../../../../etc/samples/",
str(input_file)))
with open(yaml_file, 'r') as f:
yaml_file_dict = yaml.safe_load(f)
return yaml_file_dict
def test_create_initial_param_dict(self):
base_hot_dict = self._read_file("hot_lcm_user_data.yaml")
initial_param_dict = utils.create_initial_param_dict(base_hot_dict)
self.assertEqual(example_initial_param_dict, initial_param_dict)
def test_create_initial_param_dict_empty_argument(self):
base_hot_dict = {}
initial_param_dict = utils.create_initial_param_dict(base_hot_dict)
self.assertEqual(default_initial_param_dict, initial_param_dict)
def test_create_final_param_dict(self):
initial_param_dict = {
'nfv': {
'CP': {
'CP1': {
'network': 'cp1_network_id'
}
},
'VDU': {
'VDU1': {
'image': 'vdu1_image_uuid',
'flavor': {
'ram': 'vdu1_flavor_ram'
}
}
}
}
}
vdu_flavor_dict = {'VDU1': {'ram': 'vdu1_flavor_ram_change'}}
vdu_image_dict = {'VDU1': 'vdu1_image_uuid_change'}
cpd_vl_dict = {'CP1': {'cp1_network_id_change_1',
'cp1_network_id_change_2'}}
expected_final_param_dict = {
'nfv': {
'CP': {
'CP1': {
'network': {
'cp1_network_id_change_1',
'cp1_network_id_change_2'
}
}
},
'VDU': {
'VDU1': {
'image': 'vdu1_image_uuid_change',
'flavor': {
'ram': 'vdu1_flavor_ram_change'
}
}
}
}
}
actual_final_param_dict = utils.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
self.assertEqual(expected_final_param_dict, actual_final_param_dict)
def test_create_final_param_dict_empty_value(self):
initial_param_dict = {'nfv': {'VDU': '', 'CP': ''}}
expected_final_param_dict = {'nfv': {'VDU': '', 'CP': ''}}
vdu_flavor_dict = {}
vdu_image_dict = {}
cpd_vl_dict = {}
actual_final_param_dict = utils.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
self.assertEqual(expected_final_param_dict, actual_final_param_dict)
def test_create_final_param_dict_empty_argument(self):
initial_param_dict = {}
expected_final_param_dict = {}
vdu_flavor_dict = {}
vdu_image_dict = {}
cpd_vl_dict = {}
actual_final_param_dict = utils.create_final_param_dict(
initial_param_dict, vdu_flavor_dict, vdu_image_dict, cpd_vl_dict)
self.assertEqual(expected_final_param_dict, actual_final_param_dict)
def test_create_vdu_flavor_dict(self):
vnfd_dict = self._read_file('vnfd_lcm_user_data.yaml')
test_vnfd_dict = {'VNF': {}, 'VDU1':
{'ram': 512, 'vcpus': 1, 'disk': 1}, 'CP1': {}}
vdu_flavor_dict = utils.create_vdu_flavor_dict(vnfd_dict)
self.assertEqual(test_vnfd_dict, vdu_flavor_dict)
def test_create_vdu_flavor_dict_empty_argument(self):
vnfd_dict = {}
vdu_flavor_dict = utils.create_vdu_flavor_dict(vnfd_dict)
self.assertEqual({}, vdu_flavor_dict)
def test_create_vdu_image_dict(self):
vnf_resource = type('', (), {})
resource_identifier = constants.INVALID_UUID
vnf_resource.resource_identifier = resource_identifier
grant_info = {'vdu_name': {vnf_resource}}
vdu_image_dict = utils.create_vdu_image_dict(grant_info)
self.assertEqual({'vdu_name': resource_identifier}, vdu_image_dict)
def test_create_vdu_image_dict_empty_argument(self):
grant_info = {}
vdu_image_dict = utils.create_vdu_image_dict(grant_info)
self.assertEqual({}, vdu_image_dict)
def test_create_cpd_vl_dict(self):
base_hot_dict = {'resources': {'dummy_cpd_id': "101010_d"}}
inst_req_info = instantiate_vnf_req.InstantiateVnfRequest()
ext_virtual_links_test_value = instantiate_vnf_req.ExtVirtualLinkData()
ext_virtual_links_test_value.resource_id = 'dummy_resource_id'
ext_virtual_links_ext_cps = []
ext_virtual_links_ext_cps_value = instantiate_vnf_req.VnfExtCpData()
ext_virtual_links_ext_cps_value.cpd_id = 'dummy_cpd_id'
ext_virtual_links_ext_cps.append(ext_virtual_links_ext_cps_value)
ext_virtual_links_test_value.ext_cps = ext_virtual_links_ext_cps
inst_req_info.ext_virtual_links.append(ext_virtual_links_test_value)
cpd_vl_dict = utils.create_cpd_vl_dict(base_hot_dict, inst_req_info)
self.assertEqual({'dummy_cpd_id': 'dummy_resource_id'}, cpd_vl_dict)
def test_create_cpd_vl_dict_no_cp_resource(self):
base_hot_dict = {'resources': {'dummy_cpd_id': "101010_d"}}
inst_req_info = instantiate_vnf_req.InstantiateVnfRequest()
ext_virtual_links_test_value = instantiate_vnf_req.ExtVirtualLinkData()
ext_virtual_links_test_value.resource_id = 'dummy_resource_id'
ext_virtual_links_ext_cps = []
ext_virtual_links_ext_cps_value = instantiate_vnf_req.VnfExtCpData()
ext_virtual_links_ext_cps_value.cpd_id = ""
ext_virtual_links_ext_cps.append(ext_virtual_links_ext_cps_value)
ext_virtual_links_test_value.ext_cps = ext_virtual_links_ext_cps
inst_req_info.ext_virtual_links.append(ext_virtual_links_test_value)
cpd_vl_dict = utils.create_cpd_vl_dict(base_hot_dict, inst_req_info)
self.assertEqual({}, cpd_vl_dict)
def test_create_cpd_vl_dict_empty_argument(self):
base_hot_dict = {}
inst_req_info = type('', (), {})
inst_req_info.ext_virtual_links = None
cpd_vl_dict = utils.create_cpd_vl_dict(base_hot_dict, inst_req_info)
self.assertEqual({}, cpd_vl_dict)

25
tacker/vnflcm/utils.py
View File

@ -729,3 +729,28 @@ def _convert_desired_capacity(inst_level_id, vnfd_dict, vdu):
desired_capacity = initial_delta + delta_num * level_num
return desired_capacity
def _get_vnf_package_path(context, vnfd_id):
vnf_package_id = _get_vnf_package_id(context, vnfd_id)
vnf_package_base_path = cfg.CONF.vnf_package.vnf_package_csar_path
vnf_package_path = vnf_package_base_path + '/' + vnf_package_id
return vnf_package_path
def _get_base_hot_dict(context, vnfd_id):
vnf_package_id = _get_vnf_package_id(context, vnfd_id)
vnf_package_base_path = cfg.CONF.vnf_package.vnf_package_csar_path
vnf_package_csar_path = vnf_package_base_path + '/' + vnf_package_id
base_hot_dir = 'BaseHOT'
ext = [".yaml", ".yml"]
base_hot_path = vnf_package_csar_path + '/' + base_hot_dir
base_hot_dict = None
if os.path.exists(base_hot_path):
for file in os.listdir(base_hot_path):
if file.endswith(tuple(ext)):
source_file_path = os.path.join(base_hot_path, file)
base_hot_dict = yaml.safe_load(open(source_file_path))
LOG.debug("Loaded base hot: %s", base_hot_dict)
return base_hot_dict

8
tacker/vnflcm/vnflcm_driver.py
View File

@ -137,6 +137,12 @@ class VnfLcmDriver(abstract_driver.VnfInstanceAbstractDriver):
instantiate_vnf_req):
vnfd_dict = vnflcm_utils._get_vnfd_dict(context, vnf_instance.vnfd_id,
instantiate_vnf_req.flavour_id)
base_hot_dict = vnflcm_utils._get_base_hot_dict(
context, vnf_instance.vnfd_id)
vnf_package_path = None
if base_hot_dict is not None:
vnf_package_path = vnflcm_utils._get_vnf_package_path(
context, vnf_instance.vnfd_id)
param_for_subs_map = vnflcm_utils._get_param_data(vnfd_dict,
instantiate_vnf_req)
@ -167,6 +173,8 @@ class VnfLcmDriver(abstract_driver.VnfInstanceAbstractDriver):
context=context, vnf_instance=vnf_instance,
vnfd_dict=final_vnf_dict, grant_response=vnf_resources,
vim_connection_info=vim_connection_info,
base_hot_dict=base_hot_dict,
vnf_package_path=vnf_package_path,
instantiate_vnf_req=instantiate_vnf_req)
except Exception as exp:
with excutils.save_and_reraise_exception():

145
tacker/vnfm/infra_drivers/openstack/openstack.py
View File

@ -15,8 +15,13 @@
# under the License.
import eventlet
import importlib
import os
import sys
import time
from collections import OrderedDict
from oslo_config import cfg
from oslo_log import log as logging
from oslo_serialization import jsonutils
@ -31,6 +36,7 @@ from tacker.common import utils
from tacker.extensions import vnflcm
from tacker.extensions import vnfm
from tacker import objects
from tacker.tosca.utils import represent_odict
from tacker.vnfm.infra_drivers import abstract_driver
from tacker.vnfm.infra_drivers.openstack import constants as infra_cnst
from tacker.vnfm.infra_drivers.openstack import glance_client as gc
@ -38,7 +44,9 @@ from tacker.vnfm.infra_drivers.openstack import heat_client as hc
from tacker.vnfm.infra_drivers.openstack import translate_template
from tacker.vnfm.infra_drivers.openstack import vdu
from tacker.vnfm.infra_drivers import scale_driver
from tacker.vnfm.lcm_user_data.constants import USER_DATA_TIMEOUT
eventlet.monkey_patch(time=True)
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
@ -106,17 +114,140 @@ class OpenStack(abstract_driver.VnfAbstractDriver,
@log.log
def create(self, plugin, context, vnf, auth_attr,
base_hot_dict=None, vnf_package_path=None,
inst_req_info=None, grant_info=None):
LOG.debug('vnf %s', vnf)
region_name = vnf.get('placement_attr', {}).get('region_name', None)
heatclient = hc.HeatClient(auth_attr, region_name)
additional_param = None
if inst_req_info is not None:
additional_param = inst_req_info.additional_params
user_data_path = None
user_data_class = None
if additional_param is not None:
LOG.debug('additional_param: %s', additional_param)
user_data_path = additional_param.get(
'lcm-operation-user-data')
user_data_class = additional_param.get(
'lcm-operation-user-data-class')
LOG.debug('UserData path: %s', user_data_path)
LOG.debug('UserData class: %s', user_data_class)
if user_data_path is not None and user_data_class is not None:
LOG.info('Execute user data and create heat-stack.')
if base_hot_dict is None:
error_reason = _("failed to get Base HOT.")
raise vnfm.LCMUserDataFailed(reason=error_reason)
vnfd_str = vnf['vnfd']['attributes']['vnfd']
vnfd_dict = yaml.safe_load(vnfd_str)
LOG.debug('VNFD: %s', vnfd_dict)
LOG.debug('VNF package path: %s', vnf_package_path)
sys.path.append(vnf_package_path)
user_data_module = os.path.splitext(
user_data_path.lstrip('./'))[0].replace('/', '.')
LOG.debug('UserData module: %s', user_data_module)
LOG.debug('Append sys.path: %s', sys.path)
try:
module = importlib.import_module(user_data_module)
LOG.debug('Append sys.modules: %s', sys.modules)
except Exception:
self._delete_user_data_module(user_data_module)
error_reason = _(
"failed to get UserData path based on "
"lcm-operation-user-data from additionalParams.")
raise vnfm.LCMUserDataFailed(reason=error_reason)
finally:
sys.path.remove(vnf_package_path)
LOG.debug('Remove sys.path: %s', sys.path)
try:
klass = getattr(module, user_data_class)
except Exception:
self._delete_user_data_module(user_data_module)
error_reason = _(
"failed to get UserData class based on "
"lcm-operation-user-data-class from additionalParams.")
raise vnfm.LCMUserDataFailed(reason=error_reason)
# Set the timeout and execute the UserData script.
hot_param_dict = None
with eventlet.timeout.Timeout(USER_DATA_TIMEOUT, False):
try:
hot_param_dict = klass.instantiate(
base_hot_dict, vnfd_dict, inst_req_info, grant_info)
except Exception:
raise
finally:
self._delete_user_data_module(user_data_module)
if hot_param_dict is not None:
LOG.info('HOT input parameter: %s', hot_param_dict)
else:
error_reason = _(
"fails due to timeout[sec]: %s") % USER_DATA_TIMEOUT
raise vnfm.LCMUserDataFailed(reason=error_reason)
if not isinstance(hot_param_dict, dict):
error_reason = _(
"return value as HOT parameter from UserData "
"is not in dict format.")
raise vnfm.LCMUserDataFailed(reason=error_reason)
# Create heat-stack with BaseHOT and parameters
stack = self._create_stack_with_user_data(
heatclient, vnf, base_hot_dict, hot_param_dict)
elif user_data_path is None and user_data_class is None:
LOG.info('Execute heat-translator and create heat-stack.')
tth = translate_template.TOSCAToHOT(vnf, heatclient,
inst_req_info, grant_info)
tth.generate_hot()
stack = self._create_stack(heatclient, tth.vnf, tth.fields)
else:
error_reason = _(
"failed to get lcm-operation-user-data or "
"lcm-operation-user-data-class from additionalParams.")
raise vnfm.LCMUserDataFailed(reason=error_reason)
tth = translate_template.TOSCAToHOT(vnf, heatclient,
inst_req_info, grant_info)
tth.generate_hot()
stack = self._create_stack(heatclient, tth.vnf, tth.fields)
return stack['stack']['id']
@log.log
def _delete_user_data_module(self, user_data_module):
# Delete module recursively.
mp_list = user_data_module.split('.')
while True:
del_module = '.'.join(mp_list)
print(del_module)
if del_module in sys.modules:
del sys.modules[del_module]
if len(mp_list) == 1:
break
mp_list = mp_list[0:-1]
LOG.debug('Remove sys.modules: %s', sys.modules)
@log.log
def _create_stack_with_user_data(self, heatclient, vnf,
base_hot_dict, hot_param_dict):
fields = {}
fields['stack_name'] = ("vnflcm_" + vnf["id"])
fields['template'] = self._format_base_hot(base_hot_dict)
fields['parameters'] = hot_param_dict
LOG.debug('fields: %s', fields)
LOG.debug('template: %s', fields['template'])
stack = heatclient.create(fields)
return stack
@log.log
def _format_base_hot(self, base_hot_dict):
yaml.SafeDumper.add_representer(OrderedDict,
lambda dumper, value: represent_odict(dumper,
u'tag:yaml.org,2002:map', value))
return yaml.safe_dump(base_hot_dict)
@log.log
def _create_stack(self, heatclient, vnf, fields):
if 'stack_name' not in fields:
@ -603,7 +734,8 @@ class OpenStack(abstract_driver.VnfAbstractDriver,
def instantiate_vnf(self, context, vnf_instance, vnfd_dict,
vim_connection_info, instantiate_vnf_req,
grant_response):
grant_response,
base_hot_dict=None, vnf_package_path=None):
access_info = vim_connection_info.access_info
region_name = access_info.get('region')
placement_attr = vnfd_dict.get('placement_attr', {})
@ -611,7 +743,8 @@ class OpenStack(abstract_driver.VnfAbstractDriver,
vnfd_dict['placement_attr'] = placement_attr
instance_id = self.create(None, context, vnfd_dict,
access_info, inst_req_info=instantiate_vnf_req,
access_info, base_hot_dict, vnf_package_path,
inst_req_info=instantiate_vnf_req,
grant_info=grant_response)
vnfd_dict['instance_id'] = instance_id
return instance_id

0
tacker/vnfm/lcm_user_data/__init__.py Normal file
View File

30
tacker/vnfm/lcm_user_data/abstract_user_data.py Normal file
View File

@ -0,0 +1,30 @@
#
# 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 abc
import six
from tacker.extensions import vnfm
@six.add_metaclass(abc.ABCMeta)
class AbstractUserData(object):
@staticmethod
@abc.abstractmethod
def instantiate(vnfd_dict=None,
vdu_image_dict=None,
cp_vl_dict=None):
error_reason = _(
"failed to execute UserData because not implemented.")
raise vnfm.LCMUserDataFailed(reason=error_reason)

14
tacker/vnfm/lcm_user_data/constants.py Normal file
View File

@ -0,0 +1,14 @@
#
# 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.
USER_DATA_TIMEOUT = 60

172
tacker/vnfm/lcm_user_data/utils.py Normal file
View File

@ -0,0 +1,172 @@
#
# 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 oslo_log import log as logging
from tacker.common.utils import MemoryUnit
"""Define util functions that can be used in UserData.
As for how to use the function (`create _ * _ dict`), dict can be obtained
as a return value by setting the required arguments and calling.
For detailed usage, check the docstring of each function.
NOTE: The functions defined here are limited to common and basic ones.
Please note that not all conversion logics are offered in Tacker,
different from Heat-Translator.
"""
LOG = logging.getLogger(__name__)
HOT_NOVA_SERVER = 'OS::Nova::Server'
HOT_NOVA_FLAVOR = 'OS::Nova::Flavor'
HOT_NEUTRON_PORT = 'OS::Neutron::Port'
SUPPORTED_HOT_TYPE = [HOT_NOVA_SERVER, HOT_NOVA_FLAVOR, HOT_NEUTRON_PORT]
def create_initial_param_dict(base_hot_dict):
"""Create initial dict containing information about get_param resources.
:param base_hot_dict: dict(Base HOT dict format)
:return: dict('nfv', Initial HOT resource dict)
NOTE: 'nfv' is a fixed value for 1st element.
'VDU' and 'CP' are supported for 2nd element.
3rd and 4th element are mandatory.
"""
initial_param_dict = {
'nfv': {
'VDU': {
},
'CP': {
}
}
}
resources = base_hot_dict.get('resources', {})
for resource_name, resource_val in resources.items():
resource_type = resource_val.get('type')
if resource_type in SUPPORTED_HOT_TYPE:
resource_props = resource_val.get('properties', {})
for prop_key, prop_val in resource_props.items():
if isinstance(prop_val, dict) and 'get_param' in prop_val:
param_list = prop_val['get_param']
if len(param_list) == 4:
resource_info = initial_param_dict.get(
param_list[0], {}).get(
param_list[1], {})
if param_list[2] not in resource_info:
resource_info[param_list[2]] = {}
LOG.info('initial_param_dict: %s', initial_param_dict)
return initial_param_dict
def create_final_param_dict(param_dict, vdu_flavor_dict,
vdu_image_dict, cpd_vl_dict):
"""Create final dict containing information about HOT input parameter.
:param param_dict: dict('nfv', Initial HOT resource dict)
:param vdu_flavor_dict: dict(VDU name, VDU flavor dict)
:param vdu_iamge_dict: dict(VDU name, Glance-image uuid)
:param cpd_vl_dict: dict(external CPD ID, Neutron-network uuid)
:return: dict('nfv', Final HOT resource dict)
"""
final_param_dict = copy.deepcopy(param_dict)
vdus = final_param_dict.get('nfv', {}).get('VDU', {})
for target_vdu in vdus:
vdus[target_vdu]['flavor'] = vdu_flavor_dict.get(target_vdu)
vdus[target_vdu]['image'] = vdu_image_dict.get(target_vdu)
cps = final_param_dict.get('nfv', {}).get('CP', {})
for target_cp in cps:
cps[target_cp]['network'] = cpd_vl_dict.get(target_cp)
LOG.info('final_param_dict: %s', final_param_dict)
return final_param_dict
def create_vdu_flavor_dict(vnfd_dict):
"""Create a dict containing information about VDU's flavor.
:param vnfd_dict: dict(VNFD dict format)
:return: dict(VDU name, VDU flavor dict)
"""
vdu_flavor_dict = {}
node_templates = vnfd_dict.get(
'topology_template', {}).get(
'node_templates', {})
for vdu_name, val in node_templates.items():
vdu_flavor_props = val.get(
'capabilities', {}).get(
'virtual_compute', {}).get('properties', {})
if vdu_flavor_props is not {}:
flavor_dict = {}
for key, val in vdu_flavor_props.items():
if key == 'virtual_cpu':
flavor_dict['vcpus'] = val['num_virtual_cpu']
elif key == 'virtual_memory':
# Convert to MiB
flavor_dict['ram'] = MemoryUnit.convert_unit_size_to_num(
val['virtual_mem_size'], 'MiB')
elif key == 'virtual_local_storage':
# Convert to GiB
flavor_dict['disk'] = MemoryUnit.convert_unit_size_to_num(
val[0]['size_of_storage'], 'GiB')
vdu_flavor_dict[vdu_name] = flavor_dict
LOG.info('vdu_flavor_dict: %s', vdu_flavor_dict)
return vdu_flavor_dict
def create_vdu_image_dict(grant_info):
"""Create a dict containing information about VDU's image.
:param grant_info: dict(Grant information format)
:return: dict(VDU name, Glance-image uuid)
"""
vdu_image_dict = {}
for vdu_name, resources in grant_info.items():
for vnf_resource in resources:
vdu_image_dict[vdu_name] = vnf_resource.resource_identifier
LOG.info('vdu_image_dict: %s', vdu_image_dict)
return vdu_image_dict
def create_cpd_vl_dict(base_hot_dict, inst_req_info):
"""Create a dict containing information about CPD and VL.
:param base_hot_dict: dict(Base HOT dict format)
:param inst_req_info: dict(Instantiation request information format)
:return: dict(external CPD ID, Neutron-network uuid)
"""
cpd_vl_dict = {}
ext_vls = inst_req_info.ext_virtual_links
if ext_vls is not None:
for ext_vl in ext_vls:
ext_cps = ext_vl.ext_cps
vl_uuid = ext_vl.resource_id
for ext_cp in ext_cps:
cp_resource = base_hot_dict['resources'].get(
ext_cp.cpd_id)
if cp_resource is None:
continue
cpd_vl_dict[ext_cp.cpd_id] = vl_uuid
LOG.info('cpd_vl_dict: %s', cpd_vl_dict)
return cpd_vl_dict