python-tackerclient/tackerclient/tests/unit/vm/test_cli10_vnf.py

216 lines
8.1 KiB
Python

# 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 mox
import sys
from tackerclient import shell
from tackerclient.tacker import v1_0 as tackerV1_0
from tackerclient.tacker.v1_0.vnfm import vnf
from tackerclient.tests.unit import test_cli10
API_VERSION = "1.0"
FORMAT = 'json'
TOKEN = 'testtoken'
ENDURL = 'localurl'
class CLITestV10VmVNFJSON(test_cli10.CLITestV10Base):
_RESOURCE = 'vnf'
_RESOURCES = 'vnfs'
_VNF_RESOURCES = 'vnf_resources'
def setUp(self):
plurals = {'vnfs': 'vnf',
'resources': 'resource'}
super(CLITestV10VmVNFJSON, self).setUp(plurals=plurals)
def _test_create_resource(self, resource, cmd,
name, myid, args,
position_names, position_values, tenant_id=None,
tags=None, admin_state_up=True, extra_body=None,
**kwargs):
self.mox.StubOutWithMock(cmd, "get_client")
self.mox.StubOutWithMock(self.client.httpclient, "request")
cmd.get_client().MultipleTimes().AndReturn(self.client)
non_admin_status_resources = ['vnfd', 'vnf']
if (resource in non_admin_status_resources):
body = {resource: {}, }
else:
body = {resource: {'admin_state_up': admin_state_up, }, }
if tenant_id:
body[resource].update({'tenant_id': tenant_id})
if tags:
body[resource].update({'tags': tags})
if extra_body:
body[resource].update(extra_body)
body[resource].update(kwargs)
for i in range(len(position_names)):
body[resource].update({position_names[i]: position_values[i]})
ress = {resource:
{self.id_field: myid}, }
if name:
ress[resource].update({'name': name})
self.client.format = self.format
resstr = self.client.serialize(ress)
# url method body
resource_plural = tackerV1_0._get_resource_plural(resource,
self.client)
path = getattr(self.client, resource_plural + "_path")
# Work around for LP #1217791. XML deserializer called from
# MyComparator does not decodes XML string correctly.
if self.format == 'json':
mox_body = test_cli10.MyComparator(body, self.client)
else:
mox_body = self.client.serialize(body)
self.client.httpclient.request(
test_cli10.end_url(path, format=self.format), 'POST',
body=mox_body,
headers=mox.ContainsKeyValue(
'X-Auth-Token', TOKEN)).AndReturn((
test_cli10.MyResp(200), resstr))
args.extend(['--request-format', self.format])
args.extend(['--vnfd-id', 'vnfd'])
self.mox.ReplayAll()
cmd_parser = cmd.get_parser('create_' + resource)
shell.run_command(cmd, cmd_parser, args)
self.mox.VerifyAll()
def test_create_vnf_all_params(self):
cmd = vnf.CreateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
vnfd_id = 'vnfd'
vim_id = 'vim_id'
description = 'my-description'
region_name = 'region'
key = 'key'
value = 'value'
args = [
name,
'--vnfd-id', vnfd_id,
'--vim-id', vim_id,
'--description', description,
'--vim-region-name', region_name,
'--%s' % key, value]
position_names = [
'name',
'vnfd_id',
'vim_id',
'description',
'attributes'
]
position_values = [
name,
vnfd_id,
vim_id,
description,
{}
]
extra_body = {key: value, 'placement_attr': {'region_name':
region_name}}
self._test_create_resource(self._RESOURCE, cmd, name, my_id,
args, position_names, position_values,
extra_body=extra_body)
def test_create_vnf_with_mandatory_params(self):
cmd = vnf.CreateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
vnfd_id = 'vnfd'
args = [
name,
'--vnfd-id', vnfd_id,
]
position_names = ['name', 'vnfd_id', 'attributes']
position_values = [name, vnfd_id, {}]
self._test_create_resource(self._RESOURCE, cmd, name, my_id,
args, position_names, position_values)
def test_create_vnf_with_description_param(self):
cmd = vnf.CreateVNF(test_cli10.MyApp(sys.stdout), None)
name = 'my_name'
my_id = 'my-id'
vnfd_id = 'vnfd'
description = 'my-description'
args = [
name,
'--vnfd-id', vnfd_id,
'--description', description,
]
position_names = ['name', 'vnfd_id', 'description',
'attributes']
position_values = [name, vnfd_id, description, {}]
self._test_create_resource(self._RESOURCE, cmd, None, my_id,
args, position_names, position_values)
def test_list_vnfs(self):
cmd = vnf.ListVNF(test_cli10.MyApp(sys.stdout), None)
self._test_list_resources(self._RESOURCES, cmd, True)
def test_list_vnfs_pagenation(self):
cmd = vnf.ListVNF(test_cli10.MyApp(sys.stdout), None)
self._test_list_resources(self._RESOURCES, cmd, True)
def test_show_vnf_id(self):
cmd = vnf.ShowVNF(test_cli10.MyApp(sys.stdout), None)
args = ['--fields', 'id', self.test_id]
self._test_show_resource(self._RESOURCE, cmd, self.test_id, args,
['id'])
def test_show_vnf_id_name(self):
cmd = vnf.ShowVNF(test_cli10.MyApp(sys.stdout), None)
args = ['--fields', 'id', '--fields', 'name', self.test_id]
self._test_show_resource(self._RESOURCE, cmd, self.test_id,
args, ['id', 'name'])
def test_update_vnf(self):
cmd = vnf.UpdateVNF(test_cli10.MyApp(sys.stdout), None)
my_id = 'my-id'
key = 'new_key'
value = 'new-value'
self._test_update_resource(self._RESOURCE, cmd, my_id,
[my_id, '--%s' % key, value],
{key: value})
def test_delete_vnf(self):
cmd = vnf.DeleteVNF(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_list_vnf_resources(self):
cmd = vnf.ListVNFResources(test_cli10.MyApp(sys.stdout), None)
base_args = [self.test_id]
response = [{'name': 'CP11', 'id': 'id1', 'type': 'NeutronPort'},
{'name': 'CP12', 'id': 'id2', 'type': 'NeutronPort'}]
val = self._test_list_sub_resources(self._VNF_RESOURCES, 'resources',
cmd, self.test_id,
response_contents=response,
detail=True, base_args=base_args)
self.assertIn('id1', val)
self.assertIn('NeutronPort', val)
self.assertIn('CP11', val)
def test_multi_delete_vnf(self):
cmd = vnf.DeleteVNF(test_cli10.MyApp(sys.stdout), None)
vnf_ids = 'vnf1 vnf2 vnf3'
args = [vnf_ids]
self._test_delete_resource(self._RESOURCE, cmd, vnf_ids, args)