openstack/tacker
Code Issues Proposed changes

FT of multi tenants for CNF

Browse Source 
This patch adds functional test cases to validate instantiate CNF
functionality in a multi-tenant environment.
Validates CNF instantiation is only allowed when CNF and VIM
belongs to same tenant.

Implements: blueprint k8s-namespace
Change-Id: I800b497ea6fa8f4978eb551558f0257e3d82a4ee
This commit is contained in:
Yi Feng 2022-02-25 22:00:07 +09:00
parent 30b6f74c64
commit 226281e828
9 changed files with 823 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
.zuul.yaml
View File

@ -506,6 +506,15 @@
- ^releasenotes/.*$
- ^contrib/.*$
- job:
name: tacker-functional-devstack-multinode-sol-kubernetes-multi-tenant
parent: tacker-functional-devstack-multinode-sol-kubernetes
description: |
Multinodes job for SOL Kubernetes Multi tenant devstack-based functional tests
host-vars:
controller-tacker:
tox_envlist: dsvm-functional-sol-kubernetes-multi-tenant
- job:
name: tacker-functional-devstack-multinode-libs-master
parent: tacker-functional-devstack-multinode-legacy
@ -532,3 +541,4 @@
- tacker-functional-devstack-multinode-libs-master
- tacker-functional-devstack-multinode-sol-v2
- tacker-functional-devstack-multinode-sol-multi-tenant
- tacker-functional-devstack-multinode-sol-kubernetes-multi-tenant

6
roles/setup-multi-tenant-vim/defaults/main.yaml
View File

@ -5,6 +5,9 @@ os_domain_tenant1: Default
os_vim_name_tenant1: VIM_TEST
os_vim_conf_name_tenant1: local-tenant1-vim.yaml
os_vim_conf_path_tenant1: /tmp/local-tenant1-vim.yaml
k8s_vim_name_tenant1: vim-kubernetes-t1
k8s_vim_conf_name_tenant1: local-tenant1-k8s-vim.yaml
k8s_vim_conf_path_tenant1: /tmp/local-tenant1-k8s-vim.yaml
os_username_tenant2: test_user_2
os_password_tenant2: devstack
os_project_tenant2: test_tenant_2
@ -12,4 +15,7 @@ os_domain_tenant2: Default
os_vim_name_tenant2: VIM_DEMO
os_vim_conf_name_tenant2: local-tenant2-vim.yaml
os_vim_conf_path_tenant2: /tmp/local-tenant2-vim.yaml
k8s_vim_name_tenant2: vim-kubernetes-t2
k8s_vim_conf_name_tenant2: local-tenant2-k8s-vim.yaml
k8s_vim_conf_path_tenant2: /tmp/local-tenant2-k8s-vim.yaml
os_member_role: member

87
roles/setup-multi-tenant-vim/tasks/main.yaml
View File

@ -127,3 +127,90 @@
when:
- inventory_hostname == 'controller-tacker'
- block:
- name: Get stackenv from devstack environment
slurp:
src: "{{ devstack_base_dir }}/devstack/.stackenv"
register: stackenv
- name: Set a keystone authentication uri
set_fact:
auth_uri: "{{
stackenv.content
| b64decode
| regex_replace('\n', ' ')
| regex_replace('^.*KEYSTONE_SERVICE_URI=([^ ]+).*$', '\\1')
}}"
- name: Generate Kubernetes VIM config for first tenant
shell: >
{{ zuul_work_dir }}/tools/gen_vim_config.sh --type kubernetes
--os-user {{ os_username_tenant1 }}
--os-password {{ os_password_tenant1 }}
--project {{ os_project_tenant1 }}
--os-project-domain {{ os_domain_tenant1 }}
--os-user-domain {{ os_domain_tenant1 }}
--endpoint {{ kuryr_k8s_api_url }} --os-disable-cert-verify
--k8s-token {{ hostvars['controller-k8s'].admin_token.stdout }}
-o {{ k8s_vim_conf_path_tenant1 }}
- name: Cat Kubernetes VIM config for first tenant
shell: cat {{ k8s_vim_conf_path_tenant1 }}
- name: Register Kubernetes VIM for first tenant
shell: >
openstack vim register
--os-username {{ os_username_tenant1 }}
--os-password {{ os_password_tenant1 }}
--os-project-name {{ os_project_tenant1 }}
--os-auth-url {{ auth_uri }}
--os-project-domain-name {{ os_domain_tenant1 }}
--os-user-domain-name {{ os_domain_tenant1 }}
--config-file {{ k8s_vim_conf_path_tenant1 }}
--description "Kubernetes VIM for testing multi tenant"
{{ k8s_vim_name_tenant1 }}
- name: Copy first tenant vim config file
copy:
remote_src=True
src={{ k8s_vim_conf_path_tenant1 }}
dest={{ zuul_work_dir }}/tacker/tests/etc/samples/{{ k8s_vim_conf_name_tenant1 }}
- name: Generate Kubernetes VIM config for second tenant
shell: >
{{ zuul_work_dir }}/tools/gen_vim_config.sh --type kubernetes
--os-user {{ os_username_tenant2 }}
--os-password {{ os_password_tenant2 }}
--project {{ os_project_tenant2 }}
--os-project-domain {{ os_domain_tenant2 }}
--os-user-domain {{ os_domain_tenant2 }}
--endpoint {{ kuryr_k8s_api_url }} --os-disable-cert-verify
--k8s-token {{ hostvars['controller-k8s'].admin_token.stdout }}
-o {{ k8s_vim_conf_path_tenant2 }}
- name: Cat Kubernetes VIM config
shell: cat {{ k8s_vim_conf_path_tenant2 }}
- name: Register Kubernetes VIM for second tenant
shell: >
openstack vim register
--os-username {{ os_username_tenant2 }}
--os-password {{ os_password_tenant2 }}
--os-project-name {{ os_project_tenant2 }}
--os-auth-url {{ auth_uri }}
--os-project-domain-name {{ os_domain_tenant2 }}
--os-user-domain-name {{ os_domain_tenant2 }}
--config-file {{ k8s_vim_conf_path_tenant2 }}
--description "Kubernetes VIM for testing multi tenant"
{{ k8s_vim_name_tenant2 }}
- name: Copy second tenant vim config file
copy:
remote_src=True
src={{ k8s_vim_conf_path_tenant2 }}
dest={{ zuul_work_dir }}/tacker/tests/etc/samples/{{ k8s_vim_conf_name_tenant2 }}
when:
- inventory_hostname == 'controller-tacker'
- kuryr_k8s_api_url is defined

0
tacker/tests/functional/sol_kubernetes_multi_tenant/__init__.py Normal file
View File

0
tacker/tests/functional/sol_kubernetes_multi_tenant/vnflcm/__init__.py Normal file
View File

230
tacker/tests/functional/sol_kubernetes_multi_tenant/vnflcm/base.py Normal file
View File

@ -0,0 +1,230 @@
#
# 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 time
from oslo_serialization import jsonutils
from tacker.objects import fields
from tacker.tests.functional import base
from tacker.tests.functional.common.fake_server import FakeServerManager
from tacker.tests.functional.sol.vnflcm import base as vnflcm_base
from tacker.tests import utils
FAKE_SERVER_MANAGER_T1 = FakeServerManager()
FAKE_SERVER_PORT_T1 = 9995
FAKE_SERVER_MANAGER_T2 = FakeServerManager()
FAKE_SERVER_PORT_T2 = 9996
VNF_PACKAGE_UPLOAD_TIMEOUT = 300
WAIT_TIMEOUT_ERR_MSG = ("Failed to %(action)s, process could not be completed"
" within %(timeout)s seconds")
RETRY_WAIT_TIME = 5
class BaseVnfLcmKubernetesMultiTenantTest(vnflcm_base.BaseVnfLcmTest):
prepare_fake_server = False
@classmethod
def setUpClass(cls):
super(BaseVnfLcmKubernetesMultiTenantTest, cls).setUpClass()
cls.tacker_client_tenant1 = base.BaseTackerTest.tacker_http_client(
'local-tenant1-vim.yaml')
cls.tacker_client_tenant2 = base.BaseTackerTest.tacker_http_client(
'local-tenant2-vim.yaml')
cls.base_subscriptions_url = "/vnflcm/v1/subscriptions"
cls.base_vnf_instances_url = "/vnflcm/v1/vnf_instances"
cls.base_vnf_package_url = "/vnfpkgm/v1/vnf_packages"
cls.base_vnf_lcm_op_occs_url = "/vnflcm/v1/vnf_lcm_op_occs"
# Set up fake NFVO server for tenant1 and tenant2
cls.servers = {FAKE_SERVER_PORT_T1: FAKE_SERVER_MANAGER_T1,
FAKE_SERVER_PORT_T2: FAKE_SERVER_MANAGER_T2}
# NOTE: Create both server in parallel, otherwise they can
# cause (especially server start) job timeout.
for port, manager in cls.servers.items():
cls._prepare_start_fake_server(manager, port)
@classmethod
def tearDownClass(cls):
super(BaseVnfLcmKubernetesMultiTenantTest, cls).tearDownClass()
for _, manager in cls.servers.items():
manager.stop_server()
def setUp(self):
super(BaseVnfLcmKubernetesMultiTenantTest, self).setUp()
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
self._clear_history_and_set_callback(
FAKE_SERVER_MANAGER_T1, callback_url)
self._clear_history_and_set_callback(
FAKE_SERVER_MANAGER_T2, callback_url)
vim_list = self.client.list_vims()
vim_name_t1 = 'vim-kubernetes-t1'
self.vim_tenant1 = self.get_vim(vim_list, vim_name_t1)
vim_name_t2 = 'vim-kubernetes-t2'
self.vim_tenant2 = self.get_vim(vim_list, vim_name_t2)
def create_and_upload_vnf_package(
self, tacker_client, csar_package_name, user_defined_data):
# create vnf package
body = jsonutils.dumps({"userDefinedData": user_defined_data})
_, vnf_package = tacker_client.do_request(
self.base_vnf_package_url, "POST", body=body)
vnf_pkg_id = vnf_package['id']
# upload vnf package
csar_package_path = ("../../../etc/samples/etsi/nfv/"
f"{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, _ = utils.create_csar_with_unique_vnfd_id(file_path)
with open(file_path, 'rb') as file_object:
tacker_client.do_request(
f"{self.base_vnf_package_url}/{vnf_pkg_id}/package_content",
"PUT", body=file_object, content_type='application/zip')
# wait for onboard
start_time = int(time.time())
show_url = os.path.join(self.base_vnf_package_url, vnf_pkg_id)
while True:
_, body = tacker_client.do_request(show_url, "GET")
if body['onboardingState'] == "ONBOARDED":
vnfd_id = body['vnfdId']
break
if (int(time.time()) - start_time) > VNF_PACKAGE_UPLOAD_TIMEOUT:
raise Exception(
WAIT_TIMEOUT_ERR_MSG %
{
"action": "onboard vnf package",
"timeout": VNF_PACKAGE_UPLOAD_TIMEOUT
}
)
time.sleep(RETRY_WAIT_TIME)
# remove temporarily created CSAR file
os.remove(file_path)
return vnf_package['id'], vnfd_id
def _filter_notify_history(self, callback_url, vnf_instance_id,
fake_server_manager=None, clear=True):
notify_histories = fake_server_manager.get_history(
callback_url)
if clear:
fake_server_manager.clear_history(callback_url)
return [
h for h in notify_histories
if h.request_body.get('vnfInstanceId') == vnf_instance_id]
def assert_instantiate(
self, resp, vnf_instance_id, http_client, fake_server_manager):
self.assertEqual(202, resp.status_code)
resp, vnf_instance = self._show_vnf_instance(vnf_instance_id,
http_client)
self.assert_vnf_state(vnf_instance)
# FT-checkpoint: Notification
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
notify_mock_responses = self._filter_notify_history(
callback_url, vnf_instance_id,
fake_server_manager=fake_server_manager, clear=False)
self.assertEqual(3, len(notify_mock_responses))
self.assert_notification_mock_response(
notify_mock_responses[0],
'VnfLcmOperationOccurrenceNotification',
'STARTING')
self.assert_notification_mock_response(
notify_mock_responses[1],
'VnfLcmOperationOccurrenceNotification',
'PROCESSING')
self.assert_notification_mock_response(
notify_mock_responses[2],
'VnfLcmOperationOccurrenceNotification',
'COMPLETED')
def assert_terminate(self, resp, vnf_instance_id,
http_client=None, fake_server_manager=None):
self.assertEqual(202, resp.status_code)
resp, vnf_instance = self._show_vnf_instance(
vnf_instance_id, http_client)
self.assert_instantiation_state(
vnf_instance,
fields.VnfInstanceState.NOT_INSTANTIATED)
# FT-checkpoint: Notification
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
notify_mock_responses = self._filter_notify_history(
callback_url, vnf_instance_id,
fake_server_manager=fake_server_manager, clear=False)
self.assertEqual(3, len(notify_mock_responses))
self.assert_notification_mock_response(
notify_mock_responses[0],
'VnfLcmOperationOccurrenceNotification',
'STARTING')
self.assert_notification_mock_response(
notify_mock_responses[1],
'VnfLcmOperationOccurrenceNotification',
'PROCESSING')
self.assert_notification_mock_response(
notify_mock_responses[2],
'VnfLcmOperationOccurrenceNotification',
'COMPLETED')
def delete_vnf_package(self, tacker_client, vnf_package_id):
url = f'/vnfpkgm/v1/vnf_packages/{vnf_package_id}'
# Update vnf package before delete
req_body = jsonutils.dumps({"operationalState": "DISABLED"})
tacker_client.do_request(url, "PATCH", body=req_body)
# Delete vnf package before delete
tacker_client.do_request(url, "DELETE")
def assert_occ_show(
self, resp, op_occs_info, http_code, vnf_instance_id, operation):
self.assertEqual(http_code, resp.status_code)
if http_code == 200:
self.assertEqual(
vnf_instance_id, op_occs_info.get('vnfInstanceId'))
self.assertEqual(
operation, op_occs_info.get('operation'))
def assert_occ_only_one_vnf(
self, vnf_instance_id_t1, op_occs_info, vnf_instance_id_t2):
for op_occ in op_occs_info:
if vnf_instance_id_t1 != op_occ.get('vnfInstanceId'):
self.assertNotEqual(
vnf_instance_id_t2, op_occ.get('vnfInstanceId'))

109
tacker/tests/functional/sol_kubernetes_multi_tenant/vnflcm/fake_vnflcm.py Normal file
View File

@ -0,0 +1,109 @@
#
# 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.tests import uuidsentinel
class Subscription:
@staticmethod
def make_create_request_body(callback_uri):
"""Parameter selection policy.
Set all Notification types and all life cycle types for filter.
Specify OAuth2 for authentication → do not set authentication.
Args:
callback_uri (str): Notification URI.
Returns:
dict: Request body
"""
return {
"filter": {
"vnfInstanceSubscriptionFilter": {
"vnfdIds": ["b1bb0ce7-ebca-4fa7-95ed-4840d7000000"],
"vnfProductsFromProviders": [{
"vnfProvider": "Company",
"vnfProducts": [
{
"vnfProductName": "Sample VNF",
"versions": [
{
"vnfSoftwareVersion": "1.0",
"vnfdVersions": ["1.0"]
}
]
}
]
}]
},
"notificationTypes": [
"VnfLcmOperationOccurrenceNotification",
"VnfIdentifierCreationNotification",
"VnfIdentifierDeletionNotification"
],
"operationTypes": [
"INSTANTIATE",
"SCALE",
"TERMINATE",
"HEAL",
"MODIFY_INFO",
"CHANGE_EXT_CONN"
],
"operationStates": ["STARTING"]
},
"callbackUri": callback_uri
}
class VnfInstances:
@staticmethod
def make_create_request_body(vnfd_id):
return {
"vnfdId": vnfd_id,
"vnfInstanceName": "",
"vnfInstanceDescription": "Sample VNF",
"metadata": {
"samplekey": "samplevalue"
}
}
@staticmethod
def make_inst_request_body(vim_id, additional_params):
data = {
"flavourId": "simple",
"vimConnectionInfo": [{
"id": uuidsentinel.vim_connection_id,
"vimType": "kubernetes",
"vimId": vim_id
}],
"additionalParams": additional_params
}
return data
@staticmethod
def make_term_request_body():
"""Parameter selection policy.
As all parameters are set, GRACEFUL is specified for terminationType.
(to specify gracefulTerminationTimeout)
Returns:
dict: Request body
"""
return {
"terminationType": "FORCEFUL"
}

375
tacker/tests/functional/sol_kubernetes_multi_tenant/vnflcm/test_kubernetes_multi_tenant.py Normal file
View File

@ -0,0 +1,375 @@
#
# 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
from tacker.tests.functional.sol.vnflcm import base as vnflcm_base
from tacker.tests.functional.sol_kubernetes_multi_tenant.vnflcm import (
fake_vnflcm)
from tacker.tests.functional.sol_kubernetes_multi_tenant.vnflcm import base
class VnfLcmKubernetesWithMultiTenant(base.
BaseVnfLcmKubernetesMultiTenantTest):
@classmethod
def setUpClass(cls):
super(VnfLcmKubernetesWithMultiTenant, cls).setUpClass()
def test_lcm_functionality(self):
"""Test CNF instantiate and terminate with member role users.
In this test case, we do following steps.
Note: User A belongs to Tenant 1.
User B belongs to Tenant 2.
- Create subscription.
- User A registers Subscription A(Notification Server A).
- User B registers Subscription B(Notification Server B).
- Create and Upload VNF Package
- User A creates and uploads VNF Package A.
- User B creates and uploads VNF Package B.
- Create VNF Instance
- User A creates VNF Instance A using VNF Package A.
- User B creates VNF Instance B using VNF Package B.
- Instantiate VNF
- User A fails to instantiate VNF Instance B.
- User B fails to instantiate VNF Instance A.
- User A instantiates VNF Instance A.
- User B instantiates VNF Instance B.
- List LCM operation occurrence
- User A only sees lcm_op_occ of VNF Instance A.
- User B only sees lcm_op_occ of VNF Instance B.
- Show VNF LCM operation occurrence
- User A succeeds to show lcm_op_occ of VNF Instance A.
- User A fails to show lcm_op_occ of VNF Instance B.
- User B succeeds to show lcm_op_occ of VNF Instance B.
- User B fails to show lcm_op_occ of VNF Instance A.
- Terminate VNF
- User A fails to terminate VNF Instance B.
- User B fails to terminate VNF Instance A.
- User A succeeds to terminate VNF Instance A.
- User B succeeds to terminate VNF Instance B.
- List LCM operation occurrence
- User A only sees lcm_op_occ of VNF Instance A.
- User B only sees lcm_op_occ of VNF Instance B.
- Show VNF LCM operation occurrence
- User A succeeds to show lcm_op_occ of VNF Instance A.
- User A fails to show lcm_op_occ of VNF Instance B.
- User B succeeds to show lcm_op_occ of VNF Instance B.
- User B fails to show lcm_op_occ of VNF Instance A.
- Delete VNF Instance
- User A deletes VNF Instance A.
- User B deletes VNF Instance B.
- Delete VNF Package
- User A deletes VNF Package A.
- User B deletes VNF Package B.
"""
# Pre-Setting
# User A registers Subscription A.
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL,
self._testMethodName)
req_body = fake_vnflcm.Subscription.make_create_request_body(
f'http://localhost:'
f'{base.FAKE_SERVER_MANAGER_T1.SERVER_PORT_T1}'
f'{callback_url}')
resp_t1, resp_body_t1 = self._register_subscription(
req_body, self.tacker_client_tenant1)
self.assertEqual(201, resp_t1.status_code)
self.assert_http_header_location_for_subscription(resp_t1.headers)
self.assert_notification_get(
callback_url, base.FAKE_SERVER_MANAGER_T1)
subscription_id_t1 = resp_body_t1.get('id')
self.addCleanup(self._delete_subscription, subscription_id_t1,
self.tacker_client_tenant1)
# User B registers Subscription B.
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
req_body_t2 = fake_vnflcm.Subscription.make_create_request_body(
f'http://localhost:'
f'{base.FAKE_SERVER_MANAGER_T1.SERVER_PORT_T2}'
f'{callback_url}')
resp_t2, resp_body_t2 = self._register_subscription(
req_body_t2, self.tacker_client_tenant2)
self.assertEqual(201, resp_t2.status_code)
self.assert_http_header_location_for_subscription(resp_t2.headers)
self.assert_notification_get(
callback_url, base.FAKE_SERVER_MANAGER_T2)
subscription_id_t2 = resp_body_t2.get('id')
self.addCleanup(self._delete_subscription, subscription_id_t2,
self.tacker_client_tenant2)
# User A Create and Upload VNF Package A
sample_name_t1 = 'test_cnf_multi_ns'
vnf_package_id_t1, vnfd_id_t1 = self.create_and_upload_vnf_package(
self.tacker_client_tenant1, sample_name_t1,
{"key": "multi_tenant_t1_functional"})
# User B Create and Upload VNF Package B
sample_name_t2 = 'test_cnf_multi_ns'
vnf_package_id_t2, vnfd_id_t2 = self.create_and_upload_vnf_package(
self.tacker_client_tenant2, sample_name_t2,
{"key": "multi_tenant_t2_functional"})
# Create VNF Instance
# User A creates VNF Instance A using VNF Package A
vnf_instance_name_t1 = "multi_tenant_cnf_t1"
vnf_instance_description_t1 = "multi tenant cnf t1"
resp_t1, vnf_instance_t1 = self._create_vnf_instance(
vnfd_id_t1, vnf_instance_name_t1,
vnf_instance_description_t1, self.tacker_client_tenant1)
vnf_instance_id_t1 = vnf_instance_t1.get('id')
self._wait_lcm_done(
vnf_instance_id=vnf_instance_id_t1,
fake_server_manager=base.FAKE_SERVER_MANAGER_T1)
self.assert_create_vnf(
resp_t1, vnf_instance_t1, base.FAKE_SERVER_MANAGER_T1)
# User B creates VNF Instance B using VNF Package B
vnf_instance_name_t2 = "multi_tenant_cnf_t2"
vnf_instance_description_t2 = "multi tenant cnf t2"
resp_t2, vnf_instance_t2 = self._create_vnf_instance(
vnfd_id_t2, vnf_instance_name_t2,
vnf_instance_description_t2, self.tacker_client_tenant2)
vnf_instance_id_t2 = vnf_instance_t2.get('id')
self._wait_lcm_done(
vnf_instance_id=vnf_instance_id_t2,
fake_server_manager=base.FAKE_SERVER_MANAGER_T2)
self.assert_create_vnf(
resp_t2, vnf_instance_t2, base.FAKE_SERVER_MANAGER_T2)
# Instantiate vnf instance
# User A unable to instantiate VNF Instance B
additional_params_t1 = {
"lcm-kubernetes-def-files": [
"Files/kubernetes/deployment_has_namespace.yaml",
"Files/kubernetes/namespace01.yaml"
],
"namespace": "multi-namespace01"
}
inst_req_body_t1 = fake_vnflcm.VnfInstances.make_inst_request_body(
self.vim_tenant1['id'], additional_params_t1)
resp, _ = self._instantiate_vnf_instance(
vnf_instance_id_t2, inst_req_body_t1, self.tacker_client_tenant1)
self.assertEqual(404, resp.status_code)
# User B unable to instantiate VNF Instance A
additional_params_t2 = {
"lcm-kubernetes-def-files": [
"Files/kubernetes/deployment_has_namespace.yaml",
"Files/kubernetes/namespace02.yaml"
],
"namespace": "multi-namespace02"
}
inst_req_body_t2 = fake_vnflcm.VnfInstances.make_inst_request_body(
self.vim_tenant2['id'], additional_params_t2)
resp, _ = self._instantiate_vnf_instance(
vnf_instance_id_t1, inst_req_body_t2, self.tacker_client_tenant2)
self.assertEqual(404, resp.status_code)
# User A instantiate VNF Instance A
resp, _ = self._instantiate_vnf_instance(
vnf_instance_id_t1, inst_req_body_t1, self.tacker_client_tenant1)
self._wait_lcm_done(
'COMPLETED', vnf_instance_id=vnf_instance_id_t1,
fake_server_manager=base.FAKE_SERVER_MANAGER_T1)
self.assert_instantiate(
resp, vnf_instance_id_t1, self.tacker_client_tenant1,
base.FAKE_SERVER_MANAGER_T1)
# User B instantiate VNF Instance B
resp, _ = self._instantiate_vnf_instance(
vnf_instance_id_t2, inst_req_body_t2, self.tacker_client_tenant2)
self._wait_lcm_done(
'COMPLETED', vnf_instance_id=vnf_instance_id_t2,
fake_server_manager=base.FAKE_SERVER_MANAGER_T2)
self.assert_instantiate(
resp, vnf_instance_id_t2, self.tacker_client_tenant2,
base.FAKE_SERVER_MANAGER_T2)
# get vnflcm_op_occ_id for User A
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
notify_mock_responses = (base.FAKE_SERVER_MANAGER_T1.
get_history(callback_url))
base.FAKE_SERVER_MANAGER_T1.clear_history(callback_url)
vnflcm_op_occ_id_t1 = notify_mock_responses[0].request_body.get(
'vnfLcmOpOccId')
self.assertIsNotNone(vnflcm_op_occ_id_t1)
# get vnflcm_op_occ_id for User B
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
notify_mock_responses = (base.FAKE_SERVER_MANAGER_T2.
get_history(callback_url))
base.FAKE_SERVER_MANAGER_T2.clear_history(callback_url)
vnflcm_op_occ_id_t2 = notify_mock_responses[0].request_body.get(
'vnfLcmOpOccId')
self.assertIsNotNone(vnflcm_op_occ_id_t2)
# List LcmOpOccs
# User A gets LcmOpOccs List, and should get LcmOpOccs of
# VNF Instance A
resp, op_occs_info = self._list_op_occs(
filter_string='', http_client=self.tacker_client_tenant1)
self._assert_occ_list(resp, op_occs_info)
self.assert_occ_only_one_vnf(
vnf_instance_id_t1, op_occs_info, vnf_instance_id_t2)
# User B gets LcmOpOccs List, and should get LcmOpOccs of
# VNF Instance B
resp, op_occs_info = self._list_op_occs(
filter_string='', http_client=self.tacker_client_tenant2)
self._assert_occ_list(resp, op_occs_info)
self.assert_occ_only_one_vnf(
vnf_instance_id_t2, op_occs_info, vnf_instance_id_t1)
# Show LcmOpOccs
# User A able to show LcmOpOccs List of VNF Instance A
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t1, self.tacker_client_tenant1)
self.assert_occ_show(
resp, op_occs_info, 200, vnf_instance_id_t1, 'INSTANTIATE')
# User A unable to show LcmOpOccs List of VNF Instance B
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t2, self.tacker_client_tenant1)
self.assert_occ_show(resp, op_occs_info, 404, None, None)
# User B able to show LcmOpOccs List of VNF Instance B
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t2, self.tacker_client_tenant2)
self.assert_occ_show(
resp, op_occs_info, 200, vnf_instance_id_t2, 'INSTANTIATE')
# User B unable to show LcmOpOccs List of VNF Instance A
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t1, self.tacker_client_tenant2)
self.assert_occ_show(resp, op_occs_info, 404, None, None)
# Terminate VNF
# User A unable to terminate VNF B
terminate_req_body_t1 = (fake_vnflcm.VnfInstances
.make_term_request_body())
resp, _ = self._terminate_vnf_instance(
vnf_instance_id_t2, terminate_req_body_t1,
self.tacker_client_tenant1)
self.assertEqual(404, resp.status_code)
# User B unable to terminate VNF A
terminate_req_body_t2 = (fake_vnflcm.VnfInstances
.make_term_request_body())
resp, _ = self._terminate_vnf_instance(
vnf_instance_id_t1, terminate_req_body_t2,
self.tacker_client_tenant2)
self.assertEqual(404, resp.status_code)
# User A terminates VNF A
resp, _ = self._terminate_vnf_instance(
vnf_instance_id_t1, terminate_req_body_t1,
self.tacker_client_tenant1)
self._wait_lcm_done(
'COMPLETED', vnf_instance_id=vnf_instance_id_t1,
fake_server_manager=base.FAKE_SERVER_MANAGER_T1)
self.assert_terminate(
resp, vnf_instance_id_t1, self.tacker_client_tenant1,
fake_server_manager=base.FAKE_SERVER_MANAGER_T1)
# User B terminates VNF B
resp, _ = self._terminate_vnf_instance(
vnf_instance_id_t2, terminate_req_body_t2,
self.tacker_client_tenant2)
self._wait_lcm_done(
'COMPLETED', vnf_instance_id=vnf_instance_id_t2,
fake_server_manager=base.FAKE_SERVER_MANAGER_T2)
self.assert_terminate(
resp, vnf_instance_id_t2, self.tacker_client_tenant2,
fake_server_manager=base.FAKE_SERVER_MANAGER_T2)
# get vnflcm_op_occ_id for User A
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
notify_mock_responses = (base.FAKE_SERVER_MANAGER_T1.
get_history(callback_url))
base.FAKE_SERVER_MANAGER_T1.clear_history(callback_url)
vnflcm_op_occ_id_t1 = notify_mock_responses[0].request_body.get(
'vnfLcmOpOccId')
self.assertIsNotNone(vnflcm_op_occ_id_t1)
# get vnflcm_op_occ_id for User B
callback_url = os.path.join(
vnflcm_base.MOCK_NOTIFY_CALLBACK_URL, self._testMethodName)
notify_mock_responses = (base.FAKE_SERVER_MANAGER_T2.
get_history(callback_url))
base.FAKE_SERVER_MANAGER_T2.clear_history(callback_url)
vnflcm_op_occ_id_t2 = notify_mock_responses[0].request_body.get(
'vnfLcmOpOccId')
self.assertIsNotNone(vnflcm_op_occ_id_t2)
# List LcmOpOccs
# User A gets LcmOpOccs List, and should get LcmOpOccs of
# VNF Instance A
resp, op_occs_info = self._list_op_occs(
filter_string='', http_client=self.tacker_client_tenant1)
self._assert_occ_list(resp, op_occs_info)
self.assert_occ_only_one_vnf(
vnf_instance_id_t1, op_occs_info, vnf_instance_id_t2)
# User B gets LcmOpOccs List, and should get LcmOpOccs of
# VNF Instance B
resp, op_occs_info = self._list_op_occs(
filter_string='', http_client=self.tacker_client_tenant2)
self._assert_occ_list(resp, op_occs_info)
self.assert_occ_only_one_vnf(
vnf_instance_id_t2, op_occs_info, vnf_instance_id_t1)
# Show LcmOpOccs
# User A able to show LcmOpOccs List of VNF Instance A
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t1, self.tacker_client_tenant1)
self.assert_occ_show(
resp, op_occs_info, 200, vnf_instance_id_t1, 'TERMINATE')
# User A unable to show LcmOpOccs List of VNF Instance B
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t2, self.tacker_client_tenant1)
self.assert_occ_show(resp, op_occs_info, 404, None, None)
# User B able to show LcmOpOccs List of VNF Instance B
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t2, self.tacker_client_tenant2)
self.assert_occ_show(
resp, op_occs_info, 200, vnf_instance_id_t2, 'TERMINATE')
# User B unable to show LcmOpOccs List of VNF Instance A
resp, op_occs_info = self._show_op_occs(
vnflcm_op_occ_id_t1, self.tacker_client_tenant2)
self.assert_occ_show(resp, op_occs_info, 404, None, None)
# Delete VNF
# User A deletes VNF Instance A
resp, _ = self._delete_vnf_instance(
vnf_instance_id_t1, self.tacker_client_tenant1)
self.assertEqual(vnf_instance_t1['id'], vnf_instance_id_t1)
# User B deletes VNF Instance B
resp, _ = self._delete_vnf_instance(
vnf_instance_id_t2, self.tacker_client_tenant2)
self.assertEqual(vnf_instance_t2['id'], vnf_instance_id_t2)
# Deleting vnf package.
self.delete_vnf_package(self.tacker_client_tenant1, vnf_package_id_t1)
self.delete_vnf_package(self.tacker_client_tenant2, vnf_package_id_t2)

6
tox.ini
View File

@ -67,6 +67,12 @@ setenv = {[testenv]setenv}
commands =
stestr --test-path=./tacker/tests/functional/sol_multi_tenant run --slowest --concurrency 1 {posargs}
[testenv:dsvm-functional-sol-kubernetes-multi-tenant]
setenv = {[testenv]setenv}
commands =
stestr --test-path=./tacker/tests/functional/sol_kubernetes_multi_tenant run --slowest --concurrency 1 {posargs}
[testenv:debug]
commands = oslo_debug_helper {posargs}