openstack/tacker
Code Issues Proposed changes

Support change_vnfpkg v2 API to change external CP

Browse Source 
This patch enables a change of extVirtualLinks in change_vnfpkg
v2 API. Previously it is not supported but it is available now.

Implements: blueprint enhance-change-package
Change-Id: Id61b7faae31a3cb8182437786eea62d179116956
This commit is contained in:
Itsuro Oda 2022-08-15 07:31:04 +00:00
parent b025a1bfce
commit 4a162a3ffc
11 changed files with 310 additions and 138 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
releasenotes/notes/support-changevnfpkg-to-change-external-cp-336e349c6ac0d7a9.yaml Normal file
View File

@ -0,0 +1,6 @@
---
features:
- |
Support Change Current VNF Package API to change external CP.
We add new functions to Openstack and UserData class, and
``extVirtualLinks`` parameter to ChangeCurrentVnfPkgRequest.

51
tacker/sol_refactored/infra_drivers/openstack/openstack.py
View File

@ -75,10 +75,10 @@ def _get_vdu_idx(vdu_with_idx):
return int(part[2])
def _vdu_with_idx(vdu, vdu_idx):
if vdu_idx is None:
return vdu
return f'{vdu}-{vdu_idx}'
def _rsc_with_idx(rsc, rsc_idx):
if rsc_idx is None:
return rsc
return f'{rsc}-{rsc_idx}'
class Openstack(object):
@ -330,16 +330,16 @@ class Openstack(object):
heat_client)
def _get_flavor_from_vdu_dict(self, vnfc, vdu_dict):
vdu_name = _vdu_with_idx(vnfc.vduId, vnfc.metadata.get('vdu_idx'))
vdu_name = _rsc_with_idx(vnfc.vduId, vnfc.metadata.get('vdu_idx'))
return vdu_dict.get(vdu_name, {}).get('computeFlavourId')
def _get_images_from_vdu_dict(self, vnfc, storage_infos, vdu_dict):
vdu_idx = vnfc.metadata.get('vdu_idx')
vdu_name = _vdu_with_idx(vnfc.vduId, vdu_idx)
vdu_name = _rsc_with_idx(vnfc.vduId, vdu_idx)
vdu_names = [vdu_name]
if vnfc.obj_attr_is_set('storageResourceIds'):
vdu_names += [
_vdu_with_idx(storage_info.virtualStorageDescId, vdu_idx)
_rsc_with_idx(storage_info.virtualStorageDescId, vdu_idx)
for storage_info in storage_infos
if storage_info.id in vnfc.storageResourceIds
]
@ -372,6 +372,7 @@ class Openstack(object):
return templates[parent_stack_id]
vdu_dict = fields['parameters']['nfv']['VDU']
cp_dict = fields['parameters']['nfv']['CP']
stack_name = heat_utils.get_stack_name(inst)
for vnfc in vnfcs:
@ -392,26 +393,39 @@ class Openstack(object):
"template": template
}
LOG.debug("stack fields: %s", vdu_fields)
# NOTE: for VMs with AutoScalingGroups, rolling-update does
# not update CPs one by one. There are updated in once after
# returning this method.
heat_client.update_stack(parent_stack_id, vdu_fields)
else:
# pickup 'vcImageId' and 'computeFlavourId' from vdu_dict
vdu_name = _vdu_with_idx(vnfc.vduId,
vnfc.metadata.get('vdu_idx'))
params = {}
vdu_idx = vnfc.metadata.get('vdu_idx')
vdu_name = _rsc_with_idx(vnfc.vduId, vdu_idx)
new_vdus = {}
flavor = self._get_flavor_from_vdu_dict(vnfc, vdu_dict)
if flavor:
params[vdu_name] = {'computeFlavourId': flavor}
new_vdus[vdu_name] = {'computeFlavourId': flavor}
storage_infos = (
inst.instantiatedVnfInfo.virtualStorageResourceInfo
if vnfc.obj_attr_is_set('storageResourceIds') else [])
images = self._get_images_from_vdu_dict(vnfc,
storage_infos, vdu_dict)
for vdu_name, image in images.items():
params.setdefault(vdu_name, {})
params[vdu_name]['vcImageId'] = image
update_nfv_dict = {'nfv': {'VDU': params}}
update_fields = {'parameters': update_nfv_dict}
new_vdus.setdefault(vdu_name, {})
new_vdus[vdu_name]['vcImageId'] = image
# pickup 'CP' updates
cp_names = vnfd.get_vdu_cps(
inst.instantiatedVnfInfo.flavourId, vnfc.vduId)
new_cps = {}
for cp_name in cp_names:
cp_name = _rsc_with_idx(cp_name, vdu_idx)
if cp_name in cp_dict:
new_cps[cp_name] = cp_dict[cp_name]
update_fields = {
'parameters': {'nfv': {'VDU': new_vdus, 'CP': new_cps}}
}
update_fields = self._update_fields(heat_client, stack_name,
update_fields)
LOG.debug("stack fields: %s", update_fields)
@ -727,6 +741,10 @@ class Openstack(object):
req_ext_vls = grant.extVirtualLinks
elif req.obj_attr_is_set('extVirtualLinks'):
req_ext_vls = req.extVirtualLinks
else:
# may happen in case of change_vnfpkg
return self._make_ext_vl_info_from_inst(old_inst_info,
ext_cp_infos)
req_ext_vl_ids = {ext_vl.id for ext_vl in req_ext_vls}
inst_ext_vl_ids = set()
@ -1077,7 +1095,8 @@ class Openstack(object):
flavour_id, req, grant)
else:
old_inst_vnf_info = inst.instantiatedVnfInfo
if (op == v2fields.LcmOperationType.CHANGE_EXT_CONN and
if ((op == v2fields.LcmOperationType.CHANGE_EXT_CONN or
op == v2fields.LcmOperationType.CHANGE_VNFPKG) and
not is_rollback):
ext_vl_infos = self._make_ext_vl_info_from_req_and_inst(
req, grant, old_inst_vnf_info, ext_cp_infos)

126
tacker/sol_refactored/infra_drivers/openstack/userdata_default.py
View File

@ -20,6 +20,69 @@ from tacker.sol_refactored.common import vnf_instance_utils as inst_utils
from tacker.sol_refactored.infra_drivers.openstack import userdata_utils
def _get_new_cps_from_req(cps, req, grant):
# used by change_ext_conn and change_vnfpkg
new_cps = {}
for cp_name, cp_value in cps.items():
if 'network' in cp_value:
network = common_script_utils.get_param_network(
cp_name, grant, req)
if network is None:
continue
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = common_script_utils.get_param_fixed_ips(
cp_name, grant, req)
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['fixed_ips'] = fixed_ips
return new_cps
def _get_new_cps_from_inst(cps, inst):
# used by change_ext_conn and change_vnfpkg
new_cps = {}
for cp_name, cp_value in cps.items():
if 'network' in cp_value:
network = common_script_utils.get_param_network_from_inst(
cp_name, inst)
if network is None:
continue
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = (
common_script_utils.get_param_fixed_ips_from_inst(
cp_name, inst))
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['fixed_ips'] = fixed_ips
return new_cps
class DefaultUserData(userdata_utils.AbstractUserData):
@staticmethod
@ -165,31 +228,7 @@ class DefaultUserData(userdata_utils.AbstractUserData):
nfv_dict = common_script_utils.init_nfv_dict(top_hot)
cps = nfv_dict.get('CP', {})
new_cps = {}
for cp_name, cp_value in cps.items():
if 'network' in cp_value:
network = common_script_utils.get_param_network(
cp_name, grant, req)
if network is None:
continue
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = common_script_utils.get_param_fixed_ips(
cp_name, grant, req)
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['fixed_ips'] = fixed_ips
new_cps = _get_new_cps_from_req(cps, req, grant)
fields = {'parameters': {'nfv': {'CP': new_cps}}}
@ -211,32 +250,7 @@ class DefaultUserData(userdata_utils.AbstractUserData):
nfv_dict = common_script_utils.init_nfv_dict(top_hot)
cps = nfv_dict.get('CP', {})
new_cps = {}
for cp_name, cp_value in cps.items():
if 'network' in cp_value:
network = common_script_utils.get_param_network_from_inst(
cp_name, inst)
if network is None:
continue
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = (
common_script_utils.get_param_fixed_ips_from_inst(
cp_name, inst))
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name, {})
new_cps[cp_name]['fixed_ips'] = fixed_ips
new_cps = _get_new_cps_from_inst(cps, inst)
fields = {'parameters': {'nfv': {'CP': new_cps}}}
@ -275,8 +289,11 @@ class DefaultUserData(userdata_utils.AbstractUserData):
new_vdus.setdefault(vdu_name, {})
new_vdus[vdu_name]['vcImageId'] = image
cps = nfv_dict.get('CP', {})
new_cps = _get_new_cps_from_req(cps, req, grant)
fields = {
'parameters': {'nfv': {'VDU': new_vdus}}
'parameters': {'nfv': {'VDU': new_vdus, 'CP': new_cps}}
}
return fields
@ -316,8 +333,11 @@ class DefaultUserData(userdata_utils.AbstractUserData):
new_vdus.setdefault(vdu_name, {})
new_vdus[vdu_name]['vcImageId'] = images.get(vdu_name)
cps = nfv_dict.get('CP', {})
new_cps = _get_new_cps_from_inst(cps, inst)
fields = {
'parameters': {'nfv': {'VDU': new_vdus}}
'parameters': {'nfv': {'VDU': new_vdus, 'CP': new_cps}}
}
return fields

130
tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
View File

@ -61,6 +61,71 @@ def add_idx_to_vdu_template(vdu_template, vdu_idx):
return res
def _get_new_cps_from_req(cps, req, grant):
# used by change_ext_conn and change_vnfpkg
new_cps = {}
for cp_name_idx, cp_value in cps.items():
cp_name = rm_idx(cp_name_idx)
if 'network' in cp_value:
network = common_script_utils.get_param_network(
cp_name, grant, req)
if network is None:
continue
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = common_script_utils.get_param_fixed_ips(
cp_name, grant, req)
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['fixed_ips'] = fixed_ips
return new_cps
def _get_new_cps_from_inst(cps, inst):
# used by change_ext_conn_rollback and change_vnfpkg_rollback
new_cps = {}
for cp_name_idx, cp_value in cps.items():
cp_name = rm_idx(cp_name_idx)
if 'network' in cp_value:
network = common_script_utils.get_param_network_from_inst(
cp_name, inst)
if network is None:
continue
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = (
common_script_utils.get_param_fixed_ips_from_inst(
cp_name, inst))
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['fixed_ips'] = fixed_ips
return new_cps
class StandardUserData(userdata_utils.AbstractUserData):
@staticmethod
@ -339,32 +404,7 @@ class StandardUserData(userdata_utils.AbstractUserData):
nfv_dict = common_script_utils.init_nfv_dict(top_hot)
cps = nfv_dict.get('CP', {})
new_cps = {}
for cp_name_idx, cp_value in cps.items():
cp_name = rm_idx(cp_name_idx)
if 'network' in cp_value:
network = common_script_utils.get_param_network(
cp_name, grant, req)
if network is None:
continue
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = common_script_utils.get_param_fixed_ips(
cp_name, grant, req)
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['fixed_ips'] = fixed_ips
new_cps = _get_new_cps_from_req(cps, req, grant)
fields = {'parameters': {'nfv': {'CP': new_cps}}}
@ -395,33 +435,7 @@ class StandardUserData(userdata_utils.AbstractUserData):
nfv_dict = common_script_utils.init_nfv_dict(top_hot)
cps = nfv_dict.get('CP', {})
new_cps = {}
for cp_name_idx, cp_value in cps.items():
cp_name = rm_idx(cp_name_idx)
if 'network' in cp_value:
network = common_script_utils.get_param_network_from_inst(
cp_name, inst)
if network is None:
continue
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['network'] = network
if 'fixed_ips' in cp_value:
ext_fixed_ips = (
common_script_utils.get_param_fixed_ips_from_inst(
cp_name, inst))
fixed_ips = []
for i in range(len(ext_fixed_ips)):
if i not in cp_value['fixed_ips']:
break
ips_i = cp_value['fixed_ips'][i]
if 'subnet' in ips_i:
ips_i['subnet'] = ext_fixed_ips[i].get('subnet')
if 'ip_address' in ips_i:
ips_i['ip_address'] = ext_fixed_ips[i].get(
'ip_address')
fixed_ips.append(ips_i)
new_cps.setdefault(cp_name_idx, {})
new_cps[cp_name_idx]['fixed_ips'] = fixed_ips
new_cps = _get_new_cps_from_inst(cps, inst)
fields = {'parameters': {'nfv': {'CP': new_cps}}}
@ -477,8 +491,11 @@ class StandardUserData(userdata_utils.AbstractUserData):
if 'locationConstraints' in vdu_value:
vdu_value.pop('locationConstraints')
cps = nfv_dict.get('CP', {})
new_cps = _get_new_cps_from_req(cps, req, grant)
fields = {
'parameters': {'nfv': {'VDU': vdus}}
'parameters': {'nfv': {'VDU': vdus, 'CP': new_cps}}
}
return fields
@ -537,8 +554,11 @@ class StandardUserData(userdata_utils.AbstractUserData):
new_vdus.setdefault(vdu_name, {})
new_vdus[vdu_name]['vcImageId'] = images.get(vdu_name)
cps = nfv_dict.get('CP', {})
new_cps = _get_new_cps_from_inst(cps, inst)
fields = {
'parameters': {'nfv': {'VDU': new_vdus}}
'parameters': {'nfv': {'VDU': new_vdus, 'CP': new_cps}}
}
return fields

4
tacker/sol_refactored/objects/v2/change_current_vnf_pkg_request.py
View File

@ -28,16 +28,12 @@ class ChangeCurrentVnfPkgRequest(base.TackerObject,
fields = {
'vnfdId': fields.StringField(nullable=False),
# NOTE: 'extVirtualLinks' is not supported.
# It can be specified but make no effect at all.
'extVirtualLinks': fields.ListOfObjectsField(
'ExtVirtualLinkData', nullable=True),
# NOTE: 'extManagedVirtualLinks' is not supported.
# It can be specified but make no effect at all.
'extManagedVirtualLinks': fields.ListOfObjectsField(
'ExtManagedVirtualLinkData', nullable=True),
# NOTE: 'vimConnectionInfo' is not supported.
# It can be specified but make no effect at all.
'vimConnectionInfo': fields.DictOfObjectsField(
'VimConnectionInfo', nullable=True),
'additionalParams': fields.KeyValuePairsField(nullable=True),

24
tacker/tests/functional/sol_v2/test_individual_vnfc_mgmt.py
View File

@ -261,7 +261,8 @@ class IndividualVnfcMgmtTest(test_vnflcm_basic_common.CommonVnfLcmTest):
self._get_vnfc_cp_net_id(inst_5, 'VDU2', 0, 'VDU2_CP1'))
# 6. Change_vnfpkg operation
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(self.vnfd_id_2)
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(self.vnfd_id_2,
net_ids, subnet_ids)
resp, body = self.change_vnfpkg(inst_id, change_vnfpkg_req)
self.assertEqual(202, resp.status_code)
@ -281,6 +282,19 @@ class IndividualVnfcMgmtTest(test_vnflcm_basic_common.CommonVnfLcmTest):
self._get_vnfc_image(inst_6, 'VDU1', 1))
self.assertNotEqual(self._get_vnfc_image(inst_5, 'VDU2', 0),
self._get_vnfc_image(inst_6, 'VDU2', 0))
# check VDU2_CP1 is changed to net0 from net1. other are not changed.
self.assertEqual(net_ids['net0'],
self._get_vnfc_cp_net_id(inst_5, 'VDU1', 0, 'VDU1_CP1'))
self.assertEqual(net_ids['net0'],
self._get_vnfc_cp_net_id(inst_5, 'VDU1', 1, 'VDU1_CP1'))
self.assertEqual(net_ids['net1'],
self._get_vnfc_cp_net_id(inst_5, 'VDU2', 0, 'VDU2_CP1'))
self.assertEqual(net_ids['net0'],
self._get_vnfc_cp_net_id(inst_6, 'VDU1', 0, 'VDU1_CP1'))
self.assertEqual(net_ids['net0'],
self._get_vnfc_cp_net_id(inst_6, 'VDU1', 1, 'VDU1_CP1'))
self.assertEqual(net_ids['net0'],
self._get_vnfc_cp_net_id(inst_6, 'VDU2', 0, 'VDU2_CP1'))
# Terminate VNF instance
terminate_req = paramgen.sample4_terminate()
@ -396,7 +410,8 @@ class IndividualVnfcMgmtTest(test_vnflcm_basic_common.CommonVnfLcmTest):
# 3. Change_vnfpkg operation
self._put_fail_file('change_vnfpkg')
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(self.vnfd_id_2)
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(self.vnfd_id_2,
net_ids, subnet_ids)
resp, body = self.change_vnfpkg(inst_id, change_vnfpkg_req)
self.assertEqual(202, resp.status_code)
@ -420,6 +435,11 @@ class IndividualVnfcMgmtTest(test_vnflcm_basic_common.CommonVnfLcmTest):
self._get_vnfc_image(inst_3, 'VDU1', 0))
self.assertEqual(self._get_vnfc_image(inst_2, 'VDU2', 0),
self._get_vnfc_image(inst_3, 'VDU2', 0))
# check network of extVL cps are not changed. (i.e. net1)
self.assertEqual(net_ids['net1'],
self._get_vnfc_cp_net_id(inst_3, 'VDU1', 0, 'VDU1_CP1'))
self.assertEqual(net_ids['net1'],
self._get_vnfc_cp_net_id(inst_3, 'VDU2', 0, 'VDU2_CP1'))
# Terminate VNF instance
terminate_req = paramgen.sample3_terminate()

57
tacker/tests/functional/sol_v2_common/paramgen.py
View File

@ -935,6 +935,31 @@ def change_vnfpkg(vnfd_id):
}
def change_vnfpkg_with_ext_vl(vnfd_id, net_ids):
ext_vl_1 = {
"id": uuidutils.generate_uuid(),
"resourceId": net_ids['net1'],
"extCps": [
{
"cpdId": "VDU1_CP1",
"cpConfig": {
"VDU1_CP1_1": {
"cpProtocolData": [{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
"ipAddresses": [{
"type": "IPV4",
"numDynamicAddresses": 1}]}}]}
}
}
]
}
req = change_vnfpkg(vnfd_id)
req["extVirtualLinks"] = [ext_vl_1]
return req
# sample3 is used for tests of StandardUserData
#
def sample3_create(vnfd_id):
@ -953,13 +978,13 @@ def sample3_terminate():
def sample3_instantiate(net_ids, subnet_ids, auth_url):
ext_vl_1 = {
"id": uuidutils.generate_uuid(),
"id": "ext_vl_id_net1",
"resourceId": net_ids['net1'],
"extCps": [
{
"cpdId": "VDU1_CP1",
"cpConfig": {
"VDU1_CP2_1": {
"VDU1_CP1_1": {
"cpProtocolData": [{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
@ -971,7 +996,7 @@ def sample3_instantiate(net_ids, subnet_ids, auth_url):
{
"cpdId": "VDU2_CP1",
"cpConfig": {
"VDU2_CP2_1": {
"VDU2_CP1_1": {
"cpProtocolData": [{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
@ -1055,7 +1080,7 @@ def sample3_change_ext_conn(net_ids):
return {
"extVirtualLinks": [
{
"id": uuidutils.generate_uuid(),
"id": "ext_vl_id_net0",
"resourceId": net_ids['net0'],
"extCps": [
{
@ -1082,9 +1107,31 @@ def sample3_change_ext_conn(net_ids):
# sample4 is for change_vnfpkg test of StandardUserData
#
def sample4_change_vnfpkg(vnfd_id):
def sample4_change_vnfpkg(vnfd_id, net_ids, subnet_ids):
return {
"vnfdId": vnfd_id,
"extVirtualLinks": [
{
"id": "ext_vl_id_net0",
"resourceId": net_ids['net0'],
"extCps": [
{
"cpdId": "VDU2_CP1",
"cpConfig": {
"VDU2_CP1_1": {
"cpProtocolData": [{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
"ipAddresses": [{
"type": "IPV4",
"numDynamicAddresses": 1,
"subnetId":
subnet_ids['subnet0']}]}}]}
}
}
]
}
],
"additionalParams": {
"upgrade_type": "RollingUpdate",
"lcm-operation-coordinate-new-vnf": "./Scripts/coordinate_vnf.py",

8
tacker/tests/functional/sol_v2_common/samples/test_change_vnf_pkg_with_new_image/pkggen.py
View File

@ -40,5 +40,13 @@ shutil.rmtree(tmp_dir)
# if your sample is change VM from image to image
change_vnfpkg_req_from_image_to_image = paramgen.change_vnfpkg(vnfd_id)
net_ids = utils.get_network_ids(['net1'])
change_vnfpkg_req_with_ext_vl = paramgen.change_vnfpkg_with_ext_vl(
vnfd_id, net_ids)
with open("change_vnfpkg_req_from_image_to_image", "w", encoding='utf-8') as f:
f.write(json.dumps(change_vnfpkg_req_from_image_to_image, indent=2))
with open("change_vnfpkg_req_from_image_to_image_with_ext_vl",
"w", encoding='utf-8') as f:
f.write(json.dumps(change_vnfpkg_req_with_ext_vl, indent=2))

8
tacker/tests/functional/sol_v2_common/samples/test_change_vnf_pkg_with_new_volume/pkggen.py
View File

@ -50,3 +50,11 @@ change_vnfpkg_req_error_coodinate_vnf['additionalParams'][
with open("change_vnfpkg_req_error_coodinate_vnf", "w", encoding='utf-8') as f:
f.write(json.dumps(change_vnfpkg_req_error_coodinate_vnf, indent=2))
net_ids = utils.get_network_ids(['net1'])
change_vnfpkg_req_with_ext_vl = paramgen.change_vnfpkg_with_ext_vl(
vnfd_id, net_ids)
with open("change_vnfpkg_req_from_volume_with_ext_vl",
"w", encoding='utf-8') as f:
f.write(json.dumps(change_vnfpkg_req_with_ext_vl, indent=2))

6
tacker/tests/functional/sol_v2_common/samples/userdata_standard_change_vnfpkg/pkggen.py
View File

@ -46,7 +46,11 @@ utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(vnfd_id)
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(
vnfd_id, net_ids, subnet_ids)
with open("change_vnfpkg_req", "w") as f:
f.write(json.dumps(change_vnfpkg_req, indent=2))

28
tacker/tests/functional/sol_v2_common/test_vnflcm_basic_common.py
View File

@ -1612,6 +1612,21 @@ class CommonVnfLcmTest(base_v2.BaseSolV2Test):
resp, body = self.show_subscription(sub_id)
self.assertEqual(404, resp.status_code)
def _get_vnfc_cp_net_id(self, inst, vdu, cp):
for vnfc in inst['instantiatedVnfInfo']['vnfcResourceInfo']:
if vnfc['vduId'] == vdu:
for cp_info in vnfc['vnfcCpInfo']:
if cp_info['cpdId'] == cp:
# must be found
ext_cp_id = cp_info['vnfExtCpId']
break
break
for ext_vl in inst['instantiatedVnfInfo']['extVirtualLinkInfo']:
for port in ext_vl['extLinkPorts']:
if port['cpInstanceId'] == ext_cp_id:
# must be found
return ext_vl['resourceHandle']['resourceId']
def change_vnfpkg_from_image_to_image_common_test(self, is_nfvo=False):
"""Test ChangeCurrentVNFPackage from image to image
@ -1671,13 +1686,15 @@ class CommonVnfLcmTest(base_v2.BaseSolV2Test):
self._register_vnf_package_mock_response(
vnfd_id_1, zip_file_path_1)
create_req = paramgen.change_vnfpkg_create(vnfd_id_1)
change_vnfpkg_req = paramgen.change_vnfpkg(vnfd_id_2)
change_vnfpkg_req = paramgen.change_vnfpkg_with_ext_vl(
vnfd_id_2, self.get_network_ids(['net1']))
else:
glance_image = None
flavour_vdu_dict = None
zone_name_list = None
create_req = paramgen.change_vnfpkg_create(self.vnfd_id_1)
change_vnfpkg_req = paramgen.change_vnfpkg(self.vnfd_id_2)
change_vnfpkg_req = paramgen.change_vnfpkg_with_ext_vl(
self.vnfd_id_2, self.get_network_ids(['net1']))
# 1. Create VNF instance
resp, body = self.create_vnf_instance(create_req)
@ -1735,6 +1752,9 @@ class CommonVnfLcmTest(base_v2.BaseSolV2Test):
self.check_resp_headers_in_get(resp_1)
self.check_resp_body(body_1, expected_inst_attrs)
vdu1_cp1_net_id = self._get_vnfc_cp_net_id(body_1, 'VDU1', 'VDU1_CP1')
self.assertEqual(net_ids['net0'], vdu1_cp1_net_id)
# 4. Change Current VNF Package
if is_nfvo:
self._register_vnf_package_mock_response(
@ -1762,6 +1782,10 @@ class CommonVnfLcmTest(base_v2.BaseSolV2Test):
self.check_resp_headers_in_get(resp_2)
self.check_resp_body(body_2, expected_inst_attrs)
vdu1_cp1_net_id = self._get_vnfc_cp_net_id(body_2, 'VDU1', 'VDU1_CP1')
# changed from net0 to net1
self.assertEqual(net_ids['net1'], vdu1_cp1_net_id)
# 6. Terminate VNF
self._set_grant_response(is_nfvo, 'TERMINATE')
terminate_req = paramgen.terminate_vnf_min()