98d3f4bf31
When initializing k8s client in InfraDriverV2, the SSL CA certificate is set incorrectly. To fix the issue, the following modifies are made in this patch: * A temp file for ssl_ca_cert is created before initializing k8s client and the temp file path is set to k8s_config.ssl_ca_cert, * The temp file is deleted until the lifetime of k8s client ends. Note: This references the implementation in InfraDriverV1. If set the ssl_ca_cert in instantiate request, the validation of request is failed because of the length of ssl_ca_cert exceeds 1024. For this issue, add a new type `keyvalue_pairs_no_length_limit` which has no max length limitation to verify the request. And the interfaceInfo, accessInfo, extra are all set to the new type for unity. In Zuul test environment, when registering default vim, ssl_ca_cert is not set. So the case with ssl_ca_cert is not tested. In this patch ssl_ca_cert is set into the default vim. Closes-Bug: #1979413 Change-Id: I61dbd70690b737a72fc619e5a08b4bab51160a27
267 lines
9.8 KiB
Python
267 lines
9.8 KiB
Python
# Copyright (C) 2022 FUJITSU
|
|
# 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 ddt
|
|
import os
|
|
import time
|
|
|
|
from tacker.tests.functional.sol_kubernetes_v2 import base_v2
|
|
from tacker.tests.functional.sol_kubernetes_v2 import paramgen
|
|
|
|
|
|
@ddt.ddt
|
|
class VnfLcmKubernetesTest(base_v2.BaseVnfLcmKubernetesV2Test):
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
super(VnfLcmKubernetesTest, cls).setUpClass()
|
|
|
|
cur_dir = os.path.dirname(__file__)
|
|
|
|
test_instantiate_cnf_resources_path = os.path.join(
|
|
cur_dir, "samples/test_instantiate_cnf_resources")
|
|
cls.vnf_pkg_1, cls.vnfd_id_1 = cls.create_vnf_package(
|
|
test_instantiate_cnf_resources_path)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
super(VnfLcmKubernetesTest, cls).tearDownClass()
|
|
|
|
cls.delete_vnf_package(cls.vnf_pkg_1)
|
|
|
|
def setUp(self):
|
|
super(VnfLcmKubernetesTest, self).setUp()
|
|
|
|
def test_basic_lcms_max(self):
|
|
"""Test LCM operations with all attributes set
|
|
|
|
* About attributes:
|
|
All of the following cardinality attributes are set.
|
|
In addition, 0..N or 1..N attributes are set to 2 or more.
|
|
- 0..1 (1)
|
|
- 0..N (2 or more)
|
|
- 1
|
|
- 1..N (2 or more)
|
|
|
|
* About LCM operations:
|
|
This test includes the following operations.
|
|
- 1. Create a new VNF instance resource
|
|
- 2. Instantiate a VNF instance
|
|
- 3. Show VNF instance
|
|
- 4. Terminate a VNF instance
|
|
- 5. Delete a VNF instance
|
|
"""
|
|
|
|
# 1. Create a new VNF instance resource
|
|
# NOTE: extensions and vnfConfigurableProperties are omitted
|
|
# because they are commented out in etsi_nfv_sol001.
|
|
expected_inst_attrs = [
|
|
'id',
|
|
'vnfInstanceName',
|
|
'vnfInstanceDescription',
|
|
'vnfdId',
|
|
'vnfProvider',
|
|
'vnfProductName',
|
|
'vnfSoftwareVersion',
|
|
'vnfdVersion',
|
|
# 'vnfConfigurableProperties', # omitted
|
|
# 'vimConnectionInfo', # omitted
|
|
'instantiationState',
|
|
# 'instantiatedVnfInfo', # omitted
|
|
'metadata',
|
|
# 'extensions', # omitted
|
|
'_links'
|
|
]
|
|
create_req = paramgen.test_instantiate_cnf_resources_create(
|
|
self.vnfd_id_1)
|
|
resp, body = self.create_vnf_instance(create_req)
|
|
self.assertEqual(201, resp.status_code)
|
|
self.check_resp_headers_in_create(resp)
|
|
self.check_resp_body(body, expected_inst_attrs)
|
|
inst_id = body['id']
|
|
|
|
# check usageState of VNF Package
|
|
usage_state = self.get_vnf_package(self.vnf_pkg_1)['usageState']
|
|
self.assertEqual('IN_USE', usage_state)
|
|
|
|
# 2. Instantiate a VNF instance
|
|
instantiate_req = paramgen.max_sample_instantiate(
|
|
self.auth_url, self.bearer_token, ssl_ca_cert=self.ssl_ca_cert)
|
|
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
|
|
self.assertEqual(202, resp.status_code)
|
|
self.check_resp_headers_in_operation_task(resp)
|
|
|
|
lcmocc_id = os.path.basename(resp.headers['Location'])
|
|
self.wait_lcmocc_complete(lcmocc_id)
|
|
|
|
# 3. Show VNF instance
|
|
additional_inst_attrs = [
|
|
'vimConnectionInfo',
|
|
'instantiatedVnfInfo'
|
|
]
|
|
expected_inst_attrs.extend(additional_inst_attrs)
|
|
resp, body = self.show_vnf_instance(inst_id)
|
|
self.assertEqual(200, resp.status_code)
|
|
self.check_resp_headers_in_get(resp)
|
|
self.check_resp_body(body, expected_inst_attrs)
|
|
|
|
# check vnfc_resource_info
|
|
vnfc_resource_infos = body['instantiatedVnfInfo'].get(
|
|
'vnfcResourceInfo')
|
|
for vnfc_info in vnfc_resource_infos:
|
|
if vnfc_info['vduId'] == 'VDU1':
|
|
self.assertEqual('Pod', vnfc_info[
|
|
'computeResource']['vimLevelResourceType'])
|
|
elif vnfc_info['vduId'] == 'VDU2':
|
|
self.assertEqual('Deployment', vnfc_info[
|
|
'computeResource']['vimLevelResourceType'])
|
|
elif vnfc_info['vduId'] == 'VDU3':
|
|
self.assertEqual('ReplicaSet', vnfc_info[
|
|
'computeResource']['vimLevelResourceType'])
|
|
elif vnfc_info['vduId'] == 'VDU5':
|
|
self.assertEqual('StatefulSet', vnfc_info[
|
|
'computeResource']['vimLevelResourceType'])
|
|
elif vnfc_info['vduId'] == 'VDU6':
|
|
self.assertEqual('DaemonSet', vnfc_info[
|
|
'computeResource']['vimLevelResourceType'])
|
|
|
|
# 4. Terminate a VNF instance
|
|
terminate_req = paramgen.max_sample_terminate()
|
|
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
|
|
self.assertEqual(202, resp.status_code)
|
|
self.check_resp_headers_in_operation_task(resp)
|
|
|
|
lcmocc_id = os.path.basename(resp.headers['Location'])
|
|
self.wait_lcmocc_complete(lcmocc_id)
|
|
|
|
# wait a bit because there is a bit time lag between lcmocc DB
|
|
# update and terminate completion.
|
|
time.sleep(10)
|
|
|
|
# 5. Delete a VNF instance
|
|
resp, body = self.delete_vnf_instance(inst_id)
|
|
self.assertEqual(204, resp.status_code)
|
|
self.check_resp_headers_in_delete(resp)
|
|
|
|
# check deletion of VNF instance
|
|
resp, body = self.show_vnf_instance(inst_id)
|
|
self.assertEqual(404, resp.status_code)
|
|
|
|
# check usageState of VNF Package
|
|
usage_state = self.get_vnf_package(self.vnf_pkg_1).get('usageState')
|
|
self.assertEqual('NOT_IN_USE', usage_state)
|
|
|
|
def test_basic_lcms_min(self):
|
|
"""Test LCM operations with all attributes set
|
|
|
|
* About attributes:
|
|
Omit except for required attributes.
|
|
Only the following cardinality attributes are set.
|
|
- 1
|
|
- 1..N (1)
|
|
|
|
* About LCM operations:
|
|
This test includes the following operations.
|
|
- 1. Create a new VNF instance resource
|
|
- 2. Instantiate a VNF instance
|
|
- 3. Show VNF instance
|
|
- 4. Terminate a VNF instance
|
|
- 5. Delete a VNF instance
|
|
"""
|
|
|
|
# 1. Create a new VNF instance resource
|
|
# NOTE: extensions and vnfConfigurableProperties are omitted
|
|
# because they are commented out in etsi_nfv_sol001.
|
|
expected_inst_attrs = [
|
|
'id',
|
|
'vnfInstanceName',
|
|
'vnfInstanceDescription',
|
|
'vnfdId',
|
|
'vnfProvider',
|
|
'vnfProductName',
|
|
'vnfSoftwareVersion',
|
|
'vnfdVersion',
|
|
# 'vnfConfigurableProperties', # omitted
|
|
# 'vimConnectionInfo', # omitted
|
|
'instantiationState',
|
|
# 'instantiatedVnfInfo', # omitted
|
|
'metadata',
|
|
# 'extensions', # omitted
|
|
'_links'
|
|
]
|
|
create_req = paramgen.test_instantiate_cnf_resources_create(
|
|
self.vnfd_id_1)
|
|
resp, body = self.create_vnf_instance(create_req)
|
|
self.assertEqual(201, resp.status_code)
|
|
self.check_resp_headers_in_create(resp)
|
|
self.check_resp_body(body, expected_inst_attrs)
|
|
inst_id = body['id']
|
|
|
|
# check usageState of VNF Package
|
|
usage_state = self.get_vnf_package(self.vnf_pkg_1)['usageState']
|
|
self.assertEqual('IN_USE', usage_state)
|
|
|
|
# 2. Instantiate a VNF instance
|
|
vim_id = self.get_k8s_vim_id()
|
|
instantiate_req = paramgen.min_sample_instantiate(vim_id)
|
|
resp, body = self.instantiate_vnf_instance(inst_id, instantiate_req)
|
|
self.assertEqual(202, resp.status_code)
|
|
self.check_resp_headers_in_operation_task(resp)
|
|
|
|
lcmocc_id = os.path.basename(resp.headers['Location'])
|
|
self.wait_lcmocc_complete(lcmocc_id)
|
|
|
|
# 3. Show VNF instance
|
|
additional_inst_attrs = [
|
|
'vimConnectionInfo',
|
|
'instantiatedVnfInfo'
|
|
]
|
|
expected_inst_attrs.extend(additional_inst_attrs)
|
|
resp, body = self.show_vnf_instance(inst_id)
|
|
self.assertEqual(200, resp.status_code)
|
|
self.check_resp_headers_in_get(resp)
|
|
self.check_resp_body(body, expected_inst_attrs)
|
|
|
|
# check vnfc_resource_info
|
|
vnfc_resource_infos = body['instantiatedVnfInfo'].get(
|
|
'vnfcResourceInfo')
|
|
self.assertEqual(1, len(vnfc_resource_infos))
|
|
|
|
# 4. Terminate a VNF instance
|
|
terminate_req = paramgen.max_sample_terminate()
|
|
resp, body = self.terminate_vnf_instance(inst_id, terminate_req)
|
|
self.assertEqual(202, resp.status_code)
|
|
self.check_resp_headers_in_operation_task(resp)
|
|
|
|
lcmocc_id = os.path.basename(resp.headers['Location'])
|
|
self.wait_lcmocc_complete(lcmocc_id)
|
|
|
|
# wait a bit because there is a bit time lag between lcmocc DB
|
|
# update and terminate completion.
|
|
time.sleep(10)
|
|
|
|
# 5. Delete a VNF instance
|
|
resp, body = self.delete_vnf_instance(inst_id)
|
|
self.assertEqual(204, resp.status_code)
|
|
self.check_resp_headers_in_delete(resp)
|
|
|
|
# check deletion of VNF instance
|
|
resp, body = self.show_vnf_instance(inst_id)
|
|
self.assertEqual(404, resp.status_code)
|
|
|
|
# check usageState of VNF Package
|
|
usage_state = self.get_vnf_package(self.vnf_pkg_1).get('usageState')
|
|
self.assertEqual('NOT_IN_USE', usage_state)
|