openstack/python-tackerclient
Code Issues Proposed changes

Add command to read vnfd of vnf package API

Browse Source 
Added command support for fetching VNFD of the vnf package.

Please see results here:-
http://paste.openstack.org/show/789212/

Change-Id: Ifa15ddf7ff306ed2705cf7a170d9e9f512491fba
Implements: bp enhance-vnf-package-support-part1
This commit is contained in:
Shubham Potale 2019-09-26 19:47:46 +05:30 committed by tpatil
parent af4d8343ad
commit ac02653f21
6 changed files with 269 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
setup.cfg
View File

@ -83,6 +83,7 @@ openstack.tackerclient.v1 =
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
vnflcm_create = tackerclient.osc.v1.vnflcm.vnflcm:CreateVnfLcm
vnflcm_show = tackerclient.osc.v1.vnflcm.vnflcm:ShowVnfLcm
vnflcm_list = tackerclient.osc.v1.vnflcm.vnflcm:ListVnfLcm

7
tackerclient/client.py
View File

@ -144,6 +144,8 @@ 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
@ -163,6 +165,7 @@ 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
@ -290,6 +293,10 @@ 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):

17
tackerclient/osc/sdk_utils.py
View File

@ -106,6 +106,23 @@ class DictModel(dict):
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))

61
tackerclient/osc/v1/vnfpkgm/vnf_package.py
View File

@ -14,6 +14,7 @@
# under the License.
import logging
import sys
from osc_lib.cli import parseractions
from osc_lib.command import command
@ -258,6 +259,66 @@ class DeleteVnfPackage(command.Command):
return
class DownloadVnfPackage(command.Command):
_description = _("Read 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 vnfd data. "
"If this is not specified and there is no redirection "
"then vnfd 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 parsed_args.file:
sdk_utils.save_data(body, parsed_args.file)
else:
print(body)
else:
msg = ("Currently only download vnfd from on-boarded vnf package "
"is supported. use --vnfd")
sdk_utils.exit(msg)
class UpdateVnfPackage(command.ShowOne):
_description = _("Update information about an individual VNF package")

139
tackerclient/tests/unit/osc/v1/test_vnf_package.py
View File

@ -13,12 +13,15 @@
# License for the specific language governing permissions and limitations
# under the License.
import filecmp
import os
import shutil
import sys
import tempfile
import ddt
import mock
import zipfile
from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
@ -421,3 +424,139 @@ class TestUpdateVnfPackage(TestVnfPackage):
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)

50
tackerclient/v1_0/client.py
View File

@ -205,8 +205,11 @@ class ClientBase(object):
action, method, body=body,
content_type=self.content_type())
if ('application/json' in resp.headers.get('Content-Type',
'application/json')):
if 'application/zip' == resp.headers.get('Content-Type'):
self.format = 'zip'
elif 'text/plain' == resp.headers.get('Content-Type'):
self.format = 'text'
else:
self.format = 'json'
status_code = resp.status_code
@ -231,7 +234,7 @@ class ClientBase(object):
"""
if data is None:
return None
elif self.format == 'zip':
elif self.format in ('zip', 'text'):
return data
elif type(data) is dict:
return serializer.Serializer(
@ -242,7 +245,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):
if status_code in (204, 202) or self.format in ('zip', 'text'):
return data
return serializer.Serializer(self.get_attr_metadata()).deserialize(
data, self.content_type())['body']
@ -261,12 +264,17 @@ class ClientBase(object):
constants.EXT_NS: ns}
def content_type(self, _format=None):
"""Returns the mime-type for either 'xml' or 'json'.
"""Returns the mime-type for either 'xml', 'json, 'text', or 'zip'.
Defaults to the currently set format.
"""
_format = _format or self.format
return "application/%s" % (_format)
if self.format == 'text':
return "text/plain"
elif self.format == 'both':
return "text/plain,application/zip"
else:
return "application/%s" % (_format)
def retry_request(self, method, action, body=None,
headers=None, params=None):
@ -745,6 +753,7 @@ 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'
def build_action(self, action):
return action
@ -785,6 +794,31 @@ class VnfPackageClient(ClientBase):
base_path=self.vnfpackages_path),
body=file_data)
@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 update_vnf_package(self, vnf_package, body):
return self.patch(self.vnfpackage_path % vnf_package, body=body)
@ -1110,3 +1144,7 @@ class Client(object):
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)