openstack/tacker
Code Issues Proposed changes

Create a VNF Package Generation Tool

Browse Source 
This patch creates a tool to generate VNF Packages
for the following VIM types.
- ETSINFV.OPENSTACK_KEYSTONE.V_3
- ETSINFV.KUBERNETES.V_1
- ETSINFV.HELM.V_3

Each VNF Package is created based on the following VNF Packages.
- samples/tests/functional/
  - sol_v2_common/userdata_standard
  - sol_kubernetes_v2/test_instantiate_cnf_resources
  - sol_kubernetes_v2/test_helm_instantiate

This patch also refactors pkggen.py, which generates VNF packages
for these FTs, and the associated FTs.

Change-Id: I19fe633ae00f70ddf5fff7223f6bfe2afae80489
This commit is contained in:
Ai Hamano 2024-07-31 11:15:58 +09:00
parent 7be3fdf2a7
commit dd7d27e325
30 changed files with 802 additions and 194 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

305
doc/source/reference/gen_vnf_pkg.rst Normal file
View File

@ -0,0 +1,305 @@
===========================
VNF Package Generation Tool
===========================
For upload VNF Package, it is required to provide a zip file
via ``openstack`` command.
Here is an example of uploading a VNF Package named as
``sample_vnf_package_csar.zip`` with a zip file.
.. code-block:: console
$ openstack vnf package upload \
--path ./sample_vnf_package_csar/sample_vnf_package_csar.zip \
6e6b7a6d-0ebe-4085-96c2-b34269d837f9
You can prepare a zip file using a dedicated tool.
Prerequisites
-------------
This tool uses the following networks:
* ``net0`` network
* ``subnet0`` subnetwork for net0
* ``net1`` network
* ``subnet1`` subnetwork for net1
* ``net_mgmt`` network
These networks are created by default when you install Tacker
via devstack.
If these networks do not exist in your OpenStack environment,
please create them manually before running this tool.
You can check the networks with the following command.
.. code-block:: console
$ openstack net list
+--------------------------------------+-----------------+----------------------------------------------------------------------------+
| ID | Name | Subnets |
+--------------------------------------+-----------------+----------------------------------------------------------------------------+
| 348d1921-0cca-4730-815a-cc58c503ee65 | public | 64290109-665b-442f-83d4-0a98b8301af8, d963ddfd-3ab1-4320-9efd-311049d92575 |
| 3b069dc1-4247-4063-89dd-0898efa04384 | private | 69f64379-e72f-433d-be01-ff6723e8e955, 9e0c260c-78f7-4f15-9157-40f823433449 |
| 41ebfff9-df23-4690-bef9-a0e76f618947 | net1 | e24618bb-c7bc-4771-a9d8-eba53db3c0ae |
| ab598736-9fc9-470d-b894-b0a4d2f9d46c | lb-mgmt-net | b21032be-c681-46ac-b249-8d3be320f830 |
| d9fc27d5-d881-4cf2-9023-d9cff9df64ed | k8s-pod-net | cd4269f6-9152-4707-95d4-825b722edcf0 |
| dbe64c96-1334-4983-9f60-a90905af6ff0 | net0 | 23737d05-5e3f-4af8-8c4d-83221f794787 |
| eeb398f0-818b-4c75-9573-8f50fa5de501 | net_mgmt | bfa8e9d2-039d-4e59-85a2-8a9801d90cfd |
| ef55ac46-bb38-4abe-bdf7-2b9d3be02266 | k8s-service-net | 0d98b3b8-a6fd-4044-8acf-f0630755f956 |
+--------------------------------------+-----------------+----------------------------------------------------------------------------+
$ openstack subnet list
+--------------------------------------+-------------------------+--------------------------------------+---------------------+
| ID | Name | Network | Subnet |
+--------------------------------------+-------------------------+--------------------------------------+---------------------+
| 0d98b3b8-a6fd-4044-8acf-f0630755f956 | k8s-service-subnet-IPv4 | ef55ac46-bb38-4abe-bdf7-2b9d3be02266 | 10.0.0.128/26 |
| 23737d05-5e3f-4af8-8c4d-83221f794787 | subnet0 | dbe64c96-1334-4983-9f60-a90905af6ff0 | 10.10.0.0/24 |
| 64290109-665b-442f-83d4-0a98b8301af8 | ipv6-public-subnet | 348d1921-0cca-4730-815a-cc58c503ee65 | 2001:db8::/64 |
| 69f64379-e72f-433d-be01-ff6723e8e955 | private-subnet | 3b069dc1-4247-4063-89dd-0898efa04384 | 10.0.0.0/26 |
| 9e0c260c-78f7-4f15-9157-40f823433449 | ipv6-private-subnet | 3b069dc1-4247-4063-89dd-0898efa04384 | fd5f:5cb9:4f13::/64 |
| b21032be-c681-46ac-b249-8d3be320f830 | lb-mgmt-subnet | ab598736-9fc9-470d-b894-b0a4d2f9d46c | 192.168.0.0/24 |
| bfa8e9d2-039d-4e59-85a2-8a9801d90cfd | subnet_mgmt | eeb398f0-818b-4c75-9573-8f50fa5de501 | 192.168.120.0/24 |
| cd4269f6-9152-4707-95d4-825b722edcf0 | k8s-pod-subnet-IPv4 | d9fc27d5-d881-4cf2-9023-d9cff9df64ed | 10.0.0.64/26 |
| d963ddfd-3ab1-4320-9efd-311049d92575 | public-subnet | 348d1921-0cca-4730-815a-cc58c503ee65 | 172.24.4.0/24 |
| e24618bb-c7bc-4771-a9d8-eba53db3c0ae | subnet1 | 41ebfff9-df23-4690-bef9-a0e76f618947 | 10.10.1.0/24 |
+--------------------------------------+-------------------------+--------------------------------------+---------------------+
Usage
-----
By specifying the VIM Type as an option, a sample of the corresponding
VNF Package for Tacker v2 API is generated.
This tool support the following VIM Types:
* ETSINFV.OPENSTACK_KEYSTONE.V_3
* ETSINFV.KUBERNETES.V_1
* ETSINFV.HELM.V_3
In this document, TACKER_ROOT is the root of tacker's repository.
.. code-block:: console
$ python3 -m pip install TAKCER_ROOT
$ export PYTHONPATH=TAKCER_ROOT
$ cd TAKCER_ROOT/tools
$ python3 gen_vnf_pkg.py -h
usage: gen_vnf_pkg.py [-h] -t VIM_TYPE
Create VNF Package zip and parameter files
options:
-h, --help show this help message and exit
-t VIM_TYPE, --type VIM_TYPE
specify the vim type
* ETSINFV.OPENSTACK_KEYSTONE.V_3
* ETSINFV.KUBERNETES.V_1
* ETSINFV.HELM.V_3
The output of this tool is as follows:
* Specified VIM Type
* Generated zip file name
* List of file names included in the generated zip file
.. code-block:: console
$ python3 gen_vnf_pkg.py -t ETSINFV.OPENSTACK_KEYSTONE.V_3
VIM type = ETSINFV.OPENSTACK_KEYSTONE.V_3
Zip file: userdata_standard.zip
--------------------------------------------------
BaseHOT/
Definitions/
Files/
Scripts/
TOSCA-Metadata/
UserData/
Files/images/
Files/images/cirros-0.5.2-x86_64-disk.img
Scripts/coordinate_vnf.py
Scripts/sample_script.py
TOSCA-Metadata/TOSCA.meta
UserData/userdata_standard.py
BaseHOT/simple/
BaseHOT/simple/nested/
BaseHOT/simple/sample3.yaml
BaseHOT/simple/nested/VDU1.yaml
BaseHOT/simple/nested/VDU2.yaml
Definitions/v2_sample3_types.yaml
Definitions/v2_sample3_top.vnfd.yaml
Definitions/etsi_nfv_sol001_vnfd_types.yaml
Definitions/etsi_nfv_sol001_common_types.yaml
Definitions/v2_sample3_df_simple.yaml
--------------------------------------------------
$ python3 gen_vnf_pkg.py -t ETSINFV.KUBERNETES.V_1
VIM type: ETSINFV.KUBERNETES.V_1
Zip file: test_instantiate_cnf_resources.zip
--------------------------------------------------
Definitions/
Files/
Scripts/
TOSCA-Metadata/
Files/kubernetes/
Files/kubernetes/controller-revision.yaml
Files/kubernetes/role_rolebinding_SA.yaml
Files/kubernetes/pod-template.yaml
Files/kubernetes/deployment.yaml
Files/kubernetes/statefulset.yaml
Files/kubernetes/multiple_yaml_priority-class.yaml
Files/kubernetes/persistent-volume-0.yaml
Files/kubernetes/storage-class_pv_pvc.yaml
Files/kubernetes/multiple_yaml_network-policy.yaml
Files/kubernetes/subject-access-review.yaml
Files/kubernetes/self-subject-access-review_and_self-subject-rule-review.yaml
Files/kubernetes/bindings.yaml
Files/kubernetes/pod.yaml
Files/kubernetes/daemon-set.yaml
Files/kubernetes/job.yaml
Files/kubernetes/persistent-volume-1.yaml
Files/kubernetes/horizontal-pod-autoscaler.yaml
Files/kubernetes/multiple_yaml_lease.yaml
Files/kubernetes/namespace.yaml
Files/kubernetes/clusterrole_clusterrolebinding_SA.yaml
Files/kubernetes/storage-class.yaml
Files/kubernetes/limit-range.yaml
Files/kubernetes/local-subject-access-review.yaml
Files/kubernetes/replicaset_service_secret.yaml
Files/kubernetes/resource-quota.yaml
Files/kubernetes/deployment_fail_test.yaml
Files/kubernetes/token-review.yaml
Files/kubernetes/config-map.yaml
Scripts/sample_script.py
TOSCA-Metadata/TOSCA.meta
Definitions/sample_cnf_df_simple.yaml
Definitions/etsi_nfv_sol001_vnfd_types.yaml
Definitions/etsi_nfv_sol001_common_types.yaml
Definitions/sample_cnf_top.vnfd.yaml
Definitions/sample_cnf_types.yaml
--------------------------------------------------
$ python3 gen_vnf_pkg.py -t ETSINFV.HELM.V_3
VIM type = ETSINFV.HELM.V_3
Zip file: test_helm_instantiate.zip
--------------------------------------------------
Definitions/
Files/
Scripts/
TOSCA-Metadata/
Files/kubernetes/
Files/kubernetes/test-chart/
Files/kubernetes/test-chart-0.1.0.tgz
Files/kubernetes/test-chart/templates/
Files/kubernetes/test-chart/Chart.yaml
Files/kubernetes/test-chart/values.yaml
Files/kubernetes/test-chart/.helmignore
Files/kubernetes/test-chart/templates/service.yaml
Files/kubernetes/test-chart/templates/deployment_vdu2.yaml
Files/kubernetes/test-chart/templates/NOTES.txt
Files/kubernetes/test-chart/templates/serviceaccount.yaml
Files/kubernetes/test-chart/templates/_helpers.tpl
Files/kubernetes/test-chart/templates/deployment_vdu1.yaml
Scripts/sample_script.py
TOSCA-Metadata/TOSCA.meta
Definitions/sample_cnf_df_simple.yaml
Definitions/etsi_nfv_sol001_vnfd_types.yaml
Definitions/etsi_nfv_sol001_common_types.yaml
Definitions/sample_cnf_top.vnfd.yaml
Definitions/sample_cnf_types.yaml
--------------------------------------------------
This tool generates a VNF Package zip file and a sample request file
for each VIM Type under the output directory.
.. code-block:: console
$ ls output/
helm_instantiate test_instantiate_cnf_resources userdata_standard
$ ls output/userdata_standard/
change_ext_conn_req create_req heal_req instantiate_req scale_in_req
scale_out_req terminate_req update_req userdata_standard.zip
$ ls output/test_instantiate_cnf_resources/
create_req max_sample_heal max_sample_instantiate max_sample_scale_in
max_sample_scale_out max_sample_terminate test_instantiate_cnf_resources.zip
$ ls output/helm_instantiate
create_req helm_heal helm_instantiate_req helm_scale_in helm_scale_out
helm_terminate_req test_helm_instantiate.zip
For the following request files, ``endpoint``, ``ssl_ca_cert`` and
``bearer_token`` need to be changed by your own k8s cluster information.
* max_sample_instantiate for ETSINFV.KUBERNETES.V_1
* helm_instantiate_req for ETSINFV.HELM.V_3
.. note::
``ssl_ca_cert`` needs to be on one line as shown below.
.. code-block:: json
"ssl_ca_cert": "-----BEGIN CERTIFICATE-----\nMIIDB...BH\n3bkddspNikO1\n-----END CERTIFICATE-----\n"
Please note that line breaks are changed to '``\n``'.
You can also set your own k8s cluster information to ``auth_url``,
``barere_token``, and ``ssl_ca_cert`` in gen_vnf_pkg.py before running this tool.
.. note::
If you use a VIM that is already registered,
modify vimConnectionInfo as follows.
.. code-block:: json
"vimConnectionInfo": {
"vim1": {
"vimId": "REGISTERED_VIM_ID",
"vimType": "VIM_TYPE"
}
}
For the following request files, ``vnfcInstanceId`` need
to be changed with target vnfcInfo id.
* heal_req for ETSINFV.OPENSTACK_KEYSTONE.V_3
* max_sample_heal for ETSINFV.KUBERNETES.V_1
* helm_heal for ETSINFV.HELM.V_3
And for the following request file, ``vnfdId`` need
to be changed with target VNFD id.
* update_req for ETSINFV.OPENSTACK_KEYSTONE.V_3
.. note::
This tool generates a zip file and a request file based on the following
used in FT as a sample VNF Package.
* ETSINFV.OPENSTACK_KEYSTONE.V_3:
`samples/tests/functional/sol_v2_common/userdata_standard`_
* ETSINFV.KUBERNETES.V_1:
`samples/tests/functional/sol_kubernetes_v2/test_instantiate_cnf_resources`_
* ETSINFV.HELM.V_3:
`samples/tests/functional/sol_kubernetes_v2/test_helm_instantiate`_
Please note that if FT is changed, the output of this tool may also change.
.. _samples/tests/functional/sol_v2_common/userdata_standard:
https://opendev.org/openstack/tacker/src/branch/master/samples/tests/functional/sol_v2_common/userdata_standard
.. _samples/tests/functional/sol_kubernetes_v2/test_instantiate_cnf_resources:
https://opendev.org/openstack/tacker/src/branch/master/samples/tests/functional/sol_kubernetes_v2/test_instantiate_cnf_resources
.. _samples/tests/functional/sol_kubernetes_v2/test_helm_instantiate:
https://opendev.org/openstack/tacker/src/branch/master/samples/tests/functional/sol_kubernetes_v2/test_helm_instantiate

1
doc/source/reference/index.rst
View File

@ -25,3 +25,4 @@ Tools
:maxdepth: 1
vim_config
gen_vnf_pkg

6
samples/mgmt_driver/kubernetes/sol_v2_kubernetes_vnf_package/pkggen.py
View File

@ -18,6 +18,8 @@ import os
import shutil
import tempfile
from tacker.tests import utils
# NOTE: This package is a sample for vnflcm v2 API to deploy kubernetes cluster
# with cilium CNI using the management driver.
@ -42,9 +44,7 @@ shutil.copytree(os.path.join(".", "contents"), tmp_contents)
# add userdata script
# tacker/tests/sol_refactored/infra_drivers/openstack/userdata_standard.py
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
# mkdir UserData/ and copy userdata_path into it
file_path = os.path.join(tmp_contents, "UserData")
os.makedirs(file_path)

11
samples/tests/functional/sol_kubernetes_v2/test_cnf_container_update_after/pkggen.py
View File

@ -21,19 +21,16 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
mgmt_driver_path = utils.mgmt_drivers("container_update_mgmt_v2.py")
# samples/tests/functional/sol_kubernetes_v2/{package_name}
cur_dir = os.path.dirname(__file__)
mgmt_driver_path = os.path.join(
cur_dir, '../../../../../tacker/sol_refactored',
'mgmt_drivers/container_update_mgmt_v2.py')
utils.make_zip(".", tmp_dir, vnfd_id, mgmt_driver=mgmt_driver_path)
common_utils.make_zip(".", tmp_dir, vnfd_id, mgmt_driver=mgmt_driver_path)
shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)

11
samples/tests/functional/sol_kubernetes_v2/test_cnf_container_update_before/pkggen.py
View File

@ -21,19 +21,16 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
mgmt_driver_path = utils.mgmt_drivers("container_update_mgmt_v2.py")
# samples/tests/functional/sol_kubernetes_v2/{package_name}
cur_dir = os.path.dirname(__file__)
mgmt_driver_path = os.path.join(
cur_dir, '../../../../../tacker/sol_refactored',
'mgmt_drivers/container_update_mgmt_v2.py')
utils.make_zip(".", tmp_dir, vnfd_id, mgmt_driver=mgmt_driver_path)
common_utils.make_zip(".", tmp_dir, vnfd_id, mgmt_driver=mgmt_driver_path)
shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)

20
samples/tests/functional/sol_kubernetes_v2/test_helm_instantiate/pkggen.py
View File

@ -14,30 +14,22 @@
# under the License.
import json
import os
import shutil
import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_kubernetes_v2 import zipgen
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
vnfd_id, zip_path = zipgen.test_helm_instantiate()
# samples/tests/functional/sol_kubernetes_v2/{package_name}
utils.make_zip(".", tmp_dir, vnfd_id)
shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
shutil.move(zip_path, ".")
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = paramgen.test_helm_instantiate_create(vnfd_id)
# if you instantiate with all k8s resource
# please change auth_url and bear_token to your own k8s cluster's info
# please change auth_url, bearer_token and ssl_ca_cert
# to your own k8s cluster's info
auth_url = "https://127.0.0.1:6443"
bearer_token = "your_k8s_cluster_bearer_token"
ssl_ca_cert = "k8s_ssl_ca_cert"

23
samples/tests/functional/sol_kubernetes_v2/test_instantiate_cnf_resources/pkggen.py
View File

@ -14,34 +14,27 @@
# under the License.
import json
import os
import shutil
import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_kubernetes_v2 import zipgen
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
vnfd_id, zip_path = zipgen.test_instantiate_cnf_resources()
# samples/tests/functional/sol_kubernetes_v2/{package_name}
utils.make_zip(".", tmp_dir, vnfd_id)
shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
shutil.move(zip_path, ".")
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = paramgen.test_instantiate_cnf_resources_create(vnfd_id)
# if you instantiate with all k8s resource
# please change auth_url and bear_token to your own k8s cluster's info
# please change auth_url, bearer_token and ssl_ca_cert
# to your own k8s cluster's info
auth_url = "https://127.0.0.1:6443"
bearer_token = "your_k8s_cluster_bearer_token"
ssl_ca_cert = "k8s_ssl_ca_cert"
max_sample_instantiate = paramgen.max_sample_instantiate(
auth_url, bearer_token)
auth_url, bearer_token, ssl_ca_cert)
max_sample_terminate = paramgen.max_sample_terminate()
max_sample_scale_out = paramgen.max_sample_scale_out()

20
samples/tests/functional/sol_v2_common/basic_lcms_max_individual_vnfc/pkggen.py
View File

@ -20,8 +20,8 @@ import string
import tempfile
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
SUPPORT_STRING_FOR_VNFD_ID = string.ascii_letters + string.digits + "-._ "
@ -29,16 +29,13 @@ vnfd_id = SUPPORT_STRING_FOR_VNFD_ID + "max_vnfd_id"
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.move(os.path.join(tmp_dir, zip_file_name), ".")
@ -72,10 +69,11 @@ print('#####################################################################\n'
'# vnfc ids should be changed in heal req files by show vnf manually.#\n'
'#####################################################################')
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt', 'ft-net0',
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt', 'ft-net0',
'ft-net1'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1', 'ft-ipv4-subnet0',
'ft-ipv6-subnet0', 'ft-ipv4-subnet1', 'ft-ipv6-subnet1'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1',
'ft-ipv4-subnet0', 'ft-ipv6-subnet0', 'ft-ipv4-subnet1',
'ft-ipv6-subnet1'])
instantiate_req = paramgen.instantiate_vnf_max(
net_ids, subnet_ids, None, "http://localhost/identity/v3", user_data=True)

10
samples/tests/functional/sol_v2_common/basic_lcms_min_individual_vnfc/pkggen.py
View File

@ -20,7 +20,8 @@ import string
import tempfile
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
SUPPORT_STRING_FOR_VNFD_ID = string.ascii_letters + string.digits + "-._ "
@ -29,12 +30,9 @@ zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, userdata_path=userdata_path)
common_utils.make_zip(".", tmp_dir, vnfd_id, userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)

17
samples/tests/functional/sol_v2_common/change_vnfpkg_max_individual_vnfc/pkggen.py
View File

@ -20,7 +20,8 @@ import string
import tempfile
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
SUPPORT_STRING_FOR_VNFD_ID = string.ascii_letters + string.digits + "-._ "
@ -29,12 +30,9 @@ zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, userdata_path=userdata_path)
common_utils.make_zip(".", tmp_dir, vnfd_id, userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
@ -48,10 +46,11 @@ print('#####################################################################\n'
'# after testing, run post.py and delete them. #\n'
'#####################################################################')
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt', 'ft-net0',
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt', 'ft-net0',
'ft-net1'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1', 'ft-ipv4-subnet0',
'ft-ipv6-subnet0', 'ft-ipv4-subnet1', 'ft-ipv6-subnet1'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1',
'ft-ipv4-subnet0', 'ft-ipv6-subnet0', 'ft-ipv4-subnet1',
'ft-ipv6-subnet1'])
change_vnf_pkg_individual_vnfc_max = (
paramgen.change_vnf_pkg_individual_vnfc_max(vnfd_id, net_ids, subnet_ids))

13
samples/tests/functional/sol_v2_common/change_vnfpkg_or_update_min_individual_vnfc/pkggen.py
View File

@ -20,8 +20,8 @@ import string
import tempfile
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
SUPPORT_STRING_FOR_VNFD_ID = string.ascii_letters + string.digits + "-._ "
@ -29,16 +29,13 @@ vnfd_id = SUPPORT_STRING_FOR_VNFD_ID + "upd_new_min_vnfd_id"
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")

13
samples/tests/functional/sol_v2_common/update_vnf_max_individual_vnfc/pkggen.py
View File

@ -20,8 +20,8 @@ import string
import tempfile
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
SUPPORT_STRING_FOR_VNFD_ID = string.ascii_letters + string.digits + "-._ "
@ -29,16 +29,13 @@ vnfd_id = SUPPORT_STRING_FOR_VNFD_ID + "upd_max_vnfd_id"
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")

26
samples/tests/functional/sol_v2_common/userdata_standard/pkggen.py
View File

@ -14,35 +14,17 @@
# under the License.
import json
import os
import shutil
import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import zipgen
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
vnfd_id, zip_path = zipgen.userdata_standard()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
shutil.move(zip_path, ".")
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = paramgen.sample3_create(vnfd_id)
terminate_req = paramgen.sample3_terminate()

17
samples/tests/functional/sol_v2_common/userdata_standard_az_retry/pkggen.py
View File

@ -21,24 +21,21 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
@ -47,8 +44,8 @@ shutil.rmtree(tmp_dir)
create_req = paramgen.sample6_create(vnfd_id)
terminate_req = paramgen.sample6_terminate()
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1'])
instantiate_req = paramgen.sample6_instantiate(
net_ids, subnet_ids, "http://localhost/identity/v3")

17
samples/tests/functional/sol_v2_common/userdata_standard_change_vnfpkg/pkggen.py
View File

@ -21,31 +21,28 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1'])
change_vnfpkg_req = paramgen.sample4_change_vnfpkg(
vnfd_id, net_ids, subnet_ids)

17
samples/tests/functional/sol_v2_common/userdata_standard_change_vnfpkg_nw/pkggen.py
View File

@ -21,31 +21,28 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
shutil.rmtree(tmp_dir)
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1'])
change_vnfpkg_req = paramgen.sample5_change_vnfpkg(
vnfd_id, net_ids, subnet_ids)

17
samples/tests/functional/sol_v2_common/userdata_standard_with_non_boot_volume/pkggen.py
View File

@ -21,24 +21,21 @@ import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests import utils as test_utils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
zip_file_name = os.path.basename(os.path.abspath(".")) + '.zip'
tmp_dir = tempfile.mkdtemp()
vnfd_id = uuidutils.generate_uuid()
image_path = test_utils.test_etc_sample("etsi/nfv/common/Files/images",
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/sol_refactored/infra_drivers/openstack/userdata_standard.py
# samples/tests/functional/sol_v2_common/sampleX
userdata_dir = "../../../../../tacker/sol_refactored/infra_drivers/openstack/"
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(userdata_dir + userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
common_utils.make_zip(".", tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
shutil.copy(os.path.join(tmp_dir, zip_file_name), ".")
@ -47,8 +44,8 @@ shutil.rmtree(tmp_dir)
create_req = paramgen.sample7_create(vnfd_id)
terminate_req = paramgen.sample7_terminate()
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
net_ids = common_utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = common_utils.get_subnet_ids(['subnet0', 'subnet1'])
instantiate_req = paramgen.sample7_instantiate(
net_ids, subnet_ids, "http://localhost/identity/v3")

37
tacker/tests/functional/base_v2.py
View File

@ -168,6 +168,43 @@ class BaseTackerTestV2(base.BaseTestCase):
return pkg_id, vnfd_id
@classmethod
def upload_vnf_package(cls, zip_file_path, nfvo=False, user_data={}):
if nfvo:
return zip_file_path
path = "/vnfpkgm/v1/vnf_packages"
req_body = {'userDefinedData': user_data}
resp, body = cls.tacker_client.do_request(
path, "POST", expected_status=[201], body=req_body)
pkg_id = body['id']
with open(zip_file_path, 'rb') as fp:
path = f"/vnfpkgm/v1/vnf_packages/{pkg_id}/package_content"
resp, body = cls.tacker_client.do_request(
path, "PUT", body=fp, content_type='application/zip',
expected_status=[202])
# wait for onboard
timeout = VNF_PACKAGE_UPLOAD_TIMEOUT
start_time = int(time.time())
path = f"/vnfpkgm/v1/vnf_packages/{pkg_id}"
while True:
resp, body = cls.tacker_client.do_request(
path, "GET", expected_status=[200])
if body['onboardingState'] == "ONBOARDED":
break
if int(time.time()) - start_time > timeout:
raise Exception("Failed to onboard vnf package")
time.sleep(RETRY_WAIT_TIME)
shutil.rmtree(zip_file_path.rsplit('/', 1)[0])
return pkg_id
def get_vnf_package(self, pkg_id):
path = f"/vnfpkgm/v1/vnf_packages/{pkg_id}"
resp, body = self.tacker_client.do_request(path, "GET")

6
tacker/tests/functional/sol_kubernetes_v2/base_v2.py
View File

@ -84,6 +84,12 @@ class BaseVnfLcmKubernetesV2Test(base_v2.BaseTackerTestV2):
namespace=namespace,
mgmt_driver=mgmt_driver)
@classmethod
def upload_vnf_package(cls, zip_file_path, nfvo=False, user_data={}):
return super().upload_vnf_package(zip_file_path, nfvo=nfvo,
user_data=user_data)
@classmethod
def get_controller_tacker_ip(cls):
cur_dir = os.path.dirname(__file__)

7
tacker/tests/functional/sol_kubernetes_v2/test_helm.py
View File

@ -18,6 +18,7 @@ import time
from tacker.tests.functional.sol_kubernetes_v2 import base_v2
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests.functional.sol_kubernetes_v2 import zipgen
from tacker.tests import utils
WAIT_LCMOCC_UPDATE_TIME = 3
@ -29,10 +30,8 @@ class VnfLcmHelmTest(base_v2.BaseVnfLcmKubernetesV2Test):
def setUpClass(cls):
super(VnfLcmHelmTest, cls).setUpClass()
test_helm_instantiate_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_helm_instantiate")
cls.helm_pkg, cls.helm_vnfd_id = cls.create_vnf_package(
test_helm_instantiate_path)
cls.helm_vnfd_id, zip_path = zipgen.test_helm_instantiate()
cls.helm_pkg = cls.upload_vnf_package(zip_path)
test_helm_change_vnf_pkg_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_helm_change_vnf_pkg")

5
tacker/tests/functional/sol_kubernetes_v2/test_kubernetes_container_update.py
View File

@ -25,13 +25,10 @@ class VnfLcmKubernetesContainerUpdate(base_v2.BaseVnfLcmKubernetesV2Test):
@classmethod
def setUpClass(cls):
super(VnfLcmKubernetesContainerUpdate, cls).setUpClass()
cur_dir = os.path.dirname(__file__)
test_cnf_container_update_before_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_cnf_container_update_before")
mgmt_driver_path = os.path.join(
cur_dir,
"../../../sol_refactored/mgmt_drivers/container_update_mgmt_v2.py")
mgmt_driver_path = utils.mgmt_drivers("container_update_mgmt_v2.py")
cls.vnf_package_id_before, cls.vnfd_id_before = cls.create_vnf_package(
test_cnf_container_update_before_path,
mgmt_driver=mgmt_driver_path)

8
tacker/tests/functional/sol_kubernetes_v2/test_vnflcm_basic.py
View File

@ -18,7 +18,7 @@ import os
from tacker.tests.functional.sol_kubernetes_v2 import base_v2
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
from tacker.tests import utils
from tacker.tests.functional.sol_kubernetes_v2 import zipgen
@ddt.ddt
@ -28,10 +28,8 @@ class VnfLcmKubernetesTest(base_v2.BaseVnfLcmKubernetesV2Test):
def setUpClass(cls):
super(VnfLcmKubernetesTest, cls).setUpClass()
test_instantiate_cnf_resources_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_instantiate_cnf_resources")
cls.cnf_pkg, cls.cnf_vnfd_id = cls.create_vnf_package(
test_instantiate_cnf_resources_path)
cls.cnf_vnfd_id, zip_path = zipgen.test_instantiate_cnf_resources()
cls.cnf_pkg = cls.upload_vnf_package(zip_path)
@classmethod
def tearDownClass(cls):

53
tacker/tests/functional/sol_kubernetes_v2/zipgen.py Normal file
View File

@ -0,0 +1,53 @@
# Copyright (C) 2024 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
def test_instantiate_cnf_resources(vnfd_id=None):
if vnfd_id is None:
vnfd_id = uuidutils.generate_uuid()
tmp_dir = tempfile.mkdtemp()
sample_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_instantiate_cnf_resources")
common_utils.make_zip(sample_path, tmp_dir, vnfd_id)
zip_file_name = f"{os.path.basename(os.path.abspath(sample_path))}.zip"
zip_file_path = os.path.join(tmp_dir, zip_file_name)
return vnfd_id, zip_file_path
def test_helm_instantiate(vnfd_id=None):
if vnfd_id is None:
vnfd_id = uuidutils.generate_uuid()
tmp_dir = tempfile.mkdtemp()
sample_path = utils.test_sample(
"functional/sol_kubernetes_v2/test_helm_instantiate")
common_utils.make_zip(sample_path, tmp_dir, vnfd_id)
zip_file_name = f"{os.path.basename(os.path.abspath(sample_path))}.zip"
zip_file_path = os.path.join(tmp_dir, zip_file_name)
return vnfd_id, zip_file_path

17
tacker/tests/functional/sol_v2/basic/test_individual_vnfc_mgmt.py
View File

@ -16,10 +16,10 @@
import os
import subprocess
from tacker.common import utils as tacker_utils
import tacker.conf
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import test_vnflcm_basic_common
from tacker.tests.functional.sol_v2_common import zipgen
from tacker.tests import utils
@ -49,20 +49,11 @@ class IndividualVnfcMgmtTest(test_vnflcm_basic_common.CommonVnfLcmTest):
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/tests/functional/sol_v2(here)
# /sol_refactored
cur_dir = os.path.dirname(__file__)
userdata_dir = os.path.join(
cur_dir, "{}/tacker/sol_refactored/infra_drivers/openstack".format(
tacker_utils.proj_root()))
userdata_file = "userdata_standard.py"
userdata_path = os.path.join(userdata_dir, userdata_file)
userdata_path = utils.userdata("userdata_standard.py")
# main vnf package for StandardUserData test
pkg_path_1 = utils.test_sample("functional/sol_v2_common",
"userdata_standard")
cls.standard_pkg, cls.standard_vnfd_id = cls.create_vnf_package(
pkg_path_1, image_path=image_path, userdata_path=userdata_path)
cls.standard_vnfd_id, zip_path = zipgen.userdata_standard()
cls.standard_pkg = cls.upload_vnf_package(zip_path)
# for change_vnfpkg test
pkg_path_2 = utils.test_sample("functional/sol_v2_common",

9
tacker/tests/functional/sol_v2_az_retry/test_az_retry.py
View File

@ -28,14 +28,7 @@ class AzRetryTest(test_vnflcm_basic_common.CommonVnfLcmTest):
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/tests/functional/sol_v2_az_retry(here)
# /sol_refactored
cur_dir = os.path.dirname(__file__)
userdata_dir = os.path.join(
cur_dir, "../../../sol_refactored/infra_drivers/openstack")
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(
os.path.join(userdata_dir, userdata_file))
userdata_path = utils.userdata("userdata_standard.py")
# for update_stack_retry test
pkg_path_1 = utils.test_sample(

6
tacker/tests/functional/sol_v2_common/base_v2.py
View File

@ -56,6 +56,12 @@ class BaseSolV2Test(base_v2.BaseTackerTestV2):
userdata_path=userdata_path,
provider=provider, vnfd_id=vnfd_id)
@classmethod
def upload_vnf_package(cls, zip_file_path, nfvo=False, user_data={}):
return super().upload_vnf_package(zip_file_path, nfvo=nfvo,
user_data=user_data)
def get_network_ids(self, networks):
path = "/v2.0/networks"
resp, body = self.neutron_client.do_request(path, "GET")

44
tacker/tests/functional/sol_v2_common/zipgen.py Normal file
View File

@ -0,0 +1,44 @@
# Copyright (C) 2024 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import tempfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_v2_common import utils as common_utils
from tacker.tests import utils
def userdata_standard(vnfd_id=None):
if vnfd_id is None:
vnfd_id = uuidutils.generate_uuid()
tmp_dir = tempfile.mkdtemp()
sample_path = utils.test_sample(
"functional/sol_v2_common/userdata_standard")
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
userdata_path = utils.userdata("userdata_standard.py")
common_utils.make_zip(sample_path, tmp_dir, vnfd_id, image_path=image_path,
userdata_path=userdata_path)
zip_file_name = f"{os.path.basename(os.path.abspath(sample_path))}.zip"
zip_file_path = os.path.join(tmp_dir, zip_file_name)
return vnfd_id, zip_file_path

9
tacker/tests/functional/sol_v2_individual_vnfc_mgmt/test_individual_vnfc_mgmt_basic.py
View File

@ -34,17 +34,10 @@ class IndividualVnfcMgmtBasicTest(base_v2.BaseSolV2Test):
@classmethod
def setUpClass(cls):
super(IndividualVnfcMgmtBasicTest, cls).setUpClass()
cur_dir = os.path.dirname(__file__)
image_path = utils.test_etc_sample("etsi/nfv/common/Files/images",
"cirros-0.5.2-x86_64-disk.img")
# tacker/tests/functional/sol_v2_individual_vnfc_mgmt(here)
# /sol_refactored
userdata_dir = os.path.join(
cur_dir, "../../../sol_refactored/infra_drivers/openstack")
userdata_file = "userdata_standard.py"
userdata_path = os.path.abspath(
os.path.join(userdata_dir, userdata_file))
userdata_path = utils.userdata("userdata_standard.py")
# vnf package for basic lcms tests max pattern
pkg_path_1 = utils.test_sample("functional/sol_v2_common",

25
tacker/tests/utils.py
View File

@ -26,14 +26,15 @@ import zipfile
from oslo_log import log as logging
from oslo_utils import uuidutils
from tacker.common import utils
LOG = logging.getLogger(__name__)
def sample_root():
return os.path.abspath(os.path.join(os.path.dirname(__file__),
'../../samples'))
return os.path.abspath(os.path.join(utils.proj_root(),
'samples'))
def _test_sample_root():
@ -50,6 +51,26 @@ def test_etc_sample(*p):
return test_sample('etc/samples', *p)
def _userdata_root():
# {tacker_root}/tacker/sol_refactored/infra_drivers/openstack
return os.path.abspath(os.path.join(utils.proj_root(),
'tacker/sol_refactored/infra_drivers/openstack'))
def userdata(*p):
return os.path.join(_userdata_root(), *p)
def _mgmt_drivers_root():
# {tacker_root}/tacker/sol_refactored/mgmt_drivers
return os.path.abspath(os.path.join(utils.proj_root(),
'tacker/sol_refactored/mgmt_drivers'))
def mgmt_drivers(*p):
return os.path.join(_mgmt_drivers_root(), *p)
def read_file(input_file):
yaml_file = test_etc_sample(input_file)
with open(yaml_file, 'r') as f:

226
tools/gen_vnf_pkg.py Normal file
View File

@ -0,0 +1,226 @@
# Copyright (C) 2024 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import argparse
import glob
import json
import os
import shutil
import tempfile
import textwrap
import yaml
import zipfile
from oslo_utils import uuidutils
from tacker.tests.functional.sol_kubernetes_v2 import paramgen as k8s_paramgen
from tacker.tests.functional.sol_kubernetes_v2 import zipgen as k8s_zipgen
from tacker.tests.functional.sol_v2_common import paramgen
from tacker.tests.functional.sol_v2_common import utils
from tacker.tests.functional.sol_v2_common import zipgen
STANDARD_OUTPUT="./output/userdata_standard/"
K8S_OUTPUT="./output/test_instantiate_cnf_resources/"
HELM_OUTPUT="./output/helm_instantiate/"
parser = argparse.ArgumentParser(
formatter_class=argparse.RawTextHelpFormatter,
description='Create VNF Package zip and parameter files',
add_help=True
)
parser.add_argument('-t', '--type',
help=textwrap.dedent('''\
specify the vim type
* ETSINFV.OPENSTACK_KEYSTONE.V_3
* ETSINFV.KUBERNETES.V_1
* ETSINFV.HELM.V_3
'''),
type=str, metavar='VIM_TYPE',
choices=['ETSINFV.OPENSTACK_KEYSTONE.V_3', 'ETSINFV.KUBERNETES.V_1', 'ETSINFV.HELM.V_3'],
required=True)
args = parser.parse_args()
if args.type == 'ETSINFV.OPENSTACK_KEYSTONE.V_3':
print(f"VIM type = {args.type}")
os.makedirs(STANDARD_OUTPUT, exist_ok=True)
vnfd_id, zip_path = zipgen.userdata_standard()
print(f"Zip file: {zip_path.rsplit('/', 1)[1]}")
shutil.move(zip_path, STANDARD_OUTPUT)
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = paramgen.sample3_create(vnfd_id)
terminate_req = paramgen.sample3_terminate()
net_ids = utils.get_network_ids(['net0', 'net1', 'net_mgmt'])
subnet_ids = utils.get_subnet_ids(['subnet0', 'subnet1'])
instantiate_req = paramgen.sample3_instantiate(
net_ids, subnet_ids, "http://localhost/identity/v3")
scale_out_req = paramgen.sample3_scale_out()
scale_in_req = paramgen.sample3_scale_in()
heal_req = paramgen.sample3_heal()
change_ext_conn_req = paramgen.sample3_change_ext_conn(net_ids)
update_req = paramgen.sample3_update_vnf_vnfd_id("replace real vnfd id")
with open(f"{STANDARD_OUTPUT}create_req", "w") as f:
f.write(json.dumps(create_req, indent=2))
with open(f"{STANDARD_OUTPUT}terminate_req", "w") as f:
f.write(json.dumps(terminate_req, indent=2))
with open(f"{STANDARD_OUTPUT}instantiate_req", "w") as f:
f.write(json.dumps(instantiate_req, indent=2))
with open(f"{STANDARD_OUTPUT}scale_out_req", "w") as f:
f.write(json.dumps(scale_out_req, indent=2))
with open(f"{STANDARD_OUTPUT}scale_in_req", "w") as f:
f.write(json.dumps(scale_in_req, indent=2))
# NOTE: vnfcInstanceId should be filled by hand
with open(f"{STANDARD_OUTPUT}heal_req", "w") as f:
f.write(json.dumps(heal_req, indent=2))
with open(f"{STANDARD_OUTPUT}change_ext_conn_req", "w") as f:
f.write(json.dumps(change_ext_conn_req, indent=2))
with open(f"{STANDARD_OUTPUT}update_req", "w") as f:
f.write(json.dumps(update_req, indent=2))
print("--------------------------------------------------")
zip_file = zipfile.ZipFile(f"{STANDARD_OUTPUT}userdata_standard.zip")
files = zip_file.namelist()
for file in files:
print(file)
zip_file.close()
print("--------------------------------------------------")
elif args.type == 'ETSINFV.KUBERNETES.V_1':
print(f"VIM type: {args.type}")
os.makedirs(K8S_OUTPUT, exist_ok=True)
vnfd_id, zip_path = k8s_zipgen.test_instantiate_cnf_resources()
print(f"Zip file: {zip_path.rsplit('/', 1)[1]}")
shutil.move(zip_path, K8S_OUTPUT)
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = k8s_paramgen.test_instantiate_cnf_resources_create(vnfd_id)
# please change auth_url, bearer_token and ssl_ca_cert
# to your own k8s cluster's info
auth_url = "https://127.0.0.1:6443"
bearer_token = "your_k8s_cluster_bearer_token"
ssl_ca_cert = "k8s_ssl_ca_cert"
max_sample_instantiate = k8s_paramgen.max_sample_instantiate(
auth_url, bearer_token, ssl_ca_cert)
max_sample_terminate = k8s_paramgen.max_sample_terminate()
max_sample_scale_out = k8s_paramgen.max_sample_scale_out()
max_sample_scale_in = k8s_paramgen.max_sample_scale_in()
max_sample_heal = k8s_paramgen.max_sample_heal(["replace real vnfc ids"])
with open(f"{K8S_OUTPUT}create_req", "w", encoding='utf-8') as f:
f.write(json.dumps(create_req, indent=2))
with open(f"{K8S_OUTPUT}max_sample_instantiate", "w", encoding='utf-8') as f:
f.write(json.dumps(max_sample_instantiate, indent=2))
with open(f"{K8S_OUTPUT}max_sample_terminate", "w", encoding='utf-8') as f:
f.write(json.dumps(max_sample_terminate, indent=2))
with open(f"{K8S_OUTPUT}max_sample_scale_out", "w", encoding='utf-8') as f:
f.write(json.dumps(max_sample_scale_out, indent=2))
with open(f"{K8S_OUTPUT}max_sample_scale_in", "w", encoding='utf-8') as f:
f.write(json.dumps(max_sample_scale_in, indent=2))
with open(f"{K8S_OUTPUT}max_sample_heal", "w", encoding='utf-8') as f:
f.write(json.dumps(max_sample_heal, indent=2))
print("--------------------------------------------------")
zip_file = zipfile.ZipFile(f"{K8S_OUTPUT}test_instantiate_cnf_resources.zip")
files = zip_file.namelist()
for file in files:
print(file)
zip_file.close()
print("--------------------------------------------------")
elif args.type == 'ETSINFV.HELM.V_3':
print(f"VIM type = {args.type}")
os.makedirs(HELM_OUTPUT, exist_ok=True)
vnfd_id, zip_path = k8s_zipgen.test_helm_instantiate()
print(f"Zip file: {zip_path.rsplit('/', 1)[1]}")
shutil.move(zip_path, HELM_OUTPUT)
shutil.rmtree(zip_path.rsplit('/', 1)[0])
create_req = k8s_paramgen.test_helm_instantiate_create(vnfd_id)
# please change auth_url, bearer_token and ssl_ca_cert
# to your own k8s cluster's info
auth_url = "https://127.0.0.1:6443"
bearer_token = "your_k8s_cluster_bearer_token"
ssl_ca_cert = "k8s_ssl_ca_cert"
helm_instantiate_req = k8s_paramgen.helm_instantiate(
auth_url, bearer_token, ssl_ca_cert)
helm_terminate_req = k8s_paramgen.helm_terminate()
helm_scale_out = k8s_paramgen.helm_scale_out()
helm_scale_in = k8s_paramgen.helm_scale_in()
helm_heal = k8s_paramgen.helm_heal(["replace real vnfc ids"])
with open(f"{HELM_OUTPUT}create_req", "w", encoding='utf-8') as f:
f.write(json.dumps(create_req, indent=2))
with open(f"{HELM_OUTPUT}helm_instantiate_req", "w", encoding='utf-8') as f:
f.write(json.dumps(helm_instantiate_req, indent=2))
with open(f"{HELM_OUTPUT}helm_terminate_req", "w", encoding='utf-8') as f:
f.write(json.dumps(helm_terminate_req, indent=2))
with open(f"{HELM_OUTPUT}helm_scale_out", "w", encoding='utf-8') as f:
f.write(json.dumps(helm_scale_out, indent=2))
with open(f"{HELM_OUTPUT}helm_scale_in", "w", encoding='utf-8') as f:
f.write(json.dumps(helm_scale_in, indent=2))
with open(f"{HELM_OUTPUT}helm_heal", "w", encoding='utf-8') as f:
f.write(json.dumps(helm_heal, indent=2))
print("--------------------------------------------------")
zip_file = zipfile.ZipFile(f"{HELM_OUTPUT}test_helm_instantiate.zip")
files = zip_file.namelist()
for file in files:
print(file)
zip_file.close()
print("--------------------------------------------------")