Refactor functional base test classes
The functional base test class should be implemented under os-vif. Other projects will inherit from this project to implement functional test cases. Change-Id: Ib5873687032bfb55d1847d4948f6e9679cf6d31e Closes-Bug: #1817938
This commit is contained in:
Rodolfo Alonso Hernandez
parent
9059c7eb7e
commit
c63366d67c
@ -40,14 +40,24 @@ def _get_test_log_path():
|
||||
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
|
||||
def wait_until_true(predicate, timeout=15, sleep=1):
|
||||
"""Wait until callable predicate is evaluated as True
|
||||
|
||||
:param predicate: Callable deciding whether waiting should continue.
|
||||
Best practice is to instantiate predicate with
|
||||
``functools.partial()``.
|
||||
:param timeout: Timeout in seconds how long should function wait.
|
||||
:param sleep: Polling interval for results in seconds.
|
||||
:return: True if the predicate is evaluated as True within the timeout,
|
||||
False in case of timeout evaluating the predicate.
|
||||
"""
|
||||
try:
|
||||
with eventlet.Timeout(timeout):
|
||||
while not predicate():
|
||||
eventlet.sleep(sleep)
|
||||
except eventlet.Timeout:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
class _CatchTimeoutMetaclass(abc.ABCMeta):
|
||||
@ -58,13 +68,22 @@ class _CatchTimeoutMetaclass(abc.ABCMeta):
|
||||
# both unbound methods (python2) and functions (python3)
|
||||
cls, predicate=inspect.isroutine):
|
||||
if name.startswith('test_'):
|
||||
setattr(cls, name, _catch_timeout(method))
|
||||
setattr(cls, name, cls._catch_timeout(method))
|
||||
|
||||
@staticmethod
|
||||
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
|
||||
|
||||
|
||||
def setup_logging():
|
||||
def setup_logging(component_name):
|
||||
"""Sets up the logging options for a log with supplied name."""
|
||||
product_name = "os_vif"
|
||||
logging.setup(cfg.CONF, product_name)
|
||||
logging.setup(cfg.CONF, component_name)
|
||||
LOG.info("Logging enabled!")
|
||||
LOG.info("%(prog)s version %(version)s",
|
||||
{'prog': sys.argv[0], 'version': osvif_version.__version__})
|
||||
@ -84,21 +103,25 @@ def sanitize_log_path(path):
|
||||
class BaseFunctionalTestCase(base.BaseTestCase):
|
||||
"""Base class for functional tests."""
|
||||
|
||||
COMPONENT_NAME = 'os_vif'
|
||||
PRIVILEGED_GROUP = 'os_vif_privileged'
|
||||
|
||||
def setUp(self):
|
||||
super(BaseFunctionalTestCase, self).setUp()
|
||||
logging.register_options(CONF)
|
||||
setup_logging()
|
||||
setup_logging(self.COMPONENT_NAME)
|
||||
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)
|
||||
self.flags(log_file=log_file)
|
||||
privsep_helper = os.path.join(
|
||||
os.getenv('VIRTUAL_ENV'), 'bin', 'privsep-helper')
|
||||
self.config(
|
||||
os.getenv('VIRTUAL_ENV', os.path.dirname(sys.executable)[:-4]),
|
||||
'bin', 'privsep-helper')
|
||||
self.flags(
|
||||
helper_command=' '.join(['sudo', '-E', privsep_helper]),
|
||||
group='os_vif_privileged')
|
||||
group=self.PRIVILEGED_GROUP)
|
||||
|
||||
def config(self, **kw):
|
||||
def flags(self, **kw):
|
||||
"""Override some configuration values.
|
||||
|
||||
The keyword arguments are the names of configuration options to
|
||||
|
||||
@ -1,6 +1,3 @@
|
||||
# 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
|
||||
@ -13,128 +10,14 @@
|
||||
# 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
|
||||
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
|
||||
from os_vif.tests.functional import base as os_vif_base
|
||||
|
||||
|
||||
CONF = cfg.CONF
|
||||
LOG = logging.getLogger(__name__)
|
||||
wait_until_true = os_vif_base.wait_until_true
|
||||
|
||||
|
||||
def _get_test_log_path():
|
||||
return os.environ.get('OS_LOG_PATH', '/tmp')
|
||||
class VifPlugOvsBaseFunctionalTestCase(os_vif_base.BaseFunctionalTestCase):
|
||||
"""Base class for vif_plug_ovs functional tests."""
|
||||
|
||||
|
||||
# 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 wait_until_true(predicate, timeout=15, sleep=1):
|
||||
"""Wait until callable predicate is evaluated as True
|
||||
|
||||
:param predicate: Callable deciding whether waiting should continue.
|
||||
Best practice is to instantiate predicate with
|
||||
``functools.partial()``.
|
||||
:param timeout: Timeout in seconds how long should function wait.
|
||||
:param sleep: Polling interval for results in seconds.
|
||||
:return: True if the predicate is evaluated as True within the timeout,
|
||||
False in case of timeout evaluating the predicate.
|
||||
"""
|
||||
try:
|
||||
with eventlet.Timeout(timeout):
|
||||
while not predicate():
|
||||
eventlet.sleep(sleep)
|
||||
except eventlet.Timeout:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
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, cls._catch_timeout(method))
|
||||
|
||||
@staticmethod
|
||||
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
|
||||
|
||||
|
||||
def setup_logging():
|
||||
"""Sets up the logging options for a log with supplied name."""
|
||||
product_name = "vif_plug_ovs"
|
||||
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
|
||||
replace_map = {' ': '-', '(': '_', ')': '_'}
|
||||
for s, r in replace_map.items():
|
||||
path = path.replace(s, r)
|
||||
return path
|
||||
|
||||
|
||||
@six.add_metaclass(_CatchTimeoutMetaclass)
|
||||
class BaseFunctionalTestCase(base.BaseTestCase):
|
||||
"""Base class for functional tests.
|
||||
|
||||
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).
|
||||
"""
|
||||
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.flags(log_file=log_file)
|
||||
privsep_helper = os.path.join(
|
||||
os.getenv('VIRTUAL_ENV', os.path.dirname(sys.executable)[:-4]),
|
||||
'bin', 'privsep-helper')
|
||||
self.flags(
|
||||
helper_command=' '.join(['sudo', '-E', privsep_helper]),
|
||||
group='vif_plug_ovs_privileged')
|
||||
|
||||
def flags(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)
|
||||
COMPONENT_NAME = 'vif_plug_ovs'
|
||||
PRIVILEGED_GROUP = 'vif_plug_ovs_privileged'
|
||||
|
||||
@ -35,7 +35,8 @@ def run_privileged(*full_args):
|
||||
return processutils.execute(*full_args)[0].rstrip()
|
||||
|
||||
|
||||
class TestOVSDBLib(testscenarios.WithScenarios, base.BaseFunctionalTestCase):
|
||||
class TestOVSDBLib(testscenarios.WithScenarios,
|
||||
base.VifPlugOvsBaseFunctionalTestCase):
|
||||
|
||||
scenarios = [
|
||||
('native', {'interface': 'native'}),
|
||||
|
||||
Loading…
Reference in New Issue
Block a user