diff --git a/tobiko/__init__.py b/tobiko/__init__.py index da5f2bd40..89c9c09d3 100644 --- a/tobiko/__init__.py +++ b/tobiko/__init__.py @@ -105,8 +105,9 @@ MultipleObjectsFound = _select.MultipleObjectsFound SkipException = _skip.SkipException skip_if = _skip.skip_if -skip_test = skip = _skip.skip_test +skip_test = _skip.skip_test skip_unless = _skip.skip_unless +skip = _skip.skip BaseTestCase = _testcase.TestCase assert_test_case_was_skipped = _testcase.assert_test_case_was_skipped diff --git a/tobiko/common/_operation.py b/tobiko/common/_operation.py index e24fede5f..3bb8d9209 100644 --- a/tobiko/common/_operation.py +++ b/tobiko/common/_operation.py @@ -124,9 +124,8 @@ class RunsOperationProperty(object): 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)) + decorator = _skip.skip_unless(f"Before operation {self.name}", + is_before) return self.with_operation(decorator(obj)) @@ -135,9 +134,8 @@ class RunsOperationProperty(object): 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)) + decorator = _skip.skip_unless(f"After operation {self.name}", + is_after) return self.with_operation(decorator(obj)) diff --git a/tobiko/common/_skip.py b/tobiko/common/_skip.py index 306fd21d8..c893809b1 100644 --- a/tobiko/common/_skip.py +++ b/tobiko/common/_skip.py @@ -15,70 +15,76 @@ from __future__ import absolute_import import functools import inspect -import unittest import typing # noqa +import fixtures import testtools -from tobiko.common import _fixture - SkipException = testtools.TestCase.skipException # type: typing.Type +SkipTarget = typing.Union[typing.Callable, + typing.Type[testtools.TestCase], + typing.Type[fixtures.Fixture]] +SkipDecorator = typing.Callable[[SkipTarget], SkipTarget] -def skip_test(reason, *args, **kwargs): - if args or kwargs: - reason = reason.format(*args, **kwargs) + +def skip_test(reason: str): + """Interrupt test case execution marking it as skipped for given reason""" raise SkipException(reason) -def skip_if(reason, predicate, *args, **kwargs): - return skip_if_match(reason, bool, predicate, *args, **kwargs) +def skip(reason: str) -> SkipDecorator: + """Mark test case for being skipped for a given reason""" + return _skip_unless(reason, None, True) -def skip_unless(reason, predicate, *args, **kwargs): - return skip_if_match(reason, lambda x: bool(not x), predicate, *args, - **kwargs) +def skip_if(reason: str, function: typing.Callable, *args, **kwargs) -> \ + SkipDecorator: + """Mark test case for being skipped for a given reason if it matches""" + return _skip_unless(reason, function, False, *args, **kwargs) -def skip_if_match(reason, match, predicate, *args, **kwargs): +def skip_unless(reason: str, function: typing.Callable, *args, **kwargs) -> \ + SkipDecorator: + """Mark test case for being skipped for a given reason unless it matches""" + return _skip_unless(reason, function, True, *args, **kwargs) - if not callable(predicate): - args = (predicate,) + args - predicate = bool - def decorator(obj): +def _skip_unless(reason: str, function: typing.Optional[typing.Callable], + unless: bool, *args, **kwargs) -> SkipDecorator: + """Mark test case for being skipped for a given reason unless it matches""" + + def decorator(obj: SkipTarget) -> SkipTarget: method = _get_decorated_method(obj) @functools.wraps(method) - def wrapped_method(*_args, **_kwargs): - return_value = predicate(*args, **kwargs) - if match(return_value): + def skipping_unless(*_args, **_kwargs): + if function is not None: + return_value = function(*args, **kwargs) + if unless is bool(return_value): + return method(*_args, **_kwargs) if '{return_value' in reason: - skip_test(reason, return_value=return_value) - else: - skip_test(reason) - return method(*_args, **_kwargs) + skip_test(reason.format(return_value=return_value)) + skip_test(reason) if obj is method: - return wrapped_method + return skipping_unless else: - setattr(obj, method.__name__, wrapped_method) + setattr(obj, method.__name__, skipping_unless) return obj return decorator -def _get_decorated_method(obj): +def _get_decorated_method(obj: typing.Any) -> typing.Callable: if inspect.isclass(obj): - if issubclass(obj, (unittest.TestCase, testtools.TestCase)): - return obj.setUp - elif _fixture.is_fixture(obj): - return obj.setUp + setup_method = getattr(obj, 'setUp', None) + if callable(setup_method): + return setup_method else: - raise TypeError("Cannot decorate class {!r}".format(obj)) + raise TypeError(f"Class {obj} does not implement setUp method") + elif callable(obj): + return obj else: - if callable(obj): - return obj - else: - raise TypeError("Cannot decorate object {!r}".format(obj)) + raise TypeError(f"Object {obj} is not a class or a function") diff --git a/tobiko/openstack/glance/_lzma.py b/tobiko/openstack/glance/_lzma.py index b514b8af2..607184d68 100644 --- a/tobiko/openstack/glance/_lzma.py +++ b/tobiko/openstack/glance/_lzma.py @@ -39,11 +39,9 @@ def open_file(filename, mode): try: lzma = import_lzma() except ImportError: - tobiko.skip("Package lzma or backports.lzma is required to decompress " - "{filename!r} (mode={mode!r}) XZ image file " - "({python_version!r}).", - filename=filename, - mode=mode, - python_version=sys.version) + tobiko.skip_test( + "Package lzma or backports.lzma is required to decompress " + f"{filename!r} (mode={mode!r}) XZ image file " + f"({sys.version!r}).") return lzma.LZMAFile(filename=filename, mode=mode) diff --git a/tobiko/openstack/stacks/_nova.py b/tobiko/openstack/stacks/_nova.py index 9213612d0..7c12c8dd6 100644 --- a/tobiko/openstack/stacks/_nova.py +++ b/tobiko/openstack/stacks/_nova.py @@ -214,20 +214,20 @@ class ServerStackFixture(heat.HeatStackFixture): different_host_hypervisors = nova.get_different_host_hypervisors( self.same_host, hypervisor) if different_host_hypervisors: - tobiko.skip("server {!r} of stack {!r} created on " - "different hypervisor host from servers:\n{!r}", - self.server_id, self.stack_name, - different_host_hypervisors) + tobiko.skip_test(f"Server {self.server_id} of stack " + f"{self.stack_name} created on different " + "hypervisor host from servers:\n" + f"{different_host_hypervisors}") def validate_different_host_scheduler_hints(self, hypervisor): if self.different_host: same_host_hypervisors = nova.get_same_host_hypervisors( self.different_host, hypervisor) if same_host_hypervisors: - tobiko.skip("server {!r} of stack {!r} created on the same " - "hypervisor host as servers:\n{!r}", - self.server_id, self.stack_name, - same_host_hypervisors) + tobiko.skip_test(f"Server {self.server_id} of stack " + f"{self.stack_name} created on the same " + "hypervisor host as servers:\n" + f"{same_host_hypervisors}") @property def server_details(self): @@ -301,8 +301,8 @@ class ServerStackFixture(heat.HeatStackFixture): tobiko.reset_fixture(self) return nova.wait_for_server_status(self.server_id, 'ACTIVE') else: - tobiko.skip(f"{type(self).__name__}.ensure_server_status " - "method not implemented") + tobiko.skip_test(f"{type(self).__name__}.ensure_server_status " + "method not implemented") class ExternalServerStackFixture(ServerStackFixture): diff --git a/tobiko/tests/faults/containers/container_ops.py b/tobiko/tests/faults/containers/container_ops.py index 760a3f9a9..e82a0c2bf 100644 --- a/tobiko/tests/faults/containers/container_ops.py +++ b/tobiko/tests/faults/containers/container_ops.py @@ -374,7 +374,7 @@ def rotate_logs(node): """ containers = get_filtered_node_containers(node, ['logrotate.*', ]) if not containers: - tobiko.skip('No logrotate container has been found') + tobiko.skip_test('No logrotate container has been found') else: container = containers[0] sh.execute(f'docker exec -u root {container} logrotate ' diff --git a/tobiko/tests/functional/docker/test_client.py b/tobiko/tests/functional/docker/test_client.py index 168078f3b..69ab32e2b 100644 --- a/tobiko/tests/functional/docker/test_client.py +++ b/tobiko/tests/functional/docker/test_client.py @@ -38,8 +38,9 @@ class DockerNodeFixture(tobiko.SharedFixture): break if self.node is None: - tobiko.skip('Docker server is not running in any of nodes {}', - ' '.join(node.name for node in nodes)) + nodes_text = ' '.join(node.name for node in nodes) + tobiko.skip_test("Docker server is not running in any of nodes " + f"{nodes_text}") @keystone.skip_unless_has_keystone_credentials() diff --git a/tobiko/tests/functional/openstack/stacks/test_neutron.py b/tobiko/tests/functional/openstack/stacks/test_neutron.py index d732130d2..eee648a62 100644 --- a/tobiko/tests/functional/openstack/stacks/test_neutron.py +++ b/tobiko/tests/functional/openstack/stacks/test_neutron.py @@ -44,37 +44,38 @@ class NetworkTest(testtools.TestCase): def test_ipv4_subnet_cidr(self): if not self.stack.has_ipv4: - tobiko.skip('Stack {!s} has no ipv4 subnet', self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no ipv4 " + "subnet") subnet = neutron.find_subnet(cidr=str(self.stack.ipv4_subnet_cidr)) self.assertEqual(neutron.get_subnet(self.stack.ipv4_subnet_id), subnet) def test_ipv6_subnet_cidr(self): if not self.stack.has_ipv6: - tobiko.skip('Stack {!s} has no ipv6 subnet', self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no ipv6 " + "subnet") subnet = neutron.find_subnet(cidr=str(self.stack.ipv6_subnet_cidr)) self.assertEqual(neutron.get_subnet(self.stack.ipv6_subnet_id), subnet) def test_gateway_network(self): if not self.stack.has_gateway: - tobiko.skip('Stack {!s} has no gateway', - self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no gateway") self.assertEqual( self.stack.gateway_network_id, self.stack.gateway_details['external_gateway_info']['network_id']) def test_ipv4_subnet_gateway_ip(self): if not self.stack.has_ipv4 or not self.stack.has_gateway: - tobiko.skip('Stack {!s} has no IPv4 gateway', - self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no IPv4 " + "gateway") self.assertIn( self.stack.ipv4_subnet_gateway_ip, self.stack.ipv4_gateway_addresses) def test_ipv6_subnet_gateway_ip(self): if not self.stack.has_ipv6 or not self.stack.has_gateway: - tobiko.skip('Stack {!s} has no IPv6 gateway', - self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no IPv6 " + "gateway") self.assertIn( self.stack.ipv6_subnet_gateway_ip, self.stack.ipv6_gateway_addresses) diff --git a/tobiko/tests/functional/openstack/test_neutron.py b/tobiko/tests/functional/openstack/test_neutron.py index 685ff8c30..550b6b2d1 100644 --- a/tobiko/tests/functional/openstack/test_neutron.py +++ b/tobiko/tests/functional/openstack/test_neutron.py @@ -44,7 +44,7 @@ class NeutronApiTest(testtools.TestCase): def test_find_floating_network(self): floating_network = CONF.tobiko.neutron.floating_network if not floating_network: - tobiko.skip('floating_network not configured') + tobiko.skip_test('floating_network not configured') network = neutron.find_network(name=floating_network) self.assertIn(floating_network, [network['name'], network['id']]) self.assertEqual(self.stack.gateway_network_id, network['id']) @@ -102,23 +102,23 @@ class NeutronApiTest(testtools.TestCase): def test_get_router(self): if not self.stack.has_gateway: - tobiko.skip("Stack {stack} has no gateway router", - stack=self.stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.stack_name} has no gateway " + "router") router = neutron.get_router(self.stack.gateway_id) self.assertEqual(self.stack.gateway_id, router['id']) def test_get_ipv4_subnet(self): if not self.stack.has_ipv4: - tobiko.skip("Stack {stack} has no IPv4 subnet", - stack=self.stack.stack_name) + tobiko.skip_test( + "Stack {self.stack.stack_name} has no IPv4 subnet") subnet = neutron.get_subnet(self.stack.ipv4_subnet_id) self.assertEqual(self.stack.ipv4_subnet_id, subnet['id']) self.assertEqual(self.stack.ipv4_subnet_details, subnet) def test_get_ipv6_subnet(self): if not self.stack.has_ipv6: - tobiko.skip("Stack {stack} has no IPv6 subnet", - stack=self.stack.stack_name) + tobiko.skip_test( + "Stack {self.stack.stack_name} has no IPv6 subnet") subnet = neutron.get_subnet(self.stack.ipv6_subnet_id) self.assertEqual(self.stack.ipv6_subnet_id, subnet['id']) self.assertEqual(self.stack.ipv6_subnet_details, subnet) diff --git a/tobiko/tests/functional/podman/test_client.py b/tobiko/tests/functional/podman/test_client.py index 38dab57cf..acfce8c21 100644 --- a/tobiko/tests/functional/podman/test_client.py +++ b/tobiko/tests/functional/podman/test_client.py @@ -38,8 +38,9 @@ class PodmanNodeFixture(tobiko.SharedFixture): break if self.node is None: - tobiko.skip('Podman server is not running in any of nodes {}', - ' '.join(node.name for node in nodes)) + nodes_text = ' '.join(node.name for node in nodes) + tobiko.skip_test("Podman server is not running in any of nodes " + f"{nodes_text}") @keystone.skip_unless_has_keystone_credentials() diff --git a/tobiko/tests/scenario/neutron/test_network.py b/tobiko/tests/scenario/neutron/test_network.py index 78e570853..cf8cabc13 100644 --- a/tobiko/tests/scenario/neutron/test_network.py +++ b/tobiko/tests/scenario/neutron/test_network.py @@ -57,6 +57,7 @@ class NetworkTest(testtools.TestCase): gateway['ha']) +@tobiko.skip("The test is not able to allocate required resources") class SameHostNetworkTest(NetworkTest): #: Resources stack with Nova server to send messages to @@ -72,6 +73,7 @@ class SameHostNetworkTest(NetworkTest): getattr(receiver, 'OS-EXT-SRV-ATTR:host')) +@tobiko.skip("The test is not able to allocate required resources") @nova.skip_if_missing_hypervisors(count=2, state='up', status='enabled') class DifferentHostNetworkTest(NetworkTest): diff --git a/tobiko/tests/scenario/neutron/test_router.py b/tobiko/tests/scenario/neutron/test_router.py index 60b811579..8f771b8f6 100644 --- a/tobiko/tests/scenario/neutron/test_router.py +++ b/tobiko/tests/scenario/neutron/test_router.py @@ -37,8 +37,8 @@ class RouterTest(testtools.TestCase): def setUp(self): super(RouterTest, self).setUp() if not self.stack.network_stack.has_gateway: - tobiko.skip('Stack {!s} has no gateway', - self.stack.network_stack.stack_name) + tobiko.skip_test( + f"Stack {self.stack.network_stack.stack_name} has no gateway") @property def router(self): @@ -47,15 +47,15 @@ class RouterTest(testtools.TestCase): @property def ipv4_subnet_gateway_ip(self): if not self.stack.network_stack.has_ipv4: - tobiko.skip('Stack {!s} has no ipv4 subnet', - self.stack.network_stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.network_stack.stack_name} " + "has no ipv4 subnet") return self.stack.network_stack.ipv4_subnet_gateway_ip @property def ipv6_subnet_gateway_ip(self): if not self.stack.network_stack.has_ipv6: - tobiko.skip('Stack {!s} has no ipv6 subnet', - self.stack.network_stack.stack_name) + tobiko.skip_test(f"Stack {self.stack.network_stack.stack_name} " + "has no ipv6 subnet") return self.stack.network_stack.ipv6_subnet_gateway_ip @property diff --git a/tobiko/tests/unit/test_fixture.py b/tobiko/tests/unit/test_fixture.py index c6ee9bed3..ba85b99ca 100644 --- a/tobiko/tests/unit/test_fixture.py +++ b/tobiko/tests/unit/test_fixture.py @@ -42,10 +42,10 @@ class MyBaseFixture(tobiko.SharedFixture): class MySkyppingFixture(tobiko.SharedFixture): def setup_fixture(self): - tobiko.skip('some-reason') + tobiko.skip_test('some-reason') def cleanup_fixture(self): - tobiko.skip('some-reason') + tobiko.skip_test('some-reason') class MyFixture(MyBaseFixture): diff --git a/tobiko/tests/unit/test_skip.py b/tobiko/tests/unit/test_skip.py index 6ef6fcea8..e268b491a 100644 --- a/tobiko/tests/unit/test_skip.py +++ b/tobiko/tests/unit/test_skip.py @@ -24,8 +24,8 @@ def condition(value): class PositiveSkipMethodTest(unit.TobikoUnitTest): - @tobiko.skip_if('condition value was true', True) - def test_skip_if_condition(self): + @tobiko.skip('must always skip') + def test_skip(self): self.fail('Not skipped') @tobiko.skip_if('condition value was true', condition, True) @@ -58,10 +58,6 @@ class NegativeSkipBase(unit.TobikoUnitTest): class NegativeSkipMethodTest(NegativeSkipBase): - @tobiko.skip_if('condition value was false', False) - def test_skip_if_conditions(self): - self.test_method_called = True - @tobiko.skip_if('condition value was false', condition, False) def test_skip_if_condition_called_with_args(self): self.test_method_called = True