Add commands for list, show and delete vnf package API's

Added support for below three commands and their unit test cases.
1.vnf package list
2.vnf package show
3.vnf package delete

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

Change-Id: I47e3e99a42aa36ab286fa01802999deb964a168f
Implements: bp tosca-csar-mgmt-driver
This commit is contained in:
Shubham Potale 2019-04-25 11:29:25 +05:30 committed by dharmendra kushwaha
parent 9940682bc8
commit 11e23eed22
5 changed files with 392 additions and 22 deletions

View File

@ -81,6 +81,9 @@ openstack.tackerclient.v1 =
vnf_chain_list = tackerclient.osc.v1.nfvo.vnffg:ListSFC
vnf_chain_show = tackerclient.osc.v1.nfvo.vnffg:ShowSFC
vnf_package_create = tackerclient.osc.v1.vnfpkgm.vnf_package:CreateVnfPackage
vnf_package_list = tackerclient.osc.v1.vnfpkgm.vnf_package:ListVnfPackage
vnf_package_show = tackerclient.osc.v1.vnfpkgm.vnf_package:ShowVnfPackage
vnf_package_delete = tackerclient.osc.v1.vnfpkgm.vnf_package:DeleteVnfPackage
[build_releasenotes]

View File

@ -19,16 +19,31 @@ from osc_lib.cli import parseractions
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
LOG = logging.getLogger(__name__)
_mixed_case_fields = ('onboardingState', 'operationalState',
'usageState', 'userDefinedData')
_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)
)
def _get_columns(item):
_mixed_case_fields = ('usageState', 'onboardingState', 'operationalState',
'vnfProductName', 'softwareImages', 'userDefinedData',
'vnfdId', 'vnfdVersion', 'vnfSoftwareVersion',
'vnfProvider', 'artifactPath', 'imagePath',
'diskFormat', 'userMetadata')
def _get_columns(vnf_package_obj):
column_map = {
'_links': 'Links',
'onboardingState': 'Onboarding State',
@ -37,7 +52,19 @@ def _get_columns(item):
'userDefinedData': 'User Defined Data',
'id': 'ID'
}
return sdk_utils.get_osc_show_columns_for_sdk_resource(item, column_map)
if vnf_package_obj['onboardingState'] == 'ONBOARDED':
column_map.update({
'softwareImages': 'Software Images',
'vnfProvider': 'VNF Provider',
'vnfSoftwareVersion': 'VNF Software Version',
'vnfProductName': 'VNF Product Name',
'vnfdId': 'VNFD ID',
'vnfdVersion': 'VNFD Version'
})
return sdk_utils.get_osc_show_columns_for_sdk_resource(vnf_package_obj,
column_map)
class CreateVnfPackage(command.ShowOne):
@ -69,3 +96,103 @@ class CreateVnfPackage(command.ShowOne):
sdk_utils.DictModel(vnf_package),
columns, mixed_case_fields=_mixed_case_fields)
return (display_columns, data)
class ListVnfPackage(command.Lister):
_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)
return parser
def take_action(self, parsed_args):
_params = {}
client = self.app.client_manager.tackerclient
data = client.list_vnf_packages(**_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 data['vnf_packages']))
class ShowVnfPackage(command.ShowOne):
_description = _("Show VNF Package Details")
def get_parser(self, prog_name):
LOG.debug('get_parser(%s)', prog_name)
parser = super(ShowVnfPackage, self).get_parser(prog_name)
parser.add_argument(
'vnf_package',
metavar="<vnf-package>",
help=_("VNF package ID")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
vnf_package = client.show_vnf_package(parsed_args.vnf_package)
display_columns, columns = _get_columns(vnf_package)
data = utils.get_item_properties(
sdk_utils.DictModel(vnf_package),
columns, mixed_case_fields=_mixed_case_fields)
return (display_columns, data)
class DeleteVnfPackage(command.Command):
"""Vnf package delete
Delete class supports bulk deletion of vnf packages, and error
handling.
"""
_description = _("Delete VNF Package")
resource = 'vnf-package'
def get_parser(self, prog_name):
LOG.debug('get_parser(%s)', prog_name)
parser = super(DeleteVnfPackage, self).get_parser(prog_name)
parser.add_argument(
'vnf-package',
metavar="<vnf-package>",
nargs="+",
help=_("Vnf package(s) ID to delete")
)
return parser
def take_action(self, parsed_args):
client = self.app.client_manager.tackerclient
failure = False
deleted_ids = []
failed_items = {}
resources = getattr(parsed_args, self.resource, [])
for resource_id in resources:
try:
vnf_package = client.show_vnf_package(resource_id)
client.delete_vnf_package(vnf_package['id'])
deleted_ids.append(resource_id)
except Exception as e:
failure = True
failed_items[resource_id] = e
if failure:
msg = ''
if deleted_ids:
msg = (_('Successfully deleted %(resource)s(s):'
' %(deleted_list)s') % {'deleted_list':
', '.join(deleted_ids),
'resource': self.resource})
err_msg = _("\n\nUnable to delete the below"
" 'vnf_package'(s):")
for failed_id, error in failed_items.items():
err_msg += (_('\n Cannot delete %(failed_id)s: %(error)s')
% {'failed_id': failed_id,
'error': error})
msg += err_msg
raise exceptions.CommandError(msg)
else:
print((_('All specified %(resource)s(s) deleted successfully')
% {'resource': self.resource}))
return

View File

@ -16,18 +16,38 @@
import ddt
import mock
from tackerclient.common import exceptions
from tackerclient.osc import utils as tacker_osc_utils
from tackerclient.osc.v1.vnfpkgm import vnf_package
from tackerclient.tests.unit.osc import base
from tackerclient.tests.unit.osc.v1.fixture_data import client
from tackerclient.tests.unit.osc.v1 import vnf_package_fakes
def _get_columns_vnf_package(action='list', vnf_package_obj=None):
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(['Links', 'VNFD ID', 'VNF Provider',
'VNF Software Version', 'VNFD Version',
'Software Images'])
else:
columns.extend(['Links'])
columns.remove('VNF Product Name')
return columns
class TestVnfPackage(base.FixturedTestCase):
client_fixture_class = client.ClientFixture
def setUp(self):
super(TestVnfPackage, 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
@ -37,9 +57,6 @@ class TestVnfPackage(base.FixturedTestCase):
@ddt.ddt
class TestCreateVnfPackage(TestVnfPackage):
columns = ('ID', 'Links', 'Onboarding State', 'Operational State',
'Usage State', 'User Defined Data')
def setUp(self):
super(TestCreateVnfPackage, self).setUp()
self.create_vnf_package = vnf_package.CreateVnfPackage(
@ -53,7 +70,6 @@ class TestCreateVnfPackage(TestVnfPackage):
# command param
parsed_args = self.check_parser(self.create_vnf_package, arglist,
verifylist)
header = {'content-type': 'application/json'}
if arglist:
json = vnf_package_fakes.vnf_package_obj(
@ -62,9 +78,138 @@ class TestCreateVnfPackage(TestVnfPackage):
json = vnf_package_fakes.vnf_package_obj()
self.requests_mock.register_uri(
'POST', self.url + '/vnfpkgm/v1/vnf_packages',
json=json, headers=header)
json=json, headers=self.header)
columns, data = (self.create_vnf_package.take_action(parsed_args))
self.assertEqual(self.columns, columns)
self.assertItemsEqual(_get_columns_vnf_package(action='create'),
columns)
self.assertItemsEqual(vnf_package_fakes.get_vnf_package_data(json),
data)
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')
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(
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.assertItemsEqual(_get_columns_vnf_package(), actual_columns)
self.assertItemsEqual(expected_data, list(data))
@ddt.ddt
class TestShowVnfPackage(TestVnfPackage):
def setUp(self):
super(TestShowVnfPackage, self).setUp()
self.show_vnf_package = vnf_package.ShowVnfPackage(
self.app, self.app_args, cmd_name='vnf package show')
@ddt.data(True, False)
def test_take_action(self, onboarded):
vnf_package_obj = vnf_package_fakes.vnf_package_obj(
onboarded_state=onboarded)
arglist = [vnf_package_obj['id']]
verifylist = [('vnf_package', vnf_package_obj['id'])]
parsed_args = self.check_parser(self.show_vnf_package, arglist,
verifylist)
url = self.url + '/vnfpkgm/v1/vnf_packages/' + vnf_package_obj['id']
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.assertItemsEqual(_get_columns_vnf_package(
vnf_package_obj=vnf_package_obj, action='show'), columns)
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,
self.show_vnf_package, [], [])
class TestDeleteVnfPackage(TestVnfPackage):
def setUp(self):
super(TestDeleteVnfPackage, self).setUp()
self.delete_vnf_package = vnf_package.DeleteVnfPackage(
self.app, self.app_args, cmd_name='vnf package delete')
# The Vnf Package to delete
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'])
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={})
def test_delete_one_vnf_package(self):
arglist = [self._vnf_package['vnf_packages'][0]['id']]
verifylist = [('vnf-package', [self._vnf_package['vnf_packages']
[0]['id']])]
parsed_args = self.check_parser(self.delete_vnf_package, arglist,
verifylist)
self._mock_request_url_for_delete(0)
result = self.delete_vnf_package.take_action(parsed_args)
self.assertIsNone(result)
def test_delete_multiple_vnf_package(self):
arglist = []
for vnf_pkg in self._vnf_package['vnf_packages']:
arglist.append(vnf_pkg['id'])
verifylist = [('vnf-package', arglist)]
parsed_args = self.check_parser(self.delete_vnf_package, arglist,
verifylist)
for i in range(0, 3):
self._mock_request_url_for_delete(i)
result = self.delete_vnf_package.take_action(parsed_args)
self.assertIsNone(result)
def test_delete_multiple_vnf_package_exception(self):
arglist = [
self._vnf_package['vnf_packages'][0]['id'],
'xxxx-yyyy-zzzz',
self._vnf_package['vnf_packages'][1]['id'],
]
verifylist = [
('vnf-package', arglist),
]
parsed_args = self.check_parser(self.delete_vnf_package,
arglist, verifylist)
self._mock_request_url_for_delete(0)
url = (self.url + '/vnfpkgm/v1/vnf_packages/' + 'xxxx-yyyy-zzzz')
body = {"error": exceptions.NotFound('404')}
self.requests_mock.register_uri('GET', url, body=body,
status_code=404, headers=self.header)
self._mock_request_url_for_delete(1)
self.assertRaises(exceptions.CommandError,
self.delete_vnf_package.take_action,
parsed_args)

View File

@ -13,8 +13,10 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_utils import uuidutils
def vnf_package_obj(attrs=None):
def vnf_package_obj(attrs=None, onboarded_state=False):
"""Create a fake vnf package.
:param Dictionary attrs:
@ -27,7 +29,6 @@ def vnf_package_obj(attrs=None):
# Set default attributes.
fake_vnf_package = {"id": "60a6ac16-b50d-4e92-964b-b3cf98c7cf5c",
"_links": {"self": {"href": "string"},
"vnfd": {"href": "string"},
"packageContent": {"href": "string"}
},
"onboardingState": "CREATED",
@ -35,24 +36,93 @@ def vnf_package_obj(attrs=None):
"usageState": "NOT_IN_USE",
"userDefinedData": None}
if onboarded_state:
fake_vnf_package = {"id": "60a6ac16-b50d-4e92-964b-b3cf98c7cf5c",
"vnfdId": "string",
"vnfProvider": "string",
"vnfProductName": "string",
"vnfSoftwareVersion": "string",
"vnfdVersion": "string",
"softwareImages": [
{
"id": "string",
"name": "string",
"provider": "string",
"version": "string",
"checksum": {
"algorithm": "string",
"hash": "string"
},
"containerFormat": "AKI",
"diskFormat": "AKI",
"createdAt": "2015-06-03T18:49:19.000000",
"minDisk": '0',
"minRam": '0',
"size": '0',
"userMetadata": {},
"imagePath": "string"
}
],
"onboardingState": "ONBOARDED",
"operationalState": "ENABLED",
"usageState": "IN_USE",
"userDefinedData": None,
"_links": {
"self": {
"href": "string"
},
"vnfd": {
"href": "string"
},
"packageContent": {
"href": "string"
}
}}
# Overwrite default attributes.
fake_vnf_package.update(attrs)
return fake_vnf_package
def get_vnf_package_data(vnf_package=None):
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:
A FakeVnfPackage dict object
:return:
A tuple which may include the following values:
(u"packageContent='{'href': 'string'}', self='{'href': 'string'}',
vnfd='{'href': 'string'}'", '60a6ac16-b50d-4e92-964b-b3cf98c7cf5c',
'CREATED', 'DISABLED', 'NOT_IN_USE', u"Test_key='Test_value'")
A list which may include the following values:
[{'packageContent': {'href': 'string'}, 'self': {'href': 'string'},
'vnfd': {'href': 'string'}}, '60a6ac16-b50d-4e92-964b-b3cf98c7cf5c',
'CREATED', 'DISABLED', 'NOT_IN_USE', {'Test_key': 'Test_value'}]
"""
data_list = []
if vnf_package is not None:
for x in sorted(vnf_package.keys()):
data_list.append(vnf_package[x])
return tuple(data_list)
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.get('vnfProductName'):
vnf_package['vnfProductName'] = ''
# return the list of data as per column order
if columns:
return tuple([vnf_package[key] for key in columns])
return tuple([vnf_package[key] for key in sorted(vnf_package.keys())])
def create_vnf_packages(count=2):
"""Create multiple fake vnf packages.
:param Dictionary attrs:
A dictionary with all attributes
:param int count:
The number of vnf_packages to fake
:return:
A list of fake vnf packages dictionary
"""
vnf_packages = []
for i in range(0, count):
unique_id = uuidutils.generate_uuid()
vnf_packages.append(vnf_package_obj(attrs={'id': unique_id}))
return {'vnf_packages': vnf_packages}

View File

@ -726,6 +726,7 @@ class VnfPackageClient(ClientBase):
"""
vnfpackages_path = '/vnfpkgm/v1/vnf_packages'
vnfpackage_path = '/vnfpkgm/v1/vnf_packages/%s'
def build_action(self, action):
return action
@ -734,6 +735,20 @@ class VnfPackageClient(ClientBase):
def create_vnf_package(self, body):
return self.post(self.vnfpackages_path, body=body)
@APIParamsCall
def list_vnf_packages(self, retrieve_all=True, **_params):
vnf_package = self.list("vnf_packages", self.vnfpackages_path,
retrieve_all, **_params)
return vnf_package
@APIParamsCall
def show_vnf_package(self, vnf_package, **_params):
return self.get(self.vnfpackage_path % vnf_package, params=_params)
@APIParamsCall
def delete_vnf_package(self, vnf_package):
return self.delete(self.vnfpackage_path % vnf_package)
class Client(object):
"""Unified interface to interact with multiple applications of tacker service.
@ -945,3 +960,13 @@ 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, **_params):
return self.vnf_package_client.list_vnf_packages(
retrieve_all=retrieve_all, **_params)
def show_vnf_package(self, vnf_package, **_params):
return self.vnf_package_client.show_vnf_package(vnf_package, **_params)
def delete_vnf_package(self, vnf_package):
return self.vnf_package_client.delete_vnf_package(vnf_package)