0a8dced06e
This patch fixes the output of v2 Tacker so that vnfPkgId of VnfInstance that are only used in v1 Tacker are not displayed as CLI output. This is because it is confusing for v2 users that unused vnfPkgId (VNF Package ID) is always displayed as empty. The vnfPkgId (VNF Package ID) remain displayed in the v1 Tacker CLI output. Change-Id: I4adea34e9c2c7fc51e087d2a00f47db77d67fe4a
1008 lines
41 KiB
Python
1008 lines
41 KiB
Python
# Copyright (C) 2020 NTT DATA
|
|
# 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 copy
|
|
from io import StringIO
|
|
import os
|
|
import sys
|
|
from unittest import mock
|
|
|
|
import ddt
|
|
from oslo_utils.fixture import uuidsentinel
|
|
|
|
from tackerclient import client as root_client
|
|
from tackerclient.common import exceptions
|
|
from tackerclient.osc import utils as tacker_osc_utils
|
|
from tackerclient.osc.v1.vnflcm import vnflcm
|
|
from tackerclient.tests.unit.osc import base
|
|
from tackerclient.tests.unit.osc.v1.fixture_data import client
|
|
from tackerclient.tests.unit.osc.v1 import vnflcm_fakes
|
|
from tackerclient.tests.unit.test_cli10 import MyResp
|
|
from tackerclient.v1_0 import client as proxy_client
|
|
|
|
|
|
class TestVnfLcm(base.FixturedTestCase):
|
|
client_fixture_class = client.ClientFixture
|
|
|
|
def setUp(self):
|
|
super(TestVnfLcm, self).setUp()
|
|
self.url = client.TACKER_URL
|
|
self.header = {'content-type': 'application/json'}
|
|
self.app = mock.Mock()
|
|
self.app_args = mock.Mock()
|
|
self.client_manager = self.cs
|
|
self.app.client_manager.tackerclient = self.client_manager
|
|
|
|
|
|
def _get_columns_vnflcm(action='create'):
|
|
columns = ['ID', 'Instantiation State', 'VNF Instance Description',
|
|
'VNF Instance Name', 'VNF Product Name', 'VNF Provider',
|
|
'VNF Software Version', 'VNFD ID', 'VNFD Version', 'Links',
|
|
'VNF Configurable Properties', 'VNF Package ID']
|
|
if action == 'show':
|
|
columns.extend(['Instantiated Vnf Info', 'VIM Connection Info'])
|
|
if action == 'list':
|
|
columns = [ele for ele in columns if ele not in
|
|
['VNFD Version', 'VNF Instance Description',
|
|
'VNF Configurable Properties', 'VNF Package ID']]
|
|
columns.remove('Links')
|
|
return columns
|
|
|
|
|
|
@ddt.ddt
|
|
class TestCreateVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestCreateVnfLcm, self).setUp()
|
|
self.create_vnf_lcm = vnflcm.CreateVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm create')
|
|
|
|
def test_create_no_args(self):
|
|
self.assertRaises(base.ParserException, self.check_parser,
|
|
self.create_vnf_lcm, [], [])
|
|
|
|
@ddt.data({"optional_arguments": True, "instantiate": True},
|
|
{"optional_arguments": False, "instantiate": False})
|
|
@ddt.unpack
|
|
def test_take_action(self, optional_arguments, instantiate):
|
|
arglist = [uuidsentinel.vnf_package_vnfd_id]
|
|
verifylist = [('vnfd_id', uuidsentinel.vnf_package_vnfd_id)]
|
|
|
|
if optional_arguments:
|
|
arglist.extend(['--name', 'test',
|
|
'--description', 'test'])
|
|
verifylist.extend([('name', 'test'),
|
|
('description', 'test')])
|
|
|
|
# command param
|
|
if instantiate:
|
|
param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"instantiate_vnf_instance_param_sample.json")
|
|
|
|
arglist.extend(['--I', param_file])
|
|
verifylist.append(('I', param_file))
|
|
|
|
parsed_args = self.check_parser(self.create_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
json = vnflcm_fakes.vnf_instance_response()
|
|
self.requests_mock.register_uri(
|
|
'POST', os.path.join(self.url, 'vnflcm/v1/vnf_instances'),
|
|
json=json, headers=self.header)
|
|
|
|
if instantiate:
|
|
self.requests_mock.register_uri(
|
|
'POST', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
json['id'], 'instantiate'),
|
|
json={}, headers=self.header)
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
actual_columns, data = (self.create_vnf_lcm.take_action(parsed_args))
|
|
|
|
headers, attributes = vnflcm._get_columns(json)
|
|
|
|
expected_message = (
|
|
'VNF Instance ' + json['id'] + ' is created and instantiation '
|
|
'request has been accepted.')
|
|
if instantiate:
|
|
self.assertEqual(expected_message, buffer.getvalue().strip())
|
|
|
|
self.assertCountEqual(_get_columns_vnflcm(),
|
|
actual_columns)
|
|
self.assertListItemsEqual(vnflcm_fakes.get_vnflcm_data(
|
|
json, columns=attributes), data)
|
|
|
|
|
|
class TestShowVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestShowVnfLcm, self).setUp()
|
|
self.show_vnf_lcm = vnflcm.ShowVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm show')
|
|
|
|
def test_take_action(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response(
|
|
instantiation_state='INSTANTIATED')
|
|
|
|
arglist = [vnf_instance['id']]
|
|
verifylist = [('vnf_instance', vnf_instance['id'])]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.show_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
self.requests_mock.register_uri(
|
|
'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id']),
|
|
json=vnf_instance, headers=self.header)
|
|
|
|
columns, data = (self.show_vnf_lcm.take_action(parsed_args))
|
|
self.assertCountEqual(_get_columns_vnflcm(action='show'),
|
|
columns)
|
|
headers, attributes = vnflcm._get_columns(vnf_instance, action='show')
|
|
self.assertListItemsEqual(
|
|
vnflcm_fakes.get_vnflcm_data(vnf_instance, columns=attributes),
|
|
data)
|
|
|
|
|
|
class TestListVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestListVnfLcm, self).setUp()
|
|
self.list_vnf_instance = vnflcm.ListVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm list')
|
|
|
|
def test_take_action(self):
|
|
vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
|
|
parsed_args = self.check_parser(self.list_vnf_instance, [], [])
|
|
self.requests_mock.register_uri(
|
|
'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances'),
|
|
json=vnf_instances, headers=self.header)
|
|
actual_columns, data = self.list_vnf_instance.take_action(parsed_args)
|
|
|
|
headers, columns = tacker_osc_utils.get_column_definitions(
|
|
vnflcm._attr_map, long_listing=True)
|
|
|
|
expected_data = []
|
|
for vnf_instance_obj in vnf_instances:
|
|
expected_data.append(vnflcm_fakes.get_vnflcm_data(
|
|
vnf_instance_obj, columns=columns, list_action=True))
|
|
|
|
self.assertCountEqual(_get_columns_vnflcm(action='list'),
|
|
actual_columns)
|
|
self.assertCountEqual(expected_data, list(data))
|
|
|
|
def test_take_action_with_pagination(self):
|
|
vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
|
|
next_links_num = 3
|
|
parsed_args = self.check_parser(self.list_vnf_instance, [], [])
|
|
path = os.path.join(self.url, 'vnflcm/v1/vnf_instances')
|
|
|
|
links = [0] * next_links_num
|
|
link_headers = [0] * next_links_num
|
|
|
|
for i in range(next_links_num):
|
|
links[i] = (
|
|
'{base_url}?nextpage_opaque_marker={vnf_instance_id}'.format(
|
|
base_url=path,
|
|
vnf_instance_id=vnf_instances[i]['id']))
|
|
link_headers[i] = copy.deepcopy(self.header)
|
|
link_headers[i]['Link'] = '<{link_url}>; rel="next"'.format(
|
|
link_url=links[i])
|
|
|
|
self.requests_mock.register_uri(
|
|
'GET', path, json=[vnf_instances[0]], headers=link_headers[0])
|
|
self.requests_mock.register_uri(
|
|
'GET', links[0], json=[vnf_instances[1]],
|
|
headers=link_headers[1])
|
|
self.requests_mock.register_uri(
|
|
'GET', links[1], json=[vnf_instances[2]],
|
|
headers=link_headers[2])
|
|
self.requests_mock.register_uri(
|
|
'GET', links[2], json=[], headers=self.header)
|
|
|
|
actual_columns, data = self.list_vnf_instance.take_action(parsed_args)
|
|
|
|
headers, columns = tacker_osc_utils.get_column_definitions(
|
|
vnflcm._attr_map, long_listing=True)
|
|
|
|
expected_data = []
|
|
for vnf_instance_obj in vnf_instances:
|
|
expected_data.append(vnflcm_fakes.get_vnflcm_data(
|
|
vnf_instance_obj, columns=columns, list_action=True))
|
|
|
|
self.assertCountEqual(_get_columns_vnflcm(action='list'),
|
|
actual_columns)
|
|
self.assertCountEqual(expected_data, list(data))
|
|
|
|
|
|
class TestInstantiateVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestInstantiateVnfLcm, self).setUp()
|
|
self.instantiate_vnf_lcm = vnflcm.InstantiateVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm instantiate')
|
|
|
|
def test_take_action(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"instantiate_vnf_instance_param_sample.json")
|
|
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('instantiation_request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'instantiate')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
with mock.patch.object(proxy_client.ClientBase,
|
|
'_handle_fault_response') as m:
|
|
self.instantiate_vnf_lcm.take_action(parsed_args)
|
|
# check no fault response is received
|
|
self.assertNotCalled(m)
|
|
self.assertEqual(
|
|
'Instantiate request for VNF Instance ' + vnf_instance['id'] +
|
|
' has been accepted.', buffer.getvalue().strip())
|
|
|
|
def test_take_action_vnf_instance_not_found(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"instantiate_vnf_instance_param_sample.json")
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('instantiation_request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'instantiate')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, status_code=404, json={})
|
|
|
|
self.assertRaises(exceptions.TackerClientException,
|
|
self.instantiate_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
def test_take_action_param_file_not_exists(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./not_exists.json"
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('instantiation_request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(exceptions.InvalidInput,
|
|
self.instantiate_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
expected_msg = ("Invalid input: "
|
|
"User does not have read privileges to it")
|
|
self.assertEqual(expected_msg, str(ex))
|
|
|
|
@mock.patch("os.open")
|
|
@mock.patch("os.access")
|
|
def test_take_action_invalid_format_param_file(self, mock_open,
|
|
mock_access):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./invalid_param_file.json"
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('instantiation_request_file', sample_param_file)]
|
|
|
|
mock_open.return_value = "invalid_json_data"
|
|
# command param
|
|
parsed_args = self.check_parser(self.instantiate_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(exceptions.InvalidInput,
|
|
self.instantiate_vnf_lcm.take_action,
|
|
parsed_args)
|
|
expected_msg = "Failed to load parameter file."
|
|
self.assertIn(expected_msg, str(ex))
|
|
|
|
|
|
@ddt.ddt
|
|
class TestHealVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestHealVnfLcm, self).setUp()
|
|
self.heal_vnf_lcm = vnflcm.HealVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm heal')
|
|
|
|
_heal_sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"heal_vnf_instance_param_sample.json")
|
|
|
|
@ddt.data((['--cause', 'test-cause', "--vnfc-instance",
|
|
'vnfc-id-1', 'vnfc-id-2'],
|
|
[('cause', 'test-cause'),
|
|
('vnfc_instance', ['vnfc-id-1', 'vnfc-id-2'])]),
|
|
(['--cause', 'test-cause'],
|
|
[('cause', 'test-cause')]),
|
|
(["--vnfc-instance", 'vnfc-id-1', 'vnfc-id-2'],
|
|
[('vnfc_instance', ['vnfc-id-1', 'vnfc-id-2'])]),
|
|
(["--additional-param-file", _heal_sample_param_file],
|
|
[('additional_param_file', _heal_sample_param_file)]),
|
|
([], []))
|
|
@ddt.unpack
|
|
def test_take_action(self, arglist, verifylist):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
arglist.insert(0, vnf_instance['id'])
|
|
verifylist.extend([('vnf_instance', vnf_instance['id'])])
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'heal')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
result_error = self.heal_vnf_lcm.take_action(parsed_args)
|
|
|
|
self.assertIsNone(result_error)
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = ("Heal request for VNF Instance %s has been "
|
|
"accepted.") % vnf_instance['id']
|
|
self.assertIn(expected_message, actual_message)
|
|
|
|
def test_take_action_vnf_instance_not_found(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
arglist = [vnf_instance['id']]
|
|
verifylist = [('vnf_instance', vnf_instance['id'])]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'heal')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, status_code=404, json={})
|
|
|
|
self.assertRaises(exceptions.TackerClientException,
|
|
self.heal_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
def test_take_action_param_file_not_exists(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./not_exists.json"
|
|
arglist = [vnf_instance['id'],
|
|
'--additional-param-file', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('additional_param_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.heal_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(exceptions.InvalidInput,
|
|
self.heal_vnf_lcm.take_action, parsed_args)
|
|
|
|
expected_msg = ("Invalid input: "
|
|
"User does not have read privileges to it")
|
|
self.assertEqual(expected_msg, str(ex))
|
|
|
|
|
|
@ddt.ddt
|
|
class TestTerminateVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestTerminateVnfLcm, self).setUp()
|
|
self.terminate_vnf_instance = vnflcm.TerminateVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm terminate')
|
|
|
|
@ddt.data({'termination_type': 'GRACEFUL', 'delete_vnf': True},
|
|
{'termination_type': 'FORCEFUL', 'delete_vnf': False})
|
|
@ddt.unpack
|
|
def test_take_action(self, termination_type, delete_vnf):
|
|
# argument 'delete_vnf' decides deletion of vnf instance post
|
|
# termination.
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
arglist = ['--termination-type', termination_type, vnf_instance['id']]
|
|
|
|
verifylist = [('termination_type', termination_type),
|
|
('vnf_instance', vnf_instance['id'])]
|
|
|
|
if delete_vnf:
|
|
arglist.extend(['--D'])
|
|
verifylist.extend([('D', True)])
|
|
|
|
if termination_type == 'GRACEFUL':
|
|
arglist.extend(['--graceful-termination-timeout', '60'])
|
|
verifylist.append(('graceful_termination_timeout', 60))
|
|
parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
|
|
verifylist)
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'terminate')
|
|
|
|
with mock.patch.object(proxy_client.ClientBase,
|
|
'_handle_fault_response') as m:
|
|
self.requests_mock.register_uri('POST', url, json={},
|
|
headers=self.header)
|
|
if delete_vnf:
|
|
self.requests_mock.register_uri(
|
|
'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id']),
|
|
json=vnf_instance, headers=self.header)
|
|
self.requests_mock.register_uri(
|
|
'DELETE', os.path.join(
|
|
self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id']), json={}, headers=self.header)
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
result = self.terminate_vnf_instance.take_action(parsed_args)
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = ("Terminate request for VNF Instance '%s'"
|
|
" has been accepted.") % vnf_instance['id']
|
|
self.assertIn(expected_message, actual_message)
|
|
|
|
if delete_vnf:
|
|
expected_message = ("VNF Instance '%s' is deleted successfully"
|
|
% vnf_instance['id'])
|
|
self.assertIn(expected_message, actual_message)
|
|
|
|
self.assertIsNone(result)
|
|
self.assertNotCalled(m)
|
|
|
|
def test_take_action_terminate_and_delete_wait_failed(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
termination_type = 'GRACEFUL'
|
|
arglist = ['--termination-type', termination_type, '--D',
|
|
'--graceful-termination-timeout', '5', vnf_instance['id']]
|
|
|
|
verifylist = [('termination_type', termination_type), ('D', True),
|
|
('graceful_termination_timeout', 5),
|
|
('vnf_instance', vnf_instance['id'])]
|
|
|
|
parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
|
|
verifylist)
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'terminate')
|
|
|
|
self.requests_mock.register_uri('POST', url, json={},
|
|
headers=self.header)
|
|
# set the instantiateState to "INSTANTIATED", so that the
|
|
# _wait_until_vnf_is_terminated will fail
|
|
vnf_instance['instantiationState'] = 'INSTANTIATED'
|
|
|
|
self.requests_mock.register_uri(
|
|
'GET', os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id']),
|
|
json=vnf_instance, headers=self.header)
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
with mock.patch.object(self.app.client_manager.tackerclient,
|
|
'delete_vnf_instance') as mock_delete:
|
|
result = self.assertRaises(
|
|
exceptions.CommandError,
|
|
self.terminate_vnf_instance.take_action, parsed_args)
|
|
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
# Terminate vnf instance verification
|
|
expected_message = ("Terminate request for VNF Instance '%s'"
|
|
" has been accepted.") % vnf_instance['id']
|
|
self.assertIn(expected_message, actual_message)
|
|
|
|
# Verify it fails to wait for termination before delete
|
|
expected_message = ("Couldn't verify vnf instance is terminated "
|
|
"within '%(timeout)s' seconds. Unable to "
|
|
"delete vnf instance %(id)s"
|
|
% {'timeout': 15, 'id': vnf_instance['id']})
|
|
|
|
self.assertIn(expected_message, str(result))
|
|
self.assertNotCalled(mock_delete)
|
|
|
|
def test_terminate_no_options(self):
|
|
self.assertRaises(base.ParserException, self.check_parser,
|
|
self.terminate_vnf_instance, [], [])
|
|
|
|
def test_take_action_vnf_instance_not_found(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
termination_type = 'GRACEFUL'
|
|
arglist = ['--termination-type', termination_type, '--D',
|
|
'--graceful-termination-timeout', '5', vnf_instance['id']]
|
|
|
|
verifylist = [('termination_type', termination_type), ('D', True),
|
|
('graceful_termination_timeout', 5),
|
|
('vnf_instance', vnf_instance['id'])]
|
|
|
|
parsed_args = self.check_parser(self.terminate_vnf_instance, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'terminate')
|
|
self.requests_mock.register_uri('POST', url, headers=self.header,
|
|
status_code=404, json={})
|
|
|
|
self.assertRaises(exceptions.TackerClientException,
|
|
self.terminate_vnf_instance.take_action,
|
|
parsed_args)
|
|
|
|
|
|
class TestDeleteVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestDeleteVnfLcm, self).setUp()
|
|
self.delete_vnf_instance = vnflcm.DeleteVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm delete')
|
|
|
|
# Vnf Instance to delete
|
|
self.vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
|
|
|
|
def _mock_request_url_for_delete(self, vnf_index):
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
self.vnf_instances[vnf_index]['id'])
|
|
|
|
json = self.vnf_instances[vnf_index]
|
|
|
|
self.requests_mock.register_uri('GET', url, json=json,
|
|
headers=self.header)
|
|
self.requests_mock.register_uri('DELETE', url,
|
|
headers=self.header, json={})
|
|
|
|
def test_delete_one_vnf_instance(self):
|
|
arglist = [self.vnf_instances[0]['id']]
|
|
verifylist = [('vnf_instances',
|
|
[self.vnf_instances[0]['id']])]
|
|
|
|
parsed_args = self.check_parser(self.delete_vnf_instance, arglist,
|
|
verifylist)
|
|
|
|
self._mock_request_url_for_delete(0)
|
|
sys.stdout = buffer = StringIO()
|
|
result = self.delete_vnf_instance.take_action(parsed_args)
|
|
self.assertIsNone(result)
|
|
self.assertEqual(("Vnf instance '%s' is deleted successfully")
|
|
% self.vnf_instances[0]['id'],
|
|
buffer.getvalue().strip())
|
|
|
|
def test_delete_multiple_vnf_instance(self):
|
|
arglist = []
|
|
for vnf_pkg in self.vnf_instances:
|
|
arglist.append(vnf_pkg['id'])
|
|
verifylist = [('vnf_instances', arglist)]
|
|
parsed_args = self.check_parser(self.delete_vnf_instance, arglist,
|
|
verifylist)
|
|
for i in range(0, 3):
|
|
self._mock_request_url_for_delete(i)
|
|
sys.stdout = buffer = StringIO()
|
|
result = self.delete_vnf_instance.take_action(parsed_args)
|
|
self.assertIsNone(result)
|
|
self.assertEqual('All specified vnf instances are deleted '
|
|
'successfully', buffer.getvalue().strip())
|
|
|
|
def test_delete_multiple_vnf_instance_exception(self):
|
|
arglist = [
|
|
self.vnf_instances[0]['id'],
|
|
'xxxx-yyyy-zzzz',
|
|
self.vnf_instances[1]['id'],
|
|
]
|
|
verifylist = [('vnf_instances', arglist)]
|
|
parsed_args = self.check_parser(self.delete_vnf_instance,
|
|
arglist, verifylist)
|
|
|
|
self._mock_request_url_for_delete(0)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
'xxxx-yyyy-zzzz')
|
|
self.requests_mock.register_uri(
|
|
'GET', url, exc=exceptions.ConnectionFailed)
|
|
|
|
self._mock_request_url_for_delete(1)
|
|
exception = self.assertRaises(exceptions.CommandError,
|
|
self.delete_vnf_instance.take_action,
|
|
parsed_args)
|
|
|
|
self.assertEqual('Failed to delete 1 of 3 vnf instances.',
|
|
exception.message)
|
|
|
|
|
|
class TestUpdateVnfLcm(TestVnfLcm):
|
|
def setUp(self):
|
|
super(TestUpdateVnfLcm, self).setUp()
|
|
self.update_vnf_lcm = vnflcm.UpdateVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm modify')
|
|
|
|
def test_take_action(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"update_vnf_instance_param_sample.json")
|
|
|
|
arglist = [vnf_instance['id'], '--I', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('I', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(
|
|
self.update_vnf_lcm, arglist, verifylist)
|
|
url = os.path.join(
|
|
self.url,
|
|
'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'])
|
|
|
|
self.requests_mock.register_uri(
|
|
'PATCH', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
self.update_vnf_lcm.take_action(parsed_args)
|
|
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = ('Update vnf:' + vnf_instance['id'])
|
|
|
|
self.assertEqual(expected_message, actual_message)
|
|
|
|
@mock.patch.object(proxy_client.ClientBase, 'deserialize')
|
|
def test_take_action_check_content_type(self, mock_des):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ('./tackerclient/osc/v1/vnflcm/samples/'
|
|
'update_vnf_instance_param_sample.json')
|
|
|
|
arglist = [vnf_instance['id'], '--I', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('I', sample_param_file)]
|
|
mock_des.return_value = {}
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(
|
|
self.update_vnf_lcm, arglist, verifylist)
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'])
|
|
|
|
self.requests_mock.register_uri(
|
|
'PATCH', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
|
|
with mock.patch.object(root_client.HTTPClient,
|
|
'do_request') as mock_req:
|
|
headers = {'Content-Type': 'application/json'}
|
|
mock_req.return_value = (MyResp(202, headers=headers), None)
|
|
self.update_vnf_lcm.take_action(parsed_args)
|
|
# check content_type
|
|
mock_req.assert_called_once_with(
|
|
f'/vnflcm/v1/vnf_instances/{vnf_instance["id"]}', 'PATCH',
|
|
body=mock.ANY, headers=mock.ANY,
|
|
content_type='application/merge-patch+json', accept='json')
|
|
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = f'Update vnf:{vnf_instance["id"]}'
|
|
|
|
self.assertEqual(expected_message, actual_message)
|
|
|
|
def test_take_action_param_file_not_exists(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./not_exists.json"
|
|
arglist = [vnf_instance['id'], '--I', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('I', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.update_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
self.assertRaises(exceptions.InvalidInput,
|
|
self.update_vnf_lcm.take_action, parsed_args)
|
|
|
|
|
|
@ddt.ddt
|
|
class TestScaleVnfLcm(TestVnfLcm):
|
|
def setUp(self):
|
|
super(TestScaleVnfLcm, self).setUp()
|
|
self.scale_vnf_lcm = vnflcm.ScaleVnfLcm(
|
|
self.app, self.app_args, cmd_name='vnflcm scale')
|
|
|
|
@ddt.data('SCALE_IN', 'SCALE_OUT')
|
|
def test_take_action(self, scale_type):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"scale_vnf_instance_param_sample.json")
|
|
|
|
arglist = [vnf_instance['id'],
|
|
'--aspect-id', uuidsentinel.aspect_id,
|
|
'--number-of-steps', '1',
|
|
'--type', scale_type,
|
|
'--additional-param-file', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('aspect_id', uuidsentinel.aspect_id),
|
|
('number_of_steps', 1),
|
|
('type', scale_type),
|
|
('additional_param_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
|
|
verifylist)
|
|
url = os.path.join(
|
|
self.url,
|
|
'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'],
|
|
'scale')
|
|
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
self.scale_vnf_lcm.take_action(parsed_args)
|
|
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = ("Scale request for VNF Instance %s has been "
|
|
"accepted.") % vnf_instance['id']
|
|
|
|
self.assertEqual(expected_message, actual_message)
|
|
|
|
@ddt.data('SCALE_IN', 'SCALE_OUT')
|
|
def test_take_action_no_param_file(self, scale_type):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
|
|
arglist = [vnf_instance['id'],
|
|
'--aspect-id', uuidsentinel.aspect_id,
|
|
'--number-of-steps', '1',
|
|
'--type', scale_type]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('aspect_id', uuidsentinel.aspect_id),
|
|
('number_of_steps', 1),
|
|
('type', scale_type)]
|
|
|
|
parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'scale')
|
|
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
self.scale_vnf_lcm.take_action(parsed_args)
|
|
|
|
actual_message = buffer.getvalue().strip()
|
|
|
|
expected_message = ("Scale request for VNF Instance %s has been "
|
|
"accepted.") % vnf_instance['id']
|
|
|
|
self.assertEqual(expected_message, actual_message)
|
|
|
|
@ddt.data('SCALE_IN', 'SCALE_OUT')
|
|
def test_take_action_param_file_not_exists(self, scale_type):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./not_exists.json"
|
|
arglist = [vnf_instance['id'],
|
|
'--aspect-id', uuidsentinel.aspect_id,
|
|
'--number-of-steps', '2',
|
|
'--type', scale_type,
|
|
'--additional-param-file', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('aspect_id', uuidsentinel.aspect_id),
|
|
('number_of_steps', 2),
|
|
('type', scale_type),
|
|
('additional_param_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(exceptions.InvalidInput,
|
|
self.scale_vnf_lcm.take_action, parsed_args)
|
|
|
|
expected_msg = ("Invalid input: "
|
|
"User does not have read privileges to it")
|
|
self.assertEqual(expected_msg, str(ex))
|
|
|
|
@ddt.data('SCALE_IN', 'SCALE_OUT')
|
|
def test_take_action_vnf_instance_not_found(self, scale_type):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"update_vnf_instance_param_sample.json")
|
|
arglist = [vnf_instance['id'],
|
|
'--aspect-id', uuidsentinel.aspect_id,
|
|
'--number-of-steps', '3',
|
|
'--type', scale_type,
|
|
'--additional-param-file', sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('aspect_id', uuidsentinel.aspect_id),
|
|
('number_of_steps', 3),
|
|
('type', scale_type),
|
|
('additional_param_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.scale_vnf_lcm, arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(
|
|
self.url,
|
|
'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'])
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, status_code=404, json={})
|
|
|
|
self.assertRaises(exceptions.TackerClientException,
|
|
self.scale_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
|
|
class TestChangeExtConnVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestChangeExtConnVnfLcm, self).setUp()
|
|
self.change_ext_conn_vnf_lcm = vnflcm.ChangeExtConnVnfLcm(
|
|
self.app, self.app_args,
|
|
cmd_name='vnflcm change-ext-conn')
|
|
|
|
def test_take_action(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"change_ext_conn_vnf_instance_param_sample.json")
|
|
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
|
|
arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'change_ext_conn')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, json={})
|
|
|
|
sys.stdout = buffer = StringIO()
|
|
with mock.patch.object(proxy_client.ClientBase,
|
|
'_handle_fault_response') as m:
|
|
self.change_ext_conn_vnf_lcm.take_action(parsed_args)
|
|
# check no fault response is received
|
|
self.assertNotCalled(m)
|
|
self.assertEqual(
|
|
('Change External VNF Connectivity for VNF Instance {0} '
|
|
'has been accepted.'.format(vnf_instance['id'])),
|
|
buffer.getvalue().strip())
|
|
|
|
def test_take_action_vnf_instance_not_found(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v1/vnflcm/samples/"
|
|
"change_ext_conn_vnf_instance_param_sample.json")
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
|
|
arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'change_ext_conn')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, status_code=404, json={})
|
|
|
|
self.assertRaises(exceptions.TackerClientException,
|
|
self.change_ext_conn_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
def test_take_action_param_file_not_exists(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./not_exists.json"
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(
|
|
self.change_ext_conn_vnf_lcm,
|
|
arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(
|
|
exceptions.InvalidInput,
|
|
self.change_ext_conn_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
expected_msg = ("Invalid input: "
|
|
"User does not have read privileges to it")
|
|
self.assertEqual(expected_msg, str(ex))
|
|
|
|
@mock.patch("os.open")
|
|
@mock.patch("os.access")
|
|
def test_take_action_invalid_format_param_file(self, mock_open,
|
|
mock_access):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = "./invalid_param_file.json"
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('request_file', sample_param_file)]
|
|
|
|
mock_open.return_value = "invalid_json_data"
|
|
# command param
|
|
parsed_args = self.check_parser(self.change_ext_conn_vnf_lcm,
|
|
arglist,
|
|
verifylist)
|
|
|
|
ex = self.assertRaises(
|
|
exceptions.InvalidInput,
|
|
self.change_ext_conn_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
expected_msg = "Failed to load parameter file."
|
|
self.assertIn(expected_msg, str(ex))
|
|
|
|
|
|
class TestChangeVnfPkgVnfLcm(TestVnfLcm):
|
|
|
|
def setUp(self):
|
|
super(TestChangeVnfPkgVnfLcm, self).setUp()
|
|
self.change_vnfpkg_vnf_lcm = vnflcm.ChangeVnfPkgVnfLcm(
|
|
self.app, self.app_args,
|
|
cmd_name='vnflcm change-vnfpkg')
|
|
|
|
def test_take_action_with_v1_version(self):
|
|
vnf_instance = vnflcm_fakes.vnf_instance_response()
|
|
sample_param_file = ("./tackerclient/osc/v2/vnflcm/samples/"
|
|
"change_vnfpkg_vnf_instance_param_sample.json")
|
|
arglist = [vnf_instance['id'], sample_param_file]
|
|
verifylist = [('vnf_instance', vnf_instance['id']),
|
|
('request_file', sample_param_file)]
|
|
|
|
# command param
|
|
parsed_args = self.check_parser(self.change_vnfpkg_vnf_lcm,
|
|
arglist,
|
|
verifylist)
|
|
|
|
url = os.path.join(self.url, 'vnflcm/v1/vnf_instances',
|
|
vnf_instance['id'], 'change_vnfpkg')
|
|
self.requests_mock.register_uri(
|
|
'POST', url, headers=self.header, status_code=400, json={})
|
|
|
|
ex = self.assertRaises(exceptions.UnsupportedCommandVersion,
|
|
self.change_vnfpkg_vnf_lcm.take_action,
|
|
parsed_args)
|
|
|
|
expected_msg = "This command is not supported in version 1"
|
|
self.assertEqual(expected_msg, str(ex))
|
|
|
|
|
|
class TestVnfLcmV1(base.FixturedTestCase):
|
|
client_fixture_class = client.ClientFixture
|
|
api_version = '1'
|
|
|
|
def setUp(self):
|
|
super(TestVnfLcmV1, self).setUp()
|
|
|
|
def test_client_v2(self):
|
|
self.assertEqual(self.cs.vnf_lcm_client.headers,
|
|
{'Version': '1.3.0'})
|
|
self.assertEqual(self.cs.vnf_lcm_client.vnf_instances_path,
|
|
'/vnflcm/v1/vnf_instances')
|
|
# check of other paths is omitted.
|