269 lines
8.7 KiB
Python
Executable File
269 lines
8.7 KiB
Python
Executable File
#
|
|
# Copyright (c) 2019 Wind River Systems, Inc.
|
|
#
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
#
|
|
|
|
|
|
from pytest import fixture
|
|
|
|
from utils.tis_log import LOG
|
|
from utils import exceptions
|
|
|
|
from consts.auth import Tenant
|
|
from keywords import nova_helper, vm_helper, cinder_helper, glance_helper, \
|
|
network_helper, system_helper, heat_helper
|
|
from testfixtures.fixture_resources import ResourceCleanup, GuestLogs
|
|
|
|
|
|
# SIMPLEX_RECOVERED = False
|
|
|
|
|
|
@fixture(scope='function', autouse=True)
|
|
def delete_resources_func(request):
|
|
"""
|
|
Function level fixture to delete created resources after each caller
|
|
testcase.
|
|
|
|
Notes: Auto used fixture - import it to a conftest.py file under a
|
|
feature directory to auto use it on all children testcases.
|
|
|
|
Examples:
|
|
- see nova/conftest.py for importing
|
|
- see ResourceCleanup.add function usages in nova/test_shared_cpu.py
|
|
for adding resources to cleanups
|
|
|
|
Args:
|
|
request: pytest param present caller test function
|
|
|
|
"""
|
|
|
|
def delete_():
|
|
_delete_resources(ResourceCleanup._get_resources('function'),
|
|
scope='function')
|
|
ResourceCleanup._reset('function')
|
|
|
|
request.addfinalizer(delete_)
|
|
|
|
|
|
@fixture(scope='class', autouse=True)
|
|
def delete_resources_class(request):
|
|
"""
|
|
Class level fixture to delete created resources after each caller testcase.
|
|
|
|
Notes: Auto used fixture - import it to a conftest.py file under a
|
|
feature directory to auto use it on all children
|
|
testcases.
|
|
|
|
Examples:
|
|
- see nova/conftest.py for importing
|
|
- see ResourceCleanup.add function usages in nova/test_shared_cpu.py
|
|
for adding resources to cleanups
|
|
|
|
Args:
|
|
request: pytest param present caller test function
|
|
|
|
"""
|
|
|
|
def delete_():
|
|
_delete_resources(ResourceCleanup._get_resources('class'),
|
|
scope='class')
|
|
ResourceCleanup._reset('class')
|
|
|
|
request.addfinalizer(delete_)
|
|
|
|
|
|
@fixture(scope='module', autouse=True)
|
|
def delete_resources_module(request):
|
|
"""
|
|
Module level fixture to delete created resources after each caller testcase.
|
|
|
|
Notes: Auto used fixture - import it to a conftest.py file under a
|
|
feature directory to auto use it on all children
|
|
testcases.
|
|
|
|
Examples:
|
|
- see nova/conftest.py for importing
|
|
- see ResourceCleanup.add function usages in nova/test_shared_cpu.py
|
|
for adding resources to cleanups
|
|
|
|
Args:
|
|
request: pytest param present caller test function
|
|
|
|
"""
|
|
|
|
def delete_():
|
|
_delete_resources(ResourceCleanup._get_resources('module'),
|
|
scope='module')
|
|
ResourceCleanup._reset('module')
|
|
|
|
request.addfinalizer(delete_)
|
|
|
|
|
|
@fixture(scope='session', autouse=True)
|
|
def delete_resources_session(request):
|
|
"""
|
|
Module level fixture to delete created resources after each caller testcase.
|
|
|
|
Notes: Auto used fixture - import it to a conftest.py file under a
|
|
feature directory to auto use it on all children
|
|
testcases.
|
|
|
|
Examples:
|
|
- see nova/conftest.py for importing
|
|
- see ResourceCleanup.add function usages in nova/test_shared_cpu.py
|
|
for adding resources to cleanups
|
|
|
|
Args:
|
|
request: pytest param present caller test function
|
|
|
|
"""
|
|
|
|
def delete_():
|
|
_delete_resources(ResourceCleanup._get_resources('session'),
|
|
scope='session')
|
|
ResourceCleanup._reset('session')
|
|
|
|
request.addfinalizer(delete_)
|
|
|
|
|
|
@fixture(scope='module')
|
|
def flavor_id_module():
|
|
"""
|
|
Create basic flavor and volume to be used by test cases as test setup,
|
|
at the beginning of the test module.
|
|
Delete the created flavor and volume as test teardown, at the end of the
|
|
test module.
|
|
"""
|
|
flavor = nova_helper.create_flavor()[1]
|
|
ResourceCleanup.add('flavor', resource_id=flavor, scope='module')
|
|
|
|
return flavor
|
|
|
|
|
|
def _delete_resources(resources, scope):
|
|
# global SIMPLEX_RECOVERED
|
|
# if not SIMPLEX_RECOVERED and system_helper.is_simplex():
|
|
# LOG.fixture_step('{} Ensure simplex host is up before cleaning
|
|
# up'.format(scope))
|
|
# host_helper.recover_simplex(fail_ok=True)
|
|
# SIMPLEX_RECOVERED = True
|
|
|
|
def __del_aggregate(aggregate_, **kwargs):
|
|
nova_helper.remove_hosts_from_aggregate(aggregate=aggregate_,
|
|
check_first=False, **kwargs)
|
|
return nova_helper.delete_aggregates(names=aggregate_, **kwargs)
|
|
|
|
# List resources in proper order if there are dependencies!
|
|
del_list = [
|
|
# resource, del_fun, fun_params, whether to delete all resources
|
|
# together.
|
|
('port_chain', network_helper.delete_sfc_port_chain,
|
|
{'check_first': True}, False),
|
|
('flow_classifier', network_helper.delete_flow_classifier,
|
|
{'check_first': True}, False),
|
|
('vm', vm_helper.delete_vms, {'delete_volumes': False}, True),
|
|
('vm_with_vol', vm_helper.delete_vms, {'delete_volumes': True}, True),
|
|
('vol_snapshot', cinder_helper.delete_volume_snapshots, {}, True),
|
|
('volume', cinder_helper.delete_volumes, {}, True),
|
|
('volume_type', cinder_helper.delete_volume_types, {}, True),
|
|
('volume_qos', cinder_helper.delete_volume_qos, {}, True),
|
|
('flavor', nova_helper.delete_flavors, {}, True),
|
|
('image', glance_helper.delete_images, {}, True),
|
|
('server_group', nova_helper.delete_server_groups, {}, True),
|
|
('floating_ip', network_helper.delete_floating_ips, {}, True),
|
|
('trunk', network_helper.delete_trunks, {}, True),
|
|
('heat_stack', heat_helper.delete_stack, {}, False),
|
|
('port_pair_group', network_helper.delete_sfc_port_pair_group,
|
|
{'check_first': True}, False),
|
|
('port_pair', network_helper.delete_sfc_port_pairs,
|
|
{'check_first': True}, True),
|
|
('port', network_helper.delete_port, {}, False),
|
|
('router', network_helper.delete_router, {}, False),
|
|
('subnet', network_helper.delete_subnets, {}, True),
|
|
('network_qos', network_helper.delete_qos, {}, False),
|
|
('network', network_helper.delete_network, {}, False),
|
|
('security_group_rule', network_helper.delete_security_group_rules, {},
|
|
True),
|
|
('security_group', network_helper.delete_security_group, {}, False),
|
|
('aggregate', __del_aggregate, {}, False),
|
|
('datanetwork', system_helper.delete_data_network, {}, False),
|
|
]
|
|
|
|
err_msgs = []
|
|
for item in del_list:
|
|
resource_type, del_fun, fun_kwargs, del_all = item
|
|
resource_ids = resources.get(resource_type, [])
|
|
if not resource_ids:
|
|
continue
|
|
|
|
LOG.fixture_step("({}) Attempt to delete following {}: "
|
|
"{}".format(scope, resource_type, resource_ids))
|
|
if 'auth_info' not in fun_kwargs:
|
|
fun_kwargs['auth_info'] = Tenant.get('admin')
|
|
|
|
if del_all:
|
|
resource_ids = [resource_ids]
|
|
for resource_id in resource_ids:
|
|
try:
|
|
code, msg = del_fun(resource_id, fail_ok=True, **fun_kwargs)[
|
|
0:2]
|
|
if code > 0:
|
|
err_msgs.append(msg)
|
|
except exceptions.TiSError as e:
|
|
err_msgs.append(e.__str__())
|
|
|
|
# Attempt all deletions before raising exception.
|
|
if err_msgs:
|
|
LOG.error("ERROR: Failed to delete resource(s). \nDetails: {}".format(
|
|
err_msgs))
|
|
# raise exceptions.CommonError("Failed to delete resource(s).
|
|
# Details: {}".format(err_msgs))
|
|
|
|
|
|
@fixture(scope='function', autouse=True)
|
|
def guest_logs_func(request):
|
|
"""
|
|
Collect guest logs for guests in collect list. Applicable to guest
|
|
heartbeat, server group, vm scaling test cases.
|
|
- Use fixture_resources.GuestLogs.add() to add a guest to collect list
|
|
- Use fixture_resources.GuestLogs.remove() to remove a guest from
|
|
collect list if test passed
|
|
|
|
Examples:
|
|
see testcases/functional/mtc/guest_heartbeat/test_vm_voting
|
|
.py for usage
|
|
|
|
"""
|
|
|
|
def _collect():
|
|
_collect_guest_logs(scope='function')
|
|
|
|
request.addfinalizer(_collect)
|
|
|
|
|
|
@fixture(scope='class', autouse=True)
|
|
def guest_logs_class(request):
|
|
def _collect():
|
|
_collect_guest_logs(scope='class')
|
|
|
|
request.addfinalizer(_collect)
|
|
|
|
|
|
@fixture(scope='module', autouse=True)
|
|
def guest_logs_module(request):
|
|
def _collect():
|
|
_collect_guest_logs(scope='module')
|
|
|
|
request.addfinalizer(_collect)
|
|
|
|
|
|
def _collect_guest_logs(scope):
|
|
guests = GuestLogs._get_guests(scope=scope)
|
|
if guests:
|
|
LOG.fixture_step(
|
|
"({}) Attempt to collect guest logs for: {}".format(scope, guests))
|
|
for guest in guests:
|
|
vm_helper.collect_guest_logs(vm_id=guest)
|
|
GuestLogs._reset(scope=scope)
|