Add API for defining cloud operations
Change-Id: Ifafa87fc447b34b9e6e686015c6bd7b0dc5f45c5
This commit is contained in:
Federico Ressi
parent
0afe476d7c
commit
c8c71e55eb
|
|
@ -20,6 +20,7 @@ from tobiko.common import _fixture
|
|||
from tobiko.common import _logging
|
||||
from tobiko.common.managers import testcase as testcase_manager
|
||||
from tobiko.common.managers import loader as loader_manager
|
||||
from tobiko.common import _operation
|
||||
from tobiko.common import _os
|
||||
from tobiko.common import _select
|
||||
from tobiko.common import _skip
|
||||
|
|
@ -61,6 +62,16 @@ load_module = loader_manager.load_module
|
|||
makedirs = _os.makedirs
|
||||
open_output_file = _os.open_output_file
|
||||
|
||||
runs_operation = _operation.runs_operation
|
||||
before_operation = _operation.before_operation
|
||||
after_operation = _operation.after_operation
|
||||
with_operation = _operation.with_operation
|
||||
RunsOperations = _operation.RunsOperations
|
||||
Operation = _operation.Operation
|
||||
get_operation = _operation.get_operation
|
||||
get_operation_name = _operation.get_operation_name
|
||||
operation_config = _operation.operation_config
|
||||
|
||||
discover_testcases = testcase_manager.discover_testcases
|
||||
|
||||
Selection = _select.Selection
|
||||
|
|
|
|||
|
|
@ -0,0 +1,236 @@
|
|||
# Copyright 2019 Red Hat
|
||||
#
|
||||
# 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 __future__ import absolute_import
|
||||
|
||||
import inspect
|
||||
|
||||
from oslo_log import log
|
||||
import testtools
|
||||
|
||||
from tobiko.common import _exception
|
||||
from tobiko.common import _fixture
|
||||
from tobiko.common import _skip
|
||||
|
||||
|
||||
LOG = log.getLogger(__name__)
|
||||
|
||||
|
||||
def runs_operation(operation_object):
|
||||
return RunsOperationProperty(operation_object)
|
||||
|
||||
|
||||
class RunsOperations(object):
|
||||
|
||||
def setUp(self):
|
||||
super(RunsOperations, self).setUp() # pylint: disable=no-member
|
||||
for operation_name in get_runs_operation_names(self):
|
||||
get_operation(operation_name)
|
||||
|
||||
|
||||
def add_runs_operation_name(obj, name):
|
||||
required_names = getattr(obj, '__tobiko_runs_operations__', None)
|
||||
if not isinstance(required_names, set):
|
||||
if required_names:
|
||||
required_names = set(required_names)
|
||||
else:
|
||||
required_names = set()
|
||||
# cache list for later use
|
||||
obj.__tobiko_runs_operations__ = required_names
|
||||
required_names.add(name)
|
||||
|
||||
|
||||
def get_runs_operation_names(obj):
|
||||
'''Get operation names required by given :param obj:'''
|
||||
|
||||
name, obj = _fixture.get_name_and_object(obj)
|
||||
|
||||
operations = getattr(obj, '__tobiko_runs_operations__', set())
|
||||
if isinstance(operations, tuple):
|
||||
return operations
|
||||
|
||||
try:
|
||||
# froze required names to avoid re-discovering them again
|
||||
obj.__tobiko_runs_operations__ = tuple()
|
||||
except AttributeError:
|
||||
# ignore this object because not defined by Python code
|
||||
return tuple()
|
||||
|
||||
LOG.debug('Discover operations for object %r', name)
|
||||
if operations:
|
||||
operations = set(operations)
|
||||
|
||||
if inspect.ismethod(obj) or inspect.isfunction(obj):
|
||||
pass # decorated functions have __tobiko_runs_operations__ defined
|
||||
|
||||
else:
|
||||
if not inspect.isclass(obj):
|
||||
obj = type(obj)
|
||||
operations.update(get_runs_operation_names_from_class(obj))
|
||||
|
||||
# Return every operation in alphabetical order and
|
||||
# froze required names to avoid re-discovering them again
|
||||
operations = tuple(sorted(operations))
|
||||
obj.__tobiko_runs_operations__ = operations
|
||||
return operations
|
||||
|
||||
|
||||
def get_runs_operation_names_from_class(cls):
|
||||
"""Get list of members of type RequiredFixtureProperty of given class"""
|
||||
|
||||
# inspect.getmembers() would iterate over such many testtools.TestCase
|
||||
# members too, so let exclude members from those very common base classes
|
||||
# that we know doesn't have members of type RunsOperationProperty
|
||||
base_classes = cls.__mro__
|
||||
for base_class in [testtools.TestCase, _fixture.SharedFixture]:
|
||||
if issubclass(cls, base_class):
|
||||
base_classes = base_classes[:base_classes.index(base_class)]
|
||||
break
|
||||
|
||||
# Get all members for selected class without calling properties or methods
|
||||
members = {}
|
||||
for base_class in reversed(base_classes):
|
||||
members.update(base_class.__dict__)
|
||||
|
||||
# Return all member operation names
|
||||
operations = set()
|
||||
for name, member in sorted(members.items()):
|
||||
if not name.startswith('__'):
|
||||
operations.update(get_runs_operation_names(obj=member))
|
||||
return operations
|
||||
|
||||
|
||||
class RunsOperationProperty(object):
|
||||
|
||||
def __init__(self, obj):
|
||||
self.name, self.obj = _fixture.get_name_and_object(obj)
|
||||
|
||||
@property
|
||||
def __tobiko_runs_operations__(self):
|
||||
return (self.name,)
|
||||
|
||||
def before_operation(self, obj):
|
||||
|
||||
def is_before():
|
||||
return self.get_operation().is_before
|
||||
|
||||
decorator = _skip.skip_unless(
|
||||
predicate=is_before,
|
||||
reason="Before operation {name!r}".format(name=self.name))
|
||||
|
||||
return self.with_operation(decorator(obj))
|
||||
|
||||
def after_operation(self, obj):
|
||||
|
||||
def is_after():
|
||||
return self.get_operation().is_after
|
||||
|
||||
decorator = _skip.skip_unless(
|
||||
predicate=is_after,
|
||||
reason="After operation {name!r}".format(name=self.name))
|
||||
|
||||
return self.with_operation(decorator(obj))
|
||||
|
||||
def with_operation(self, obj):
|
||||
add_runs_operation_name(obj, self.name)
|
||||
return obj
|
||||
|
||||
def __get__(self, instance, _):
|
||||
if instance is None:
|
||||
return self
|
||||
else:
|
||||
return self.get_operation()
|
||||
|
||||
def get_operation(self):
|
||||
return get_operation(self.obj)
|
||||
|
||||
|
||||
def before_operation(obj):
|
||||
return RunsOperationProperty(obj).before_operation
|
||||
|
||||
|
||||
def after_operation(obj):
|
||||
return RunsOperationProperty(obj).after_operation
|
||||
|
||||
|
||||
def with_operation(obj):
|
||||
return RunsOperationProperty(obj).with_operation
|
||||
|
||||
|
||||
def get_operation(obj):
|
||||
operation = _fixture.get_fixture(obj)
|
||||
_exception.check_valid_type(operation, Operation)
|
||||
|
||||
if operation.is_before and operation_config().run_operations:
|
||||
operation = _fixture.setup_fixture(obj)
|
||||
|
||||
return operation
|
||||
|
||||
|
||||
def get_operation_name(obj):
|
||||
return _fixture.get_object_name(obj)
|
||||
|
||||
|
||||
def operation_config():
|
||||
return _fixture.setup_fixture(OperationsConfigFixture)
|
||||
|
||||
|
||||
class OperationsConfigFixture(_fixture.SharedFixture):
|
||||
|
||||
after_operations = None
|
||||
run_operations = None
|
||||
|
||||
def setup_fixture(self):
|
||||
from tobiko import config
|
||||
self.after_operations = set(
|
||||
config.get_list_env('TOBIKO_AFTER_OPERATIONS'))
|
||||
self.run_operations = config.get_bool_env(
|
||||
'TOBIKO_RUN_OPERATIONS') or False
|
||||
|
||||
|
||||
class Operation(_fixture.SharedFixture):
|
||||
|
||||
@property
|
||||
def operation_name(self):
|
||||
return _fixture.get_fixture_name(self)
|
||||
|
||||
@property
|
||||
def is_before(self):
|
||||
return not self.is_after
|
||||
|
||||
_is_after = None
|
||||
|
||||
@property
|
||||
def is_after(self):
|
||||
is_after = self._is_after
|
||||
if is_after is None:
|
||||
config = operation_config()
|
||||
is_after = self.operation_name in config.after_operations
|
||||
self._is_after = is_after
|
||||
return is_after
|
||||
|
||||
def setup_fixture(self):
|
||||
if not self.is_after:
|
||||
LOG.debug('Executing operation: %r', self.operation_name)
|
||||
try:
|
||||
self.run_operation()
|
||||
except Exception as ex:
|
||||
LOG.debug('Operation %r failed: %r', self.operation_name, ex)
|
||||
raise
|
||||
else:
|
||||
LOG.debug('Operation executed: %r', self.operation_name)
|
||||
finally:
|
||||
self._is_after = True
|
||||
|
||||
def run_operation(self):
|
||||
raise NotImplementedError
|
||||
|
|
@ -347,3 +347,11 @@ def get_bool_env(name):
|
|||
LOG.exception("Environment variable %r is not a boolean: %r",
|
||||
name, value)
|
||||
return None
|
||||
|
||||
|
||||
def get_list_env(name, separator=','):
|
||||
value = get_env(name)
|
||||
if value:
|
||||
return value.split(separator)
|
||||
else:
|
||||
return []
|
||||
|
|
|
|||
|
|
@ -0,0 +1,144 @@
|
|||
# Copyright 2018 Red Hat
|
||||
#
|
||||
# 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 __future__ import absolute_import
|
||||
|
||||
import typing # noqa
|
||||
|
||||
import tobiko
|
||||
from tobiko.tests import unit
|
||||
|
||||
|
||||
class Operation(tobiko.Operation):
|
||||
executed = False
|
||||
|
||||
def run_operation(self):
|
||||
self.executed = True
|
||||
|
||||
|
||||
class OperationForDecorator(Operation):
|
||||
pass
|
||||
|
||||
|
||||
class OperationTest(unit.TobikoUnitTest):
|
||||
|
||||
operation = tobiko.runs_operation(Operation)
|
||||
expect_after_operations = set() # type: typing.Set[str]
|
||||
expect_executed_operations = set() # type: typing.Set[str]
|
||||
|
||||
def expect_is_after(self, operation):
|
||||
name = tobiko.get_operation_name(operation)
|
||||
return name in self.expect_after_operations
|
||||
|
||||
def expect_is_executed(self, operation):
|
||||
name = tobiko.get_operation_name(operation)
|
||||
return name in self.expect_executed_operations
|
||||
|
||||
def test_operation_config(self):
|
||||
config = tobiko.operation_config()
|
||||
self.assertFalse(config.run_operations)
|
||||
self.assertEqual(set(), config.after_operations)
|
||||
|
||||
def test_operation(self, operation=None):
|
||||
operation = operation or self.operation
|
||||
expect_is_after = self.expect_is_after(operation)
|
||||
self.assertIs(not expect_is_after, operation.is_before)
|
||||
self.assertIs(expect_is_after, operation.is_after)
|
||||
expect_is_executed = self.expect_is_executed(operation)
|
||||
self.assertIs(expect_is_executed, operation.executed)
|
||||
|
||||
@tobiko.with_operation(OperationForDecorator)
|
||||
def test_with_operation(self):
|
||||
operation = tobiko.get_operation(OperationForDecorator)
|
||||
self.test_operation(operation)
|
||||
|
||||
@tobiko.before_operation(OperationForDecorator)
|
||||
def test_before_operation(self):
|
||||
operation = tobiko.get_operation(OperationForDecorator)
|
||||
self.test_operation(operation)
|
||||
|
||||
@tobiko.after_operation(OperationForDecorator)
|
||||
def test_after_operation(self):
|
||||
operation = tobiko.get_operation(OperationForDecorator)
|
||||
if self.expect_is_after(operation):
|
||||
self.test_operation(operation)
|
||||
else:
|
||||
self.fail('Test method not skipped')
|
||||
|
||||
@operation.with_operation
|
||||
def test_with_operation_method(self):
|
||||
self.test_operation()
|
||||
|
||||
@operation.before_operation
|
||||
def test_before_operation_method(self):
|
||||
self.test_operation()
|
||||
|
||||
@operation.after_operation
|
||||
def test_after_operation_method(self):
|
||||
if self.expect_is_after(self.operation):
|
||||
self.test_operation()
|
||||
else:
|
||||
self.fail('Test method not skipped')
|
||||
|
||||
|
||||
class DontRunOperationsTest(OperationTest):
|
||||
|
||||
patch_environ = {
|
||||
'TOBIKO_RUN_OPERATIONS': 'false'
|
||||
}
|
||||
|
||||
|
||||
class DoRunOperationsTest(OperationTest):
|
||||
|
||||
patch_environ = {
|
||||
'TOBIKO_RUN_OPERATIONS': 'true'
|
||||
}
|
||||
|
||||
expect_run_operations = True
|
||||
expect_after_operations = {
|
||||
tobiko.get_operation_name(Operation),
|
||||
tobiko.get_operation_name(OperationForDecorator)}
|
||||
expect_executed_operations = expect_after_operations
|
||||
|
||||
def test_operation_config(self):
|
||||
config = tobiko.operation_config()
|
||||
self.assertTrue(config.run_operations)
|
||||
self.assertEqual(set(), config.after_operations)
|
||||
|
||||
|
||||
class AfterOperationsTest(OperationTest):
|
||||
|
||||
patch_environ = {
|
||||
'TOBIKO_AFTER_OPERATIONS': tobiko.get_fixture_name(Operation)
|
||||
}
|
||||
|
||||
expect_after_operations = {tobiko.get_operation_name(Operation)}
|
||||
|
||||
def test_operation_config(self):
|
||||
config = tobiko.operation_config()
|
||||
self.assertFalse(config.run_operations)
|
||||
self.assertEqual(self.expect_after_operations,
|
||||
config.after_operations)
|
||||
|
||||
|
||||
class RunOperationsMixinTest(tobiko.RunsOperations, unit.TobikoUnitTest):
|
||||
|
||||
patch_environ = {
|
||||
'TOBIKO_RUN_OPERATIONS': 'true'
|
||||
}
|
||||
|
||||
operation = tobiko.runs_operation(Operation)
|
||||
|
||||
def test_operation(self):
|
||||
self.assertTrue(self.operation.is_after)
|
||||
self.assertTrue(self.operation.executed)
|
||||
Loading…
Reference in New Issue