openstack
/
tacker
Code Issues Proposed changes

Add functional tests for ConnectionPoint features 

Functional tests exercising static_ip address and anti_spoof_protection
(port_security_enabled) properties of ConnectionPoint.

Closes-Bug: #1585788

Change-Id: I3f90aaa5be08cd1cc67d0663b559a85c4603f12c
This commit is contained in:
Santosh Kodicherla 2016-07-25 16:35:02 +00:00 committed by Sridhar Ramaswamy
parent bd127c9e87
commit 56d595d7bc
6 changed files with 259 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
tacker/tests/etc/samples/local-vim.yaml
View File

@ -2,4 +2,4 @@ auth_url: http://127.0.0.1:5000
username: nfv_user
password: devstack
project_name: nfv
domain_name: Default
domain_name: Default

72
tacker/tests/etc/samples/sample-tosca-vnfd-static-ip.yaml Normal file
View File

@ -0,0 +1,72 @@
tosca_definitions_version: tosca_simple_profile_for_nfv_1_0_0
description: VNFD with predefined properties.
topology_template:
node_templates:
VDU1:
type: tosca.nodes.nfv.VDU.Tacker
capabilities:
nfv_compute:
properties:
num_cpus: 1
mem_size: 512 MB
disk_size: 1 GB
properties:
image: cirros-0.3.4-x86_64-uec
availability_zone: nova
mgmt_driver: noop
config: |
param0: key1
param1: key2
CP1:
type: tosca.nodes.nfv.CP.Tacker
properties:
management: true
ip_address: 192.168.120.225
anti_spoofing_protection: true
requirements:
- virtualLink:
node: VL1
- virtualBinding:
node: VDU1
CP2:
type: tosca.nodes.nfv.CP.Tacker
properties:
anti_spoofing_protection: false
requirements:
- virtualLink:
node: VL2
- virtualBinding:
node: VDU1
CP3:
type: tosca.nodes.nfv.CP.Tacker
properties:
anti_spoofing_protection: true
requirements:
- virtualLink:
node: VL3
- virtualBinding:
node: VDU1
VL1:
type: tosca.nodes.nfv.VL
properties:
vendor: ACME
network_name: net_mgmt
VL2:
type: tosca.nodes.nfv.VL
properties:
network_name: net0
vendor: Tacker
VL3:
type: tosca.nodes.nfv.VL
properties:
network_name: net1
vendor: Tacker

21
tacker/tests/functional/base.py
View File

@ -13,20 +13,22 @@
# under the License.
import time
import yaml
from neutronclient.v2_0 import client as neutron_client
from novaclient import client as nova_client
from oslo_config import cfg
from tempest.lib import base
import yaml
from tacker.plugins.common import constants as evt_constants
from tacker.tests import constants
from tacker.tests.functional import clients
from tacker.tests.utils import read_file
from tacker import version
from tackerclient.v1_0 import client as tacker_client
CONF = cfg.CONF
@ -44,6 +46,7 @@ class BaseTackerTest(base.BaseTestCase):
**kwargs)
cls.client = cls.tackerclient()
cls.h_client = cls.heatclient()
@classmethod
def get_credentials(cls):
@ -72,6 +75,15 @@ class BaseTackerTest(base.BaseTestCase):
vim_params = cls.get_credentials()
return neutron_client.Client(**vim_params)
@classmethod
def heatclient(cls):
data = yaml.load(read_file('local-vim.yaml'))
data['auth_url'] = data['auth_url'] + '/v3'
domain_name = data.pop('domain_name')
data['user_domain_name'] = domain_name
data['project_domain_name'] = domain_name
return clients.OpenstackClients(auth_attr=data).heat
def wait_until_vnf_status(self, vnf_id, target_status, timeout,
sleep_interval):
start_time = int(time.time())
@ -179,3 +191,10 @@ class BaseTackerTest(base.BaseTestCase):
if vim['name'] == vim_name:
return vim
return None
def verify_antispoofing_in_stack(self, stack_id, resource_name):
resource_types = self.h_client.resources
resource_details = resource_types.get(stack_id=stack_id,
resource_name=resource_name)
resource_dict = resource_details.to_dict()
self.assertTrue(resource_dict['attributes']['port_security_enabled'])

52
tacker/tests/functional/clients.py Normal file
View File

@ -0,0 +1,52 @@
# 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 heatclient import client as heatclient
from tacker.tests.functional import keystone
class OpenstackClients(object):
def __init__(self, auth_attr, region_name=None):
super(OpenstackClients, self).__init__()
self.keystone_plugin = keystone.Keystone()
self.heat_client = None
self.keystone_client = None
self.region_name = region_name
self.auth_attr = auth_attr
def _keystone_client(self):
version = self.auth_attr['auth_url'].rpartition('/')[2]
return self.keystone_plugin.initialize_client(version,
**self.auth_attr)
def _heat_client(self):
endpoint = self.keystone_session.get_endpoint(
service_type='orchestration', region_name=self.region_name)
return heatclient.Client('1', endpoint=endpoint,
session=self.keystone_session)
@property
def keystone_session(self):
return self.keystone.session
@property
def keystone(self):
if not self.keystone_client:
self.keystone_client = self._keystone_client()
return self.keystone_client
@property
def heat(self):
if not self.heat_client:
self.heat_client = self._heat_client()
return self.heat_client

55
tacker/tests/functional/keystone.py Normal file
View File

@ -0,0 +1,55 @@
# Copyright 2016 Brocade Communications System, Inc.
# 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 keystoneclient.auth import identity
from keystoneclient import client
from keystoneclient import exceptions
from keystoneclient import session
from oslo_config import cfg
from oslo_log import log as logging
LOG = logging.getLogger(__name__)
CONF = cfg.CONF
class Keystone(object):
"""Keystone module for OpenStack VIM
Handles identity operations for a given OpenStack
instance such as version, session and client
"""
def get_version(self, base_url=None):
try:
keystone_client = client.Client(auth_url=base_url)
except exceptions.ConnectionRefused:
raise
return keystone_client.version
def get_session(self, auth_plugin):
ses = session.Session(auth=auth_plugin)
return ses
def get_endpoint(self, ses, service_type, region_name=None):
return ses.get_endpoint(service_type, region_name)
def initialize_client(self, version, **kwargs):
from keystoneclient.v3 import client
auth_plugin = identity.v3.Password(**kwargs)
ses = self.get_session(auth_plugin=auth_plugin)
cli = client.Client(session=ses)
return cli

80
tacker/tests/functional/vnfm/test_tosca_vnf.py
View File

@ -12,9 +12,10 @@
# License for the specific language governing permissions and limitations
# under the License.
import yaml
from novaclient import exceptions
from oslo_config import cfg
import yaml
from tacker.plugins.common import constants as evt_constants
from tacker.tests import constants
@ -27,12 +28,13 @@ VNF_CIRROS_CREATE_TIMEOUT = 120
class VnfTestToscaCreate(base.BaseTackerTest):
def test_create_delete_vnf_tosca_no_monitoring(self):
input_yaml = read_file('sample-tosca-vnfd.yaml')
vnfd_name = 'test_tosca_vnf_with_cirros_no_monitoring'
tosca_dict = yaml.safe_load(input_yaml)
tosca_arg = {'vnfd': {'name': vnfd_name,
'attributes': {'vnfd': tosca_dict}}}
def _test_create_vnf(self, vnfd_file, vnf_name):
data = dict()
values_str = read_file(vnfd_file)
data['tosca'] = values_str
toscal = data['tosca']
tosca_arg = {'vnfd': {'name': vnf_name,
'attributes': {'vnfd': toscal}}}
# Create vnfd with tosca template
vnfd_instance = self.client.create_vnfd(body=tosca_arg)
@ -40,7 +42,6 @@ class VnfTestToscaCreate(base.BaseTackerTest):
# Create vnf with vnfd_id
vnfd_id = vnfd_instance['vnfd']['id']
vnf_name = 'test_tosca_vnf_with_cirros_no_monitoring'
vnf_arg = {'vnf': {'vnfd_id': vnfd_id, 'name': vnf_name}}
vnf_instance = self.client.create_vnf(body=vnf_arg)
@ -51,12 +52,53 @@ class VnfTestToscaCreate(base.BaseTackerTest):
vnf_id,
constants.VNF_CIRROS_CREATE_TIMEOUT,
constants.ACTIVE_SLEEP_TIME)
self.assertIsNotNone(self.client.show_vnf(vnf_id)['vnf']['mgmt_url'])
vnf_show_out = self.client.show_vnf(vnf_id)['vnf']
self.assertIsNotNone(vnf_show_out['mgmt_url'])
input_dict = yaml.load(values_str)
prop_dict = input_dict['topology_template']['node_templates'][
'CP1']['properties']
# Verify if ip_address is static, it is same as in show_vnf
if prop_dict.get('ip_address'):
mgmt_url_input = prop_dict.get('ip_address')
mgmt_info = yaml.load(
vnf_show_out['mgmt_url'])
self.assertEqual(mgmt_url_input, mgmt_info['VDU1'])
# Verify anti spoofing settings
stack_id = vnf_show_out['instance_id']
template_dict = input_dict['topology_template']['node_templates']
for field in template_dict.keys():
prop_dict = template_dict[field]['properties']
if prop_dict.get('anti_spoofing_protection'):
self.verify_antispoofing_in_stack(stack_id=stack_id,
resource_name=field)
self.verify_vnf_crud_events(
vnf_id, evt_constants.RES_EVT_CREATE,
vnf_instance['vnf'][evt_constants.RES_EVT_CREATED_FLD])
return vnfd_id, vnf_id
def _test_delete_vnf(self, vnf_id):
# Delete vnf_instance with vnf_id
try:
self.client.delete_vnf(vnf_id)
except Exception:
assert False, "vnf Delete failed"
self.verify_vnf_crud_events(vnf_id, evt_constants.RES_EVT_DELETE)
def _test_cleanup_vnfd(self, vnfd_id, vnf_id):
# 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_tosca(self):
vnfd_id, vnf_id = self._test_create_vnf(
'sample-tosca-vnfd.yaml',
'test_tosca_vnf_with_cirros')
servers = self.novaclient().servers.list()
vdus = []
for server in servers:
@ -68,19 +110,15 @@ class VnfTestToscaCreate(base.BaseTackerTest):
for port in port_list:
vdu_ports.append(port['name'])
self.assertIn('test-cp', vdu_ports)
self._test_delete_vnf(vnf_id)
self._test_cleanup_vnfd(vnfd_id, vnf_id)
# Delete vnf_instance with vnf_id
try:
self.client.delete_vnf(vnf_id)
except Exception:
assert False, "vnf Delete failed"
self.verify_vnf_crud_events(vnf_id, evt_constants.RES_EVT_DELETE)
# 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_static_ip(self):
vnfd_id, vnf_id = self._test_create_vnf(
'sample-tosca-vnfd-static-ip.yaml',
'test_tosca_vnf_with_cirros_no_monitoring')
self._test_delete_vnf(vnf_id)
self._test_cleanup_vnfd(vnfd_id, vnf_id)
class VnfTestToscaCreateFlavorCreation(base.BaseTackerTest):