Compare commits

..

8 Commits

Author SHA1 Message Date
Zuul
ad36c1f5b6 Merge "Update TOX/UPPER_CONSTRAINTS_FILE for stable/train" into stable/train 2019-09-23 05:51:35 +00:00
Zuul
ef82c12b25 Merge "Update .gitreview for stable/train" into stable/train 2019-09-23 05:51:14 +00:00
Zuul
8f2bdf326f Merge "Add restFul methods to fix backward compatibility issue" into stable/train 2019-09-23 05:40:21 +00:00
nirajsingh
57692126a0 Add restFul methods to fix backward compatibility issue
Tacker functional tests and applications using old version
of tackerclient might be using post/get/delete/put/list
method interfaces provided by Client in their applications.
These interfaces were removed in the recent changes [1]
because of which tacker tests are failing.

Added post/delete/put/get/list methods back to the Client class.

[1] : https://review.opendev.org/#/c/679956

Closes-Bug: #1844625

Change-Id: I803298dbc16dea9e58ec2a0a5fe3afe431c8532c
(cherry picked from commit 31b7690858)
2019-09-22 19:49:22 +00:00
Hiroya Nakaya
d5a591eec0 fix:osc scale is failed.
parameter which isn't used is removed.
changed Command type.

Closes-Bug:#1834548

Change-Id: Id884f9ebd26f311ebc213bcadff6019aaee5cd3a
(cherry picked from commit 983a2402a4)
2019-09-22 19:49:01 +00:00
Hiroya Nakaya
b98ef94e46 fix:"openstack vnf set" command can't execute
internal function's args is wrong.

Closes-Bug: #1839232

Change-Id: Iea1becdc0ee5dea3a94d843c558f28bce38e4d1c
(cherry picked from commit 9a1020fc84)
2019-09-22 15:40:24 +00:00
be6b25e49a Update TOX/UPPER_CONSTRAINTS_FILE for stable/train
Update the URL to the upper-constraints file to point to the redirect
rule on releases.openstack.org so that anyone working on this branch
will switch to the correct upper-constraints list automatically when
the requirements repository branches.

Until the requirements repository has as stable/train branch, tests will
continue to use the upper-constraints list on master.

Change-Id: I937e88fdb199839fc25122835bee43f63c7e3820
2019-09-20 17:43:22 +00:00
f716570413 Update .gitreview for stable/train
Change-Id: I8a3f66189e172a8517462f3c48ed3eb3fed6cf64
2019-09-20 17:43:20 +00:00
97 changed files with 457 additions and 7153 deletions

View File

@@ -2,4 +2,4 @@
host=review.opendev.org
port=29418
project=openstack/python-tackerclient.git
defaultbranch=stable/xena
defaultbranch=stable/train

View File

@@ -1,6 +1,8 @@
- project:
templates:
- check-requirements
- openstack-python3-xena-jobs
- openstack-lower-constraints-jobs
- openstack-python-jobs
- openstack-python3-train-jobs
- publish-openstack-docs-pti
- release-notes-jobs-python3

View File

@@ -2,6 +2,7 @@
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
# These are needed for docs generation
sphinx>=2.0.0,!=2.1.0 # BSD
openstackdocstheme>=2.2.1 # Apache-2.0
reno>=3.1.0 # Apache-2.0
sphinx!=1.6.6,!=1.6.7,>=1.6.2,<2.0.0;python_version=='2.7' # BSD
sphinx!=1.6.6,!=1.6.7,>=1.6.2;python_version>='3.4' # BSD
openstackdocstheme>=1.18.1 # Apache-2.0
reno>=2.5.0 # Apache-2.0

View File

@@ -44,6 +44,7 @@ source_suffix = '.rst'
master_doc = 'index'
# General information about the project.
project = 'python-tackerclient'
copyright = 'OpenStack Contributors'
# If true, '()' will be appended to :func: etc. cross-reference text.
@@ -54,7 +55,7 @@ add_function_parentheses = True
add_module_names = True
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'native'
pygments_style = 'sphinx'
# -- Options for HTML output --------------------------------------------------
@@ -67,20 +68,21 @@ htmlhelp_basename = 'tackerclientdoc'
# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
html_last_updated_fmt = '%Y-%m-%d %H:%M'
# -- Options for manual page output -------------------------------------------
man_pages = [
('cli/index', 'tacker', 'Client for Tacker API',
['OpenStack Contributors'], 1),
('cli/index', 'tacker', u'Client for Tacker API',
[u'OpenStack Contributors'], 1),
]
# -- Options for openstackdocstheme -------------------------------------------
openstackdocs_repo_name = 'openstack/python-tackerclient'
openstackdocs_bug_project = 'python-tackerclient'
openstackdocs_bug_tag = 'doc'
repository_name = 'openstack/python-tackerclient'
bug_project = 'python-tackerclient'
bug_tag = 'doc'
# -- Options for cliff.sphinxext plugin ---------------------------------------
autoprogram_cliff_application = 'openstack'
autoprogram_cliff_application = 'openstack'

View File

@@ -52,4 +52,5 @@ Indices and Tables
------------------
* :ref:`genindex`
* :ref:`modindex`
* :ref:`search`

View File

@@ -11,53 +11,59 @@
License for the specific language governing permissions and limitations
under the License.
Convention for heading levels in Neutron devref:
======= Heading 0 (reserved for the title in a document)
------- Heading 1
~~~~~~~ Heading 2
+++++++ Heading 3
''''''' Heading 4
(Avoid deeper levels because they do not render well.)
============
Installation
============
This document describes how to install python-tackerclient.
**Note:** The paths we are using for configuration files in these steps
are with reference to Ubuntu Operating System. The paths may vary for
other Operating Systems.
.. note::
This installation guide contents are specific to Ubuntu distro.
The branch_name which is used in commands, specify the branch_name
as stable/<branch> for any stable branch installation. For eg:
stable/queens, stable/pike. If unspecified the default will be
master branch.
Using python install
====================
1. Clone python-tackerclient repository.
#. Clone python-tackerclient repository.
::
You can use -b for specific release, optionally.
$ cd ~/
$ git clone https://github.com/openstack/python-tackerclient -b <branch_name>
.. code-block:: console
$ cd ~/
$ git clone https://opendev.org/openstack/python-tackerclient -b <branch_name>
2. Install python-tackerclient.
.. note::
::
Make sure to replace the ``<branch_name>`` in command example with
specific branch name, such as ``stable/victoria``.
$ cd python-tackerclient
$ sudo python setup.py install
#. Install python-tackerclient.
.. code-block:: console
$ cd python-tackerclient
$ sudo python3 setup.py install
Using pip
=========
You can also install the latest version by using ``pip`` command:
.. code-block:: console
::
$ pip install python-tackerclient
$ pip3 install python-tackerclient
Or, if it is needed to install ``python-tackerclient`` from master branch,
type
.. code-block:: console
::
$ pip3 install git+https://opendev.org/openstack/python-tackerclient
$ pip install git+https://github.com/openstack/python-tackerclient.git

63
lower-constraints.txt Normal file
View File

@@ -0,0 +1,63 @@
appdirs==1.3.0
Babel==2.3.4
cliff==2.8.0
cmd2==0.8.0
coverage==4.0
ddt==1.0.1
debtcollector==1.2.0
decorator==3.4.0
deprecation==1.0
dogpile.cache==0.6.2
extras==1.0.0
fixtures==3.0.0
flake8==2.5.5
hacking==0.12.0
iso8601==0.1.11
jmespath==0.9.0
jsonpatch==1.16
jsonpointer==1.13
keystoneauth1==3.4.0
linecache2==1.0.0
mccabe==0.2.1
mock==2.0.0
monotonic==0.6
msgpack-python==0.4.0
munch==2.1.0
netaddr==0.7.18
netifaces==0.10.4
openstacksdk==0.11.2
os-client-config==1.28.0
os-service-types==1.2.0
osc-lib==1.8.0
oslo.config==5.2.0
oslo.context==2.19.2
oslo.i18n==3.15.3
oslo.log==3.36.0
oslo.serialization==2.18.0
oslo.utils==3.33.0
pbr==2.0.0
pep8==1.5.7
positional==1.2.1
prettytable==0.7.2
pyflakes==0.8.1
pyinotify==0.9.6
pyparsing==2.1.0
pyperclip==1.5.27
python-dateutil==2.5.3
python-keystoneclient==3.8.0
python-mimeparse==1.6.0
python-subunit==1.0.0
pytz==2013.6
PyYAML==3.12
requests==2.14.2
requests-mock==1.2.0
requestsexceptions==1.2.0
rfc3986==0.3.1
simplejson==3.5.1
six==1.10.0
stestr==2.0.0
stevedore==1.20.0
testtools==2.2.0
traceback2==1.4.0
unittest2==1.1.0
wrapt==1.7.0

View File

@@ -1,6 +0,0 @@
---
upgrade:
- |
Python 2.7 support has been dropped. Last release of python-tackerclient
to support python 2.7 is OpenStack Train. The minimum version of Python now
supported by python-tackerclient is Python 3.6.

View File

@@ -53,8 +53,8 @@ source_suffix = '.rst'
master_doc = 'index'
# General information about the project.
project = 'Tacker Client Release Notes'
copyright = '2016, Tacker Developers'
project = u'Tacker Client Release Notes'
copyright = u'2016, Tacker Developers'
# Release notes are version independent.
release = ''
@@ -90,7 +90,7 @@ exclude_patterns = []
# show_authors = False
# The name of the Pygments (syntax highlighting) style to use.
pygments_style = 'native'
pygments_style = 'sphinx'
# A list of ignored prefixes for module index sorting.
# modindex_common_prefix = []
@@ -131,6 +131,10 @@ html_theme = 'openstackdocs'
# so a file named "default.css" will overwrite the builtin "default.css".
html_static_path = ['_static']
# If not '', a 'Last updated on:' timestamp is inserted at every page bottom,
# using the given strftime format.
# html_last_updated_fmt = '%b %d, %Y'
# If true, SmartyPants will be used to convert quotes and dashes to
# typographically correct entities.
# html_use_smartypants = True
@@ -190,8 +194,8 @@ latex_elements = {
# [howto/manual]).
latex_documents = [
('index', 'TackerClientReleaseNotes.tex',
'Tacker Client Release Notes Documentation',
'Tacker Developers', 'manual'),
u'Tacker Client Release Notes Documentation',
u'Tacker Developers', 'manual'),
]
# The name of an image file (relative to this directory) to place at the top of
@@ -221,8 +225,8 @@ latex_documents = [
# (source start file, name, description, authors, manual section).
man_pages = [
('index', 'tackerreleasenotes',
'Tacker Client Release Notes Documentation',
['Tacker Developers'], 1)
u'Tacker Client Release Notes Documentation',
[u'Tacker Developers'], 1)
]
# If true, show URL addresses after external links.
@@ -236,8 +240,8 @@ man_pages = [
# dir menu entry, description, category)
texinfo_documents = [
('index', 'TackerClientReleaseNotes',
'Tacker Client Release Notes Documentation',
'Tacker Developers', 'TackerClientReleaseNotes',
u'Tacker Client Release Notes Documentation',
u'Tacker Developers', 'TackerClientReleaseNotes',
'Tacker Client Project.',
'Miscellaneous'),
]
@@ -255,7 +259,6 @@ texinfo_documents = [
locale_dirs = ['locale/']
# -- Options for openstackdocstheme -------------------------------------------
openstackdocs_repo_name = 'openstack/python-tackerclient'
openstackdocs_bug_project = 'python-tackerclient'
openstackdocs_bug_tag = ''
openstackdocs_auto_name = False
repository_name = 'openstack/python-tackerclient'
bug_project = 'python-tackerclient'
bug_tag = ''

View File

@@ -7,9 +7,6 @@ Contents:
:maxdepth: 2
unreleased
victoria
ussuri
train
stein
rocky
queens

View File

@@ -1,6 +0,0 @@
==========================
Train Series Release Notes
==========================
.. release-notes::
:branch: stable/train

View File

@@ -1,6 +0,0 @@
===========================
Ussuri Series Release Notes
===========================
.. release-notes::
:branch: stable/ussuri

View File

@@ -1,6 +0,0 @@
=============================
Victoria Series Release Notes
=============================
.. release-notes::
:branch: stable/victoria

View File

@@ -8,10 +8,11 @@ netaddr>=0.7.18 # BSD
requests>=2.14.2 # Apache-2.0
python-keystoneclient>=3.8.0 # Apache-2.0
simplejson>=3.5.1 # MIT
six>=1.10.0 # MIT
stevedore>=1.20.0 # Apache-2.0
Babel!=2.4.0,>=2.3.4 # BSD
oslo.i18n>=3.15.3 # Apache-2.0
osc-lib>=1.8.0 # Apache-2.0
oslo.log>=3.36.0 # Apache-2.0
oslo.utils>=3.40.0 # Apache-2.0
oslo.utils>=3.33.0 # Apache-2.0
oslo.serialization!=2.19.1,>=2.18.0 # Apache-2.0

View File

@@ -6,7 +6,6 @@ description-file =
author = OpenStack
author-email = openstack-discuss@lists.openstack.org
home-page = https://docs.openstack.org/python-tackerclient/
python-requires = >=3.6
classifier =
Environment :: OpenStack
Intended Audience :: Developers
@@ -15,8 +14,8 @@ classifier =
License :: OSI Approved :: Apache Software License
Operating System :: POSIX :: Linux
Programming Language :: Python
Programming Language :: Python :: Implementation :: CPython
Programming Language :: Python :: 3 :: Only
Programming Language :: Python :: 2
Programming Language :: Python :: 2.7
Programming Language :: Python :: 3
Programming Language :: Python :: 3.6
Programming Language :: Python :: 3.7
@@ -25,6 +24,10 @@ classifier =
packages =
tackerclient
[global]
setup-hooks =
pbr.hooks.setup_hook
[entry_points]
console_scripts =
tacker = tackerclient.shell:main
@@ -82,32 +85,12 @@ openstack.tackerclient.v1 =
vnf_package_show = tackerclient.osc.v1.vnfpkgm.vnf_package:ShowVnfPackage
vnf_package_upload = tackerclient.osc.v1.vnfpkgm.vnf_package:UploadVnfPackage
vnf_package_delete = tackerclient.osc.v1.vnfpkgm.vnf_package:DeleteVnfPackage
vnf_package_update = tackerclient.osc.v1.vnfpkgm.vnf_package:UpdateVnfPackage
vnf_package_download = tackerclient.osc.v1.vnfpkgm.vnf_package:DownloadVnfPackage
vnf_package_artifact_download = tackerclient.osc.v1.vnfpkgm.vnf_package:DownloadVnfPackageArtifact
vnflcm_create = tackerclient.osc.v1.vnflcm.vnflcm:CreateVnfLcm
vnflcm_show = tackerclient.osc.v1.vnflcm.vnflcm:ShowVnfLcm
vnflcm_list = tackerclient.osc.v1.vnflcm.vnflcm:ListVnfLcm
vnflcm_instantiate = tackerclient.osc.v1.vnflcm.vnflcm:InstantiateVnfLcm
vnflcm_terminate = tackerclient.osc.v1.vnflcm.vnflcm:TerminateVnfLcm
vnflcm_delete = tackerclient.osc.v1.vnflcm.vnflcm:DeleteVnfLcm
vnflcm_heal = tackerclient.osc.v1.vnflcm.vnflcm:HealVnfLcm
vnflcm_update = tackerclient.osc.v1.vnflcm.vnflcm:UpdateVnfLcm
vnflcm_scale = tackerclient.osc.v1.vnflcm.vnflcm:ScaleVnfLcm
vnflcm_change-ext-conn = tackerclient.osc.v1.vnflcm.vnflcm:ChangeExtConnVnfLcm
vnflcm_op_rollback = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:RollbackVnfLcmOp
vnflcm_op_fail = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:FailVnfLcmOp
vnflcm_op_retry = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:RetryVnfLcmOp
vnflcm_op_list = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:ListVnfLcmOp
vnflcm_op_show = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:ShowVnfLcmOp
vnflcm_versions = tackerclient.osc.common.vnflcm.vnflcm_versions:VnfLcmVersions
openstack.tackerclient.v2 =
vnflcm_create = tackerclient.osc.v1.vnflcm.vnflcm:CreateVnfLcm
vnflcm_show = tackerclient.osc.v1.vnflcm.vnflcm:ShowVnfLcm
vnflcm_list = tackerclient.osc.v1.vnflcm.vnflcm:ListVnfLcm
vnflcm_instantiate = tackerclient.osc.v1.vnflcm.vnflcm:InstantiateVnfLcm
vnflcm_terminate = tackerclient.osc.v1.vnflcm.vnflcm:TerminateVnfLcm
vnflcm_delete = tackerclient.osc.v1.vnflcm.vnflcm:DeleteVnfLcm
vnflcm_op_list = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:ListVnfLcmOp
vnflcm_op_show = tackerclient.osc.v1.vnflcm.vnflcm_op_occs:ShowVnfLcmOp
vnflcm_versions = tackerclient.osc.common.vnflcm.vnflcm_versions:VnfLcmVersions
[build_releasenotes]
all_files = 1
build-dir = releasenotes/build
source-dir = releasenotes/source
[wheel]
universal = 1

View File

@@ -13,8 +13,17 @@
# See the License for the specific language governing permissions and
# limitations under the License.
# THIS FILE IS MANAGED BY THE GLOBAL REQUIREMENTS REPO - DO NOT EDIT
import setuptools
# In python < 2.7.4, a lazy loading of package `pbr` will break
# setuptools if some other modules registered functions in `atexit`.
# solution from: http://bugs.python.org/issue15881#msg170215
try:
import multiprocessing # noqa
except ImportError:
pass
setuptools.setup(
setup_requires=['pbr>=2.0.0'],
pbr=True)

View File

@@ -144,8 +144,6 @@ class HTTPClient(object):
verify=self.verify_cert,
timeout=self.timeout,
**kwargs)
if resp.headers.get('content-type') == 'application/zip':
return resp, resp.content
return resp, resp.text
@@ -165,7 +163,6 @@ class HTTPClient(object):
# re-authenticate and try again. If it still fails, bail.
try:
kwargs.setdefault('headers', {})
kwargs.setdefault('content_type', kwargs.get('content_type'))
if self.auth_token is None:
self.auth_token = ""
kwargs['headers']['X-Auth-Token'] = self.auth_token
@@ -293,10 +290,6 @@ class SessionClient(adapter.Adapter):
headers.setdefault('Content-Type', content_type)
resp = super(SessionClient, self).request(*args, **kwargs)
if resp.headers.get('content-type') == 'application/zip':
return resp, resp.content
return resp, resp.text
def _check_uri_length(self, url):

View File

@@ -37,4 +37,4 @@ _P = _translators.plural_form
def get_available_languages():
return oslo_i18n.get_available_languages(DOMAIN)
return oslo_i18n.get_available_languages(DOMAIN)

View File

@@ -28,6 +28,7 @@ ATOM_LINK_NOTATION = "{%s}link" % ATOM_NAMESPACE
TYPE_BOOL = "bool"
TYPE_INT = "int"
TYPE_LONG = "long"
TYPE_FLOAT = "float"
TYPE_LIST = "list"
TYPE_DICT = "dict"

View File

@@ -18,6 +18,7 @@ from xml.etree import ElementTree as etree
from xml.parsers import expat
from oslo_serialization import jsonutils
import six
from tackerclient.common import constants
from tackerclient.common import exceptions as exception
@@ -25,6 +26,9 @@ from tackerclient.i18n import _
LOG = logging.getLogger(__name__)
if six.PY3:
long = int
class ActionDispatcher(object):
"""Maps method name to local methods through action name."""
@@ -54,7 +58,7 @@ class JSONDictSerializer(DictSerializer):
def default(self, data):
def sanitizer(obj):
return str(obj)
return six.text_type(obj)
return jsonutils.dumps(data, default=sanitizer)
@@ -89,7 +93,7 @@ class XMLDictSerializer(DictSerializer):
root_key = constants.VIRTUAL_ROOT_KEY
root_value = None
else:
link_keys = [k for k in data.keys() or []
link_keys = [k for k in six.iterkeys(data) or []
if k.endswith('_links')]
if link_keys:
links = data.pop(link_keys[0], None)
@@ -179,6 +183,10 @@ class XMLDictSerializer(DictSerializer):
result.set(
constants.TYPE_ATTR,
constants.TYPE_INT)
elif isinstance(data, long):
result.set(
constants.TYPE_ATTR,
constants.TYPE_LONG)
elif isinstance(data, float):
result.set(
constants.TYPE_ATTR,
@@ -186,7 +194,7 @@ class XMLDictSerializer(DictSerializer):
LOG.debug("Data %(data)s type is %(type)s",
{'data': data,
'type': type(data)})
result.text = str(data)
result.text = six.text_type(data)
return result
def _create_link_nodes(self, xml_doc, links):
@@ -315,6 +323,8 @@ class XMLDeserializer(TextDeserializer):
lambda x: x.lower() == 'true',
constants.TYPE_INT:
lambda x: int(x),
constants.TYPE_LONG:
lambda x: long(x),
constants.TYPE_FLOAT:
lambda x: float(x)}
if attrType and attrType in converters:

View File

@@ -24,6 +24,7 @@ import os
from oslo_log import versionutils
from oslo_utils import encodeutils
from oslo_utils import importutils
import six
from tackerclient.common import exceptions
from tackerclient.i18n import _
@@ -56,7 +57,7 @@ def get_client_class(api_name, version, version_map):
"one of: %(map_keys)s")
msg = msg % {'api_name': api_name, 'version': version,
'map_keys': ', '.join(version_map.keys())}
raise exceptions.UnsupportedVersion(message=msg)
raise exceptions.UnsupportedVersion(msg)
return importutils.import_class(client_path)
@@ -140,7 +141,7 @@ def http_log_resp(_logger, resp, body):
def _safe_encode_without_obj(data):
if isinstance(data, str):
if isinstance(data, six.string_types):
return encodeutils.safe_encode(data)
return data

View File

@@ -54,7 +54,7 @@ def validate_int_range(parsed_args, attr_name, min_value=None, max_value=None):
{'attr_name': attr_name.replace('_', '-'),
'val': val})
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
def validate_ip_subnet(parsed_args, attr_name):
@@ -65,5 +65,5 @@ def validate_ip_subnet(parsed_args, attr_name):
netaddr.IPNetwork(val)
except (netaddr.AddrFormatError, ValueError):
raise exceptions.CommandError(
message=(_('%(attr_name)s "%(val)s" is not a valid CIDR.') %
{'attr_name': attr_name.replace('_', '-'), 'val': val}))
(_('%(attr_name)s "%(val)s" is not a valid CIDR.') %
{'attr_name': attr_name.replace('_', '-'), 'val': val}))

View File

@@ -1,49 +0,0 @@
# Copyright (C) 2021 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from osc_lib.command import command
from tackerclient.common import exceptions
from tackerclient.i18n import _
SUPPORTED_VERSIONS = [1, 2]
class VnfLcmVersions(command.ShowOne):
_description = _("Show VnfLcm Api versions")
def get_parser(self, prog_name):
parser = super(VnfLcmVersions, self).get_parser(prog_name)
parser.add_argument(
'--major-version',
metavar="<major-version>",
type=int,
help=_('Show only specify major version.'))
return parser
def take_action(self, parsed_args):
v = None
if parsed_args.major_version:
if parsed_args.major_version not in SUPPORTED_VERSIONS:
msg = _("Major version %d is not supported")
reason = msg % parsed_args.major_version
raise exceptions.InvalidInput(reason=reason)
v = "v{}".format(parsed_args.major_version)
client = self.app.client_manager.tackerclient
data = client.show_vnf_lcm_versions(v)
return (tuple(data.keys()), tuple(data.values()))

View File

@@ -26,17 +26,15 @@ API_NAME = 'tackerclient'
API_VERSION_OPTION = 'os_tacker_api_version'
API_VERSIONS = {
'1': 'tackerclient.v1_0.client.Client',
'2': 'tackerclient.v1_0.client.Client',
}
def make_client(instance):
"""Returns a client to the ClientManager."""
api_version = instance._api_version[API_NAME]
tacker_client = utils.get_client_class(
API_NAME,
api_version,
instance._api_version[API_NAME],
API_VERSIONS)
LOG.debug('Instantiating tacker client: %s', tacker_client)
@@ -44,8 +42,7 @@ def make_client(instance):
'region_name': instance._region_name,
'endpoint_type': instance._interface,
'interface': instance._interface,
'session': instance.session,
'api_version': api_version
'session': instance.session
}
client = tacker_client(**kwargs)

View File

@@ -10,10 +10,6 @@
# License for the specific language governing permissions and limitations
# under the License.
import sys
from oslo_utils import encodeutils
def get_osc_show_columns_for_sdk_resource(
sdk_resource,
@@ -104,26 +100,3 @@ class DictModel(dict):
def __str__(self):
pairs = ['%s=%s' % (k, v) for k, v in self.items()]
return ', '.join(sorted(pairs))
def save_data(data, path):
"""Save data to the specified path.
:param data: binary or string data
:param path: file path to save data
"""
if path is None:
vnfpackage = getattr(sys.stdout, 'buffer', sys.stdout)
else:
mode = 'wb' if isinstance(data, bytes) else 'w'
vnfpackage = open(path, mode)
try:
vnfpackage.write(data)
finally:
vnfpackage.close()
def exit(msg=None, exit_code=1):
if msg:
print(encodeutils.safe_decode(msg))
sys.exit(exit_code)

View File

@@ -22,7 +22,6 @@ to this module. They should go to tackerclient.osc.v1.utils.
import operator
from cliff import columns as cliff_columns
from keystoneclient import exceptions as identity_exc
from keystoneclient.v3 import domains
from keystoneclient.v3 import projects
@@ -206,9 +205,3 @@ def _find_identity_resource(identity_client_manager, name_or_id,
# The above are borrowed from openstackclient.identity.common.
class FormatComplexDataColumn(cliff_columns.FormattableColumn):
def human_readable(self):
return format_dict_with_indention(self._value)

View File

@@ -109,10 +109,9 @@ class CreateNS(command.ShowOne):
template = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not template:
raise exceptions.InvalidInput(
reason='The nsd file is empty')
raise exceptions.InvalidInput('The nsd file is empty')
body[_NS]['nsd_template'] = template
if parsed_args.param_file:
@@ -122,10 +121,9 @@ class CreateNS(command.ShowOne):
param_yaml = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not param_yaml:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
raise exceptions.InvalidInput('The parameter file is empty')
body[_NS]['attributes'] = {'param_values': param_yaml}
tackerV10.update_dict(parsed_args, body[_NS],
['tenant_id', 'name', 'description',
@@ -197,12 +195,12 @@ class DeleteNS(command.Command):
'resource': _NS})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _NS
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _NS}))

View File

@@ -71,15 +71,15 @@ class CreateNSD(command.ShowOne):
body = {_NSD: {}}
nsd = None
if not parsed_args.nsd_file:
raise exceptions.InvalidInput(reason="Invalid input for nsd file")
raise exceptions.InvalidInput("Invalid input for nsd file")
with open(parsed_args.nsd_file) as f:
nsd = f.read()
try:
nsd = yaml.load(nsd, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not nsd:
raise exceptions.InvalidInput(reason="nsd file is empty")
raise exceptions.InvalidInput("nsd file is empty")
body[_NSD]['attributes'] = {'nsd': nsd}
tackerV10.update_dict(parsed_args, body[_NSD],
['tenant_id', 'name', 'description'])
@@ -133,12 +133,12 @@ class DeleteNSD(command.Command):
'resource': _NSD})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _NSD
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _NSD}))
@@ -217,7 +217,7 @@ class ShowTemplateNSD(command.ShowOne):
obj[_NSD]['attributes']['nsd'])
data = utils.get_item_properties(
sdk_utils.DictModel(obj[_NSD]),
('attributes',),
(u'attributes',),
formatters=_formatters)
data = (data or _('Unable to display NSD template!'))
return (('attributes',), data)
return ((u'attributes',), data)

View File

@@ -124,7 +124,7 @@ class CreateVIM(command.ShowOne):
config_param = yaml.load(vim_config,
Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
vim_obj = body[_VIM]
try:
auth_url = config_param.pop('auth_url')
@@ -185,12 +185,12 @@ class DeleteVIM(command.Command):
'resource': _VIM})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _VIM
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _VIM}))
@@ -230,10 +230,9 @@ class UpdateVIM(command.ShowOne):
with open(parsed_args.config_file) as f:
config_yaml = f.read()
try:
config_param = yaml.load(config_yaml,
Loader=yaml.SafeLoader)
config_param = yaml.load(config_yaml)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
vim_obj = body[_VIM]
if config_param is not None:
vim_utils.args2body_vim(config_param, vim_obj)

View File

@@ -145,10 +145,9 @@ class CreateVNFFG(command.ShowOne):
try:
template = yaml.load(template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not template:
raise exceptions.InvalidInput(
reason='The vnffgd file is empty')
raise exceptions.InvalidInput('The vnffgd file is empty')
body[_VNFFG]['vnffgd_template'] = template
if parsed_args.param_file:
@@ -158,10 +157,9 @@ class CreateVNFFG(command.ShowOne):
param_yaml = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not param_yaml:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
raise exceptions.InvalidInput('The parameter file is empty')
body[_VNFFG]['attributes'] = {'param_values': param_yaml}
tackerV10.update_dict(parsed_args, body[_VNFFG],
['tenant_id', 'name', 'vnffgd_id',
@@ -190,32 +188,18 @@ class DeleteVNFFG(command.Command):
nargs="+",
help=_("VNFFG(s) to delete (name or ID)")
)
parser.add_argument(
'--force',
default=False,
action='store_true',
help=_('Force delete VNFFG'))
return parser
def args2body(self, parsed_args):
body = dict()
if parsed_args.force:
body[_VNFFG] = dict()
body[_VNFFG]['attributes'] = dict()
body[_VNFFG]['attributes']['force'] = True
return body
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
failure = False
deleted_ids = []
failed_items = {}
body = self.args2body(parsed_args)
for resource_id in parsed_args.vnffg:
try:
obj = tackerV10.find_resourceid_by_name_or_id(
client, _VNFFG, resource_id)
client.delete_vnffg(obj, body)
client.delete_vnffg(obj)
deleted_ids.append(resource_id)
except Exception as e:
failure = True
@@ -229,12 +213,12 @@ class DeleteVNFFG(command.Command):
'resource': _VNFFG})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _VNFFG
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _VNFFG}))
@@ -265,9 +249,6 @@ class UpdateVNFFG(command.ShowOne):
parser.add_argument(
'--description',
help=_('Set a description for the VNFFG'))
parser.add_argument(
'--param-file',
help=_('YAML file with specific VNFFG parameters'))
return parser
def args2body(self, parsed_args):
@@ -293,10 +274,9 @@ class UpdateVNFFG(command.ShowOne):
template = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not template:
raise exceptions.InvalidInput(
reason='The vnffgd file is empty')
raise exceptions.InvalidInput('The vnffgd file is empty')
body[_VNFFG]['vnffgd_template'] = template
if parsed_args.param_file:
@@ -306,12 +286,11 @@ class UpdateVNFFG(command.ShowOne):
param_yaml = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not param_yaml:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
raise exceptions.InvalidInput('The parameter file is empty')
body[_VNFFG]['attributes'] = {'param_values': param_yaml}
tackerV10.update_dict(parsed_args, body[_VNFFG],
tackerV10.update_dict(parsed_args, body[self.resource],
['vnf_mapping', 'symmetrical', 'description'])
return body

View File

@@ -71,16 +71,15 @@ class CreateVNFFGD(command.ShowOne):
body = {_VNFFGD: {}}
vnffgd = None
if not parsed_args.vnffgd_file:
raise exceptions.InvalidInput(
reason="Invalid input for vnffgd file")
raise exceptions.InvalidInput("Invalid input for vnffgd file")
with open(parsed_args.vnffgd_file) as f:
vnffgd = f.read()
try:
vnffgd = yaml.load(vnffgd, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not vnffgd:
raise exceptions.InvalidInput(reason="vnffgd file is empty")
raise exceptions.InvalidInput("vnffgd file is empty")
body[_VNFFGD]['template'] = {'vnffgd': vnffgd}
tackerV10.update_dict(parsed_args, body[_VNFFGD],
['tenant_id', 'name', 'description'])
@@ -132,12 +131,12 @@ class DeleteVNFFGD(command.Command):
'resource': _VNFFGD})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _VNFFGD
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _VNFFGD}))
@@ -211,7 +210,7 @@ class ShowTemplateVNFFGD(command.ShowOne):
obj = client.show_vnffgd(obj_id)
data = utils.get_item_properties(
sdk_utils.DictModel(obj[_VNFFGD]),
('template',),
(u'template',),
formatters=_formatters)
data = (data or _('Unable to display VNFFGD template!'))
return (('template',), data)
return ((u'template',), data)

View File

@@ -1,69 +0,0 @@
{
"extVirtualLinks": [
{
"id": "ext-vl-uuid-VL1",
"resourceId": "neutron-network-uuid_VL1",
"extCps": [
{
"cpdId": "CP1",
"cpConfig": [
{
"cpProtocolData": [
{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
"ipAddresses": [
{
"type": "IPV4",
"numDynamicAddresses": 1,
"subnetId": "subnet-uuid"
}
]
}
}
]
}
]
},
{
"cpdId": "CP2",
"cpConfig": [
{
"cpProtocolData": [
{
"layerProtocol": "IP_OVER_ETHERNET",
"ipOverEthernet": {
"ipAddresses": [
{
"type": "IPV4",
"fixedAddresses": [
"10.0.0.1"
],
"subnetId": "subnet-uuid"
}
]
}
}
]
}
]
}
]
}
],
"vimConnectionInfo": [
{
"id": "vim-uuid",
"vimType": "ETSINFV.OPENSTACK_KEYSTONE.v_2",
"vimConnectionId": "dummy-vimid",
"interfaceInfo": {
"key1":"value1",
"key2":"value2"
},
"accessInfo": {
"key1":"value1",
"key2":"value2"
}
}
]
}

View File

@@ -1,79 +0,0 @@
{
"flavourId":"simple",
"instantiationLevelId":"instantiation_level_1",
"extVirtualLinks":[
{
"id":"ext-vl-uuid-VL1",
"vimConnectionId":"vim-uuid",
"resourceProviderId":"resource-provider-id",
"resourceId":"neutron-network-uuid_VL1",
"extCps":[
{
"cpdId":"CP1",
"cpConfig":[
{
"cpInstanceId":"cp-instance-id",
"linkPortId":"link-port-uuid_CP1",
"cpProtocolData":[
{
"layerProtocol":"IP_OVER_ETHERNET",
"ipOverEthernet":{
"macAddress":"00:25:96:FF:FE:12:34:56",
"ipAddresses":[
{
"addressRange":{
"minAddress":"192.168.11.01",
"maxAddress":"192.168.21.201"
},
"subnetId":"neutron-subnet-uuid_CP1"
}
]
}
}
]
}
]
}
],
"extLinkPorts":[
{
"id":"link-port-uuid_CP1",
"resourceHandle":{
"vimConnectionId":"vim-uuid",
"resourceProviderId":"resource-provider-id",
"resourceId":"neutron-port-uuid_CP1",
"vimLevelResourceType":"LINKPORT"
}
}
]
}
],
"extManagedVirtualLinks":[
{
"id":"extMngVLnk-uuid_VL3",
"vnfVirtualLinkDescId":"VL3",
"vimConnectionId":"vim-uuid",
"resourceProviderId":"resource-provider-id",
"resourceId":"neutron-network-uuid_VL3"
}
],
"vimConnectionInfo":[
{
"id":"vim-uuid",
"vimId":"dummy-vimid",
"vimType":"ETSINFV.OPENSTACK_KEYSTONE.v_2",
"interfaceInfo":{
"key1":"value1",
"key2":"value2"
},
"accessInfo":{
"key1":"value1",
"key2":"value2"
},
"extra":{
"key1":"value1",
"key2":"value2"
}
}
]
}

View File

@@ -1,3 +0,0 @@
{
"additionalParams": {"key1":"value1", "key2":"value2"}
}

View File

@@ -1,5 +0,0 @@
{
"vnfInstanceName": "sample",
"vnfInstanceDescription" : "sample_description",
"vnfdId" : "sample_id"
}

View File

@@ -1,568 +0,0 @@
# 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 json
import logging
import os
import time
from osc_lib.command import command
from osc_lib import utils
from tackerclient.common import exceptions
from tackerclient.i18n import _
from tackerclient.osc import sdk_utils
from tackerclient.osc import utils as tacker_osc_utils
_attr_map = (
('id', 'ID', tacker_osc_utils.LIST_BOTH),
('vnfInstanceName', 'VNF Instance Name', tacker_osc_utils.LIST_BOTH),
('instantiationState', 'Instantiation State', tacker_osc_utils.LIST_BOTH),
('vnfProvider', 'VNF Provider', tacker_osc_utils.LIST_BOTH),
('vnfSoftwareVersion', 'VNF Software Version', tacker_osc_utils.LIST_BOTH),
('vnfProductName', 'VNF Product Name', tacker_osc_utils.LIST_BOTH),
('vnfdId', 'VNFD ID', tacker_osc_utils.LIST_BOTH)
)
LOG = logging.getLogger(__name__)
_mixed_case_fields = ('vnfInstanceName', 'vnfInstanceDescription', 'vnfdId',
'vnfProvider', 'vnfProductName', 'vnfSoftwareVersion',
'vnfdVersion', 'instantiationState',
'vimConnectionInfo', 'instantiatedVnfInfo',
'vnfConfigurableProperties')
_VNF_INSTANCE = 'vnf_instance'
VNF_INSTANCE_TERMINATION_TIMEOUT = 300
EXTRA_WAITING_TIME = 10
SLEEP_TIME = 1
formatters = {'vimConnectionInfo': tacker_osc_utils.FormatComplexDataColumn,
'instantiatedVnfInfo': tacker_osc_utils.FormatComplexDataColumn,
'_links': tacker_osc_utils.FormatComplexDataColumn}
def _get_columns(vnflcm_obj, action=None):
column_map = {
'id': 'ID',
'vnfInstanceName': 'VNF Instance Name',
'vnfInstanceDescription': 'VNF Instance Description',
'vnfdId': 'VNFD ID',
'vnfProvider': 'VNF Provider',
'vnfProductName': 'VNF Product Name',
'vnfSoftwareVersion': 'VNF Software Version',
'vnfdVersion': 'VNFD Version',
'instantiationState': 'Instantiation State',
'_links': 'Links',
'vnfConfigurableProperties': 'VNF Configurable Properties',
}
if action == 'show':
if vnflcm_obj['instantiationState'] == 'INSTANTIATED':
column_map.update(
{'instantiatedVnfInfo': 'Instantiated Vnf Info'}
)
column_map.update(
{'vimConnectionInfo': 'VIM Connection Info',
'_links': 'Links'}
)
return sdk_utils.get_osc_show_columns_for_sdk_resource(vnflcm_obj,
column_map)
class CreateVnfLcm(command.ShowOne):
_description = _("Create a new VNF Instance")
def get_parser(self, prog_name):
parser = super(CreateVnfLcm, self).get_parser(prog_name)
parser.add_argument(
'vnfd_id',
metavar="<vnfd-id>",
help=_('Identifier that identifies the VNFD which defines the '
'VNF instance to be created.'))
parser.add_argument(
'--name',
metavar="<vnf-instance-name>",
help=_('Name of the VNF instance to be created.'))
parser.add_argument(
'--description',
metavar="<vnf-instance-description>",
help=_('Description of the VNF instance to be created.'))
parser.add_argument(
'--I',
metavar="<param-file>",
help=_("Instantiate VNF subsequently after it's creation. "
"Specify instantiate request parameters in a json file."))
return parser
def args2body(self, parsed_args, file_path=None):
body = {}
if file_path:
return jsonfile2body(file_path)
body['vnfdId'] = parsed_args.vnfd_id
if parsed_args.description:
body['vnfInstanceDescription'] = parsed_args.description
if parsed_args.name:
body['vnfInstanceName'] = parsed_args.name
return body
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
vnf = client.create_vnf_instance(self.args2body(parsed_args))
if parsed_args.I:
# Instantiate VNF instance.
result = client.instantiate_vnf_instance(
vnf['id'],
self.args2body(parsed_args, file_path=parsed_args.I))
if not result:
print((_('VNF Instance %(id)s is created and instantiation'
' request has been accepted.') % {'id': vnf['id']}))
display_columns, columns = _get_columns(vnf)
data = utils.get_item_properties(sdk_utils.DictModel(vnf),
columns, formatters=formatters,
mixed_case_fields=_mixed_case_fields)
return (display_columns, data)
class ShowVnfLcm(command.ShowOne):
_description = _("Display VNF instance details")
def get_parser(self, prog_name):
parser = super(ShowVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_("VNF instance ID to display"))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
obj = client.show_vnf_instance(parsed_args.vnf_instance)
display_columns, columns = _get_columns(obj, action='show')
data = utils.get_item_properties(
sdk_utils.DictModel(obj),
columns, mixed_case_fields=_mixed_case_fields,
formatters=formatters)
return (display_columns, data)
class ListVnfLcm(command.Lister):
_description = _("List VNF Instance")
def get_parser(self, prog_name):
parser = super(ListVnfLcm, self).get_parser(prog_name)
return parser
def take_action(self, parsed_args):
_params = {}
client = self.app.client_manager.tackerclient
vnf_instances = client.list_vnf_instances(**_params)
headers, columns = tacker_osc_utils.get_column_definitions(
_attr_map, long_listing=True)
return (headers,
(utils.get_dict_properties(
s, columns, mixed_case_fields=_mixed_case_fields,
) for s in vnf_instances))
def jsonfile2body(file_path):
if file_path is not None and os.access(file_path, os.R_OK) is False:
msg = _("File %s does not exist or user does not have read "
"privileges to it")
reason = msg % file_path
raise exceptions.InvalidInput(reason=reason)
try:
with open(file_path) as f:
body = json.load(f)
except (IOError, ValueError) as ex:
msg = _("Failed to load parameter file. Error: %s")
reason = msg % ex
raise exceptions.InvalidInput(reason=reason)
if not body:
reason = _('The parameter file is empty')
raise exceptions.InvalidInput(reason=reason)
return body
class InstantiateVnfLcm(command.Command):
_description = _("Instantiate a VNF Instance")
def get_parser(self, prog_name):
parser = super(InstantiateVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_("VNF instance ID to instantiate"))
parser.add_argument(
'instantiation_request_file',
metavar="<param-file>",
help=_('Specify instantiate request parameters in a json file.'))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
result = client.instantiate_vnf_instance(
parsed_args.vnf_instance, jsonfile2body(
parsed_args.instantiation_request_file))
if not result:
print((_('Instantiate request for VNF Instance %(id)s has been'
' accepted.') % {'id': parsed_args.vnf_instance}))
class HealVnfLcm(command.Command):
_description = _("Heal VNF Instance")
def get_parser(self, prog_name):
parser = super(HealVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_("VNF instance ID to heal"))
parser.add_argument(
'--cause',
help=_('Specify the reason why a healing procedure is required.'))
parser.add_argument(
'--vnfc-instance',
metavar="<vnfc-instance-id>",
nargs="+",
help=_("List of VNFC instances requiring a healing action.")
)
return parser
def args2body(self, parsed_args):
body = {}
if parsed_args.cause:
body['cause'] = parsed_args.cause
if parsed_args.vnfc_instance:
body['vnfcInstanceId'] = parsed_args.vnfc_instance
return body
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
result = client.heal_vnf_instance(
parsed_args.vnf_instance, self.args2body(parsed_args))
if not result:
print((_('Heal request for VNF Instance %(id)s has been'
' accepted.') % {'id': parsed_args.vnf_instance}))
class TerminateVnfLcm(command.Command):
_description = _("Terminate a VNF instance")
def get_parser(self, prog_name):
parser = super(TerminateVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_("VNF instance ID to terminate"))
parser.add_argument(
"--termination-type",
default='GRACEFUL',
metavar="<termination-type>",
choices=['GRACEFUL', 'FORCEFUL'],
help=_("Termination type can be 'GRACEFUL' or 'FORCEFUL'. "
"Default is 'GRACEFUL'"))
parser.add_argument(
'--graceful-termination-timeout',
metavar="<graceful-termination-timeout>",
type=int,
help=_('This attribute is only applicable in case of graceful '
'termination. It defines the time to wait for the VNF to be'
' taken out of service before shutting down the VNF and '
'releasing the resources. The unit is seconds.'))
parser.add_argument(
'--D',
action='store_true',
default=False,
help=_("Delete VNF Instance subsequently after it's termination"),
)
return parser
def args2body(self, parsed_args):
body = {}
body['terminationType'] = parsed_args.termination_type
if parsed_args.graceful_termination_timeout:
if parsed_args.termination_type == 'FORCEFUL':
exceptions.InvalidInput(reason='--graceful-termination-timeout'
' argument is invalid for "FORCEFUL"'
' termination')
body['gracefulTerminationTimeout'] = parsed_args.\
graceful_termination_timeout
return body
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
result = client.terminate_vnf_instance(parsed_args.vnf_instance,
self.args2body(parsed_args))
if not result:
print(_("Terminate request for VNF Instance '%(id)s' has been"
" accepted.") % {'id': parsed_args.vnf_instance})
if parsed_args.D:
print(_("Waiting for vnf instance to be terminated before "
"deleting"))
self._wait_until_vnf_is_terminated(
client, parsed_args.vnf_instance,
graceful_timeout=parsed_args.graceful_termination_timeout)
result = client.delete_vnf_instance(parsed_args.vnf_instance)
if not result:
print(_("VNF Instance '%(id)s' deleted successfully") %
{'id': parsed_args.vnf_instance})
def _wait_until_vnf_is_terminated(self, client, vnf_instance_id,
graceful_timeout=None):
# wait until vnf instance 'instantiationState' is set to
# 'NOT_INSTANTIATED'
if graceful_timeout:
# If graceful_termination_timeout is provided,
# terminate vnf will start after this timeout period.
# Hence, it should wait for extra time of 10 seconds
# after this graceful_termination_timeout period.
timeout = graceful_timeout + EXTRA_WAITING_TIME
else:
timeout = VNF_INSTANCE_TERMINATION_TIMEOUT
start_time = int(time.time())
while True:
vnf_instance = client.show_vnf_instance(vnf_instance_id)
if vnf_instance['instantiationState'] == 'NOT_INSTANTIATED':
break
if ((int(time.time()) - start_time) > timeout):
msg = _("Couldn't verify vnf instance is terminated within "
"'%(timeout)s' seconds. Unable to delete vnf instance "
"%(id)s")
raise exceptions.CommandError(
message=msg % {'timeout': timeout, 'id': vnf_instance_id})
time.sleep(SLEEP_TIME)
class DeleteVnfLcm(command.Command):
"""Vnf lcm delete
DeleteVnfLcm class supports bulk deletion of vnf instances, and error
handling.
"""
_description = _("Delete VNF Instance(s)")
def get_parser(self, prog_name):
parser = super(DeleteVnfLcm, self).get_parser(prog_name)
parser.add_argument(
'vnf_instances',
metavar="<vnf-instance>",
nargs="+",
help=_("VNF instance ID(s) to delete"))
return parser
def take_action(self, parsed_args):
error_count = 0
client = self.app.client_manager.tackerclient
vnf_instances = parsed_args.vnf_instances
for vnf_instance in vnf_instances:
try:
client.delete_vnf_instance(vnf_instance)
except Exception as e:
error_count += 1
LOG.error(_("Failed to delete vnf instance with "
"ID '%(vnf)s': %(e)s"),
{'vnf': vnf_instance, 'e': e})
total = len(vnf_instances)
if (error_count > 0):
msg = (_("Failed to delete %(error_count)s of %(total)s "
"vnf instances.") % {'error_count': error_count,
'total': total})
raise exceptions.CommandError(message=msg)
else:
if total > 1:
print(_('All specified vnf instances are deleted '
'successfully'))
else:
print(_("Vnf instance '%s' deleted "
"successfully") % vnf_instances[0])
class UpdateVnfLcm(command.Command):
_description = _("Update VNF Instance")
def get_parser(self, prog_name):
"""Add arguments to parser.
Args:
prog_name ([string]): program name
Returns:
parser([ArgumentParser]): [description]
"""
parser = super(UpdateVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_('VNF instance ID to update.'))
parser.add_argument(
'--I',
metavar="<param-file>",
help=_("Specify update request parameters in a json file."))
return parser
def args2body(self, file_path=None):
"""Call jsonfile2body to store request body to body(dict)
Args:
file_path ([string], optional): file path of param file(json).
Defaults to None.
Returns:
body ([dict]): Request body is stored
"""
body = {}
if file_path:
return jsonfile2body(file_path)
return body
def take_action(self, parsed_args):
"""Execute update_vnf_instance and output result comment
Args:
parsed_args ([Namespace]): [description]
"""
client = self.app.client_manager.tackerclient
if parsed_args.I:
# Update VNF instance.
result = client.update_vnf_instance(
parsed_args.vnf_instance,
self.args2body(file_path=parsed_args.I))
if not result:
print((_('Update vnf:%(id)s ') %
{'id': parsed_args.vnf_instance}))
class ScaleVnfLcm(command.Command):
_description = _("Scale a VNF Instance")
def get_parser(self, prog_name):
parser = super(ScaleVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_('VNF instance ID to scale'))
parser.add_argument(
'--number-of-steps',
metavar="<number-of-steps>",
type=int,
help=_("Number of scaling steps to be executed as part of "
"this Scale VNF operation."))
parser.add_argument(
'--additional-param-file',
metavar="<additional-param-file>",
help=_("Additional parameters passed by the NFVO as input "
"to the scaling process."))
scale_require_parameters = parser.add_argument_group(
"require arguments"
)
scale_require_parameters.add_argument(
'--type',
metavar="<type>",
required=True,
choices=['SCALE_OUT', 'SCALE_IN'],
help=_("SCALE_OUT or SCALE_IN for type of scale operation."))
scale_require_parameters.add_argument(
'--aspect-id',
required=True,
metavar="<aspect-id>",
help=_("Identifier of the scaling aspect."))
return parser
def args2body(self, parsed_args):
"""To store request body, call jsonfile2body.
Args:
parsed_args ([Namespace]): arguments of CLI.
Returns:
body ([dict]): Request body is stored
"""
body = {'type': parsed_args.type, 'aspectId': parsed_args.aspect_id}
if parsed_args.number_of_steps:
body['numberOfSteps'] = parsed_args.number_of_steps
if parsed_args.additional_param_file:
body.update(jsonfile2body(parsed_args.additional_param_file))
return body
def take_action(self, parsed_args):
"""Execute scale_vnf_instance and output result comment.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
client = self.app.client_manager.tackerclient
result = client.scale_vnf_instance(
parsed_args.vnf_instance,
self.args2body(parsed_args))
if not result:
print((_('Scale request for VNF Instance %s has been accepted.')
% parsed_args.vnf_instance))
class ChangeExtConnVnfLcm(command.Command):
_description = _("Change External VNF Connectivity")
def get_parser(self, prog_name):
parser = super(ChangeExtConnVnfLcm, self).get_parser(prog_name)
parser.add_argument(
_VNF_INSTANCE,
metavar="<vnf-instance>",
help=_("VNF instance ID to Change External VNF Connectivity"))
parser.add_argument(
'request_file',
metavar="<param-file>",
help=_("Specify change-ext-conn request parameters "
"in a json file."))
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
result = client.change_ext_conn_vnf_instance(
parsed_args.vnf_instance, jsonfile2body(
parsed_args.request_file))
if not result:
print((_('Change External VNF Connectivity for VNF Instance %s '
'has been accepted.') % parsed_args.vnf_instance))

View File

@@ -1,306 +0,0 @@
# 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 osc_lib.command import command
from osc_lib import utils
from tackerclient.i18n import _
from tackerclient.osc import sdk_utils
from tackerclient.osc import utils as tacker_osc_utils
_VNF_LCM_OP_OCC_ID = 'vnf_lcm_op_occ_id'
_MIXED_CASE_FIELDS = ['operationState', 'stateEnteredTime', 'startTime',
'vnfInstanceId', 'grantId', 'isAutomaticInvocation',
'isCancelPending', 'cancelMode', 'operationParams',
'resourceChanges', 'changedInfo',
'changedExtConnectivity']
_FORMATTERS = {
'operationParams': tacker_osc_utils.FormatComplexDataColumn,
'error': tacker_osc_utils.FormatComplexDataColumn,
'resourceChanges': tacker_osc_utils.FormatComplexDataColumn,
'changedInfo': tacker_osc_utils.FormatComplexDataColumn,
'changedExtConnectivity': tacker_osc_utils.FormatComplexDataColumn,
'_links': tacker_osc_utils.FormatComplexDataColumn
}
_ATTR_MAP = (
('id', 'id', tacker_osc_utils.LIST_BOTH),
('operationState', 'operationState', tacker_osc_utils.LIST_BOTH),
('vnfInstanceId', 'vnfInstanceId', tacker_osc_utils.LIST_BOTH),
('operation', 'operation', tacker_osc_utils.LIST_BOTH)
)
def _get_columns(vnflcm_op_occ_obj, action=None):
column_map = {
'id': 'ID',
'operationState': 'Operation State',
'stateEnteredTime': 'State Entered Time',
'startTime': 'Start Time',
'vnfInstanceId': 'VNF Instance ID',
'operation': 'Operation',
'isAutomaticInvocation': 'Is Automatic Invocation',
'isCancelPending': 'Is Cancel Pending',
'error': 'Error',
'_links': 'Links'
}
if action == 'show':
column_map.update(
{'operationParams': 'Operation Parameters',
'grantId': 'Grant ID',
'resourceChanges': 'Resource Changes',
'changedInfo': 'Changed Info',
'cancelMode': 'Cancel Mode',
'changedExtConnectivity': 'Changed External Connectivity'}
)
return sdk_utils.get_osc_show_columns_for_sdk_resource(vnflcm_op_occ_obj,
column_map)
class RollbackVnfLcmOp(command.Command):
def get_parser(self, prog_name):
"""Add arguments to parser.
Args:
prog_name ([type]): program name
Returns:
parser([ArgumentParser]):
"""
parser = super(RollbackVnfLcmOp, self).get_parser(prog_name)
parser.add_argument(
_VNF_LCM_OP_OCC_ID,
metavar="<vnf-lcm-op-occ-id>",
help=_('VNF lifecycle management operation occurrence ID.'))
return parser
def take_action(self, parsed_args):
"""Execute rollback_vnf_instance and output comment.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
client = self.app.client_manager.tackerclient
result = client.rollback_vnf_instance(parsed_args.vnf_lcm_op_occ_id)
if not result:
print((_('Rollback request for LCM operation %(id)s has been'
' accepted') % {'id': parsed_args.vnf_lcm_op_occ_id}))
class FailVnfLcmOp(command.ShowOne):
_description = _("Fail VNF Instance")
def get_parser(self, prog_name):
"""Add arguments to parser.
Args:
prog_name ([type]): program name
Returns:
parser([ArgumentParser]):
"""
parser = super(FailVnfLcmOp, self).get_parser(prog_name)
parser.add_argument(
_VNF_LCM_OP_OCC_ID,
metavar="<vnf-lcm-op-occ-id>",
help=_('VNF lifecycle management operation occurrence ID.'))
return parser
def take_action(self, parsed_args):
"""Execute fail_vnf_instance and output response.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
client = self.app.client_manager.tackerclient
obj = client.fail_vnf_instance(parsed_args.vnf_lcm_op_occ_id)
display_columns, columns = _get_columns(obj)
data = utils.get_item_properties(
sdk_utils.DictModel(obj),
columns, formatters=_FORMATTERS,
mixed_case_fields=_MIXED_CASE_FIELDS)
return (display_columns, data)
class RetryVnfLcmOp(command.Command):
_description = _("Retry VNF Instance")
def get_parser(self, prog_name):
"""Add arguments to parser.
Args:
prog_name ([type]): program name
Returns:
parser([ArgumentParser]):
"""
parser = super(RetryVnfLcmOp, self).get_parser(prog_name)
parser.add_argument(
_VNF_LCM_OP_OCC_ID,
metavar="<vnf-lcm-op-occ-id>",
help=_('VNF lifecycle management operation occurrence ID.'))
return parser
def take_action(self, parsed_args):
"""Execute retry_vnf_instance and output comment.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
client = self.app.client_manager.tackerclient
result = client.retry_vnf_instance(parsed_args.vnf_lcm_op_occ_id)
if not result:
print((_('Retry request for LCM operation %(id)s has been'
' accepted') % {'id': parsed_args.vnf_lcm_op_occ_id}))
class ListVnfLcmOp(command.Lister):
_description = _("List LCM Operation Occurrences")
def get_parser(self, program_name):
"""Add arguments to parser.
Args:
program_name ([type]): program name
Returns:
parser([ArgumentParser]):
"""
parser = super(ListVnfLcmOp, self).get_parser(program_name)
parser.add_argument(
"--filter",
metavar="<filter>",
help=_("Attribute-based-filtering parameters"),
)
fields_exclusive_group = parser.add_mutually_exclusive_group(
required=False)
fields_exclusive_group.add_argument(
"--fields",
metavar="<fields>",
help=_("Complex attributes to be included into the response"),
)
fields_exclusive_group.add_argument(
"--exclude-fields",
metavar="<exclude-fields>",
help=_("Complex attributes to be excluded from the response"),
)
return parser
def get_attributes(self):
"""Get attributes.
Returns:
attributes([attributes]): a list of table entry definitions.
Each entry should be a tuple consisting of
(API attribute name, header name, listing mode).
"""
fields = [
{
"key": "id",
"value": "ID"
},
{
"key": "operationState",
"value": "Operation State"
},
{
"key": "vnfInstanceId",
"value": "VNF Instance ID"
},
{
"key": "operation",
"value": "Operation"
}
]
attributes = []
for field in fields:
attributes.extend([(field['key'], field['value'],
tacker_osc_utils.LIST_BOTH)])
return tuple(attributes)
def take_action(self, parsed_args):
"""Execute list_vnflcm_op_occs and output response.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
params = {}
exclude_fields = []
extra_fields = []
if parsed_args.filter:
params['filter'] = parsed_args.filter
if parsed_args.fields:
params['fields'] = parsed_args.fields
fields = parsed_args.fields.split(',')
for field in fields:
extra_fields.append(field.split('/')[0])
if parsed_args.exclude_fields:
params['exclude-fields'] = parsed_args.exclude_fields
fields = parsed_args.exclude_fields.split(',')
exclude_fields.extend(fields)
client = self.app.client_manager.tackerclient
vnflcm_op_occs = client.list_vnf_lcm_op_occs(**params)
headers, columns = tacker_osc_utils.get_column_definitions(
self.get_attributes(),
long_listing=True)
dictionary_properties = (utils.get_dict_properties(
s, columns, mixed_case_fields=_MIXED_CASE_FIELDS)
for s in vnflcm_op_occs
)
return (headers, dictionary_properties)
class ShowVnfLcmOp(command.ShowOne):
_description = _("Display Operation Occurrence details")
def get_parser(self, program_name):
"""Add arguments to parser.
Args:
program_name ([type]): program name
Returns:
parser([ArgumentParser]):
"""
parser = super(ShowVnfLcmOp, self).get_parser(program_name)
parser.add_argument(
_VNF_LCM_OP_OCC_ID,
metavar="<vnf-lcm-op-occ-id>",
help=_('VNF lifecycle management operation occurrence ID.'))
return parser
def take_action(self, parsed_args):
"""Execute show_vnf_lcm_op_occs and output response.
Args:
parsed_args ([Namespace]): arguments of CLI.
"""
client = self.app.client_manager.tackerclient
obj = client.show_vnf_lcm_op_occs(parsed_args.vnf_lcm_op_occ_id)
display_columns, columns = _get_columns(obj, action='show')
data = utils.get_item_properties(
sdk_utils.DictModel(obj),
columns, formatters=_FORMATTERS,
mixed_case_fields=_MIXED_CASE_FIELDS)
return (display_columns, data)

View File

@@ -18,7 +18,6 @@ import yaml
from osc_lib.command import command
from osc_lib import utils
from oslo_utils import encodeutils
from tackerclient.common import exceptions
from tackerclient.i18n import _
@@ -120,7 +119,7 @@ class CreateVNF(command.ShowOne):
config = yaml.load(
config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if config:
body[_VNF]['attributes'] = {'config': config}
@@ -146,10 +145,9 @@ class CreateVNF(command.ShowOne):
template = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not template:
raise exceptions.InvalidInput(
reason='The vnfd file is empty')
raise exceptions.InvalidInput('The vnfd file is empty')
body[_VNF]['vnfd_template'] = template
if parsed_args.param_file:
@@ -159,10 +157,9 @@ class CreateVNF(command.ShowOne):
param_yaml = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not param_yaml:
raise exceptions.InvalidInput(
reason='The parameter file is empty')
raise exceptions.InvalidInput('The parameter file is empty')
body[_VNF]['attributes'] = {'param_values': param_yaml}
tackerV10.update_dict(parsed_args, body[_VNF],
['tenant_id', 'name', 'description',
@@ -231,12 +228,12 @@ class DeleteVNF(command.Command):
'resource': _VNF})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _VNF
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _VNF}))
@@ -360,16 +357,13 @@ class UpdateVNF(command.ShowOne):
def get_parser(self, prog_name):
parser = super(UpdateVNF, self).get_parser(prog_name)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument(
config_group = parser.add_mutually_exclusive_group(required=True)
config_group.add_argument(
'--config-file',
help=_('YAML file with VNF configuration'))
group.add_argument(
config_group.add_argument(
'--config',
help=_('YAML data with VNF configuration'))
group.add_argument(
'--param-file',
help=_('YAML file with VNF parameter'))
help=_('Specify config YAML data'))
parser.add_argument(
_VNF,
metavar="<VNF>",
@@ -387,33 +381,17 @@ class UpdateVNF(command.ShowOne):
try:
config = yaml.load(config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The config file is empty')
raise exceptions.InvalidInput(e)
if parsed_args.config:
decoded_config = encodeutils.safe_decode(parsed_args.config)
try:
config = yaml.load(decoded_config, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The parameter is empty')
config = parsed_args.config
if isinstance(config, str) or isinstance(config, unicode):
config_str = parsed_args.config.decode('unicode_escape')
try:
config = yaml.load(config_str, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(e)
if 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'])
return body

View File

@@ -71,16 +71,16 @@ class CreateVNFD(command.ShowOne):
body = {_VNFD: {}}
vnfd = None
if not parsed_args.vnfd_file:
raise exceptions.InvalidInput(reason="Invalid input for vnfd file")
raise exceptions.InvalidInput("Invalid input for vnfd file")
with open(parsed_args.vnfd_file) as f:
vnfd = f.read()
try:
vnfd = yaml.load(vnfd, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
msg = _("yaml failed to load vnfd file. %s") % e
raise exceptions.InvalidInput(reason=msg)
raise exceptions.InvalidInput(msg)
if not vnfd:
raise exceptions.InvalidInput(reason="vnfd file is empty")
raise exceptions.InvalidInput("vnfd file is empty")
body[_VNFD]['attributes'] = {'vnfd': vnfd}
tackerV10.update_dict(parsed_args, body[_VNFD],
['tenant_id', 'name', 'description'])
@@ -134,12 +134,12 @@ class DeleteVNFD(command.Command):
'resource': _VNFD})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % _VNFD
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': _VNFD}))
@@ -218,7 +218,7 @@ class ShowTemplateVNFD(command.ShowOne):
obj[_VNFD]['attributes']['vnfd'])
data = utils.get_item_properties(
sdk_utils.DictModel(obj[_VNFD]),
('attributes',),
(u'attributes',),
formatters=_formatters)
data = (data or _('Unable to display VNFD template!'))
return (('attributes',), data)
return ((u'attributes',), data)

View File

@@ -13,9 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
from functools import reduce
import logging
import sys
from osc_lib.cli import parseractions
from osc_lib.command import command
@@ -28,18 +26,21 @@ from tackerclient.osc import utils as tacker_osc_utils
LOG = logging.getLogger(__name__)
formatters = {'softwareImages': tacker_osc_utils.FormatComplexDataColumn,
'checksum': tacker_osc_utils.FormatComplexDataColumn,
'_links': tacker_osc_utils.FormatComplexDataColumn,
'userDefinedData': tacker_osc_utils.FormatComplexDataColumn,
'additionalArtifacts': tacker_osc_utils.FormatComplexDataColumn}
_attr_map = (
('id', 'ID', tacker_osc_utils.LIST_BOTH),
('vnfProductName', 'VNF Product Name', tacker_osc_utils.LIST_BOTH),
('onboardingState', 'Onboarding State', tacker_osc_utils.LIST_BOTH),
('usageState', 'Usage State', tacker_osc_utils.LIST_BOTH),
('operationalState', 'Operational State', tacker_osc_utils.LIST_BOTH),
('userDefinedData', 'User Defined Data', tacker_osc_utils.LIST_BOTH)
)
_mixed_case_fields = ('usageState', 'onboardingState', 'operationalState',
'vnfProductName', 'softwareImages', 'userDefinedData',
'vnfdId', 'vnfdVersion', 'vnfSoftwareVersion',
'vnfProvider', 'additionalArtifacts')
'vnfProvider', 'artifactPath', 'imagePath',
'diskFormat', 'userMetadata')
def _get_columns(vnf_package_obj):
@@ -59,9 +60,7 @@ def _get_columns(vnf_package_obj):
'vnfSoftwareVersion': 'VNF Software Version',
'vnfProductName': 'VNF Product Name',
'vnfdId': 'VNFD ID',
'vnfdVersion': 'VNFD Version',
'checksum': 'Checksum',
'additionalArtifacts': 'Additional Artifacts'
'vnfdVersion': 'VNFD Version'
})
return sdk_utils.get_osc_show_columns_for_sdk_resource(vnf_package_obj,
@@ -95,123 +94,27 @@ class CreateVnfPackage(command.ShowOne):
display_columns, columns = _get_columns(vnf_package)
data = utils.get_item_properties(
sdk_utils.DictModel(vnf_package),
columns, formatters=formatters,
mixed_case_fields=_mixed_case_fields)
columns, mixed_case_fields=_mixed_case_fields)
return (display_columns, data)
class ListVnfPackage(command.Lister):
_description = _("List VNF Packages")
_description = _("List VNF Package")
def get_parser(self, prog_name):
LOG.debug('get_parser(%s)', prog_name)
parser = super(ListVnfPackage, self).get_parser(prog_name)
parser.add_argument(
"--filter",
metavar="<filter>",
help=_("Atrribute-based-filtering parameters"),
)
fields_exclusive_group = parser.add_mutually_exclusive_group(
required=False)
fields_exclusive_group.add_argument(
"--all_fields",
action="store_true",
default=False,
help=_("Include all complex attributes in the response"),
)
fields_exclusive_group.add_argument(
"--fields",
metavar="fields",
help=_("Complex attributes to be included into the response"),
)
fields_exclusive_group.add_argument(
"--exclude_fields",
metavar="exclude-fields",
help=_("Complex attributes to be excluded from the response"),
)
parser.add_argument(
"--exclude_default",
action="store_true",
default=False,
help=_("Indicates to exclude all complex attributes"
" from the response. This argument can be used alone or"
" with --fields and --filter. For all other combinations"
" tacker server will throw bad request error"),
)
return parser
def case_modify(self, field):
return reduce(
lambda x, y: x + (' ' if y.isupper() else '') + y, field).title()
def get_attributes(self, extra_fields=None, all_fields=False,
exclude_fields=None, exclude_default=False):
fields = ['id', 'vnfProductName', 'onboardingState',
'usageState', 'operationalState', '_links']
complex_fields = [
'checksum',
'softwareImages',
'userDefinedData',
'additionalArtifacts']
simple_fields = ['vnfdVersion', 'vnfProvider', 'vnfSoftwareVersion',
'vnfdId']
if extra_fields:
fields.extend(extra_fields)
if exclude_fields:
fields.extend([field for field in complex_fields
if field not in exclude_fields])
if all_fields:
fields.extend(complex_fields)
fields.extend(simple_fields)
if exclude_default:
fields.extend(simple_fields)
attrs = []
for field in fields:
if field == '_links':
attrs.extend([(field, 'Links', tacker_osc_utils.LIST_BOTH)])
else:
attrs.extend([(field, self.case_modify(field),
tacker_osc_utils.LIST_BOTH)])
return tuple(attrs)
def take_action(self, parsed_args):
_params = {}
extra_fields = []
exclude_fields = []
all_fields = False
exclude_default = False
if parsed_args.filter:
_params['filter'] = parsed_args.filter
if parsed_args.fields:
_params['fields'] = parsed_args.fields
fields = parsed_args.fields.split(',')
for field in fields:
extra_fields.append(field.split('/')[0])
if parsed_args.exclude_fields:
_params['exclude_fields'] = parsed_args.exclude_fields
fields = parsed_args.exclude_fields.split(',')
exclude_fields.extend(fields)
if parsed_args.exclude_default:
_params['exclude_default'] = None
exclude_default = True
if parsed_args.all_fields:
_params['all_fields'] = None
all_fields = True
client = self.app.client_manager.tackerclient
data = client.list_vnf_packages(**_params)
headers, columns = tacker_osc_utils.get_column_definitions(
self.get_attributes(extra_fields, all_fields, exclude_fields,
exclude_default), long_listing=True)
_attr_map, long_listing=True)
return (headers,
(utils.get_dict_properties(
s, columns, formatters=formatters,
mixed_case_fields=_mixed_case_fields,
s, columns, mixed_case_fields=_mixed_case_fields,
) for s in data['vnf_packages']))
@@ -234,8 +137,7 @@ class ShowVnfPackage(command.ShowOne):
display_columns, columns = _get_columns(vnf_package)
data = utils.get_item_properties(
sdk_utils.DictModel(vnf_package),
columns, formatters=formatters,
mixed_case_fields=_mixed_case_fields)
columns, mixed_case_fields=_mixed_case_fields)
return (display_columns, data)
@@ -349,174 +251,8 @@ class DeleteVnfPackage(command.Command):
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': self.resource}))
return
class DownloadVnfPackage(command.Command):
_description = _("Download VNF package contents or VNFD of an on-boarded "
"VNF package.")
def get_parser(self, prog_name):
parser = super(DownloadVnfPackage, self).get_parser(prog_name)
parser.add_argument(
"vnf_package",
metavar="<vnf-package>",
help=_("VNF package ID")
)
parser.add_argument(
"--file",
metavar="<FILE>",
help=_("Local file to save downloaded VNF Package or VNFD data. "
"If this is not specified and there is no redirection "
"then data will not be saved.")
)
parser.add_argument(
"--vnfd",
action="store_true",
default=False,
help=_("Download VNFD of an on-boarded vnf package."),
)
parser.add_argument(
"--type",
default="application/zip",
metavar="<type>",
choices=["text/plain", "application/zip", "both"],
help=_("Provide text/plain when VNFD is implemented as a single "
"YAML file otherwise use application/zip. If you are not "
"aware whether VNFD is a single or multiple yaml files, "
"then you can specify 'both' option value. "
"Provide this option only when --vnfd is set.")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
if parsed_args.vnfd:
if sys.stdout.isatty() and not (parsed_args.file and
parsed_args.type != "text/plain"):
msg = ("No redirection or local file specified for downloaded "
"VNFD data. Please specify a local file with --file to "
"save downloaded VNFD data or use redirection.")
sdk_utils.exit(msg)
body = client.download_vnfd_from_vnf_package(
parsed_args.vnf_package, parsed_args.type)
if not parsed_args.file:
print(body)
return
else:
body = client.download_vnf_package(parsed_args.vnf_package)
sdk_utils.save_data(body, parsed_args.file)
class DownloadVnfPackageArtifact(command.Command):
_description = _("Download VNF package artifact of an on-boarded "
"VNF package.")
def get_parser(self, prog_name):
parser = super(DownloadVnfPackageArtifact, self).get_parser(prog_name)
parser.add_argument(
"vnf_package",
metavar="<vnf-package>",
help=_("VNF package ID")
)
parser.add_argument(
"artifact_path",
metavar="<artifact-path>",
help=_("The artifact file's path")
)
parser.add_argument(
"--file",
metavar="<FILE>",
help=_("Local file to save downloaded VNF Package artifact "
"file data. If this is not specified and "
"there is no redirection then data will not be saved.")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
if sys.stdout.isatty() and not (parsed_args.file):
msg = (
"No redirection or local file specified for downloaded "
"vnf package artifact data. Please specify a "
"local file with --file to "
"save downloaded vnf package artifact data "
"or use redirection.")
sdk_utils.exit(msg)
body = client.download_artifact_from_vnf_package(
parsed_args.vnf_package, parsed_args.artifact_path)
if not parsed_args.file:
print(body)
return
else:
sdk_utils.save_data(body, parsed_args.file)
class UpdateVnfPackage(command.ShowOne):
_description = _("Update information about an individual VNF package")
def get_parser(self, prog_name):
LOG.debug('get_parser(%s)', prog_name)
parser = super(UpdateVnfPackage, self).get_parser(prog_name)
parser.add_argument(
'vnf_package',
metavar="<vnf-package>",
help=_("VNF package ID")
)
parser.add_argument(
'--operational-state',
metavar="<operational-state>",
choices=['ENABLED', 'DISABLED'],
help=_("Change the operational state of VNF Package, Valid values"
" are 'ENABLED' or 'DISABLED'.")
)
parser.add_argument(
'--user-data',
metavar='<key=value>',
action=parseractions.KeyValueAction,
help=_('User defined data for the VNF package '
'(repeat option to set multiple user defined data)'),
)
return parser
def get_columns(self, updated_values):
column_map = {}
if updated_values.get('userDefinedData'):
column_map.update({'userDefinedData': 'User Defined Data'})
if updated_values.get('operationalState'):
column_map.update({'operationalState': 'Operational State'})
return sdk_utils.get_osc_show_columns_for_sdk_resource(updated_values,
column_map)
def args2body(self, parsed_args):
body = {}
if not parsed_args.user_data and not parsed_args.operational_state:
msg = ('Provide at least one of the argument from "--user-data"'
' or "--operational-state"')
sdk_utils.exit(msg)
if parsed_args.user_data:
body["userDefinedData"] = parsed_args.user_data
if parsed_args.operational_state:
body["operationalState"] = parsed_args.operational_state
return body
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
updated_values = client.update_vnf_package(
parsed_args.vnf_package, self.args2body(parsed_args))
display_columns, columns = self.get_columns(updated_values)
data = utils.get_item_properties(
sdk_utils.DictModel(updated_values),
columns, formatters=formatters,
mixed_case_fields=_mixed_case_fields)
return (display_columns, data)

View File

@@ -18,6 +18,8 @@
Command-line interface to the Tacker APIs
"""
from __future__ import print_function
import argparse
import getpass
import inspect
@@ -25,10 +27,6 @@ import itertools
import logging
import os
import sys
from urllib import parse as urlparse
from cliff import app
from cliff import commandmanager
from keystoneclient.auth.identity import v2 as v2_auth
from keystoneclient.auth.identity import v3 as v3_auth
@@ -36,6 +34,10 @@ from keystoneclient import discover
from keystoneclient import exceptions as ks_exc
from keystoneclient import session
from oslo_utils import encodeutils
import six.moves.urllib.parse as urlparse
from cliff import app
from cliff import commandmanager
from tackerclient.common import clientmanager
from tackerclient.common import command as openstack_command
@@ -102,7 +104,6 @@ class BashCompletionCommand(openstack_command.OpenStackCommand):
"""Prints all of the commands and options for bash-completion."""
resource = "bash_completion"
COMMAND_V1 = {
'bash-completion': BashCompletionCommand,
'ext-list': extension.ListExt,

View File

@@ -54,9 +54,9 @@ def make_client(instance):
auth=instance._auth)
return client
else:
raise exceptions.UnsupportedVersion(
reason=_("API version %s is not supported") %
instance._api_version[API_NAME])
raise exceptions.UnsupportedVersion(_("API version %s is not "
"supported") %
instance._api_version[API_NAME])
def Client(api_version, *args, **kwargs):

View File

@@ -14,6 +14,8 @@
# under the License.
#
from __future__ import print_function
import abc
import argparse
import logging
@@ -23,6 +25,7 @@ from cliff.formatters import table
from cliff import lister
from cliff import show
from oslo_serialization import jsonutils
import six
from tackerclient.common._i18n import _
from tackerclient.common import command
@@ -161,9 +164,9 @@ def _process_previous_argument(current_arg, _value_number, current_type_str,
values_specs):
if current_arg is not None:
if _value_number == 0 and (current_type_str or _list_flag):
# This kind of argument should have value
raise exceptions.CommandError(
message=_("Invalid values_specs %s") % ' '.join(values_specs))
# This kind of argument should have value
raise exceptions.CommandError(
_("Invalid values_specs %s") % ' '.join(values_specs))
if _value_number > 1 or _list_flag or current_type_str == 'list':
current_arg.update({'nargs': '+'})
elif _value_number == 0:
@@ -234,8 +237,7 @@ def parse_args_to_dict(values_specs):
_value_number = 0
if _item in _options:
raise exceptions.CommandError(
message=_("Duplicated "
"options %s") % ' '.join(values_specs))
_("Duplicated options %s") % ' '.join(values_specs))
else:
_options.update({_item: {}})
current_arg = _options[_item]
@@ -243,8 +245,7 @@ def parse_args_to_dict(values_specs):
elif _item.startswith('type='):
if current_arg is None:
raise exceptions.CommandError(
message=_("Invalid "
"values_specs %s") % ' '.join(values_specs))
_("Invalid values_specs %s") % ' '.join(values_specs))
if 'type' not in current_arg:
current_type_str = _item.split('=', 2)[1]
current_arg.update({'type': eval(current_type_str)})
@@ -266,8 +267,7 @@ def parse_args_to_dict(values_specs):
if (not current_item or '=' in current_item or
_item.startswith('-') and not is_number(_item)):
raise exceptions.CommandError(
message=_("Invalid "
"values_specs %s") % ' '.join(values_specs))
_("Invalid values_specs %s") % ' '.join(values_specs))
_value_number += 1
_values_specs.append(_item)
@@ -332,7 +332,6 @@ class TableFormater(table.TableFormatter):
https://bugs.launchpad.net/python-tackerclient/+bug/1165962
"""
def emit_list(self, column_names, data, stdout, parsed_args):
if column_names:
super(TableFormater, self).emit_list(column_names, data, stdout,
@@ -353,7 +352,8 @@ class TackerCommandMeta(abc.ABCMeta):
name, bases, cls_dict)
class TackerCommand(command.OpenStackCommand, metaclass=TackerCommandMeta):
@six.add_metaclass(TackerCommandMeta)
class TackerCommand(command.OpenStackCommand):
api = 'nfv-orchestration'
values_specs = []
@@ -483,8 +483,7 @@ class UpdateCommand(TackerCommand):
body[self.resource] = _extra_values
if not body[self.resource]:
raise exceptions.CommandError(
message=_("Must specify new"
" values to update %s") % self.resource)
_("Must specify new values to update %s") % self.resource)
if self.allow_names:
_id = find_resourceid_by_name_or_id(
tacker_client, self.resource, parsed_args.id)
@@ -560,12 +559,12 @@ class DeleteCommand(TackerCommand):
'resource': self.resource})
err_msg = _("\n\nUnable to delete the below"
" %s(s):") % self.resource
for failed_id, error in failed_items.items():
for failed_id, error in failed_items.iteritems():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(message=msg)
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) %(msg)s successfully')
% {'msg': self.deleted_msg.get(self.resource, 'deleted'),

View File

@@ -82,17 +82,17 @@ class CreateNS(tackerV10.CreateCommand):
tacker_client = self.get_client()
tacker_client.format = parsed_args.request_format
if parsed_args.vim_name:
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vim',
parsed_args.
vim_name)
parsed_args.vim_id = _id
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vim',
parsed_args.
vim_name)
parsed_args.vim_id = _id
if parsed_args.nsd_name:
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'nsd',
parsed_args.
nsd_name)
parsed_args.nsd_id = _id
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'nsd',
parsed_args.
nsd_name)
parsed_args.nsd_id = _id
elif parsed_args.nsd_template:
with open(parsed_args.nsd_template) as f:
template = f.read()
@@ -100,9 +100,9 @@ class CreateNS(tackerV10.CreateCommand):
args['nsd_template'] = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not args['nsd_template']:
raise exceptions.InvalidInput(reason='The nsd file is empty')
raise exceptions.InvalidInput('The nsd file is empty')
if parsed_args.param_file:
with open(parsed_args.param_file) as f:
@@ -111,7 +111,7 @@ class CreateNS(tackerV10.CreateCommand):
args['attributes']['param_values'] = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
tackerV10.update_dict(parsed_args, body[self.resource],
['tenant_id', 'name', 'description',
'nsd_id', 'vim_id'])

View File

@@ -10,6 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import print_function
import yaml
from oslo_serialization import jsonutils

View File

@@ -71,7 +71,7 @@ class CreateVIM(tackerV10.CreateCommand):
config_param = yaml.load(vim_config,
Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
vim_obj = body[self.resource]
try:
auth_url = config_param.pop('auth_url')
@@ -117,10 +117,9 @@ class UpdateVIM(tackerV10.UpdateCommand):
with open(parsed_args.config_file) as f:
config_yaml = f.read()
try:
config_param = yaml.load(config_yaml,
Loader=yaml.SafeLoader)
config_param = yaml.load(config_yaml)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
vim_obj = body[self.resource]
if config_param is not None:
vim_utils.args2body_vim(config_param, vim_obj)

View File

@@ -13,8 +13,7 @@
# 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 urllib import parse as urlparse
import six.moves.urllib.parse as urlparse
from tackerclient.common import exceptions

View File

@@ -153,10 +153,9 @@ class CreateVNFFG(tackerV10.CreateCommand):
args['vnffgd_template'] = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not args['vnffgd_template']:
raise exceptions.InvalidInput(
reason='The vnffgd file is empty')
raise exceptions.InvalidInput('The vnffgd file is empty')
if parsed_args.param_file:
with open(parsed_args.param_file) as f:
@@ -165,7 +164,7 @@ class CreateVNFFG(tackerV10.CreateCommand):
args['attributes']['param_values'] = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
tackerV10.update_dict(parsed_args, body[self.resource],
['tenant_id', 'name', 'vnffgd_id',
@@ -218,10 +217,9 @@ class UpdateVNFFG(tackerV10.UpdateCommand):
args['vnffgd_template'] = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not args['vnffgd_template']:
raise exceptions.InvalidInput(
reason='The vnffgd template is empty')
raise exceptions.InvalidInput('The vnffgd template is empty')
tackerV10.update_dict(parsed_args, body[self.resource],
['tenant_id', 'vnf_mapping', 'symmetrical'])
@@ -232,18 +230,3 @@ class DeleteVNFFG(tackerV10.DeleteCommand):
"""Delete a given VNFFG."""
resource = _VNFFG
remove_output_fields = ["attributes"]
def add_known_arguments(self, parser):
parser.add_argument(
'--force',
default=False,
action='store_true',
help=_('Force delete VNFFG'))
def args2body(self, parsed_args):
body = dict()
if parsed_args.force:
body[self.resource] = dict()
body[self.resource]['attributes'] = {'force': True}
return body

View File

@@ -10,6 +10,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import print_function
import yaml
from oslo_serialization import jsonutils

View File

@@ -17,8 +17,6 @@
import yaml
from oslo_utils import encodeutils
from tackerclient.common import exceptions
from tackerclient.i18n import _
from tackerclient.tacker import v1_0 as tackerV10
@@ -94,7 +92,7 @@ class CreateVNF(tackerV10.CreateCommand):
config = yaml.load(
config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if config:
args['attributes']['config'] = config
@@ -105,17 +103,17 @@ class CreateVNF(tackerV10.CreateCommand):
tacker_client = self.get_client()
tacker_client.format = parsed_args.request_format
if parsed_args.vim_name:
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vim',
parsed_args.
vim_name)
parsed_args.vim_id = _id
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vim',
parsed_args.
vim_name)
parsed_args.vim_id = _id
if parsed_args.vnfd_name:
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vnfd',
parsed_args.
vnfd_name)
parsed_args.vnfd_id = _id
_id = tackerV10.find_resourceid_by_name_or_id(tacker_client,
'vnfd',
parsed_args.
vnfd_name)
parsed_args.vnfd_id = _id
elif parsed_args.vnfd_template:
with open(parsed_args.vnfd_template) as f:
template = f.read()
@@ -123,7 +121,7 @@ class CreateVNF(tackerV10.CreateCommand):
args['vnfd_template'] = yaml.load(
template, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if parsed_args.param_file:
with open(parsed_args.param_file) as f:
@@ -132,7 +130,7 @@ class CreateVNF(tackerV10.CreateCommand):
args['attributes']['param_values'] = yaml.load(
param_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
tackerV10.update_dict(parsed_args, body[self.resource],
['tenant_id', 'name', 'description',
'vnfd_id', 'vim_id'])
@@ -145,16 +143,12 @@ class UpdateVNF(tackerV10.UpdateCommand):
resource = _VNF
def add_known_arguments(self, parser):
group = parser.add_mutually_exclusive_group()
group.add_argument(
parser.add_argument(
'--config-file',
help=_('YAML file with VNF configuration'))
group.add_argument(
parser.add_argument(
'--config',
help=_('YAML data with VNF configuration'))
group.add_argument(
'--param-file',
help=_('YAML file with VNF parameter'))
help=_('Specify config yaml data'))
def args2body(self, parsed_args):
body = {self.resource: {}}
@@ -166,33 +160,17 @@ class UpdateVNF(tackerV10.UpdateCommand):
try:
config = yaml.load(config_yaml, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The config file is empty')
raise exceptions.InvalidInput(e)
if parsed_args.config:
config_param = encodeutils.safe_decode(parsed_args.config)
try:
config = yaml.load(config_param, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
if not config:
raise exceptions.InvalidInput(
reason='The parameter is empty')
config = parsed_args.config
if isinstance(config, str) or isinstance(config, unicode):
config_str = parsed_args.config.decode('unicode_escape')
try:
config = yaml.load(config_str, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(e)
if 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'])
return body

View File

@@ -15,6 +15,8 @@
# License for the specific language governing permissions and limitations
# under the License.
from __future__ import print_function
from oslo_serialization import jsonutils
import yaml
@@ -75,15 +77,15 @@ class CreateVNFD(tackerV10.CreateCommand):
body = {self.resource: {}}
vnfd = None
if not parsed_args.vnfd_file:
raise exceptions.InvalidInput(reason="Invalid input for vnfd file")
raise exceptions.InvalidInput("Invalid input for vnfd file")
with open(parsed_args.vnfd_file) as f:
vnfd = f.read()
try:
vnfd = yaml.load(vnfd, Loader=yaml.SafeLoader)
except yaml.YAMLError as e:
raise exceptions.InvalidInput(reason=e)
raise exceptions.InvalidInput(e)
if not vnfd:
raise exceptions.InvalidInput(reason="vnfd file is empty")
raise exceptions.InvalidInput("vnfd file is empty")
body[self.resource]['attributes'] = {'vnfd': vnfd}
tackerV10.update_dict(parsed_args, body[self.resource],
['tenant_id', 'name', 'description'])

View File

@@ -13,16 +13,13 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
from requests_mock.contrib import fixture as requests_mock_fixture
import testtools
from unittest import mock
from cliff import columns as cliff_columns
class FixturedTestCase(testtools.TestCase):
client_fixture_class = None
api_version = '1'
def setUp(self):
super(FixturedTestCase, self).setUp()
@@ -30,8 +27,7 @@ class FixturedTestCase(testtools.TestCase):
if self.client_fixture_class:
self.requests_mock = self.useFixture(requests_mock_fixture.
Fixture())
fix = self.client_fixture_class(self.requests_mock,
api_version=self.api_version)
fix = self.client_fixture_class(self.requests_mock)
self.cs = self.useFixture(fix).client
def check_parser(self, cmd, args, verify_args):
@@ -55,20 +51,6 @@ class FixturedTestCase(testtools.TestCase):
msg = 'method %s should not have been called' % m
self.fail(msg)
def assertListItemsEqual(self, expected, actual):
"""Assertion based on human_readable values of list items"""
self.assertEqual(len(expected), len(actual))
for col_expected, col_actual in zip(expected, actual):
if isinstance(col_actual, tuple):
self.assertListItemsEqual(col_expected, col_actual)
elif isinstance(col_expected, cliff_columns.FormattableColumn):
self.assertIsInstance(col_actual, col_expected.__class__)
self.assertEqual(col_expected.human_readable(),
col_actual.human_readable())
else:
self.assertEqual(col_expected, col_actual)
class ParserException(Exception):
pass

View File

@@ -1,101 +0,0 @@
# Copyright (C) 2021 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
import ddt
from unittest import mock
from tackerclient.common import exceptions
from tackerclient.osc.common.vnflcm import vnflcm_versions
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import 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
@ddt.ddt
class TestVnfLcmVersions(TestVnfLcm):
def setUp(self):
super(TestVnfLcmVersions, self).setUp()
self.vnflcm_versions = vnflcm_versions.VnfLcmVersions(
self.app, self.app_args, cmd_name='vnflcm versions')
def _versions_response(self, major_version=None):
if major_version is None:
return {"uriPrefix": "/vnflcm",
"apiVersions": [{"version": "1.3.0",
"isDeprecated": False},
{"version": "2.0.0",
"isDeprecated": False}]}
elif major_version == "1":
return {"uriPrefix": "/vnflcm/v1",
"apiVersions": [{"version": "1.3.0",
"isDeprecated": False}]}
elif major_version == "2":
return {"uriPrefix": "/vnflcm/v2",
"apiVersions": [{"version": "2.0.0",
"isDeprecated": False}]}
def test_invalid_major_version(self):
parser = self.vnflcm_versions.get_parser('vnflcm versions')
parsed_args = parser.parse_args(["--major-version", "3"])
self.assertRaises(exceptions.InvalidInput,
self.vnflcm_versions.take_action,
parsed_args)
def test_take_action_no_arg(self):
parser = self.vnflcm_versions.get_parser('vnflcm versions')
parsed_args = parser.parse_args([])
response = self._versions_response()
self.requests_mock.register_uri(
'GET', os.path.join(self.url, 'vnflcm/api_versions'),
json=response, headers=self.header)
colmns, data = self.vnflcm_versions.take_action(parsed_args)
self.assertEqual(colmns, tuple(response.keys()))
self.assertEqual(data, tuple(response.values()))
@ddt.data('1', '2')
def test_take_action_with_major_version(self, major_version):
parser = self.vnflcm_versions.get_parser('vnflcm versions')
parsed_args = parser.parse_args(["--major-version",
major_version])
response = self._versions_response(major_version)
self.requests_mock.register_uri(
'GET',
os.path.join(self.url,
'vnflcm/v{}/api_versions'.format(major_version)),
json=response, headers=self.header)
colmns, data = self.vnflcm_versions.take_action(parsed_args)
self.assertEqual(colmns, tuple(response.keys()))
self.assertEqual(data, tuple(response.values()))

View File

@@ -1,7 +0,0 @@
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

@@ -1,7 +0,0 @@
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

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

View File

@@ -25,8 +25,7 @@ TACKER_URL = 'http://nfv-orchestration'
class ClientFixture(fixtures.Fixture):
def __init__(self, requests_mock, identity_url=IDENTITY_URL,
api_version='1'):
def __init__(self, requests_mock, identity_url=IDENTITY_URL):
super(ClientFixture, self).__init__()
self.identity_url = identity_url
self.client = None
@@ -36,7 +35,6 @@ class ClientFixture(fixtures.Fixture):
self.discovery = fixture.V2Discovery(href=self.identity_url)
s = self.token.add_service('nfv-orchestration')
s.add_endpoint(TACKER_URL)
self.api_version = api_version
def setUp(self):
super(ClientFixture, self).setUp()
@@ -59,5 +57,4 @@ class ClientFixture(fixtures.Fixture):
region_name='RegionOne',
auth_url=self.identity_url,
token=self.token.token_id,
endpoint_url=TACKER_URL,
api_version=self.api_version)
endpoint_url=TACKER_URL)

View File

@@ -1,202 +0,0 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: ETSI NFV SOL 001 common types definitions version 2.6.1
metadata:
template_name: etsi_nfv_sol001_common_types
template_author: ETSI_NFV
template_version: 2.6.1
data_types:
tosca.datatypes.nfv.L2AddressData:
derived_from: tosca.datatypes.Root
description: Describes the information on the MAC addresses to be assigned to a connection point.
properties:
mac_address_assignment:
type: boolean
description: Specifies if the address assignment is the responsibility of management and orchestration function or not. If it is set to True, it is the management and orchestration function responsibility
required: true
tosca.datatypes.nfv.L3AddressData:
derived_from: tosca.datatypes.Root
description: Provides information about Layer 3 level addressing scheme and parameters applicable to a CP
properties:
ip_address_assignment:
type: boolean
description: Specifies if the address assignment is the responsibility of management and orchestration function or not. If it is set to True, it is the management and orchestration function responsibility
required: true
floating_ip_activated:
type: boolean
description: Specifies if the floating IP scheme is activated on the Connection Point or not
required: true
ip_address_type:
type: string
description: Defines address type. The address type should be aligned with the address type supported by the layer_protocols properties of the parent VnfExtCp
required: false
constraints:
- valid_values: [ ipv4, ipv6 ]
number_of_ip_address:
type: integer
description: Minimum number of IP addresses to be assigned
required: false
constraints:
- greater_than: 0
tosca.datatypes.nfv.AddressData:
derived_from: tosca.datatypes.Root
description: Describes information about the addressing scheme and parameters applicable to a CP
properties:
address_type:
type: string
description: Describes the type of the address to be assigned to a connection point. The content type shall be aligned with the address type supported by the layerProtocol property of the connection point
required: true
constraints:
- valid_values: [ mac_address, ip_address ]
l2_address_data:
type: tosca.datatypes.nfv.L2AddressData
description: Provides the information on the MAC addresses to be assigned to a connection point.
required: false
l3_address_data:
type: tosca.datatypes.nfv.L3AddressData
description: Provides the information on the IP addresses to be assigned to a connection point
required: false
tosca.datatypes.nfv.ConnectivityType:
derived_from: tosca.datatypes.Root
description: describes additional connectivity information of a virtualLink
properties:
layer_protocols:
type: list
description: Identifies the protocol a virtualLink gives access to (ethernet, mpls, odu2, ipv4, ipv6, pseudo-wire).The top layer protocol of the virtualLink protocol stack shall always be provided. The lower layer protocols may be included when there are specific requirements on these layers.
required: true
entry_schema:
type: string
constraints:
- valid_values: [ ethernet, mpls, odu2, ipv4, ipv6, pseudo-wire ]
flow_pattern:
type: string
description: Identifies the flow pattern of the connectivity
required: false
constraints:
- valid_values: [ line, tree, mesh ]
tosca.datatypes.nfv.LinkBitrateRequirements:
derived_from: tosca.datatypes.Root
description: describes the requirements in terms of bitrate for a virtual link
properties:
root:
type: integer # in bits per second
description: Specifies the throughput requirement in bits per second of the link (e.g. bitrate of E-Line, root bitrate of E-Tree, aggregate capacity of E-LAN).
required: true
constraints:
- greater_or_equal: 0
leaf:
type: integer # in bits per second
description: Specifies the throughput requirement in bits per second of leaf connections to the link when applicable to the connectivity type (e.g. for E-Tree and E LAN branches).
required: false
constraints:
- greater_or_equal: 0
tosca.datatypes.nfv.CpProtocolData:
derived_from: tosca.datatypes.Root
description: Describes and associates the protocol layer that a CP uses together with other protocol and connection point information
properties:
associated_layer_protocol:
type: string
required: true
description: One of the values of the property layer_protocols of the CP
constraints:
- valid_values: [ ethernet, mpls, odu2, ipv4, ipv6, pseudo-wire ]
address_data:
type: list
description: Provides information on the addresses to be assigned to the CP
entry_schema:
type: tosca.datatypes.nfv.AddressData
required: false
tosca.datatypes.nfv.VnfProfile:
derived_from: tosca.datatypes.Root
description: describes a profile for instantiating VNFs of a particular NS DF according to a specific VNFD and VNF DF.
properties:
instantiation_level:
type: string
description: Identifier of the instantiation level of the VNF DF to be used for instantiation. If not present, the default instantiation level as declared in the VNFD shall be used.
required: false
min_number_of_instances:
type: integer
description: Minimum number of instances of the VNF based on this VNFD that is permitted to exist for this VnfProfile.
required: true
constraints:
- greater_or_equal: 0
max_number_of_instances:
type: integer
description: Maximum number of instances of the VNF based on this VNFD that is permitted to exist for this VnfProfile.
required: true
constraints:
- greater_or_equal: 0
tosca.datatypes.nfv.Qos:
derived_from: tosca.datatypes.Root
description: describes QoS data for a given VL used in a VNF deployment flavour
properties:
latency:
type: scalar-unit.time #Number
description: Specifies the maximum latency
required: true
constraints:
- greater_than: 0 s
packet_delay_variation:
type: scalar-unit.time #Number
description: Specifies the maximum jitter
required: true
constraints:
- greater_or_equal: 0 s
packet_loss_ratio:
type: float
description: Specifies the maximum packet loss ratio
required: false
constraints:
- in_range: [ 0.0, 1.0 ]
capability_types:
tosca.capabilities.nfv.VirtualLinkable:
derived_from: tosca.capabilities.Node
description: A node type that includes the VirtualLinkable capability indicates that it can be pointed by tosca.relationships.nfv.VirtualLinksTo relationship type
relationship_types:
tosca.relationships.nfv.VirtualLinksTo:
derived_from: tosca.relationships.DependsOn
description: Represents an association relationship between the VduCp and VnfVirtualLink node types
valid_target_types: [ tosca.capabilities.nfv.VirtualLinkable ]
node_types:
tosca.nodes.nfv.Cp:
derived_from: tosca.nodes.Root
description: Provides information regarding the purpose of the connection point
properties:
layer_protocols:
type: list
description: Identifies which protocol the connection point uses for connectivity purposes
required: true
entry_schema:
type: string
constraints:
- valid_values: [ ethernet, mpls, odu2, ipv4, ipv6, pseudo-wire ]
role: #Name in ETSI NFV IFA011 v0.7.3: cpRole
type: string
description: Identifies the role of the port in the context of the traffic flow patterns in the VNF or parent NS
required: false
constraints:
- valid_values: [ root, leaf ]
description:
type: string
description: Provides human-readable information on the purpose of the connection point
required: false
protocol:
type: list
description: Provides information on the addresses to be assigned to the connection point(s) instantiated from this Connection Point Descriptor
required: false
entry_schema:
type: tosca.datatypes.nfv.CpProtocolData
trunk_mode:
type: boolean
description: Provides information about whether the CP instantiated from this Cp is in Trunk mode (802.1Q or other), When operating in "trunk mode", the Cp is capable of carrying traffic for several VLANs. Absence of this property implies that trunkMode is not configured for the Cp i.e. It is equivalent to boolean value "false".
required: false

View File

@@ -1,277 +0,0 @@
tosca_definitions_version: tosca_simple_yaml_1_2
description: Simple deployment flavour for Sample VNF
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
- helloworld3_types.yaml
topology_template:
inputs:
descriptor_id:
type: string
descriptor_version:
type: string
provider:
type: string
product_name:
type: string
software_version:
type: string
vnfm_info:
type: list
entry_schema:
type: string
flavour_id:
type: string
flavour_description:
type: string
substitution_mappings:
node_type: ntt.nslab.VNF
properties:
flavour_id: simple
requirements:
virtual_link_external: [ CP1, virtual_link ]
node_templates:
VNF:
type: ntt.nslab.VNF
properties:
flavour_description: A simple flavour
interfaces:
Vnflcm:
# supporting only 'instantiate', 'terminate', 'modify'
# not supporting LCM script, supporting only default LCM
instantiate: []
instantiate_start: []
instantiate_end: []
terminate: []
terminate_start: []
terminate_end: []
modify_information: []
modify_information_start: []
modify_information_end: []
# change_flavour: []
# change_flavour_start: []
# change_flavour_end: []
# change_external_connectivity: []
# change_external_connectivity_start: []
# change_external_connectivity_end: []
# operate: []
# operate_start: []
# operate_end: []
# heal: []
# heal_start: []
# heal_end: []
# scale: []
# scale_start: []
# scale_end: []
# scale_to_level: []
# scale_to_level_start: []
# scale_to_level_end: []
VDU1:
type: tosca.nodes.nfv.Vdu.Compute
properties:
name: VDU1
description: VDU1 compute node
vdu_profile:
min_number_of_instances: 1
max_number_of_instances: 1
sw_image_data:
name: Software of VDU1
version: '0.4.0'
checksum:
algorithm: sha-256
hash: b9c3036539fd7a5f87a1bf38eb05fdde8b556a1a7e664dbeda90ed3cd74b4f9d
container_format: bare
disk_format: qcow2
min_disk: 1 GB
size: 1 GB
artifacts:
sw_image:
type: tosca.artifacts.nfv.SwImage
file: cirros-0.4.0-x86_64-disk.img
repository: http://download.cirros-cloud.net/0.4.0/
capabilities:
virtual_compute:
properties:
virtual_memory:
virtual_mem_size: 512 MB
virtual_cpu:
num_virtual_cpu: 1
virtual_local_storage:
- size_of_storage: 1 GB
VDU2:
type: tosca.nodes.nfv.Vdu.Compute
properties:
name: VDU2
description: VDU2 compute node
vdu_profile:
min_number_of_instances: 1
max_number_of_instances: 3
capabilities:
virtual_compute:
properties:
virtual_memory:
virtual_mem_size: 512 MB
virtual_cpu:
num_virtual_cpu: 1
virtual_local_storage:
- size_of_storage: 1 GB
requirements:
- virtual_storage: VirtualStorage
VirtualStorage:
type: tosca.nodes.nfv.Vdu.VirtualBlockStorage
properties:
virtual_block_storage_data:
size_of_storage: 30 GB
rdma_enabled: true
sw_image_data:
name: VrtualStorage
version: '0.4.0'
checksum:
algorithm: sha-256
hash: b9c3036539fd7a5f87a1bf38eb05fdde8b556a1a7e664dbeda90ed3cd74b4f9d
container_format: bare
disk_format: qcow2
min_disk: 2 GB
min_ram: 8192 MB
size: 2 GB
artifacts:
sw_image:
type: tosca.artifacts.nfv.SwImage
file: cirros-0.4.0-x86_64-disk.img
repository: http://download.cirros-cloud.net/0.4.0/
CP1:
type: tosca.nodes.nfv.VduCp
properties:
layer_protocols: [ ipv4 ]
order: 0
vnic_type: direct-physical
requirements:
- virtual_binding: VDU1
#- virtual_link: # the target node is determined in the NSD
CP2:
type: tosca.nodes.nfv.VduCp
properties:
layer_protocols: [ ipv4 ]
order: 1
requirements:
- virtual_binding: VDU1
- virtual_link: internalVL2
CP3:
type: tosca.nodes.nfv.VduCp
properties:
layer_protocols: [ ipv4 ]
order: 2
requirements:
- virtual_binding: VDU2
- virtual_link: internalVL2
internalVL2:
type: tosca.nodes.nfv.VnfVirtualLink
properties:
connectivity_type:
layer_protocols: [ ipv4 ]
description: Internal Virtual link in the VNF
vl_profile:
max_bitrate_requirements:
root: 1048576
leaf: 1048576
min_bitrate_requirements:
root: 1048576
leaf: 1048576
virtual_link_protocol_data:
- associated_layer_protocol: ipv4
l3_protocol_data:
ip_version: ipv4
cidr: 11.11.0.0/24
policies:
- scaling_aspects:
type: tosca.policies.nfv.ScalingAspects
properties:
aspects:
worker_instance:
name: worker_instance_aspect
description: worker_instance scaling aspect
max_scale_level: 2
step_deltas:
- delta_1
- VDU2_initial_delta:
type: tosca.policies.nfv.VduInitialDelta
properties:
initial_delta:
number_of_instances: 1
targets: [ VDU2 ]
- VDU2_scaling_aspect_deltas:
type: tosca.policies.nfv.VduScalingAspectDeltas
properties:
aspect: worker_instance
deltas:
delta_2:
number_of_instances: 1
targets: [ VDU2 ]
- instantiation_levels:
type: tosca.policies.nfv.InstantiationLevels
properties:
levels:
instantiation_level_1:
description: Smallest size
scale_info:
worker_instance:
scale_level: 0
instantiation_level_2:
description: Largest size
scale_info:
worker_instance:
scale_level: 2
default_level: instantiation_level_1
- VDU1_instantiation_levels:
type: tosca.policies.nfv.VduInstantiationLevels
properties:
levels:
instantiation_level_1:
number_of_instances: 1
instantiation_level_2:
number_of_instances: 3
targets: [ VDU1 ]
- VDU2_instantiation_levels:
type: tosca.policies.nfv.VduInstantiationLevels
properties:
levels:
instantiation_level_1:
number_of_instances: 1
instantiation_level_2:
number_of_instances: 1
targets: [ VDU2 ]
- internalVL2_instantiation_levels:
type: tosca.policies.nfv.VirtualLinkInstantiationLevels
properties:
levels:
instantiation_level_1:
bitrate_requirements:
root: 1048576
leaf: 1048576
instantiation_level_2:
bitrate_requirements:
root: 1048576
leaf: 1048576
targets: [ internalVL2 ]

View File

@@ -1,32 +0,0 @@
tosca_definitions_version: tosca_simple_yaml_1_0
description: Sample VNF of NTT NS lab.
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
- helloworld3_types.yaml
- helloworld3_df_simple.yaml
# - helloworld3_df_complex.yaml
topology_template:
inputs:
selected_flavour:
type: string
description: VNF deployment flavour selected by the consumer. It is provided in the API
node_templates:
VNF:
type: ntt.nslab.VNF
properties:
flavour_id: { get_input: selected_flavour }
descriptor_id: b1bb0ce7-ebca-4fa7-95ed-4840d70a1177
provider: NTT NS lab
product_name: Sample VNF
software_version: '1.0'
descriptor_version: '1.0'
vnfm_info:
- Tacker
requirements:
#- virtual_link_external # mapped in lower-level templates
#- virtual_link_internal # mapped in lower-level templates

View File

@@ -1,53 +0,0 @@
tosca_definitions_version: tosca_simple_yaml_1_0
description: ntt.nslab.VNF type definition
imports:
- etsi_nfv_sol001_common_types.yaml
- etsi_nfv_sol001_vnfd_types.yaml
node_types:
ntt.nslab.VNF:
derived_from: tosca.nodes.nfv.VNF
properties:
descriptor_id:
type: string
constraints: [ valid_values: [ b1bb0ce7-ebca-4fa7-95ed-4840d70a1177 ] ]
default: b1bb0ce7-ebca-4fa7-95ed-4840d70a1177
descriptor_version:
type: string
constraints: [ valid_values: [ '1.0' ] ]
default: '1.0'
provider:
type: string
constraints: [ valid_values: [ 'NTT NS lab' ] ]
default: 'NTT NS lab'
product_name:
type: string
constraints: [ valid_values: [ 'Sample VNF' ] ]
default: 'Sample VNF'
software_version:
type: string
constraints: [ valid_values: [ '1.0' ] ]
default: '1.0'
vnfm_info:
type: list
entry_schema:
type: string
constraints: [ valid_values: [ Tacker ] ]
default: [ Tacker ]
flavour_id:
type: string
constraints: [ valid_values: [ simple ] ]
default: simple
flavour_description:
type: string
default: ""
requirements:
- virtual_link_external:
capability: tosca.capabilities.nfv.VirtualLinkable
- virtual_link_internal:
capability: tosca.capabilities.nfv.VirtualLinkable
interfaces:
Vnflcm:
type: tosca.interfaces.nfv.Vnflcm

View File

@@ -1,3 +0,0 @@
#!/bin/bash
echo "Hello, World!"

View File

@@ -1,8 +0,0 @@
TOSCA-Meta-File-Version: 1.0
Created-by: Dummy User
CSAR-Version: 1.1
Entry-Definitions: Definitions/helloworld3_top.vnfd.yaml
Source: Scripts/install.sh
Algorithm: SHA-256
Hash: 27bbdb25d8f4ed6d07d6f6581b86515e8b2f0059b236ef7b6f50d6674b34f02a

View File

@@ -13,15 +13,12 @@
# License for the specific language governing permissions and limitations
# under the License.
import filecmp
import os
import shutil
import sys
import tempfile
from unittest import mock
import ddt
import zipfile
import mock
from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
@@ -42,28 +39,18 @@ def _create_zip():
def _get_columns_vnf_package(action='list', vnf_package_obj=None):
columns = []
if action == 'update':
if vnf_package_obj.get('userDefinedData'):
columns.extend(['User Defined Data'])
if vnf_package_obj.get('operationalState'):
columns.extend(['Operational State'])
return columns
columns.extend(['ID', 'Onboarding State', 'Operational State',
'Usage State', 'User Defined Data', 'Links'])
columns = ['ID', 'Onboarding State', 'Operational State', 'Usage State',
'User Defined Data', 'VNF Product Name']
if action in ['show', 'create']:
if vnf_package_obj and vnf_package_obj[
'onboardingState'] == 'ONBOARDED':
columns.extend(['VNFD ID',
'VNF Provider',
'VNF Software Version',
'VNFD Version',
'Software Images',
'VNF Product Name',
'Checksum',
'Additional Artifacts'])
columns.extend(['Links', 'VNFD ID', 'VNF Provider',
'VNF Software Version', 'VNFD Version',
'Software Images'])
else:
columns.extend(['Links'])
columns.remove('VNF Product Name')
return columns
@@ -108,183 +95,38 @@ class TestCreateVnfPackage(TestVnfPackage):
json=json, headers=self.header)
columns, data = (self.create_vnf_package.take_action(parsed_args))
self.assertCountEqual(_get_columns_vnf_package(), columns)
headers, attributes = vnf_package._get_columns(json)
self.assertListItemsEqual(vnf_package_fakes.get_vnf_package_data(
json, columns=attributes), data)
self.assertItemsEqual(_get_columns_vnf_package(action='create'),
columns)
self.assertItemsEqual(vnf_package_fakes.get_vnf_package_data(json),
data)
@ddt.ddt
class TestListVnfPackage(TestVnfPackage):
_vnf_packages = vnf_package_fakes.create_vnf_packages(count=3)
def setUp(self):
super(TestListVnfPackage, self).setUp()
self.list_vnf_package = vnf_package.ListVnfPackage(
self.app, self.app_args, cmd_name='vnf package list')
self._vnf_packages = self._get_vnf_packages()
def _get_vnf_packages(self, onboarded_vnf_package=False):
return vnf_package_fakes.create_vnf_packages(
count=3, onboarded_vnf_package=onboarded_vnf_package)
def get_list_columns(self, all_fields=False, exclude_fields=None,
extra_fields=None, exclude_default=False):
columns = ['Id', 'Vnf Product Name', 'Onboarding State', 'Usage State',
'Operational State', 'Links']
complex_columns = [
'Checksum',
'Software Images',
'User Defined Data',
'Additional Artifacts']
simple_columns = ['Vnfd Version', 'Vnf Provider', 'Vnfd Id',
'Vnf Software Version']
if extra_fields:
columns.extend(extra_fields)
if exclude_fields:
columns.extend([field for field in complex_columns
if field not in exclude_fields])
if all_fields:
columns.extend(complex_columns)
columns.extend(simple_columns)
if exclude_default:
columns.extend(simple_columns)
return columns
def _get_mock_response_for_list_vnf_packages(
self, filter_attribute, json=None):
self.requests_mock.register_uri(
'GET', self.url + '/vnfpkgm/v1/vnf_packages?' + filter_attribute,
json=json if json else self._get_vnf_packages(),
headers=self.header)
def test_take_action_default_fields(self):
def test_take_action(self):
parsed_args = self.check_parser(self.list_vnf_package, [], [])
self.requests_mock.register_uri(
'GET', self.url + '/vnfpkgm/v1/vnf_packages',
json=self._vnf_packages, headers=self.header)
actual_columns, data = self.list_vnf_package.take_action(parsed_args)
expected_data = []
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnf_package.get_attributes(), long_listing=True)
vnf_package._attr_map, long_listing=True)
for vnf_package_obj in self._vnf_packages['vnf_packages']:
expected_data.append(vnf_package_fakes.get_vnf_package_data(
vnf_package_obj, columns=columns, list_action=True))
self.assertCountEqual(self.get_list_columns(), actual_columns)
self.assertListItemsEqual(expected_data, list(data))
@ddt.data('all_fields', 'exclude_default')
def test_take_action(self, arg):
parsed_args = self.check_parser(
self.list_vnf_package,
["--" + arg, "--filter", '(eq,onboardingState,ONBOARDED)'],
[(arg, True), ('filter', '(eq,onboardingState,ONBOARDED)')])
vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
self._get_mock_response_for_list_vnf_packages(
'filter=(eq,onboardingState,ONBOARDED)&' + arg, json=vnf_packages)
actual_columns, data = self.list_vnf_package.take_action(parsed_args)
expected_data = []
kwargs = {arg: True}
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnf_package.get_attributes(**kwargs), long_listing=True)
for vnf_package_obj in vnf_packages['vnf_packages']:
expected_data.append(vnf_package_fakes.get_vnf_package_data(
vnf_package_obj, columns=columns, list_action=True, **kwargs))
self.assertCountEqual(self.get_list_columns(**kwargs), actual_columns)
self.assertListItemsEqual(expected_data, list(data))
def test_take_action_with_exclude_fields(self):
parsed_args = self.check_parser(
self.list_vnf_package,
["--exclude_fields", 'softwareImages,checksum,'
'userDefinedData,additionalArtifacts',
"--filter", '(eq,onboardingState,ONBOARDED)'],
[('exclude_fields', 'softwareImages,checksum,'
'userDefinedData,additionalArtifacts'),
('filter', '(eq,onboardingState,ONBOARDED)')])
vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
updated_vnf_packages = {'vnf_packages': []}
for vnf_pkg in vnf_packages['vnf_packages']:
vnf_pkg.pop('softwareImages')
vnf_pkg.pop('checksum')
vnf_pkg.pop('userDefinedData')
vnf_pkg.pop('additionalArtifacts')
updated_vnf_packages['vnf_packages'].append(vnf_pkg)
self._get_mock_response_for_list_vnf_packages(
'filter=(eq,onboardingState,ONBOARDED)&'
'exclude_fields=softwareImages,checksum,'
'userDefinedData,additionalArtifacts',
json=updated_vnf_packages)
actual_columns, data = self.list_vnf_package.take_action(parsed_args)
expected_data = []
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnf_package.get_attributes(
exclude_fields=['softwareImages', 'checksum',
'userDefinedData', 'additionalArtifacts']),
long_listing=True)
for vnf_package_obj in updated_vnf_packages['vnf_packages']:
expected_data.append(vnf_package_fakes.get_vnf_package_data(
vnf_package_obj, columns=columns, list_action=True))
expected_columns = self.get_list_columns(
exclude_fields=['Software Images', 'Checksum',
'User Defined Data', 'Additional Artifacts'])
self.assertCountEqual(expected_columns, actual_columns)
self.assertListItemsEqual(expected_data, list(data))
@ddt.data((['--all_fields', '--fields', 'softwareImages'],
[('all_fields', True), ('fields', 'softwareImages')]),
(['--all_fields', '--exclude_fields', 'checksum'],
[('all_fields', True), ('exclude_fields', 'checksum')]),
(['--fields', 'softwareImages', '--exclude_fields', 'checksum'],
[('fields', 'softwareImages'), ('exclude_fields', 'checksum')]))
@ddt.unpack
def test_take_action_with_invalid_combination(self, arglist, verifylist):
self.assertRaises(base.ParserException, self.check_parser,
self.list_vnf_package, arglist, verifylist)
def test_take_action_with_valid_combination(self):
parsed_args = self.check_parser(
self.list_vnf_package,
["--fields", 'softwareImages,checksum', "--exclude_default"],
[('fields', 'softwareImages,checksum'), ('exclude_default', True)])
vnf_packages = self._get_vnf_packages(onboarded_vnf_package=True)
updated_vnf_packages = {'vnf_packages': []}
for vnf_pkg in vnf_packages['vnf_packages']:
vnf_pkg.pop('userDefinedData')
updated_vnf_packages['vnf_packages'].append(vnf_pkg)
self._get_mock_response_for_list_vnf_packages(
'exclude_default&fields=softwareImages,checksum',
json=updated_vnf_packages)
actual_columns, data = self.list_vnf_package.take_action(parsed_args)
expected_data = []
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnf_package.get_attributes(
extra_fields=['softwareImages', 'checksum'],
exclude_default=True),
long_listing=True)
for vnf_package_obj in updated_vnf_packages['vnf_packages']:
expected_data.append(vnf_package_fakes.get_vnf_package_data(
vnf_package_obj, columns=columns, list_action=True,
exclude_default=True))
self.assertCountEqual(self.get_list_columns(
extra_fields=['Software Images', 'Checksum'],
exclude_default=True),
actual_columns)
self.assertListItemsEqual(expected_data, list(data))
self.assertItemsEqual(_get_columns_vnf_package(), actual_columns)
self.assertItemsEqual(expected_data, list(data))
@ddt.ddt
@@ -307,13 +149,10 @@ class TestShowVnfPackage(TestVnfPackage):
self.requests_mock.register_uri('GET', url, json=vnf_package_obj,
headers=self.header)
columns, data = (self.show_vnf_package.take_action(parsed_args))
self.assertCountEqual(_get_columns_vnf_package(
self.assertItemsEqual(_get_columns_vnf_package(
vnf_package_obj=vnf_package_obj, action='show'), columns)
headers, attributes = vnf_package._get_columns(vnf_package_obj)
self.assertListItemsEqual(
vnf_package_fakes.get_vnf_package_data(vnf_package_obj,
columns=attributes), data)
self.assertItemsEqual(
vnf_package_fakes.get_vnf_package_data(vnf_package_obj), data)
def test_show_no_options(self):
self.assertRaises(base.ParserException, self.check_parser,
@@ -331,15 +170,15 @@ class TestDeleteVnfPackage(TestVnfPackage):
self._vnf_package = vnf_package_fakes.create_vnf_packages(count=3)
def _mock_request_url_for_delete(self, vnf_pkg_index):
url = (self.url + '/vnfpkgm/v1/vnf_packages/' +
self._vnf_package['vnf_packages'][vnf_pkg_index]['id'])
url = (self.url + '/vnfpkgm/v1/vnf_packages/' +
self._vnf_package['vnf_packages'][vnf_pkg_index]['id'])
json = self._vnf_package['vnf_packages'][vnf_pkg_index]
json = self._vnf_package['vnf_packages'][vnf_pkg_index]
self.requests_mock.register_uri('GET', url, json=json,
headers=self.header)
self.requests_mock.register_uri('DELETE', url,
headers=self.header, json={})
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_package(self):
arglist = [self._vnf_package['vnf_packages'][0]['id']]
@@ -487,317 +326,3 @@ class TestUploadVnfPackage(TestVnfPackage):
self.upload_vnf_package.take_action, parsed_args)
# Delete temporary folder
shutil.rmtree(temp_dir)
def test_upload_vnf_package_failed_with_404_not_found(self):
# Scenario in which vnf package is not found
zip_file, temp_dir = _create_zip()
arglist = [
'dumy-id',
"--path", zip_file
]
verifylist = [
('path', zip_file),
('vnf_package', 'dumy-id')
]
parsed_args = self.check_parser(self.upload_vnf_package, arglist,
verifylist)
error_message = "Can not find requested vnf package: dummy-id"
body = {"itemNotFound": {"message": error_message, "code": 404}}
url = self.url + '/vnfpkgm/v1/vnf_packages/dumy-id/package_content'
self.requests_mock.register_uri(
'PUT', url, json=body,
status_code=404)
exception = self.assertRaises(
exceptions.TackerClientException,
self.upload_vnf_package.take_action, parsed_args)
self.assertEqual(error_message, exception.message)
# Delete temporary folder
shutil.rmtree(temp_dir)
@ddt.ddt
class TestUpdateVnfPackage(TestVnfPackage):
def setUp(self):
super(TestUpdateVnfPackage, self).setUp()
self.update_vnf_package = vnf_package.UpdateVnfPackage(
self.app, self.app_args, cmd_name='vnf package update')
@ddt.data((["--user-data", 'Test_key=Test_value',
"--operational-state", 'DISABLED'],
[('user_data', {'Test_key': 'Test_value'}),
('operational_state', 'DISABLED')]),
(["--user-data", 'Test_key=Test_value'],
[('user_data', {'Test_key': 'Test_value'})]),
(["--operational-state", 'DISABLED'],
[('operational_state', 'DISABLED')]))
@ddt.unpack
def test_take_action(self, arglist, verifylist):
vnf_package_obj = vnf_package_fakes.vnf_package_obj(
onboarded_state=True)
arglist.append(vnf_package_obj['id'])
verifylist.append(('vnf_package', vnf_package_obj['id']))
parsed_args = self.check_parser(self.update_vnf_package, arglist,
verifylist)
url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages',
vnf_package_obj['id'])
fake_response = vnf_package_fakes.get_fake_update_vnf_package_obj(
arglist)
self.requests_mock.register_uri('PATCH', url,
json=fake_response,
headers=self.header)
columns, data = self.update_vnf_package.take_action(parsed_args)
self.assertCountEqual(_get_columns_vnf_package(
vnf_package_obj=fake_response, action='update'), columns)
self.assertListItemsEqual(
vnf_package_fakes.get_vnf_package_data(fake_response), data)
def test_update_no_options(self):
self.assertRaises(base.ParserException, self.check_parser,
self.update_vnf_package, [], [])
def test_update_without_user_data_and_operational_state(self):
vnf_package_obj = vnf_package_fakes.vnf_package_obj(
onboarded_state=True)
arglist = [vnf_package_obj['id']]
verifylist = [('vnf_package', vnf_package_obj['id'])]
parsed_args = self.check_parser(self.update_vnf_package, arglist,
verifylist)
self.assertRaises(SystemExit, self.update_vnf_package.take_action,
parsed_args)
@ddt.ddt
class TestDownloadVnfPackage(TestVnfPackage):
# The new vnf package created.
_vnf_package = vnf_package_fakes.vnf_package_obj(
attrs={'userDefinedData': {'Test_key': 'Test_value'}})
def setUp(self):
super(TestDownloadVnfPackage, self).setUp()
self.download_vnf_package = vnf_package.DownloadVnfPackage(
self.app, self.app_args, cmd_name='vnf package download')
def test_download_no_options(self):
self.assertRaises(base.ParserException, self.check_parser,
self.download_vnf_package, [], [])
def _mock_request_url_for_download_vnfd(self, content_type, vnfd_data):
self.header = {'content-type': content_type}
url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages',
self._vnf_package['id'], 'vnfd')
if content_type == 'text/plain':
self.requests_mock.register_uri('GET', url,
headers=self.header,
text=vnfd_data)
else:
self.requests_mock.register_uri('GET', url,
headers=self.header,
content=vnfd_data)
def _get_arglist_and_verifylist(self, accept_type, file_name):
arglist = [
self._vnf_package['id'],
'--vnfd',
'--type', accept_type,
'--file', file_name
]
verifylist = [
('type', accept_type),
('vnfd', True),
('vnf_package', self._vnf_package['id']),
('file', file_name)
]
return arglist, verifylist
def test_download_vnfd_from_vnf_package_for_type_text_plain(self):
test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
'sample_vnf_package/Definitions/'
'etsi_nfv_sol001_common_types.yaml')
local_file = tempfile.NamedTemporaryFile(suffix='vnfd_data.yaml')
vnfd_data = open(test_file, 'r').read()
arglist, verifylist = self._get_arglist_and_verifylist(
'text/plain', local_file.name)
parsed_args = self.check_parser(self.download_vnf_package, arglist,
verifylist)
self._mock_request_url_for_download_vnfd('text/plain', vnfd_data)
self.download_vnf_package.take_action(parsed_args)
self.assertTrue(filecmp.cmp(test_file, local_file.name),
"Downloaded contents don't match test file")
@ddt.data('application/zip', 'both')
def test_download_vnfd_from_vnf_package(self, accept_type):
test_file, temp_dir = _create_zip()
# File in which VNFD data will be stored.
# For testing purpose we are creating temporary zip file.
local_file = tempfile.NamedTemporaryFile(suffix='vnfd_data.zip')
vnfd_data = open(test_file, 'rb').read()
arglist, verifylist = self._get_arglist_and_verifylist(
accept_type, local_file.name)
parsed_args = self.check_parser(self.download_vnf_package, arglist,
verifylist)
# When --type argument is set to 'both', then 'Accept' header in
# request is set to 'text/plain,application/zip' now it is up to the
# NFVO to choose the format to return for a single-file VNFD and for
# a multi-file VNFD, a ZIP file shall be returned. Here we have taken
# the example of multi-file vnfd hence its retuning zip file and
# setting the 'Content-Type' as 'application/zip' in response header.
self._mock_request_url_for_download_vnfd('application/zip', vnfd_data)
self.download_vnf_package.take_action(parsed_args)
self.assertTrue(filecmp.cmp(test_file, local_file.name),
"Downloaded contents don't match test file")
self.assertTrue(self._check_valid_zip_file(local_file.name))
shutil.rmtree(temp_dir)
def _check_valid_zip_file(self, zip_file):
with zipfile.ZipFile(zip_file) as zf:
ret = zf.testzip()
return False if ret else True
@mock.patch('builtins.print')
def test_download_vnfd_from_vnf_package_without_file_arg(self, mock_print):
# --file argument is optional when --type is set to 'text/plain'.
arglist = [
self._vnf_package['id'],
'--vnfd',
'--type', 'text/plain',
]
verifylist = [
('type', 'text/plain'),
('vnfd', True),
('vnf_package', self._vnf_package['id']),
]
parsed_args = self.check_parser(self.download_vnf_package, arglist,
verifylist)
test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
'sample_vnf_package/Definitions/'
'etsi_nfv_sol001_common_types.yaml')
vnfd_data = open(test_file, 'r').read()
self._mock_request_url_for_download_vnfd('text/plain', vnfd_data)
self.download_vnf_package.take_action(parsed_args)
mock_print.assert_called_once_with(vnfd_data)
@ddt.data('application/zip', 'both')
def test_download_vnfd_from_vnf_package_failed_with_no_file_arg(
self, accept_type):
arglist = [
self._vnf_package['id'],
'--vnfd',
'--type', accept_type,
]
verifylist = [
('type', accept_type),
('vnfd', True),
('vnf_package', self._vnf_package['id']),
]
parsed_args = self.check_parser(self.download_vnf_package, arglist,
verifylist)
with mock.patch.object(sys.stdout, "isatty") as mock_isatty:
mock_isatty.return_value = True
self.assertRaises(SystemExit,
self.download_vnf_package.take_action,
parsed_args)
def test_download_vnf_package(self):
file_name = 'vnf_package_data.zip'
test_file, temp_dir = _create_zip()
# file in which VNF Package data will be stored.
# for testing purpose we are creating temporary zip file.
local_file = tempfile.NamedTemporaryFile(suffix=file_name)
vnf_package_data = open(test_file, 'rb').read()
arglist = [
self._vnf_package['id'],
'--file', local_file.name
]
verifylist = [
('vnf_package', self._vnf_package['id']),
('file', local_file.name)
]
parsed_args = self.check_parser(self.download_vnf_package, arglist,
verifylist)
url = os.path.join(self.url, '/vnfpkgm/v1/vnf_packages',
self._vnf_package['id'], 'package_content')
self.requests_mock.register_uri(
'GET', url, headers={'content-type': 'application/zip'},
content=vnf_package_data)
self.download_vnf_package.take_action(parsed_args)
self.assertTrue(filecmp.cmp(test_file, local_file.name),
"Downloaded contents don't match test file")
self.assertTrue(self._check_valid_zip_file(local_file.name))
shutil.rmtree(temp_dir)
@ddt.ddt
class TestDownloadVnfPackageArtifact(TestVnfPackage):
# The new vnf package created.
_vnf_package = vnf_package_fakes.vnf_package_obj(
attrs={'userDefinedData': {'Test_key': 'Test_value'}})
def setUp(self):
super(TestDownloadVnfPackageArtifact, self).setUp()
self.download_vnf_package_artifacts = vnf_package.\
DownloadVnfPackageArtifact(
self.app, self.app_args,
cmd_name='vnf package artifact download')
def test_download_no_options(self):
self.assertRaises(base.ParserException, self.check_parser,
self.download_vnf_package_artifacts, [], [])
def _mock_request_url_for_download_artifacts(
self, artifact_path, artifact_data):
self.header = {'content-type': 'text/plain'}
url = os.path.join(self.url, 'vnfpkgm/v1/vnf_packages',
self._vnf_package['id'], 'artifacts', artifact_path)
self.requests_mock.register_uri('GET', url,
headers=self.header,
text=artifact_data)
def _get_arglist_and_verifylist(self, localfile):
arglist = [
self._vnf_package['id'],
localfile.name[1:],
'--file', localfile.name
]
verifylist = [
('vnf_package', self._vnf_package['id']),
('artifact_path', localfile.name[1:]),
('file', localfile.name)
]
return arglist, verifylist
def test_download_artifacts_from_vnf_package(self):
test_file = ('./tackerclient/tests//unit/osc/v1/fixture_data/'
'sample_vnf_package_artifacts/Scripts/'
'install.sh')
local_file = tempfile.NamedTemporaryFile(suffix='install.sh')
artifact_data = open(test_file, 'r').read()
arglist, verifylist = self._get_arglist_and_verifylist(
local_file)
parsed_args = self.check_parser(
self.download_vnf_package_artifacts, arglist, verifylist)
self._mock_request_url_for_download_artifacts(
local_file.name[1:], artifact_data)
self.download_vnf_package_artifacts.take_action(parsed_args)
self.assertTrue(filecmp.cmp(test_file, local_file.name),
"Downloaded contents don't match test file")

View File

@@ -1,865 +0,0 @@
# 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.
from io import StringIO
import os
import sys
from unittest import mock
import ddt
from oslo_utils.fixture import uuidsentinel
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.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']
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']]
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):
vnf_instances = vnflcm_fakes.create_vnf_instances(count=3)
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):
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=self.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 self.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: File %s does not exist "
"or user does not have read privileges to it")
self.assertEqual(expected_msg % sample_param_file, 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')
@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'])]),
([], []))
@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)
@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' 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' 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)
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: File %s does not exist "
"or user does not have read privileges to it")
self.assertEqual(expected_msg % sample_param_file, 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: File %s does not exist "
"or user does not have read privileges to it")
self.assertEqual(expected_msg % sample_param_file, 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 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.

View File

@@ -1,503 +0,0 @@
# 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 io import StringIO
import os
import sys
from oslo_utils.fixture import uuidsentinel
from unittest import mock
from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
from tackerclient.osc.v1.vnflcm import vnflcm_op_occs
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_op_occs_fakes
def _get_columns_vnflcm_op_occs(action='show'):
if action == 'fail':
return ['ID', 'Operation State', 'State Entered Time',
'Start Time', 'VNF Instance ID', 'Operation',
'Is Automatic Invocation', 'Is Cancel Pending',
'Error', 'Links']
elif action == 'list':
return ['ID', 'Operation State', 'VNF Instance ID',
'Operation']
else:
return ['ID', 'Operation State', 'State Entered Time',
'Start Time', 'VNF Instance ID', 'Grant ID',
'Operation', 'Is Automatic Invocation',
'Operation Parameters', 'Is Cancel Pending',
'Cancel Mode', 'Error', 'Resource Changes',
'Changed Info', 'Changed External Connectivity', 'Links']
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
class TestRollbackVnfLcmOp(TestVnfLcm):
def setUp(self):
super(TestRollbackVnfLcmOp, self).setUp()
self.rollback_vnf_lcm = vnflcm_op_occs.RollbackVnfLcmOp(
self.app, self.app_args, cmd_name='vnflcm op rollback')
def test_take_action(self):
"""take_action normal system test"""
arglist = [uuidsentinel.vnf_lcm_op_occ_id]
verifylist = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
parsed_args = self.check_parser(
self.rollback_vnf_lcm, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'rollback')
self.requests_mock.register_uri(
'POST', url, headers=self.header, json={})
sys.stdout = buffer = StringIO()
self.rollback_vnf_lcm.take_action(parsed_args)
actual_message = buffer.getvalue().strip()
expected_message = (
'Rollback request for LCM operation ' +
uuidsentinel.vnf_lcm_op_occ_id +
' has been accepted')
self.assertEqual(expected_message, actual_message)
def test_take_action_vnf_lcm_op_occ_id_not_found(self):
"""take_action abnomaly system test"""
arglist = [uuidsentinel.vnf_lcm_op_occ_id]
verifylist = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
parsed_args = self.check_parser(
self.rollback_vnf_lcm, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'rollback')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=404, json={})
self.assertRaises(exceptions.TackerClientException,
self.rollback_vnf_lcm.take_action,
parsed_args)
class TestFailVnfLcmOp(TestVnfLcm):
def setUp(self):
super(TestFailVnfLcmOp, self).setUp()
self.fail_vnf_lcm = vnflcm_op_occs.FailVnfLcmOp(
self.app, self.app_args, cmd_name='vnflcm op fail')
def test_take_action(self):
"""Test of take_action()"""
vnflcm_op_occ = vnflcm_op_occs_fakes.vnflcm_op_occ_response(
action='fail')
arg_list = [vnflcm_op_occ['id']]
verify_list = [('vnf_lcm_op_occ_id', vnflcm_op_occ['id'])]
# command param
parsed_args = self.check_parser(
self.fail_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
vnflcm_op_occ['id'],
'fail')
self.requests_mock.register_uri(
'POST', url, headers=self.header, json=vnflcm_op_occ)
columns, data = (self.fail_vnf_lcm.take_action(parsed_args))
expected_columns = _get_columns_vnflcm_op_occs(action='fail')
self.assertCountEqual(expected_columns, columns)
def test_take_action_vnf_lcm_op_occ_id_not_found(self):
"""Test if vnf-lcm-op-occ-id does not find"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.fail_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'fail')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=404, json={})
self.assertRaises(exceptions.TackerClientException,
self.fail_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_state_is_conflict(self):
"""Test if vnf-lcm-op-occ state is conflict"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.fail_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'fail')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=409, json={})
self.assertRaises(exceptions.TackerClientException,
self.fail_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_internal_server_error(self):
"""Test if request is internal server error"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.fail_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'fail')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=500, json={})
self.assertRaises(exceptions.TackerClientException,
self.fail_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_missing_vnf_lcm_op_occ_id_argument(
self):
"""Test if vnflcm_op_occ_id is not provided"""
arg_list = []
verify_list = [('vnf_lcm_op_occ_id', arg_list)]
self.assertRaises(base.ParserException, self.check_parser,
self.fail_vnf_lcm, arg_list, verify_list)
class TestRetryVnfLcmOp(TestVnfLcm):
def setUp(self):
super(TestRetryVnfLcmOp, self).setUp()
self.retry_vnf_lcm = vnflcm_op_occs.RetryVnfLcmOp(
self.app, self.app_args, cmd_name='vnflcm op retry')
def test_take_action(self):
"""Test of take_action()"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.retry_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'retry')
self.requests_mock.register_uri(
'POST', url, headers=self.header, json={})
sys.stdout = buffer = StringIO()
self.retry_vnf_lcm.take_action(parsed_args)
actual_message = buffer.getvalue().strip()
expected_message = (
'Retry request for LCM operation ' +
uuidsentinel.vnf_lcm_op_occ_id +
' has been accepted')
self.assertEqual(expected_message, actual_message)
def test_take_action_vnf_lcm_op_occ_id_not_found(self):
"""Test if vnf-lcm-op-occ-id is not found."""
arglist = [uuidsentinel.vnf_lcm_op_occ_id]
verifylist = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.retry_vnf_lcm, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'retry')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=404, json={})
self.assertRaises(exceptions.TackerClientException,
self.retry_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_state_is_conflict(self):
"""Test if vnf-lcm-op-occ state is conflict"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.retry_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'retry')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=409, json={})
self.assertRaises(exceptions.TackerClientException,
self.retry_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_internal_server_error(self):
"""Test if request is internal server error"""
arg_list = [uuidsentinel.vnf_lcm_op_occ_id]
verify_list = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.retry_vnf_lcm, arg_list, verify_list)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id,
'retry')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=500, json={})
self.assertRaises(exceptions.TackerClientException,
self.retry_vnf_lcm.take_action,
parsed_args)
def test_take_action_vnf_lcm_op_occ_missing_vnf_lcm_op_occ_id_argument(
self):
"""Test if vnflcm_op_occ_id is not provided"""
arg_list = []
verify_list = [('vnf_lcm_op_occ_id', arg_list)]
self.assertRaises(base.ParserException, self.check_parser,
self.retry_vnf_lcm, arg_list, verify_list)
class TestListVnfLcmOp(TestVnfLcm):
def setUp(self):
super(TestListVnfLcmOp, self).setUp()
self.list_vnflcm_op_occ = vnflcm_op_occs.ListVnfLcmOp(
self.app, self.app_args, cmd_name='vnflcm op list')
def test_take_action(self):
vnflcm_op_occs_obj = vnflcm_op_occs_fakes.create_vnflcm_op_occs(
count=3)
parsed_args = self.check_parser(self.list_vnflcm_op_occ, [], [])
self.requests_mock.register_uri(
'GET', os.path.join(self.url,
'vnflcm/v1/vnf_lcm_op_occs'),
json=vnflcm_op_occs_obj, headers=self.header)
actual_columns, data = self.list_vnflcm_op_occ.take_action(parsed_args)
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnflcm_op_occ.get_attributes(), long_listing=True)
expected_data = []
for vnflcm_op_occ_obj_idx in vnflcm_op_occs_obj:
expected_data.append(vnflcm_op_occs_fakes.get_vnflcm_op_occ_data(
vnflcm_op_occ_obj_idx, columns=columns))
self.assertCountEqual(_get_columns_vnflcm_op_occs(action='list'),
actual_columns)
self.assertCountEqual(expected_data, list(data))
def test_take_action_with_filter(self):
vnflcm_op_occs_obj = vnflcm_op_occs_fakes.create_vnflcm_op_occs(
count=3)
parsed_args = self.check_parser(
self.list_vnflcm_op_occ,
["--filter", '(eq,operationState,STARTING)'],
[('filter', '(eq,operationState,STARTING)')])
self.requests_mock.register_uri(
'GET', os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs?'
'filter=(eq,operationState,STARTING)'),
json=vnflcm_op_occs_obj, headers=self.header)
actual_columns, data = self.list_vnflcm_op_occ.take_action(parsed_args)
headers, columns = tacker_osc_utils.get_column_definitions(
self.list_vnflcm_op_occ.get_attributes(), long_listing=True)
expected_data = []
for vnflcm_op_occ_obj_idx in vnflcm_op_occs_obj:
expected_data.append(vnflcm_op_occs_fakes.get_vnflcm_op_occ_data(
vnflcm_op_occ_obj_idx, columns=columns))
self.assertCountEqual(_get_columns_vnflcm_op_occs(action='list'),
actual_columns)
self.assertListItemsEqual(expected_data, list(data))
def test_take_action_with_incorrect_filter(self):
parsed_args = self.check_parser(
self.list_vnflcm_op_occ,
["--filter", '(operationState)'],
[('filter', '(operationState)')])
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs?filter=(operationState)')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=400, json={})
self.assertRaises(exceptions.TackerClientException,
self.list_vnflcm_op_occ.take_action,
parsed_args)
def test_take_action_internal_server_error(self):
parsed_args = self.check_parser(
self.list_vnflcm_op_occ,
["--filter", '(eq,operationState,STARTING)'],
[('filter', '(eq,operationState,STARTING)')])
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs?'
'filter=(eq,operationState,STARTING)')
self.requests_mock.register_uri(
'POST', url, headers=self.header, status_code=500, json={})
self.assertRaises(exceptions.TackerClientException,
self.list_vnflcm_op_occ.take_action,
parsed_args)
class TestShowVnfLcmOp(TestVnfLcm):
def setUp(self):
super(TestShowVnfLcmOp, self).setUp()
self.show_vnf_lcm_op_occs = vnflcm_op_occs.ShowVnfLcmOp(
self.app, self.app_args, cmd_name='vnflcm op show')
def test_take_action(self):
"""Test of take_action()"""
vnflcm_op_occ = vnflcm_op_occs_fakes.vnflcm_op_occ_response()
arglist = [vnflcm_op_occ['id']]
verifylist = [('vnf_lcm_op_occ_id', vnflcm_op_occ['id'])]
# command param
parsed_args = self.check_parser(
self.show_vnf_lcm_op_occs, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
vnflcm_op_occ['id'])
self.requests_mock.register_uri(
'GET', url, headers=self.header, json=vnflcm_op_occ)
columns, data = (self.show_vnf_lcm_op_occs.take_action(parsed_args))
self.assertCountEqual(_get_columns_vnflcm_op_occs(),
columns)
def test_take_action_vnf_lcm_op_occ_id_not_found(self):
"""Test if vnf-lcm-op-occ-id does not find."""
arglist = [uuidsentinel.vnf_lcm_op_occ_id]
verifylist = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.show_vnf_lcm_op_occs, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id)
self.requests_mock.register_uri(
'GET', url, headers=self.header, status_code=404, json={})
self.assertRaises(exceptions.TackerClientException,
self.show_vnf_lcm_op_occs.take_action,
parsed_args)
def test_take_action_internal_server_error(self):
"""Test for internal server error."""
arglist = [uuidsentinel.vnf_lcm_op_occ_id]
verifylist = [('vnf_lcm_op_occ_id', uuidsentinel.vnf_lcm_op_occ_id)]
# command param
parsed_args = self.check_parser(
self.show_vnf_lcm_op_occs, arglist, verifylist)
url = os.path.join(
self.url,
'vnflcm/v1/vnf_lcm_op_occs',
uuidsentinel.vnf_lcm_op_occ_id)
self.requests_mock.register_uri(
'GET', url, headers=self.header, status_code=500, json={})
self.assertRaises(exceptions.TackerClientException,
self.show_vnf_lcm_op_occs.take_action,
parsed_args)

View File

@@ -1,313 +0,0 @@
# 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
from unittest import mock
import fixtures
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.assertCountEqual(_get_columns_vnf_parameter(),
columns)
self.assertEqual(get_client_called_count,
self.requests_mock.call_count)
self.assertCountEqual(
ast.literal_eval(self.requests_mock.last_request.body),
ast.literal_eval(body))
self.assertCountEqual(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

@@ -15,8 +15,6 @@
from oslo_utils import uuidutils
from tackerclient.osc import utils as tacker_osc_utils
def vnf_package_obj(attrs=None, onboarded_state=False):
"""Create a fake vnf package.
@@ -36,7 +34,7 @@ def vnf_package_obj(attrs=None, onboarded_state=False):
"onboardingState": "CREATED",
"operationalState": "DISABLED",
"usageState": "NOT_IN_USE",
"userDefinedData": {'key': 'value'}}
"userDefinedData": None}
if onboarded_state:
fake_vnf_package = {"id": "60a6ac16-b50d-4e92-964b-b3cf98c7cf5c",
@@ -65,14 +63,10 @@ def vnf_package_obj(attrs=None, onboarded_state=False):
"imagePath": "string"
}
],
"checksum": {
"algorithm": "string",
"hash": "string"
},
"onboardingState": "ONBOARDED",
"operationalState": "ENABLED",
"usageState": "IN_USE",
"userDefinedData": {'key': 'value'},
"userDefinedData": None,
"_links": {
"self": {
"href": "string"
@@ -83,27 +77,17 @@ def vnf_package_obj(attrs=None, onboarded_state=False):
"packageContent": {
"href": "string"
}
},
"additionalArtifacts": [
{
"artifactPath": "string",
"metadata": {},
"checksum": {
"algorithm": "string",
"hash": "string"
}
}]
}
}}
# Overwrite default attributes.
fake_vnf_package.update(attrs)
return fake_vnf_package
def get_vnf_package_data(vnf_package_obj, **kwargs):
def get_vnf_package_data(vnf_package, list_action=False, columns=None):
"""Get the vnf package data from a FakeVnfPackage dict object.
:param vnf_package_obj:
:param vnf_package:
A FakeVnfPackage dict object
:return:
A list which may include the following values:
@@ -111,30 +95,23 @@ def get_vnf_package_data(vnf_package_obj, **kwargs):
'vnfd': {'href': 'string'}}, '60a6ac16-b50d-4e92-964b-b3cf98c7cf5c',
'CREATED', 'DISABLED', 'NOT_IN_USE', {'Test_key': 'Test_value'}]
"""
complex_attributes = ['softwareImages', 'checksum', '_links',
'userDefinedData', 'additionalArtifacts']
for attribute in complex_attributes:
if vnf_package_obj.get(attribute):
vnf_package_obj.update(
{attribute: tacker_osc_utils.FormatComplexDataColumn(
vnf_package_obj[attribute])})
if kwargs.get('list_action'):
if list_action:
vnf_package.pop('_links')
# In case of List VNF packages we get empty string as data for
# 'vnfProductName' if onboardingState is CREATED. Hence to match
# up with actual data we are adding here empty string.
if not vnf_package_obj.get('vnfProductName'):
vnf_package_obj['vnfProductName'] = ''
if not vnf_package.get('vnfProductName'):
vnf_package['vnfProductName'] = ''
if kwargs.get('columns'):
# return the list of data as per column order
return tuple([vnf_package_obj[key] for key in kwargs.get('columns')])
# return the list of data as per column order
if columns:
return tuple([vnf_package[key] for key in columns])
return tuple([vnf_package_obj[key] for key in sorted(
vnf_package_obj.keys())])
return tuple([vnf_package[key] for key in sorted(vnf_package.keys())])
def create_vnf_packages(count=2, onboarded_vnf_package=False):
def create_vnf_packages(count=2):
"""Create multiple fake vnf packages.
:param Dictionary attrs:
@@ -147,19 +124,5 @@ def create_vnf_packages(count=2, onboarded_vnf_package=False):
vnf_packages = []
for i in range(0, count):
unique_id = uuidutils.generate_uuid()
vnf_packages.append(vnf_package_obj(
attrs={'id': unique_id}, onboarded_state=onboarded_vnf_package))
vnf_packages.append(vnf_package_obj(attrs={'id': unique_id}))
return {'vnf_packages': vnf_packages}
def get_fake_update_vnf_package_obj(arglist):
fake_update_vnf_package_dict = {}
if '--user-data' in arglist:
fake_update_vnf_package_dict.update(
{"userDefinedData": {'Test_key': 'Test_value'}})
if '--operational-state' in arglist:
fake_update_vnf_package_dict.update({
"operationalState": "DISABLED",
})
return fake_update_vnf_package_dict

View File

@@ -1,154 +0,0 @@
# 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.
from oslo_utils.fixture import uuidsentinel
from oslo_utils import uuidutils
from tackerclient.osc import utils as tacker_osc_utils
def vnf_instance_response(attrs=None, instantiation_state='NOT_INSTANTIATED'):
"""Create a fake vnf instance.
:param Dictionary attrs:
A dictionary with all attributes
:return:
A vnf instance dict
"""
attrs = attrs or {}
# Set default attributes.
dummy_vnf_instance = {
"id": uuidsentinel.vnf_instance_id,
"vnfInstanceName": "Fake-VNF-Instance",
"vnfInstanceDescription": "Fake VNF",
"vnfdId": uuidsentinel.vnf_package_vnfd_id,
"vnfProvider": "NTT NS lab",
"vnfProductName": "Sample VNF",
"vnfSoftwareVersion": "1.0",
"vnfdVersion": "1.0",
"_links": "vnflcm/v1/vnf_instances/" + uuidsentinel.vnf_instance_id +
"/instantiate",
"instantiationState": instantiation_state,
"vnfConfigurableProperties": {
"test": "test_value"}}
if instantiation_state == 'INSTANTIATED':
dummy_vnf_instance.update({
"vimConnectionInfo": [{
'id': uuidsentinel.uuid,
'vimId': uuidsentinel.vimId,
'vimType': 'openstack',
'interfaceInfo': {'k': 'v'},
'accessInfo': {'k': 'v'},
'extra': {'k': 'v'}
}],
"instantiatedVnfInfo": {
"flavourId": uuidsentinel.flavourId,
"vnfState": "STARTED",
"extCpInfo": [{
'id': uuidsentinel.extCpInfo_uuid,
'cpdId': uuidsentinel.cpdId_uuid,
'cpProtocolInfo': [{
'layerProtocol': 'IP_OVER_ETHERNET',
'ipOverEthernet': '{}'
}],
'extLinkPortId': uuidsentinel.extLinkPortId_uuid,
'metadata': {'k': 'v'},
'associatedVnfcCpId': uuidsentinel.associatedVnfcCpId_uuid
}],
"extVirtualLinkInfo": [{
'id': uuidsentinel.extVirtualLinkInfo_uuid,
'resourceHandle': {},
'extLinkPorts': []
}],
"extManagedVirtualLinkInfo": [{
"id": uuidsentinel.extManagedVirtualLinkInfo_uuid,
'vnfVirtualLinkDescId': {},
'networkResource': {},
'vnfLinkPorts': []
}],
"vnfcResourceInfo": [{
'id': uuidsentinel.vnfcResourceInfo_uuid,
'vduId': uuidsentinel.vduId_uuid,
'computeResource': {},
'storageResourceIds': [],
'reservationId': uuidsentinel.reservationId,
}],
"vnfVirtualLinkResourceInfo": [{
'id': uuidsentinel.vnfVirtualLinkResourceInfo,
'vnfVirtualLinkDescId': 'VL4',
'networkResource': {},
'reservationId': uuidsentinel.reservationId,
'vnfLinkPorts': [],
'metadata': {'k': 'v'}
}],
"virtualStorageResourceInfo": [{
'id': uuidsentinel.virtualStorageResourceInfo,
'virtualStorageDescId': uuidsentinel.virtualStorageDescId,
'storageResource': {},
'reservationId': uuidsentinel.reservationId,
'metadata': {'k': 'v'}
}]
},
"_links": {
'self': 'self_link',
'indicators': None,
'instantiate': 'instantiate_link'
}
})
# Overwrite default attributes.
dummy_vnf_instance.update(attrs)
return dummy_vnf_instance
def get_vnflcm_data(vnf_instance, list_action=False, columns=None):
"""Get the vnf instance data.
:return:
A tuple object sorted based on the name of the columns.
"""
complex_attributes = ['vimConnectionInfo', 'instantiatedVnfInfo', '_links']
for attribute in complex_attributes:
if vnf_instance.get(attribute):
vnf_instance.update(
{attribute: tacker_osc_utils.FormatComplexDataColumn(
vnf_instance[attribute])})
if list_action:
for item in ['vnfInstanceDescription', 'vnfdVersion']:
vnf_instance.pop(item)
# return the list of data as per column order
if columns:
return tuple([vnf_instance[key] for key in columns])
return tuple([vnf_instance[key] for key in sorted(
vnf_instance.keys())])
def create_vnf_instances(count=2):
"""Create multiple fake vnf instances.
:param count: The number of vnf instances to fake
:return:
A list of fake vnf instances dictionary
"""
vnf_instances = []
for i in range(0, count):
unique_id = uuidutils.generate_uuid()
vnf_instances.append(vnf_instance_response(attrs={'id': unique_id}))
return vnf_instances

View File

@@ -1,107 +0,0 @@
# 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 oslo_utils.fixture import uuidsentinel
from oslo_utils import uuidutils
from tackerclient.osc import utils as tacker_osc_utils
def vnflcm_op_occ_response(attrs=None, action=''):
"""Create a fake vnflcm op occurrence.
:param Dictionary attrs:
A dictionary with all attributes
:return:
A vnf lcm op occs dict
"""
attrs = attrs or {}
# Set default attributes.
dummy_vnf_lcm_op_occ = {
"id": uuidsentinel.vnflcm_op_occ_id,
"operationState": "STARTING",
"stateEnteredTime": "2018-12-22T16:59:45.187Z",
"startTime": "2018-12-22T16:59:45.187Z",
"vnfInstanceId": "376f37f3-d4e9-4d41-8e6a-9b0ec98695cc",
"grantId": "",
"operation": "INSTANTIATE",
"isAutomaticInvocation": "true",
"operationParams": {
"flavourId": "default",
"instantiationLevelId": "n-mme-min"
},
"isCancelPending": "true",
"cancelMode": "",
"error": {
"status": "500",
"detail": "internal server error"
},
"resourceChanges": [],
"changedInfo": [],
"changedExtConnectivity": [],
"_links": {
"self": ""
}
}
if action == 'fail':
fail_not_needed_columns = [
'grantId', 'operationParams',
'cancelMode', 'resourceChanges', 'changedInfo',
'changedExtConnectivity']
for key in fail_not_needed_columns:
del dummy_vnf_lcm_op_occ[key]
# Overwrite default attributes.
dummy_vnf_lcm_op_occ.update(attrs)
return dummy_vnf_lcm_op_occ
def get_vnflcm_op_occ_data(vnf_lcm_op_occ, columns=None):
"""Get the vnflcm op occurrence.
:return:
A tuple object sorted based on the name of the columns.
"""
complex_attributes = [
'operationParams', 'error', 'resourceChanges',
'changedInfo', 'changedExtConnectivity', 'links']
for attribute in complex_attributes:
if vnf_lcm_op_occ.get(attribute):
vnf_lcm_op_occ.update(
{attribute: tacker_osc_utils.FormatComplexDataColumn(
vnf_lcm_op_occ[attribute])})
# return the list of data as per column order
if columns:
return tuple([vnf_lcm_op_occ[key] for key in columns])
return tuple([vnf_lcm_op_occ[key] for key in sorted(
vnf_lcm_op_occ.keys())])
def create_vnflcm_op_occs(count=2):
"""Create multiple fake vnflcm op occs.
:param count: The number of vnflcm op occs to fake
:return:
A list of fake vnflcm op occs dictionary
"""
vnflcm_op_occs = []
for i in range(0, count):
unique_id = uuidutils.generate_uuid()
vnflcm_op_occs.append(vnflcm_op_occ_response(attrs={'id': unique_id}))
return vnflcm_op_occs

View File

@@ -1,32 +0,0 @@
# Copyright (C) 2021 Nippon Telegraph and Telephone Corporation
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import client
class TestVnfLcmV2(base.FixturedTestCase):
client_fixture_class = client.ClientFixture
api_version = '2'
def setUp(self):
super(TestVnfLcmV2, self).setUp()
def test_client_v2(self):
self.assertEqual(self.cs.vnf_lcm_client.headers,
{'Version': '2.0.0'})
self.assertEqual(self.cs.vnf_lcm_client.vnf_instances_path,
'/vnflcm/v2/vnf_instances')
# check of other paths is omitted.

View File

@@ -16,10 +16,10 @@
import copy
import json
from unittest import mock
import uuid
from keystoneclient import exceptions as k_exceptions
import mock
import requests
import testtools

View File

@@ -14,14 +14,15 @@
# under the License.
#
import urllib
import contextlib
import fixtures
import io
import mock
import six
import six.moves.urllib.parse as urlparse
import sys
import testtools
from unittest import mock
import urllib
from urllib import parse as urlparse
from tackerclient.common import constants
from tackerclient.common import exceptions
@@ -39,7 +40,7 @@ ENDURL = 'localurl'
@contextlib.contextmanager
def capture_std_streams():
fake_stdout, fake_stderr = io.StringIO(), io.StringIO()
fake_stdout, fake_stderr = six.StringIO(), six.StringIO()
stdout, stderr = sys.stdout, sys.stderr
try:
sys.stdout, sys.stderr = fake_stdout, fake_stderr
@@ -316,7 +317,7 @@ class CLITestV10Base(testtools.TestCase):
args.append("--tag")
for tag in tags:
args.append(tag)
if isinstance(tag, str):
if isinstance(tag, six.string_types):
tag = urllib.quote(tag.encode('utf-8'))
if query:
query += "&tag=" + tag
@@ -413,7 +414,7 @@ class CLITestV10Base(testtools.TestCase):
args.append("--tag")
for tag in tags:
args.append(tag)
if isinstance(tag, str):
if isinstance(tag, six.string_types):
tag = urllib.quote(tag.encode('utf-8'))
if query:
query += "&tag=" + tag
@@ -572,7 +573,6 @@ class CLITestV10Base(testtools.TestCase):
'PUT',
body=_body,
headers=test_utils.ContainsKeyValue('X-Auth-Token', TOKEN))
self.assertEqual(get_client_called_count, mock_get.call_count)
_str = self.fake_stdout.make_string()
self.assertIn(myid, _str)
@@ -653,8 +653,8 @@ class CLITestV10Base(testtools.TestCase):
class ClientV1TestJson(CLITestV10Base):
def test_do_request_unicode(self):
self.client.format = self.format
unicode_text = '\u7f51\u7edc'
action = '/test'
unicode_text = u'\u7f51\u7edc'
action = u'/test'
params = {'test': unicode_text}
body = params
expect_body = self.client.serialize(body)
@@ -663,7 +663,7 @@ class ClientV1TestJson(CLITestV10Base):
mock_req.return_value = (MyResp(200), expect_body)
res_body = self.client.do_request('PUT', action, body=body,
params=params)
expected_uri = 'localurl/v1.0/test.json?test=%E7%BD%91%E7%BB%9C'
expected_uri = u'localurl/v1.0/test.json?test=%E7%BD%91%E7%BB%9C'
mock_req.assert_called_with(
expected_uri, 'PUT', body=expect_body,
headers={'X-Auth-Token': unicode_text,
@@ -798,27 +798,3 @@ class CLITestV10ExceptionHandler(CLITestV10Base):
exceptions.TackerClientException, 500,
expected_msg=expected_msg,
error_content=error_content)
def test_exception_handler_v10_tacker_etsi_error(self):
"""Test ETSI error response"""
known_error_map = [
({
"status": "status 1",
"detail": "sample 1"
}, 400),
({
"status": "status 2",
"detail": "sample 2"
}, 404),
({
"status": "status 3",
"detail": "sample 3"
}, 409)
]
for error_content, status_code in known_error_map:
self._test_exception_handler_v10(
exceptions.TackerClientException, status_code,
expected_msg=error_content['detail'],
error_content=error_content)

View File

@@ -17,6 +17,7 @@
import logging
import testtools
from testtools import helpers
from tackerclient.tacker import v1_0 as tackerV10
@@ -26,7 +27,7 @@ class TestCommandMeta(testtools.TestCase):
class FakeCommand(tackerV10.TackerCommand):
pass
self.assertTrue(hasattr(FakeCommand, 'log'))
self.assertTrue(helpers.safe_hasattr(FakeCommand, 'log'))
self.assertIsInstance(FakeCommand.log, logging.getLoggerClass())
self.assertEqual(FakeCommand.log.name, __name__ + ".FakeCommand")
@@ -34,5 +35,5 @@ class TestCommandMeta(testtools.TestCase):
class FakeCommand(tackerV10.TackerCommand):
log = None
self.assertTrue(hasattr(FakeCommand, 'log'))
self.assertTrue(helpers.safe_hasattr(FakeCommand, 'log'))
self.assertIsNone(FakeCommand.log)

View File

@@ -13,8 +13,8 @@
# License for the specific language governing permissions and limitations
# under the License.
import mock
import testtools
from unittest import mock
from tackerclient.client import HTTPClient
from tackerclient.common import exceptions

View File

@@ -14,17 +14,17 @@
# under the License.
import argparse
import fixtures
import io
import logging
import os
import re
import six
import sys
import fixtures
from keystoneclient import session
import mock
import testtools
from testtools import matchers
from unittest import mock
from keystoneclient import session
from tackerclient.common import clientmanager
from tackerclient import shell as openstack_shell
@@ -62,8 +62,8 @@ class ShellTest(testtools.TestCase):
clean_env = {}
_old_env, os.environ = os.environ, clean_env.copy()
try:
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()
sys.stdout = six.StringIO()
sys.stderr = six.StringIO()
_shell = openstack_shell.TackerShell(DEFAULT_API_VERSION)
_shell.run(argstr.split())
except SystemExit:
@@ -154,10 +154,10 @@ class ShellTest(testtools.TestCase):
@mock.patch.object(openstack_shell.TackerShell, 'run')
def test_main_with_unicode(self, mock_run):
mock_run.return_value = 0
unicode_text = '\u7f51\u7edc'
unicode_text = u'\u7f51\u7edc'
argv = ['net-list', unicode_text, unicode_text.encode('utf-8')]
ret = openstack_shell.main(argv=argv)
mock_run.assert_called_once_with(['net-list', unicode_text,
mock_run.assert_called_once_with([u'net-list', unicode_text,
unicode_text])
self.assertEqual(0, ret)

View File

@@ -13,10 +13,9 @@
# License for the specific language governing permissions and limitations
# under the License.
from unittest import mock
import fixtures
from keystoneclient import session
import mock
import requests
import testtools

View File

@@ -14,10 +14,9 @@
# License for the specific language governing permissions and limitations
# under the License.
import sys
from unittest import mock
from tackerclient.common import exceptions
from tackerclient.common import utils
import mock
from tackerclient import shell
from tackerclient.tacker import v1_0 as tackerV1_0
from tackerclient.tacker.v1_0 import TackerCommand
@@ -188,313 +187,7 @@ class CLITestV10VmVNFJSON(test_cli10.CLITestV10Base):
[my_id, '--%s' % key, value],
{key: value})
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):
def test_delete_vnf_without_force(self):
cmd = vnf.DeleteVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]

View File

@@ -14,13 +14,13 @@
# License for the specific language governing permissions and limitations
# under the License.
from mock import mock_open
from mock import patch
import sys
from tackerclient.common.exceptions import InvalidInput
from tackerclient.tacker.v1_0.vnfm import vnfd
from tackerclient.tests.unit import test_cli10
from unittest.mock import mock_open
from unittest.mock import patch
class CLITestV10VmVNFDJSON(test_cli10.CLITestV10Base):

View File

@@ -113,14 +113,8 @@ class CLITestV10VmVNFFGJSON(test_cli10.CLITestV10Base):
args, extra_fields,
get_client_called_count=2)
def test_delete_vnffg_without_force(self):
def test_delete_vnffg(self):
cmd = vnffg.DeleteVNFFG(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id]
self._test_delete_resource(self._RESOURCE, cmd, my_id, args)
def test_delete_vnffg_with_force(self):
cmd = vnffg.DeleteVNFFG(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
args = [my_id, '--force']
self._test_delete_resource(self._RESOURCE, cmd, my_id, args)

View File

@@ -10,12 +10,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from mock import mock_open
from mock import patch
import sys
from tackerclient.tacker.v1_0.nfvo import vnffgd
from tackerclient.tests.unit import test_cli10
from unittest.mock import mock_open
from unittest.mock import patch
class CLITestV10VmVNFFGDJSON(test_cli10.CLITestV10Base):

View File

@@ -13,11 +13,12 @@
# License for the specific language governing permissions and limitations
# under the License.
from mock import sentinel
import testtools
from tackerclient.common import exceptions
from tackerclient.tacker.v1_0.nfvo import vim_utils
from unittest.mock import sentinel
class TestVIMUtils(testtools.TestCase):

View File

@@ -19,7 +19,7 @@ import logging
import time
import requests
from urllib import parse as urlparse
import six.moves.urllib.parse as urlparse
from tackerclient import client
from tackerclient.common import constants
@@ -54,7 +54,6 @@ def exception_handler_v10(status_code, error_content):
:param status_code: HTTP error status code
:param error_content: deserialized body of error response
"""
etsi_error_content = error_content
error_dict = None
if isinstance(error_content, dict):
error_dict = error_content.get('TackerError')
@@ -95,13 +94,6 @@ def exception_handler_v10(status_code, error_content):
if message:
raise exceptions.TackerClientException(status_code=status_code,
message=message)
# ETSI error response
if isinstance(etsi_error_content, dict):
if etsi_error_content.get('status') and \
etsi_error_content.get('detail'):
message = etsi_error_content.get('detail')
raise exceptions.TackerClientException(status_code=status_code,
message=message)
# If we end up here the exception was not a tacker error
msg = "%s-%s" % (status_code, error_content)
@@ -199,53 +191,20 @@ class ClientBase(object):
action = self.action_prefix + action
return action
def _build_params_query(self, params=None):
flag_params = []
keyval_params = {}
for key, value in params.items():
if value is None:
flag_params.append(key)
else:
keyval_params[key] = value
flags_encoded = utils.safe_encode_list(flag_params) \
if flag_params else ""
keyval_encoded = utils.safe_encode_dict(keyval_params) \
if keyval_params else ""
query = ""
for flag in flags_encoded:
query = query + urlparse.quote_plus(flag) + '&'
query = query + urlparse.urlencode(keyval_encoded, doseq=1)
return query.strip('&')
def do_request(self, method, action, body=None, headers=None, params=None):
action = self.build_action(action)
# Add format and tenant_id
if type(params) is dict and params:
query = self._build_params_query(params)
action += '?' + query
params = utils.safe_encode_dict(params)
action += '?' + urlparse.urlencode(params, doseq=1)
if body or body == {}:
body = self.serialize(body)
if headers is None:
# self.httpclient.do_request is not accept 'headers=None'.
headers = {}
resp, replybody = self.httpclient.do_request(
action, method, body=body, headers=headers,
action, method, body=body,
content_type=self.content_type())
if 'application/zip' == resp.headers.get('Content-Type'):
self.format = 'zip'
elif 'text/plain' == resp.headers.get('Content-Type'):
self.format = 'text'
elif 'artifacts' in action:
self.format = 'any'
else:
self.format = 'json'
status_code = resp.status_code
if status_code in (requests.codes.ok,
requests.codes.created,
@@ -268,7 +227,7 @@ class ClientBase(object):
"""
if data is None:
return None
elif self.format in ('zip', 'text'):
elif self.format == 'zip':
return data
elif type(data) is dict:
return serializer.Serializer(
@@ -279,7 +238,7 @@ class ClientBase(object):
def deserialize(self, data, status_code):
"""Deserializes an XML or JSON string into a dictionary."""
if status_code in (204, 202) or self.format in ('zip', 'text', 'any'):
if status_code in (204, 202):
return data
return serializer.Serializer(self.get_attr_metadata()).deserialize(
data, self.content_type())['body']
@@ -298,17 +257,12 @@ class ClientBase(object):
constants.EXT_NS: ns}
def content_type(self, _format=None):
"""Returns the mime-type for either 'xml', 'json, 'text', or 'zip'.
"""Returns the mime-type for either 'xml' or 'json'.
Defaults to the currently set format.
"""
_format = _format or self.format
if self.format == 'text':
return "text/plain"
elif self.format == 'both':
return "text/plain,application/zip"
else:
return "application/%s" % (_format)
return "application/%s" % (_format)
def retry_request(self, method, action, body=None,
headers=None, params=None):
@@ -355,39 +309,26 @@ class ClientBase(object):
return self.retry_request("PUT", action, body=body,
headers=headers, params=params)
def patch(self, action, body=None, headers=None, params=None):
return self.retry_request("PATCH", action, body=body,
headers=headers, params=params)
def list(self, collection, path, retrieve_all=True, headers=None,
**params):
def list(self, collection, path, retrieve_all=True, **params):
if retrieve_all:
res = []
for r in self._pagination(collection, path, headers, **params):
if type(r) is list:
res.extend(r)
else:
res.extend(r[collection])
return {collection: res} if collection else res
for r in self._pagination(collection, path, **params):
res.extend(r[collection])
return {collection: res}
else:
return self._pagination(collection, path, headers, **params)
return self._pagination(collection, path, **params)
def _pagination(self, collection, path, headers, **params):
def _pagination(self, collection, path, **params):
if params.get('page_reverse', False):
linkrel = 'previous'
else:
linkrel = 'next'
next = True
while next:
res = self.get(path, headers=headers, params=params)
res = self.get(path, params=params)
yield res
next = False
try:
# TODO(tpatil): Handle pagination for list data type
# once it's supported by tacker.
if type(res) is list:
break
for link in res['%s_links' % collection]:
if link['rel'] == linkrel:
query_str = urlparse.urlparse(link['href']).query
@@ -631,8 +572,8 @@ class LegacyClient(ClientBase):
return self.post(self.vnffgs_path, body=body)
@APIParamsCall
def delete_vnffg(self, vnffg, body=None):
return self.delete(self.vnffg_path % vnffg, body=body)
def delete_vnffg(self, vnffg):
return self.delete(self.vnffg_path % vnffg)
@APIParamsCall
def update_vnffg(self, vnffg, body):
@@ -788,10 +729,6 @@ class VnfPackageClient(ClientBase):
vnfpackages_path = '/vnfpkgm/v1/vnf_packages'
vnfpackage_path = '/vnfpkgm/v1/vnf_packages/%s'
vnfpackage_vnfd_path = '/vnfpkgm/v1/vnf_packages/%s/vnfd'
vnfpackage_download_path = '/vnfpkgm/v1/vnf_packages/%s/package_content'
vnfpakcage_artifact_path = '/vnfpkgm/v1/vnf_packages/%(id)s/artifacts/' \
'%(artifact_path)s'
def build_action(self, action):
return action
@@ -832,164 +769,6 @@ class VnfPackageClient(ClientBase):
base_path=self.vnfpackages_path),
body=file_data)
@APIParamsCall
def download_vnf_package(self, vnf_package):
self.format = 'zip'
return self.get(self.vnfpackage_download_path % vnf_package)
@APIParamsCall
def download_vnfd_from_vnf_package(self, vnf_package, accept):
"""Read VNFD of an on-boarded VNF Package.
:param vnf_package: The value can be either the ID of a vnf package
or a :class:`~openstack.nfv_orchestration.v1.
vnf_package` instance.
:param accept: Valid values are 'text/plain', 'application/zip' and
'both'. According to these values 'Accept' header will
be set as 'text/plain', 'application/zip',
'text/plain,application/zip' respectively.
:returns: If the VNFD is implemented in the form of multiple files,
a ZIP file embedding these files shall be returned.
If the VNFD is implemented as a single file, either that
file or a ZIP file embedding that file shall be returned.
"""
if accept == 'text/plain':
self.format = 'text'
elif accept == 'application/zip':
self.format = 'zip'
else:
self.format = 'both'
return self.get(self.vnfpackage_vnfd_path % vnf_package)
@APIParamsCall
def download_artifact_from_vnf_package(self, vnf_package, artifact_path):
return self.get(self.vnfpakcage_artifact_path %
{'id': vnf_package, 'artifact_path': artifact_path})
@APIParamsCall
def update_vnf_package(self, vnf_package, body):
return self.patch(self.vnfpackage_path % vnf_package, body=body)
class VnfLCMClient(ClientBase):
"""Client for vnflcm APIs.
Purpose of this class is to create required request url for vnflcm
APIs.
"""
def __init__(self, api_version, **kwargs):
super(VnfLCMClient, self).__init__(**kwargs)
self.headers = {'Version': '1.3.0'}
sol_api_version = 'v1'
if api_version == '2':
self.headers = {'Version': '2.0.0'}
sol_api_version = 'v2'
self.vnf_instances_path = (
'/vnflcm/{}/vnf_instances'.format(sol_api_version))
self.vnf_instance_path = (
'/vnflcm/{}/vnf_instances/%s'.format(sol_api_version))
self.vnf_lcm_op_occurrences_path = (
'/vnflcm/{}/vnf_lcm_op_occs'.format(sol_api_version))
self.vnf_lcm_op_occs_path = (
'/vnflcm/{}/vnf_lcm_op_occs/%s'.format(sol_api_version))
def build_action(self, action):
return action
@APIParamsCall
def create_vnf_instance(self, body):
return self.post(self.vnf_instances_path, body=body,
headers=self.headers)
@APIParamsCall
def show_vnf_instance(self, vnf_id, **_params):
return self.get(self.vnf_instance_path % vnf_id,
headers=self.headers, params=_params)
@APIParamsCall
def list_vnf_instances(self, retrieve_all=True, **_params):
vnf_instances = self.list(None, self.vnf_instances_path,
retrieve_all, headers=self.headers,
**_params)
return vnf_instances
@APIParamsCall
def instantiate_vnf_instance(self, vnf_id, body):
return self.post((self.vnf_instance_path + "/instantiate") % vnf_id,
body=body, headers=self.headers)
@APIParamsCall
def heal_vnf_instance(self, vnf_id, body):
return self.post((self.vnf_instance_path + "/heal") % vnf_id,
body=body, headers=self.headers)
@APIParamsCall
def terminate_vnf_instance(self, vnf_id, body):
return self.post((self.vnf_instance_path + "/terminate") % vnf_id,
body=body, headers=self.headers)
@APIParamsCall
def delete_vnf_instance(self, vnf_id):
return self.delete(self.vnf_instance_path % vnf_id,
headers=self.headers)
@APIParamsCall
def update_vnf_instance(self, vnf_id, body):
return self.patch(self.vnf_instance_path % vnf_id, body=body,
headers=self.headers)
@APIParamsCall
def scale_vnf_instance(self, vnf_id, body):
return self.post((self.vnf_instance_path + "/scale") % vnf_id,
body=body, headers=self.headers)
@APIParamsCall
def rollback_vnf_instance(self, occ_id):
return self.post((self.vnf_lcm_op_occs_path + "/rollback") % occ_id,
headers=self.headers)
@APIParamsCall
def fail_vnf_instance(self, occ_id):
return self.post((self.vnf_lcm_op_occs_path + "/fail") % occ_id,
headers=self.headers)
@APIParamsCall
def change_ext_conn_vnf_instance(self, vnf_id, body):
return self.post((self.vnf_instance_path + "/change_ext_conn") %
vnf_id, body=body, headers=self.headers)
@APIParamsCall
def retry_vnf_instance(self, occ_id):
return self.post((self.vnf_lcm_op_occs_path + "/retry") % occ_id,
headers=self.headers)
@APIParamsCall
def list_vnf_lcm_op_occs(self, retrieve_all=True, **_params):
vnf_lcm_op_occs = self.list(None, self.vnf_lcm_op_occurrences_path,
retrieve_all, headers=self.headers,
**_params)
return vnf_lcm_op_occs
@APIParamsCall
def show_vnf_lcm_op_occs(self, occ_id):
return self.get(self.vnf_lcm_op_occs_path % occ_id,
headers=self.headers)
@APIParamsCall
def show_vnf_lcm_versions(self, major_version):
if major_version is None:
path = "/vnflcm/api_versions"
else:
path = "/vnflcm/{}/api_versions".format(major_version)
# NOTE: This may be called with any combination of
# --os-tacker-api-verson:[1, 2] and major_version:[None, 1, 2].
# Specifying "headers={'Version': '2.0.0'}" is most simple to
# make all cases OK.
return self.get(path, headers={'Version': '2.0.0'})
class Client(object):
"""Unified interface to interact with multiple applications of tacker service.
@@ -1011,8 +790,6 @@ class Client(object):
"""
def __init__(self, **kwargs):
api_version = kwargs.pop('api_version', '1')
self.vnf_lcm_client = VnfLCMClient(api_version, **kwargs)
self.vnf_package_client = VnfPackageClient(**kwargs)
self.legacy_client = LegacyClient(**kwargs)
@@ -1189,8 +966,8 @@ class Client(object):
def create_ns(self, body):
return self.legacy_client.create_ns(body)
def delete_ns(self, ns, body=None):
return self.legacy_client.delete_ns(ns, body=body)
def delete_ns(self, ns):
return self.legacy_client.delete_ns(ns)
def create_cluster(self, body=None):
return self.legacy_client.create_cluster(body=body)
@@ -1224,8 +1001,7 @@ class Client(object):
def create_vnf_package(self, body):
return self.vnf_package_client.create_vnf_package(body)
def list_vnf_packages(self, retrieve_all=True, query_parameter=None,
**_params):
def list_vnf_packages(self, retrieve_all=True, **_params):
return self.vnf_package_client.list_vnf_packages(
retrieve_all=retrieve_all, **_params)
@@ -1238,71 +1014,3 @@ class Client(object):
def delete_vnf_package(self, vnf_package):
return self.vnf_package_client.delete_vnf_package(vnf_package)
# VnfLCMClient methods.
def create_vnf_instance(self, body):
return self.vnf_lcm_client.create_vnf_instance(body)
def show_vnf_instance(self, vnf_instance, **_params):
return self.vnf_lcm_client.show_vnf_instance(vnf_instance,
**_params)
def list_vnf_instances(self, retrieve_all=True, **_params):
return self.vnf_lcm_client.list_vnf_instances(
retrieve_all=retrieve_all, **_params)
def instantiate_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.instantiate_vnf_instance(vnf_id, body)
def heal_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.heal_vnf_instance(vnf_id, body)
def terminate_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.terminate_vnf_instance(vnf_id, body)
def scale_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.scale_vnf_instance(vnf_id, body)
def change_ext_conn_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.change_ext_conn_vnf_instance(vnf_id, body)
def delete_vnf_instance(self, vnf_id):
return self.vnf_lcm_client.delete_vnf_instance(vnf_id)
def update_vnf_instance(self, vnf_id, body):
return self.vnf_lcm_client.update_vnf_instance(vnf_id, body)
def rollback_vnf_instance(self, occ_id):
return self.vnf_lcm_client.rollback_vnf_instance(occ_id)
def fail_vnf_instance(self, occ_id):
return self.vnf_lcm_client.fail_vnf_instance(occ_id)
def retry_vnf_instance(self, occ_id):
return self.vnf_lcm_client.retry_vnf_instance(occ_id)
def update_vnf_package(self, vnf_package, body):
return self.vnf_package_client.update_vnf_package(vnf_package, body)
def download_vnfd_from_vnf_package(self, vnf_package, accept):
return self.vnf_package_client.download_vnfd_from_vnf_package(
vnf_package, accept)
def download_artifact_from_vnf_package(self, vnf_package, artifact_path):
return self.vnf_package_client.download_artifact_from_vnf_package(
vnf_package, artifact_path
)
def download_vnf_package(self, vnf_package):
return self.vnf_package_client.download_vnf_package(vnf_package)
def list_vnf_lcm_op_occs(self, retrieve_all=True, **_params):
return self.vnf_lcm_client.list_vnf_lcm_op_occs(
retrieve_all=retrieve_all, **_params)
def show_vnf_lcm_op_occs(self, occ_id):
return self.vnf_lcm_client.show_vnf_lcm_op_occs(occ_id)
def show_vnf_lcm_versions(self, major_version):
return self.vnf_lcm_client.show_vnf_lcm_versions(major_version)

View File

@@ -2,7 +2,7 @@
# of appearance. Changing the order has an impact on the overall integration
# process, which may cause wedges in the gate later.
hacking>=4.0.0,<4.1.0 # Apache-2.0
hacking>=1.1.0,<1.2.0 # Apache-2.0
coverage!=4.4,>=4.0 # Apache-2.0
ddt>=1.0.1 # MIT
fixtures>=3.0.0 # Apache-2.0/BSD
@@ -10,3 +10,4 @@ python-subunit>=1.0.0 # Apache-2.0/BSD
requests-mock>=1.2.0 # Apache-2.0
stestr>=2.0.0 # Apache-2.0
testtools>=2.2.0 # MIT
mock>=2.0.0 # BSD

26
tox.ini
View File

@@ -1,40 +1,44 @@
[tox]
envlist = py38,py36,pep8,docs
minversion = 3.18.0
envlist = py37,py36,py27,pep8,docs
minversion = 2.0
skipsdist = True
ignore_basepython_conflict = True
[testenv]
basepython = python3
setenv = VIRTUAL_ENV={envdir}
LANG=en_US.UTF-8
LANGUAGE=en_US:en
LC_ALL=C
usedevelop = True
install_command = pip install {opts} {packages}
deps =
-c{env:TOX_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/xena}
-c{env:UPPER_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/train}
-r{toxinidir}/requirements.txt
-r{toxinidir}/test-requirements.txt
commands = stestr run --slowest {posargs}
[testenv:pep8]
basepython = python3
commands = flake8
distribute = false
[testenv:venv]
basepython = python3
commands = {posargs}
[testenv:docs]
basepython = python3
deps = -r{toxinidir}/doc/requirements.txt
commands = sphinx-build -W -b html doc/source doc/build/html
[testenv:releasenotes]
basepython = python3
deps =
-c{env:TOX_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/xena}
-c{env:UPPER_CONSTRAINTS_FILE:https://releases.openstack.org/constraints/upper/train}
-r{toxinidir}/doc/requirements.txt
commands = sphinx-build -a -E -W -d releasenotes/build/doctrees -b html releasenotes/source releasenotes/build/html
[testenv:cover]
basepython = python3
setenv =
PYTHON=coverage run --source tackerclient --parallel-mode
commands =
@@ -45,10 +49,16 @@ commands =
[flake8]
# E125 continuation line does not distinguish itself from next logical line
# W504 line break after binary operator
ignore = E125,W504
ignore = E125
show-source = true
exclude=.venv,.git,.tox,dist,doc,*lib/python*,*egg,tools
# F821 undefined name 'unicode'
# if isinstance(config, str) or isinstance(config, unicode):
builtins = unicode
[testenv:lower-constraints]
basepython = python3
deps =
-c{toxinidir}/lower-constraints.txt
-r{toxinidir}/test-requirements.txt
-r{toxinidir}/requirements.txt