116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
# Derived from: neutron/tests/functional/base.py
|
|
# neutron/tests/base.py
|
|
#
|
|
# 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 abc
|
|
import functools
|
|
import inspect
|
|
import os
|
|
import six
|
|
import sys
|
|
|
|
import eventlet.timeout
|
|
from os_vif import version as osvif_version
|
|
from oslo_config import cfg
|
|
from oslo_log import log as logging
|
|
from oslo_utils import fileutils
|
|
from oslotest import base
|
|
|
|
|
|
CONF = cfg.CONF
|
|
LOG = logging.getLogger(__name__)
|
|
|
|
|
|
def _get_test_log_path():
|
|
return os.environ.get('OS_LOG_PATH', '/tmp')
|
|
|
|
|
|
# This is the directory from which infra fetches log files for functional tests
|
|
DEFAULT_LOG_DIR = os.path.join(_get_test_log_path(), 'osvif-functional-logs')
|
|
|
|
|
|
def _catch_timeout(f):
|
|
@functools.wraps(f)
|
|
def func(self, *args, **kwargs):
|
|
try:
|
|
return f(self, *args, **kwargs)
|
|
except eventlet.Timeout as e:
|
|
self.fail('Execution of this test timed out: %s' % e)
|
|
return func
|
|
|
|
|
|
class _CatchTimeoutMetaclass(abc.ABCMeta):
|
|
def __init__(cls, name, bases, dct):
|
|
super(_CatchTimeoutMetaclass, cls).__init__(name, bases, dct)
|
|
for name, method in inspect.getmembers(
|
|
# NOTE(ihrachys): we should use isroutine because it will catch
|
|
# both unbound methods (python2) and functions (python3)
|
|
cls, predicate=inspect.isroutine):
|
|
if name.startswith('test_'):
|
|
setattr(cls, name, _catch_timeout(method))
|
|
|
|
|
|
def setup_logging():
|
|
"""Sets up the logging options for a log with supplied name."""
|
|
product_name = "os_vif"
|
|
logging.setup(cfg.CONF, product_name)
|
|
LOG.info("Logging enabled!")
|
|
LOG.info("%(prog)s version %(version)s",
|
|
{'prog': sys.argv[0], 'version': osvif_version.__version__})
|
|
LOG.debug("command line: %s", " ".join(sys.argv))
|
|
|
|
|
|
def sanitize_log_path(path):
|
|
"""Sanitize the string so that its log path is shell friendly"""
|
|
return path.replace(' ', '-').replace('(', '_').replace(')', '_')
|
|
|
|
|
|
# Test worker cannot survive eventlet's Timeout exception, which effectively
|
|
# kills the whole worker, with all test cases scheduled to it. This metaclass
|
|
# makes all test cases convert Timeout exceptions into unittest friendly
|
|
# failure mode (self.fail).
|
|
@six.add_metaclass(_CatchTimeoutMetaclass)
|
|
class BaseFunctionalTestCase(base.BaseTestCase):
|
|
"""Base class for functional tests."""
|
|
|
|
def setUp(self):
|
|
super(BaseFunctionalTestCase, self).setUp()
|
|
logging.register_options(CONF)
|
|
setup_logging()
|
|
fileutils.ensure_tree(DEFAULT_LOG_DIR, mode=0o755)
|
|
log_file = sanitize_log_path(
|
|
os.path.join(DEFAULT_LOG_DIR, "%s.txt" % self.id()))
|
|
self.config(log_file=log_file)
|
|
privsep_helper = os.path.join(
|
|
os.getenv('VIRTUAL_ENV'), 'bin', 'privsep-helper')
|
|
self.config(
|
|
helper_command=' '.join(['sudo', '-E', privsep_helper]),
|
|
group='os_vif_privileged')
|
|
|
|
def config(self, **kw):
|
|
"""Override some configuration values.
|
|
|
|
The keyword arguments are the names of configuration options to
|
|
override and their values.
|
|
|
|
If a group argument is supplied, the overrides are applied to
|
|
the specified configuration option group.
|
|
|
|
All overrides are automatically cleared at the end of the current
|
|
test by the fixtures cleanup process.
|
|
"""
|
|
group = kw.pop('group', None)
|
|
for k, v in kw.items():
|
|
CONF.set_override(k, v, group)
|