Local copy of scenario test base class
The scenario tests base class from Tempest is not a stable interface and it's going to be refactored on Tempest side, as notified in http://lists.openstack.org/pipermail/openstack-dev/2017-February/112938.html Maintain a local copy of the base class, taken from Tempest with head of master at c5f1064759fe6c75a4bc5dc251ed1661845936cb. Change-Id: Idfa5ebe18c794c51e406156fb120d128478d4f1e
This commit is contained in:
Andrea Frittoli
@@ -25,9 +25,9 @@ from oslo_log import log
|
||||
from tempest import config
|
||||
from tempest import exceptions
|
||||
from tempest.lib.common.utils import data_utils
|
||||
from tempest.scenario import manager
|
||||
|
||||
from watcher_tempest_plugin import infra_optim_clients as clients
|
||||
from watcher_tempest_plugin.tests.scenario import manager
|
||||
|
||||
LOG = log.getLogger(__name__)
|
||||
CONF = config.CONF
|
||||
|
||||
206
watcher_tempest_plugin/tests/scenario/manager.py
Normal file
206
watcher_tempest_plugin/tests/scenario/manager.py
Normal file
@@ -0,0 +1,206 @@
|
||||
# Copyright 2012 OpenStack Foundation
|
||||
# Copyright 2013 IBM Corp.
|
||||
# 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_log import log
|
||||
|
||||
from tempest.common import compute
|
||||
from tempest.common import waiters
|
||||
from tempest import config
|
||||
from tempest.lib.common.utils import data_utils
|
||||
from tempest.lib.common.utils import test_utils
|
||||
from tempest.lib import exceptions as lib_exc
|
||||
import tempest.test
|
||||
|
||||
CONF = config.CONF
|
||||
|
||||
LOG = log.getLogger(__name__)
|
||||
|
||||
|
||||
class ScenarioTest(tempest.test.BaseTestCase):
|
||||
"""Base class for scenario tests. Uses tempest own clients. """
|
||||
|
||||
credentials = ['primary']
|
||||
|
||||
@classmethod
|
||||
def setup_clients(cls):
|
||||
super(ScenarioTest, cls).setup_clients()
|
||||
# Clients (in alphabetical order)
|
||||
cls.flavors_client = cls.manager.flavors_client
|
||||
cls.compute_floating_ips_client = (
|
||||
cls.manager.compute_floating_ips_client)
|
||||
if CONF.service_available.glance:
|
||||
# Check if glance v1 is available to determine which client to use.
|
||||
if CONF.image_feature_enabled.api_v1:
|
||||
cls.image_client = cls.manager.image_client
|
||||
elif CONF.image_feature_enabled.api_v2:
|
||||
cls.image_client = cls.manager.image_client_v2
|
||||
else:
|
||||
raise lib_exc.InvalidConfiguration(
|
||||
'Either api_v1 or api_v2 must be True in '
|
||||
'[image-feature-enabled].')
|
||||
# Compute image client
|
||||
cls.compute_images_client = cls.manager.compute_images_client
|
||||
cls.keypairs_client = cls.manager.keypairs_client
|
||||
# Nova security groups client
|
||||
cls.compute_security_groups_client = (
|
||||
cls.manager.compute_security_groups_client)
|
||||
cls.compute_security_group_rules_client = (
|
||||
cls.manager.compute_security_group_rules_client)
|
||||
cls.servers_client = cls.manager.servers_client
|
||||
cls.interface_client = cls.manager.interfaces_client
|
||||
# Neutron network client
|
||||
cls.networks_client = cls.manager.networks_client
|
||||
cls.ports_client = cls.manager.ports_client
|
||||
cls.routers_client = cls.manager.routers_client
|
||||
cls.subnets_client = cls.manager.subnets_client
|
||||
cls.floating_ips_client = cls.manager.floating_ips_client
|
||||
cls.security_groups_client = cls.manager.security_groups_client
|
||||
cls.security_group_rules_client = (
|
||||
cls.manager.security_group_rules_client)
|
||||
|
||||
if CONF.volume_feature_enabled.api_v2:
|
||||
cls.volumes_client = cls.manager.volumes_v2_client
|
||||
cls.snapshots_client = cls.manager.snapshots_v2_client
|
||||
else:
|
||||
cls.volumes_client = cls.manager.volumes_client
|
||||
cls.snapshots_client = cls.manager.snapshots_client
|
||||
|
||||
# ## Test functions library
|
||||
#
|
||||
# The create_[resource] functions only return body and discard the
|
||||
# resp part which is not used in scenario tests
|
||||
|
||||
def _create_port(self, network_id, client=None, namestart='port-quotatest',
|
||||
**kwargs):
|
||||
if not client:
|
||||
client = self.ports_client
|
||||
name = data_utils.rand_name(namestart)
|
||||
result = client.create_port(
|
||||
name=name,
|
||||
network_id=network_id,
|
||||
**kwargs)
|
||||
self.assertIsNotNone(result, 'Unable to allocate port')
|
||||
port = result['port']
|
||||
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
|
||||
client.delete_port, port['id'])
|
||||
return port
|
||||
|
||||
def create_keypair(self, client=None):
|
||||
if not client:
|
||||
client = self.keypairs_client
|
||||
name = data_utils.rand_name(self.__class__.__name__)
|
||||
# We don't need to create a keypair by pubkey in scenario
|
||||
body = client.create_keypair(name=name)
|
||||
self.addCleanup(client.delete_keypair, name)
|
||||
return body['keypair']
|
||||
|
||||
def create_server(self, name=None, image_id=None, flavor=None,
|
||||
validatable=False, wait_until='ACTIVE',
|
||||
clients=None, **kwargs):
|
||||
"""Wrapper utility that returns a test server.
|
||||
|
||||
This wrapper utility calls the common create test server and
|
||||
returns a test server. The purpose of this wrapper is to minimize
|
||||
the impact on the code of the tests already using this
|
||||
function.
|
||||
"""
|
||||
|
||||
# NOTE(jlanoux): As a first step, ssh checks in the scenario
|
||||
# tests need to be run regardless of the run_validation and
|
||||
# validatable parameters and thus until the ssh validation job
|
||||
# becomes voting in CI. The test resources management and IP
|
||||
# association are taken care of in the scenario tests.
|
||||
# Therefore, the validatable parameter is set to false in all
|
||||
# those tests. In this way create_server just return a standard
|
||||
# server and the scenario tests always perform ssh checks.
|
||||
|
||||
# Needed for the cross_tenant_traffic test:
|
||||
if clients is None:
|
||||
clients = self.manager
|
||||
|
||||
if name is None:
|
||||
name = data_utils.rand_name(self.__class__.__name__ + "-server")
|
||||
|
||||
vnic_type = CONF.network.port_vnic_type
|
||||
|
||||
# If vnic_type is configured create port for
|
||||
# every network
|
||||
if vnic_type:
|
||||
ports = []
|
||||
|
||||
create_port_body = {'binding:vnic_type': vnic_type,
|
||||
'namestart': 'port-smoke'}
|
||||
if kwargs:
|
||||
# Convert security group names to security group ids
|
||||
# to pass to create_port
|
||||
if 'security_groups' in kwargs:
|
||||
security_groups = \
|
||||
clients.security_groups_client.list_security_groups(
|
||||
).get('security_groups')
|
||||
sec_dict = dict([(s['name'], s['id'])
|
||||
for s in security_groups])
|
||||
|
||||
sec_groups_names = [s['name'] for s in kwargs.pop(
|
||||
'security_groups')]
|
||||
security_groups_ids = [sec_dict[s]
|
||||
for s in sec_groups_names]
|
||||
|
||||
if security_groups_ids:
|
||||
create_port_body[
|
||||
'security_groups'] = security_groups_ids
|
||||
networks = kwargs.pop('networks', [])
|
||||
else:
|
||||
networks = []
|
||||
|
||||
# If there are no networks passed to us we look up
|
||||
# for the project's private networks and create a port.
|
||||
# The same behaviour as we would expect when passing
|
||||
# the call to the clients with no networks
|
||||
if not networks:
|
||||
networks = clients.networks_client.list_networks(
|
||||
**{'router:external': False, 'fields': 'id'})['networks']
|
||||
|
||||
# It's net['uuid'] if networks come from kwargs
|
||||
# and net['id'] if they come from
|
||||
# clients.networks_client.list_networks
|
||||
for net in networks:
|
||||
net_id = net.get('uuid', net.get('id'))
|
||||
if 'port' not in net:
|
||||
port = self._create_port(network_id=net_id,
|
||||
client=clients.ports_client,
|
||||
**create_port_body)
|
||||
ports.append({'port': port['id']})
|
||||
else:
|
||||
ports.append({'port': net['port']})
|
||||
if ports:
|
||||
kwargs['networks'] = ports
|
||||
self.ports = ports
|
||||
|
||||
tenant_network = self.get_tenant_network()
|
||||
|
||||
body, servers = compute.create_test_server(
|
||||
clients,
|
||||
tenant_network=tenant_network,
|
||||
wait_until=wait_until,
|
||||
name=name, flavor=flavor,
|
||||
image_id=image_id, **kwargs)
|
||||
|
||||
self.addCleanup(waiters.wait_for_server_termination,
|
||||
clients.servers_client, body['id'])
|
||||
self.addCleanup(test_utils.call_and_ignore_notfound_exc,
|
||||
clients.servers_client.delete_server, body['id'])
|
||||
server = clients.servers_client.show_server(body['id'])['server']
|
||||
return server
|
||||
Reference in New Issue
Block a user