Support updating VNF parameters in tackerclient

Implementation to update parameters of created VNF in python-tackerclient.

Change-Id: Ifd5b07bdd1445639db4c23f6b92a3350d4ae22a5
Blueprint: https://blueprints.launchpad.net/tacker/+spec/reservation-vnfm
This commit is contained in:
Hiroo Kitamura 2020-02-27 05:50:43 +00:00
parent e712c255d0
commit a4c2b291b2
14 changed files with 706 additions and 26 deletions

View File

@ -18,6 +18,7 @@ import yaml
from osc_lib.command import command from osc_lib.command import command
from osc_lib import utils from osc_lib import utils
from oslo_utils import encodeutils
from tackerclient.common import exceptions from tackerclient.common import exceptions
from tackerclient.i18n import _ from tackerclient.i18n import _
@ -357,13 +358,16 @@ class UpdateVNF(command.ShowOne):
def get_parser(self, prog_name): def get_parser(self, prog_name):
parser = super(UpdateVNF, self).get_parser(prog_name) parser = super(UpdateVNF, self).get_parser(prog_name)
config_group = parser.add_mutually_exclusive_group(required=True) group = parser.add_mutually_exclusive_group(required=True)
config_group.add_argument( group.add_argument(
'--config-file', '--config-file',
help=_('YAML file with VNF configuration')) help=_('YAML file with VNF configuration'))
config_group.add_argument( group.add_argument(
'--config', '--config',
help=_('Specify config YAML data')) help=_('YAML data with VNF configuration'))
group.add_argument(
'--param-file',
help=_('YAML file with VNF parameter'))
parser.add_argument( parser.add_argument(
_VNF, _VNF,
metavar="<VNF>", metavar="<VNF>",
@ -381,17 +385,33 @@ class UpdateVNF(command.ShowOne):
try: try:
config = yaml.load(config_yaml, Loader=yaml.SafeLoader) config = yaml.load(config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e: except yaml.YAMLError as e:
raise exceptions.InvalidInput(e) raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The config file is empty')
if parsed_args.config: if parsed_args.config:
config = parsed_args.config decoded_config = encodeutils.safe_decode(parsed_args.config)
if isinstance(config, str) or isinstance(config, unicode):
config_str = parsed_args.config.decode('unicode_escape')
try: try:
config = yaml.load(config_str, Loader=yaml.SafeLoader) config = yaml.load(decoded_config, Loader=yaml.SafeLoader)
except yaml.YAMLError as e: except yaml.YAMLError as e:
raise exceptions.InvalidInput(e) raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The parameter is empty')
if config: if config:
body[_VNF]['attributes'] = {'config': config} body[_VNF]['attributes'] = {'config': config}
if parsed_args.param_file:
with open(parsed_args.param_file) as f:
param_yaml = f.read()
try:
param = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not param:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
body[_VNF]['attributes'] = {'param_values': param}
tackerV10.update_dict(parsed_args, body[_VNF], ['tenant_id']) tackerV10.update_dict(parsed_args, body[_VNF], ['tenant_id'])
return body return body

View File

@ -17,6 +17,8 @@
import yaml import yaml
from oslo_utils import encodeutils
from tackerclient.common import exceptions from tackerclient.common import exceptions
from tackerclient.i18n import _ from tackerclient.i18n import _
from tackerclient.tacker import v1_0 as tackerV10 from tackerclient.tacker import v1_0 as tackerV10
@ -143,12 +145,16 @@ class UpdateVNF(tackerV10.UpdateCommand):
resource = _VNF resource = _VNF
def add_known_arguments(self, parser): def add_known_arguments(self, parser):
parser.add_argument( group = parser.add_mutually_exclusive_group()
group.add_argument(
'--config-file', '--config-file',
help=_('YAML file with VNF configuration')) help=_('YAML file with VNF configuration'))
parser.add_argument( group.add_argument(
'--config', '--config',
help=_('Specify config yaml data')) help=_('YAML data with VNF configuration'))
group.add_argument(
'--param-file',
help=_('YAML file with VNF parameter'))
def args2body(self, parsed_args): def args2body(self, parsed_args):
body = {self.resource: {}} body = {self.resource: {}}
@ -160,17 +166,33 @@ class UpdateVNF(tackerV10.UpdateCommand):
try: try:
config = yaml.load(config_yaml, Loader=yaml.SafeLoader) config = yaml.load(config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e: except yaml.YAMLError as e:
raise exceptions.InvalidInput(e) raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The config file is empty')
if parsed_args.config: if parsed_args.config:
config = parsed_args.config config_param = encodeutils.safe_decode(parsed_args.config)
if isinstance(config, str) or isinstance(config, unicode):
config_str = parsed_args.config.decode('unicode_escape')
try: try:
config = yaml.load(config_str, Loader=yaml.SafeLoader) config = yaml.load(config_param, Loader=yaml.SafeLoader)
except yaml.YAMLError as e: except yaml.YAMLError as e:
raise exceptions.InvalidInput(e) raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The parameter is empty')
if config: if config:
body[self.resource]['attributes'] = {'config': config} body[self.resource]['attributes'] = {'config': config}
if parsed_args.param_file:
with open(parsed_args.param_file) as f:
param_yaml = f.read()
try:
param = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not param:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
body[self.resource]['attributes'] = {'param_values': param}
tackerV10.update_dict(parsed_args, body[self.resource], ['tenant_id']) tackerV10.update_dict(parsed_args, body[self.resource], ['tenant_id'])
return body return body

View File

@ -0,0 +1,7 @@
auth_url: 'http://1.2.3.4:5000'
username: 'xyz'
password: '12345'
project_name: 'abc'
project_domain_name: 'prj_domain_name'
user_domain_name: 'user_domain_name'
type: 'openstack'

View File

@ -0,0 +1,7 @@
auth_url: ][
username: 'xyz'
password: '12345'
project_name: 'abc'
project_domain_name: 'prj_domain_name'
user_domain_name: 'user_domain_name'
type: 'openstack'

View File

@ -0,0 +1 @@
old_key: ][

View File

@ -0,0 +1 @@
key: new-value

View File

@ -0,0 +1,312 @@
# Copyright 2014 Intel 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 ast
import fixtures
import mock
import yaml
from tackerclient.common import exceptions
from tackerclient.common import utils
from tackerclient.osc.v1.vnfm import vnf
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import client
def _get_columns_vnf_parameter():
columns = ['attributes', 'project_id']
return columns
class TestVnfParameter(base.FixturedTestCase):
client_fixture_class = client.ClientFixture
def setUp(self):
super(TestVnfParameter, 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
class TestUpdateVNF(TestVnfParameter):
def setUp(self):
super(TestUpdateVNF, self).setUp()
self.useFixture(fixtures.MonkeyPatch(
'tackerclient.tacker.v1_0.find_resourceid_by_name_or_id',
self._find_resourceid))
self.set_vnf = vnf.UpdateVNF(
self.app, self.app_args, cmd_name='vnf set')
def _find_resourceid(self, client, resource, name_or_id):
return name_or_id
def _cmd_parser(self, cmd_parser, sub_argv):
_argv = sub_argv
index = -1
if '--' in sub_argv:
index = sub_argv.index('--')
_argv = sub_argv[:index]
known_args, _values_specs = cmd_parser.parse_known_args(_argv)
return known_args
def _get_vnf_data(self, vnf_parameter):
return tuple([vnf_parameter[key]
for key in sorted(vnf_parameter.keys())])
def _take_action(self, args, extra_fields, get_client_called_count=1):
cmd_par = self.set_vnf.get_parser("update_vnf")
parsed_args = self._cmd_parser(cmd_par, args)
body = {"vnf": extra_fields}
body = str(body)
project_id = {"tenant_id": "test-vnf-tenant_id"}
extra_fields.update(project_id)
json = {"vnf": extra_fields}
self.requests_mock.register_uri(
'PUT', self.url + '/v1.0/vnfs/my-id.json',
json=json, headers=self.header)
columns, data = (self.set_vnf.take_action(parsed_args))
self.assertItemsEqual(_get_columns_vnf_parameter(),
columns)
self.assertEqual(get_client_called_count,
self.requests_mock.call_count)
self.assertItemsEqual(
ast.literal_eval(self.requests_mock.last_request.body),
ast.literal_eval(body))
self.assertItemsEqual(self._get_vnf_data(json['vnf']), data)
def test_vnf_update_param_file(self):
my_id = 'my-id'
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
args = [my_id, '--param-file', str(param_file)]
extra_fields = {'attributes': {'param_values': {'key': 'new-value'}}}
self._take_action(args, extra_fields)
def test_vnf_update_config_file(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
p_auth_url = 'http://1.2.3.4:5000'
p_project_name = 'abc'
p_username = 'xyz'
p_project_domain_name = 'prj_domain_name'
p_type = 'openstack'
p_user_domain_name = 'user_domain_name'
p_password = '12345'
args = [my_id, '--config-file', str(config_file)]
config = {'auth_url': p_auth_url, 'project_name': p_project_name,
'username': p_username,
'project_domain_name': p_project_domain_name,
'type': p_type, 'user_domain_name': p_user_domain_name,
'password': p_password}
extra_fields = {'attributes': {'config': config}}
self._take_action(args, extra_fields)
def test_vnf_update_config(self):
my_id = 'my-id'
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
config = yaml.dump(config)
args = [my_id, '--config', str(config)]
extra_fields = {'attributes': {'config': config}}
self._take_action(args, extra_fields)
def test_vnf_update_invalid_format_param_file(self):
my_id = 'my-id'
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_invalid_format_param.yaml')
args = [my_id, '--param-file', str(param_file)]
extra_fields = {'attributes': {'param_values': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_empty_param_file(self):
my_id = 'my-id'
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_empty_param.yaml')
args = [my_id, '--param-file', str(param_file)]
extra_fields = {'attributes': {'param_values': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_invalid_format_config_file(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_invalid_format_config.yaml')
args = [my_id, '--config-file', str(config_file)]
extra_fields = {'attributes': {'config': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_empty_config_file(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_empty_config.yaml')
args = [my_id, '--config-file', str(config_file)]
extra_fields = {'attributes': {'config': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_invalid_format_config(self):
my_id = 'my_id'
config = 'test: : ]['
args = [my_id, '--config', config]
extra_fields = {'attributes': {'config': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_empty_config(self):
my_id = 'my-id'
config = ' '
args = [my_id, '--config', config]
extra_fields = {'attributes': {'config': None}}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_multi_args_config_configfile_paramfile(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
config = yaml.dump(config)
args = [my_id, '--config-file', str(config_file),
'--config', str(config), '--param-file', str(param_file)]
extra_fields = {'attributes': None}
self.assertRaises(BaseException,
self._take_action,
args, extra_fields)
def test_vnf_update_multi_args_configfile_paramfile(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
args = [my_id, '--param-file', str(param_file),
'--config-file', str(config_file)]
extra_fields = {'attributes': None}
self.assertRaises(BaseException,
self._take_action,
args, extra_fields)
def test_vnf_update_multi_args_config_configfile(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
config = yaml.dump(config)
args = [my_id, '--config-file', str(config_file),
'--config', str(config)]
extra_fields = {'attributes': None}
self.assertRaises(BaseException,
self._take_action,
args, extra_fields)
def test_vnf_update_multi_args_config_paramfile(self):
my_id = 'my-id'
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
config = yaml.dump(config)
args = [my_id, '--param-file', str(param_file),
'--config', str(config)]
extra_fields = {'attributes': None}
self.assertRaises(BaseException,
self._take_action,
args, extra_fields)
def test_vnf_update_param_file_with_empty_dict(self):
my_id = 'my-id'
param_file = utils.get_file_path(
'tests/unit/osc/samples/'
'vnf_update_param_file_with_empty_dict.yaml')
args = [my_id, '--param-file', str(param_file)]
extra_fields = {'attributes': None}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)
def test_vnf_update_config_file_with_empty_dict(self):
my_id = 'my-id'
config_file = utils.get_file_path(
'tests/unit/osc/samples/'
'vnf_update_config_file_with_empty_dict.yaml')
args = [my_id, '--config-file', str(config_file)]
extra_fields = {'attributes': None}
self.assertRaises(exceptions.InvalidInput,
self._take_action,
args, extra_fields)

View File

@ -573,6 +573,7 @@ class CLITestV10Base(testtools.TestCase):
'PUT', 'PUT',
body=_body, body=_body,
headers=test_utils.ContainsKeyValue('X-Auth-Token', TOKEN)) headers=test_utils.ContainsKeyValue('X-Auth-Token', TOKEN))
self.assertEqual(get_client_called_count, mock_get.call_count) self.assertEqual(get_client_called_count, mock_get.call_count)
_str = self.fake_stdout.make_string() _str = self.fake_stdout.make_string()
self.assertIn(myid, _str) self.assertIn(myid, _str)

View File

@ -13,10 +13,11 @@
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations # License for the specific language governing permissions and limitations
# under the License. # under the License.
import mock
import sys import sys
import mock from tackerclient.common import exceptions
from tackerclient.common import utils
from tackerclient import shell from tackerclient import shell
from tackerclient.tacker import v1_0 as tackerV1_0 from tackerclient.tacker import v1_0 as tackerV1_0
from tackerclient.tacker.v1_0 import TackerCommand from tackerclient.tacker.v1_0 import TackerCommand
@ -187,7 +188,313 @@ class CLITestV10VmVNFJSON(test_cli10.CLITestV10Base):
[my_id, '--%s' % key, value], [my_id, '--%s' % key, value],
{key: value}) {key: value})
def test_delete_vnf_without_force(self): def test_vnf_update_param_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
key = 'key'
value = 'new-value'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
args = [my_id, '--param-file', str(config_file)]
extra_fields = {'attributes': {'param_values': {key: value}}}
self._test_update_resource(self._RESOURCE, cmd, my_id, args,
extra_fields)
def test_vnf_update_config_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
args = [
name,
'--config-file', vim_config,
'--description', description]
p_auth_url = 'http://1.2.3.4:5000'
p_type = 'openstack'
p_project_name = 'abc'
p_username = 'xyz'
p_project_domain_name = 'prj_domain_name'
p_user_domain_name = 'user_domain_name'
p_password = '12345'
config = {'auth_url': p_auth_url, 'project_name': p_project_name,
'username': p_username,
'project_domain_name': p_project_domain_name,
'type': p_type, 'user_domain_name': p_user_domain_name,
'password': p_password}
extra_body = {'description': description,
'attributes': {'config': config}}
self._test_update_resource(self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_config(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
args = [
name,
'--description', description,
'--config', str(config)]
extra_body = {'description': description,
'attributes': {'config': config}}
self._test_update_resource(self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_invalid_format_param_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
name = 'my-name'
key = 'key'
value = 'new-value'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_invalid_format_param.yaml')
args = [my_id, '--param-file', str(config_file)]
extra_fields = {'attributes': {'param_values': {key: value}}}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_fields)
def test_vnf_update_empty_param_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
name = 'my-name'
key = 'key'
value = 'new-value'
config_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_empty_param.yaml')
args = [my_id, '--param-file', str(config_file)]
extra_fields = {'attributes': {'param_values': {key: value}}}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_fields)
def test_vnf_update_invalid_format_config_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_invalid_format_config.yaml')
args = [
name,
'--config-file', vim_config,
'--description', description]
p_auth_url = 'http://1.2.3.4:5000'
p_type = 'openstack'
p_project_name = 'abc'
p_username = 'xyz'
p_project_domain_name = 'prj_domain_name'
p_user_domain_name = 'user_domain_name'
p_password = '12345'
config = {'auth_url': p_auth_url, 'project_name': p_project_name,
'username': p_username,
'project_domain_name': p_project_domain_name, 'type': p_type,
'user_domain_name': p_user_domain_name,
'password': p_password}
extra_body = {'description': description,
'attributes': {'config': config}}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_empty_config_file(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_empty_config.yaml')
args = [
name,
'--config-file', vim_config,
'--description', description]
extra_body = {'description': description}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_invalid_format_config(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
name = 'vnfs/my-id'
key = 'key'
value = 'new-value'
description = 'Vim Description'
extra_fields = {'attributes': {'param_values': {key: value}}}
config = 'test: : ]['
args = [my_id,
'--config', config,
'--description', description]
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_fields)
def test_vnf_update_empty_config(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
config = {}
args = [name,
'--config', str(config),
'--description', description]
extra_body = {'description': description}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_multi_args_config_configfile_paramfile(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
key = 'key'
name = 'my-name'
value = 'new-value'
description = 'Vim Description'
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
args = [my_id,
'--config-file', str(vim_config),
'--param-file', str(param_file),
'--config', str(config),
'--description', description]
extra_fields = {'attributes': {'param_values': {key: value}}}
self.assertRaises(BaseException,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_fields)
def test_vnf_update_multi_args_configfile_paramfile(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
description = 'Vim Description'
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
args = [
my_id,
'--param-file', str(param_file),
'--config-file', vim_config,
'--description', description]
extra_body = {'description': description}
self.assertRaises(BaseException,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_multi_args_config_configfile(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
description = 'Vim Description'
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
vim_config = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_config.yaml')
args = [
my_id,
'--config-file', str(vim_config),
'--config', str(config),
'--description', description]
extra_body = {'description': description}
self.assertRaises(BaseException,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_multi_args_config_paramfile(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
description = 'Vim Description'
p_auth_url = 'https://1.2.3.4:6443'
p_type = 'kubernetes'
p_password = '12345'
p_project_name = 'default'
p_username = 'xyz'
p_ssl_ca_cert = 'abcxyz'
config = {'password': p_password, 'project_name': p_project_name,
'username': p_username, 'type': p_type,
'ssl_ca_cert': p_ssl_ca_cert, 'auth_url': p_auth_url}
param_file = utils.get_file_path(
'tests/unit/osc/samples/vnf_update_param.yaml')
args = [
my_id,
'--param-file', str(param_file),
'--config', str(config),
'--description', description]
extra_body = {'description': description}
self.assertRaises(BaseException,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_vnf_update_param_file_with_empty_dict(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
name = 'my-name'
key = 'key'
value = 'new-value'
config_file = utils.get_file_path(
'tests/unit/osc/samples/'
'vnf_update_param_file_with_empty_dict.yaml')
args = [my_id, '--param-file', str(config_file)]
extra_fields = {'attributes': {'param_values': {key: value}}}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_fields)
def test_vnf_update_config_file_with_empty_dict(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my-name'
description = 'Vim Description'
vim_config = utils.get_file_path(
'tests/unit/osc/samples/'
'vnf_update_config_file_with_empty_dict.yaml')
args = [
name,
'--config-file', vim_config,
'--description', description]
extra_body = {'description': description}
self.assertRaises(exceptions.InvalidInput,
self._test_update_resource,
self._RESOURCE, cmd, name, args, extra_body)
def test_delete_vnf(self):
cmd = vnf.DeleteVNF(test_cli10.MyApp(sys.stdout), None) cmd = vnf.DeleteVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id' my_id = 'my-id'
args = [my_id] args = [my_id]