x/tap-as-a-service-tempest-plugin
Code Issues Proposed changes

Add a scenario test and refactor API tests

Browse Source 
1. Adds a tempest scenario testcase in TaaS.

2. Refactor the existing API tests.

3. Adds a new API testcase to test the newly added Vlan Mirror input
parameter in tap-flow-create API.

This patchset is an extension to following commit which adds SRIOV
driver support in TaaS:

Change-Id: I847d7f3d841900d99c914a451c24ee20e8119618
Commit: https://review.openstack.org/#/c/603501/
This commit is contained in:
Deepak Tiwari 2019-04-08 10:53:35 -05:00
parent 799bf1c026
commit e5826f70c3
5 changed files with 308 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
neutron_taas_tempest_plugin/config.py Normal file
View File

@ -0,0 +1,37 @@
# Copyright (c) 2018 AT&T 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.
from oslo_config import cfg
from tempest import config
CONF = config.CONF
taas_plugin_group = cfg.OptGroup(name='taas_plugin_options',
title='TaaS Tempest Plugin Options')
TaaSPluginGroup = [
cfg.StrOpt('provider_physical_network',
default='',
help='Physical network to be used for creating SRIOV network.'),
cfg.StrOpt('provider_segmentation_id',
default='',
help='Segmentation-id to be used for creating SRIOV network.'),
cfg.StrOpt('vlan_filter',
default='',
help='Comma separated list of VLANs to be mirrored '
'for a Tap-Flow.'),
]

9
neutron_taas_tempest_plugin/plugin.py
View File

@ -17,10 +17,13 @@ import os
from tempest.test_discover import plugins
from neutron_taas_tempest_plugin import config as project_config
class NeutronTaaSPlugin(plugins.TempestPlugin):
def get_opt_lists(self):
return []
return [(project_config.taas_plugin_group.name,
project_config.TaaSPluginGroup)]
def load_tests(self):
this_dir = os.path.dirname(os.path.abspath(__file__))
@ -31,4 +34,6 @@ class NeutronTaaSPlugin(plugins.TempestPlugin):
return (test_dir, top_level_dir)
def register_opts(self, conf):
return
conf.register_group(project_config.taas_plugin_group)
conf.register_opts(project_config.TaaSPluginGroup,
project_config.taas_plugin_group)

15
neutron_taas_tempest_plugin/tests/api/test_taas.py
View File

@ -49,6 +49,21 @@ class TaaSExtensionTestJSON(base.BaseTaaSTest):
self.create_tap_flow(tap_service_id=tap_service['id'],
direction='BOTH', source_port=self.tf_port['id'])
@decorators.idempotent_id('897a0aaf-1b55-4ea8-9d9f-1bc0fd09cb60')
@utils.requires_ext(extension='taas-vlan-filter', service='network')
def test_create_tap_service_and_flow_vlan_filter(self):
"""create tap service with vlan_filter
Test create tap service with additional vlan_filter argument.
"""
tap_service = self.create_tap_service(port_id=self.ts_port['id'])
tap_flow = self.create_tap_flow(tap_service_id=tap_service['id'],
direction='BOTH',
source_port=self.tf_port['id'],
vlan_filter='189,279,999-1008')
self.assertEqual(tap_flow['tap_flow']['vlan_filter'],
'189,279,999-1008')
@decorators.idempotent_id('d7a2115d-16b4-41cf-95a6-dcebc3682b24')
def test_delete_tap_resources_after_ts_port_delete(self):
"""delete tap resources after ts port delete

4
neutron_taas_tempest_plugin/tests/scenario/base.py
View File

@ -14,7 +14,9 @@
# under the License.
from neutron_taas_tempest_plugin.tests.scenario import manager
from neutron_taas_tempest_plugin.tests import taas_client
class TaaSScenarioTest(manager.NetworkScenarioTest):
class TaaSScenarioTest(taas_client.TaaSClientMixin,
manager.NetworkScenarioTest):
pass

252
neutron_taas_tempest_plugin/tests/scenario/test_taas.py
View File

@ -13,25 +13,265 @@
# License for the specific language governing permissions and limitations
# under the License.
from oslo_log import log as logging
from tempest.common import utils
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib.common.utils import test_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc
import testtools
from neutron_taas_tempest_plugin.tests.scenario import base
CONF = config.CONF
LOG = logging.getLogger(__name__)
# pylint: disable=too-many-ancestors
class TestTaaS(base.TaaSScenarioTest):
"""Config Requirement in tempest.conf:
- project_network_cidr_bits- specifies the subnet range for each network
- project_network_cidr
- public_network_id.
"""
@classmethod
@utils.requires_ext(extension='taas', service='network')
@utils.requires_ext(extension='security-group', service='network')
@utils.requires_ext(extension='router', service='network')
def skip_checks(cls):
super(TestTaaS, cls).skip_checks()
@classmethod
def resource_setup(cls):
LOG.debug("Initializing TaaSScenarioTest Setup")
super(TestTaaS, cls).resource_setup()
for ext in ['taas']:
if not utils.is_extension_enabled(ext, 'network'):
msg = "%s Extension not enabled." % ext
raise cls.skipException(msg)
LOG.debug("TaaSScenarioTest Setup done.")
def _create_server(self, network, security_group=None):
"""Create a server
Creates a server having a port on given network and security group.
"""
keys = self.create_keypair()
kwargs = {}
if security_group is not None:
kwargs['security_groups'] = [{'name': security_group['name']}]
server = self.create_server(
key_name=keys['name'],
networks=[{'uuid': network['id']}],
wait_until='ACTIVE',
**kwargs)
return server, keys
def _create_test_server(self, network, security_group):
"""Create a server
Creates a server having a port on given network and security group;
Also creates a floting IP if port is not an sriov port.
"""
pub_network_id = CONF.network.public_network_id
server, keys = self._create_server(
network, security_group=security_group)
private_key = keys['private_key']
vnic_type = CONF.network.port_vnic_type
server_floating_ip = None
if vnic_type != 'direct':
server_floating_ip = self.create_floating_ip(server,
pub_network_id)
fixed_ip = list(server['addresses'].values())[0][0]['addr']
return server, private_key, fixed_ip, server_floating_ip
@testtools.skipUnless(CONF.taas_plugin_options.provider_physical_network,
'Provider physical network parameter not provided.')
@utils.requires_ext(extension="provider", service="network")
def _create_network_sriov(self, networks_client=None,
tenant_id=None,
namestart='network-smoke-sriov-',
port_security_enabled=True):
if not networks_client:
networks_client = self.networks_client
if not tenant_id:
tenant_id = networks_client.tenant_id
name = data_utils.rand_name(namestart)
network_kwargs = dict(name=name, tenant_id=tenant_id)
# Neutron disables port security by default so we have to check the
# config before trying to create the network with
# port_security_enabled
if CONF.network_feature_enabled.port_security:
network_kwargs['port_security_enabled'] = port_security_enabled
if CONF.network.port_vnic_type and \
CONF.network.port_vnic_type == 'direct':
network_kwargs['provider:network_type'] = 'vlan'
if CONF.taas_plugin_options.provider_segmentation_id:
if CONF.taas_plugin_options.provider_segmentation_id == '0':
network_kwargs['provider:network_type'] = 'flat'
else:
network_kwargs['provider:segmentation_id'] = \
CONF.taas_plugin_options.provider_segmentation_id
network_kwargs['provider:physical_network'] = \
CONF.taas_plugin_options.provider_physical_network
result = networks_client.create_network(**network_kwargs)
network = result['network']
self.assertEqual(network['name'], name)
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
networks_client.delete_network,
network['id'])
return network
@testtools.skipUnless(CONF.taas_plugin_options.provider_physical_network,
'Provider physical network parameter not provided.')
@utils.requires_ext(extension="provider", service="network")
def create_networks_sriov(self, networks_client=None,
routers_client=None, subnets_client=None,
tenant_id=None, dns_nameservers=None,
port_security_enabled=True):
"""Create a network with a subnet connected to a router.
The baremetal driver is a special case since all nodes are
on the same shared network.
:param tenant_id: id of tenant to create resources in.
:param dns_nameservers: list of dns servers to send to subnet.
:returns: network, subnet, router
"""
router = None
if CONF.network.shared_physical_network:
# NOTE(Shrews): This exception is for environments where tenant
# credential isolation is available, but network separation is
# not (the current baremetal case). Likely can be removed when
# test account mgmt is reworked:
# https://blueprints.launchpad.net/tempest/+spec/test-accounts
if not CONF.compute.fixed_network_name:
msg = 'fixed_network_name must be specified in config'
raise lib_exc.InvalidConfiguration(msg)
network = self._get_network_by_name(
CONF.compute.fixed_network_name)
subnet = None
else:
network = self._create_network_sriov(
networks_client=networks_client,
tenant_id=tenant_id,
port_security_enabled=port_security_enabled)
subnet_kwargs = dict(network=network,
subnets_client=subnets_client,
routers_client=routers_client)
# use explicit check because empty list is a valid option
if dns_nameservers is not None:
subnet_kwargs['dns_nameservers'] = dns_nameservers
subnet = self._create_subnet(**subnet_kwargs)
return network, subnet, router
def _create_topology(self):
"""Topology
+----------+ +----------+
| "server" | | "server" |
| VM-1 | | VM-2 |
| | | |
+----+-----+ +----+-----+
| |
| |
+----+----+----+----+----+----+-----+
|
|
|
+------+------+
| "server" |
| tap-service |
+-------------+
"""
LOG.debug('Starting Topology Creation')
resp = {}
# Create Network1 and Subnet1.
vnic_type = CONF.network.port_vnic_type
if vnic_type == 'direct':
self.network1, self.subnet1, self.router1 = \
self.create_networks_sriov()
else:
self.network1, self.subnet1, self.router1 = self.create_networks()
resp['network1'] = self.network1
resp['subnet1'] = self.subnet1
resp['router1'] = self.router1
# Create a security group allowing icmp and ssh traffic.
security_group = self._create_security_group()
# Create 3 VMs and assign them a floating IP each.
server1, private_key1, server_fixed_ip_1, server_floating_ip_1 = (
self._create_test_server(self.network1, security_group))
server2, private_key2, server_fixed_ip_2, server_floating_ip_2 = (
self._create_test_server(self.network1, security_group))
server3, private_key3, server_fixed_ip_3, server_floating_ip_3 = (
self._create_test_server(self.network1, security_group))
# Store the received information to be used later
resp['server1'] = server1
resp['private_key1'] = private_key1
resp['server_fixed_ip_1'] = server_fixed_ip_1
resp['server_floating_ip_1'] = server_floating_ip_1
resp['server2'] = server2
resp['private_key2'] = private_key2
resp['server_fixed_ip_2'] = server_fixed_ip_2
resp['server_floating_ip_2'] = server_floating_ip_2
resp['server3'] = server3
resp['private_key3'] = private_key3
resp['server_fixed_ip_3'] = server_fixed_ip_3
resp['server_floating_ip_3'] = server_floating_ip_3
return resp
@utils.services('network')
@utils.requires_ext(extension="taas-vlan-filter", service="network")
@decorators.attr(type='slow')
@decorators.idempotent_id('40903cbd-0e3c-464d-b311-dc77d3894e65')
def test_dummy(self):
pass
def test_tap_flow_data_mirroring(self):
"""Create test topology and TaaS resources
Creates test topology consisting of 3 servers, one routable network,
ports and TaaS resources, i.e. tap-service and tap-flow using those
ports.
"""
topology = self._create_topology()
# Fetch source port and tap-service port to be used for creating
# Tap Service and Tap flow.
source_port = self.os_admin.ports_client.list_ports(
network_id=topology['network1']['id'],
device_id=topology['server1']['id']
)['ports'][0]
tap_service_port = self.os_admin.ports_client.list_ports(
network_id=topology['network1']['id'],
device_id=topology['server3']['id']
)['ports'][0]
# Create Tap-Service.
tap_service = self.create_tap_service(port_id=tap_service_port['id'])
LOG.debug('TaaS Config options: vlan-filter: %s',
CONF.taas_plugin_options.vlan_filter)
# Create Tap-Flow.
vnic_type = CONF.network.port_vnic_type
vlan_filter = None
if vnic_type == 'direct':
vlan_filter = '108-117,126,135-144'
if CONF.taas_plugin_options.vlan_filter:
vlan_filter = CONF.taas_plugin_options.vlan_filter
elif topology['network1']['provider:segmentation_id'] != '0':
vlan_filter = topology['network1']['provider:segmentation_id']
tap_flow = self.create_tap_flow(tap_service_id=tap_service['id'],
direction='BOTH',
source_port=source_port['id'],
vlan_filter=vlan_filter)
self.assertEqual(tap_flow['tap_flow']['vlan_filter'], vlan_filter)