Adds functional tests for common services plugin

This commit adds functional tests for testing out the logging of
events for VNFD, VNF and VIM operations.

Change-Id: If517e9fc58eb7decbfb34c644e32f2750d344d3a
Implements: blueprint: audit-support
This commit is contained in:
vish 2016-08-22 14:28:58 +00:00 committed by vishwanath jayaraman
parent f9c0e7a1a6
commit fd8e37de5c
4 changed files with 275 additions and 0 deletions

View File

@ -0,0 +1,119 @@
# Copyright 2016 Brocade Communications System, Inc.
#
# 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 yaml
from tacker.plugins.common import constants as evt_constants
from tacker.tests.functional import base
from tacker.tests.utils import read_file
SECRET_PASSWORD = '***'
class VimTestCreate(base.BaseTackerTest):
def _test_create_delete_vim(self, vim_file, name, description, vim_type,
version=None):
data = yaml.load(read_file(vim_file))
password = data['password']
username = data['username']
project_name = data['project_name']
auth_url = data['auth_url']
domain_name = data.get('domain_name', None)
vim_arg = {'vim': {'name': name, 'description': description,
'type': vim_type,
'auth_url': auth_url,
'auth_cred': {'username': username,
'password': password,
'user_domain_name': domain_name},
'vim_project': {'name': project_name,
'project_domain_name':
domain_name},
'is_default': False}}
# Register vim
vim_res = self.client.create_vim(vim_arg)
vim_obj = vim_res['vim']
vim_id = vim_obj['id']
self.verify_vim(vim_obj, data, name, description, version)
params = {'resource_id': vim_id,
'event_type': evt_constants.RES_EVT_CREATE,
'timestamp': vim_res['vim'][
evt_constants.RES_EVT_CREATED_FLD]}
vim_evt_list = self.client.list_vim_events(params)
self.assertIsNotNone(vim_evt_list,
"List of VIM events are Empty after Creation")
self.assertEqual(1, len(vim_evt_list))
# Read vim
vim_show_res = self.client.show_vim(vim_id)
self.verify_vim(vim_show_res['vim'], data, name, description, version)
# Delete vim
try:
self.client.delete_vim(vim_id)
except Exception:
self.assertFalse(True, "Failed to delete vim %s" % vim_id)
params = {'resource_id': vim_id,
'event_type': evt_constants.RES_EVT_DELETE}
vim_evt_list = self.client.list_vim_events(params)
self.assertIsNotNone(vim_evt_list,
"List of vnf events are Empty after Deletion")
self.assertEqual(1, len(vim_evt_list))
def verify_vim(self, vim_instance, config_data, name, description,
version):
expected_regions = ['RegionOne']
self.assertIsNotNone(vim_instance)
self.assertEqual(description, vim_instance['description'])
self.assertEqual(name, vim_instance['name'])
self.assertIsNotNone(vim_instance['tenant_id'])
self.assertIsNotNone(vim_instance['id'])
self.assertEqual(config_data['username'],
vim_instance['auth_cred']['username'])
self.assertEqual(SECRET_PASSWORD,
vim_instance['auth_cred']['password'])
self.assertEqual(expected_regions,
vim_instance['placement_attr']['regions'])
if version:
method_name = 'verify_vim_' + version
getattr(self, method_name)(vim_instance, config_data)
def verify_vim_v2(self, vim_instance, config_data):
self.assertEqual(config_data['project_name'],
vim_instance['auth_cred']['tenant_name'])
def verify_vim_v3(self, vim_instance, config_data):
self.assertEqual(config_data['project_name'],
vim_instance['auth_cred']['project_name'])
def test_create_delete_local_vim(self):
name = 'Default vim'
description = 'Local vim description'
vim_type = 'openstack'
ks_version = 'v3'
self._test_create_delete_vim('local-vim.yaml', name, description,
vim_type, ks_version)
def test_create_delete_local_vim_keystone_v2(self):
name = 'Openstack'
description = 'OpenStack VIM with keystone v2'
vim_type = 'openstack'
ks_version = 'v2'
self._test_create_delete_vim('vim-config-ks-v2.yaml', name,
description, vim_type, ks_version)

View File

@ -0,0 +1,96 @@
# Copyright 2016 Brocade Communications System, Inc.
#
# 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_config import cfg
from tacker.plugins.common import constants as evt_constants
from tacker.tests import constants
from tacker.tests.functional import base
from tacker.tests.utils import read_file
CONF = cfg.CONF
VNF_CIRROS_CREATE_TIMEOUT = 120
class VnfTestEvent(base.BaseTackerTest):
def _test_create_delete_vnf_events(self, vnf_name, vim_id=None):
data = dict()
data['tosca'] = read_file(
'sample_cirros_vnf_no_monitoring.yaml')
vnfd_name = 'sample_cirros_vnf_no_monitoring'
toscal = data['tosca']
tosca_arg = {'vnfd': {'name': vnfd_name,
'attributes': {'vnfd': toscal}}}
# Create vnfd with tosca template
vnfd_instance = self.client.create_vnfd(body=tosca_arg)
self.assertIsNotNone(vnfd_instance)
# Create vnf with vnfd_id
vnfd_id = vnfd_instance['vnfd']['id']
vnf_arg = {'vnf': {'vnfd_id': vnfd_id, 'name': vnf_name}}
if vim_id:
vnf_arg['vnf']['vim_id'] = vim_id
vnf_instance = self.client.create_vnf(body=vnf_arg)
self.validate_vnf_instance(vnfd_instance, vnf_instance)
vnf_id = vnf_instance['vnf']['id']
vnf_current_status = self.wait_until_vnf_active(
vnf_id,
constants.VNF_CIRROS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
self.assertEqual('ACTIVE', vnf_current_status)
self.assertIsNotNone(self.client.show_vnf(vnf_id)['vnf']['mgmt_url'])
if vim_id:
self.assertEqual(vim_id, vnf_instance['vnf']['vim_id'])
vnf_id = vnf_instance['vnf']['id']
params = {'resource_id': vnf_id,
'event_type': evt_constants.RES_EVT_CREATE,
'timestamp': vnf_instance['vnf'][
evt_constants.RES_EVT_CREATED_FLD]}
vnf_evt_list = self.client.list_vnf_events(params)
self.assertIsNotNone(vnf_evt_list,
"List of vnfd events are Empty after Creation")
# Delete vnf_instance with vnf_id
try:
self.client.delete_vnf(vnf_id)
except Exception:
assert False, "vnf Delete failed"
params = {'resource_id': vnf_id,
'event_type': evt_constants.RES_EVT_DELETE}
vnf_evt_list = self.client.list_vnf_events(params)
self.assertIsNotNone(vnf_evt_list,
"List of vnf events are Empty after Deletion")
self.assertEqual(1, len(vnf_evt_list))
# Delete vnfd_instance
self.addCleanup(self.client.delete_vnfd, vnfd_id)
self.addCleanup(self.wait_until_vnf_delete, vnf_id,
constants.VNF_CIRROS_DELETE_TIMEOUT)
def test_create_delete_vnf_with_default_vim(self):
self._test_create_delete_vnf_events(
vnf_name='test_vnf_with_cirros_no_monitoring')
def test_create_delete_vnf_with_vim_id(self):
vim_list = self.client.list_vims()
vim0_id = self.get_vim(vim_list, 'VIM0')['id']
self._test_create_delete_vnf_events(
vim_id=vim0_id,
vnf_name='test_vnf_with_cirros_with_default_vim_id')

View File

@ -0,0 +1,60 @@
# Copyright 2016 Brocade Communications System, Inc.
#
# 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_config import cfg
from tacker.plugins.common import constants
from tacker.tests.functional import base
from tacker.tests.utils import read_file
CONF = cfg.CONF
class VnfdTestEvent(base.BaseTackerTest):
def _test_create_delete_vnfd_events(self, vnfd_file):
data = dict()
data['tosca'] = read_file(vnfd_file)
toscal = data['tosca']
vnfd_name = 'sample_cirros_vnf'
tosca_arg = {'vnfd': {'name': vnfd_name,
'attributes': {'vnfd': toscal}}}
vnfd_instance = self.client.create_vnfd(body=tosca_arg)
self.assertIsNotNone(vnfd_instance)
vnfd_id = vnfd_instance['vnfd']['id']
params = {'resource_id': vnfd_id,
'event_type': constants.RES_EVT_CREATE,
'timestamp': vnfd_instance['vnfd'][
constants.RES_EVT_CREATED_FLD]}
vnfd_evt_list = self.client.list_vnfd_events(params)
self.assertIsNotNone(vnfd_evt_list,
"List of vnfds events are Empty after Creation")
self.assertEqual(1, len(vnfd_evt_list))
try:
self.client.delete_vnfd(vnfd_id)
except Exception:
assert False, "vnfd Delete failed"
params = {'resource_id': vnfd_id,
'event_type': constants.RES_EVT_DELETE}
vnfd_evt_list = self.client.list_vnfd_events(params)
self.assertIsNotNone(vnfd_evt_list,
"List of vnfds events are Empty after Deletion")
self.assertEqual(1, len(vnfd_evt_list))
def test_vnfd_events(self):
self._test_create_delete_vnfd_events('sample_cirros_vnf.yaml')