Import base exceptions from main tobiko module.

Change-Id: Ieb1001acc6fa2cd86ca51a544bd34a4aaf7c78b8

tobiko/__init__.py

@@ -12,10 +12,13 @@
 #    under the License.
 from __future__ import absolute_import
 
+from tobiko.common import _asserts
+from tobiko.common import _exception
 from tobiko.common.managers import fixture
 from tobiko.common.managers import testcase as testcase_manager
 from tobiko.common.managers import loader as loader_manager
-from tobiko.common import exceptions
+from tobiko.common import _skip
+
 
 load_object = loader_manager.load_object
 load_module = loader_manager.load_module
@@ -33,5 +36,12 @@ cleanup_fixture = fixture.cleanup_fixture
 list_required_fixtures = fixture.list_required_fixtures
 SharedFixture = fixture.SharedFixture
 
-TobikoException = exceptions.TobikoException
-FailureException = exceptions.FailureException
+TobikoException = _exception.TobikoException
+
+FailureException = _asserts.FailureException
+fail = _asserts.fail
+
+SkipException = _skip.SkipException
+skip = _skip.skip
+skip_if = _skip.skip_if
+skip_until = _skip.skip_until

tobiko/cmd/create.py

@@ -17,9 +17,10 @@ import os
 import sys
 
 from oslo_log import log
+
+import tobiko
 from tobiko.cmd import base
 from tobiko.common import constants
-from tobiko.common import exceptions
 
 LOG = log.getLogger(__name__)
 
@@ -64,7 +65,7 @@ class CreateUtil(base.TobikoCMD):
         self.ansibleManager.run_playbook(playbook, mode='create')
 
 
-class NoSuchTemplateError(exceptions.TobikoException):
+class NoSuchTemplateError(tobiko.TobikoException):
     message = "No such template. Existing templates:\n%(templates)s"
 
 

tobiko/common/_asserts.py

@@ -0,0 +1,25 @@
+# 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 testtools
+
+
+FailureException = testtools.TestCase.failureException
+
+
+def fail(reason, *args, **kwargs):
+    if args or kwargs:
+        reason = reason.format(*args, **kwargs)
+    raise FailureException(reason)

tobiko/common/_exception.py

@@ -13,8 +13,6 @@
 #    under the License.
 from __future__ import absolute_import
 
-import testtools
-
 
 class TobikoException(Exception):
     """Base Tobiko Exception.
@@ -42,7 +40,3 @@ class TobikoException(Exception):
         except KeyError:
             msg = ("{!r} object has no attribute {!r}").format(self, name)
             raise AttributeError(msg)
-
-
-class FailureException(TobikoException, testtools.TestCase.failureException):
-    pass

tobiko/common/_skip.py

@@ -0,0 +1,55 @@
+# 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 functools
+
+import testtools
+
+
+SkipException = testtools.TestCase.skipException
+
+
+def skip(reason):
+    raise SkipException(reason)
+
+
+def skip_if(reason, condition, *condition_args, **condition_kwargs):
+
+    def decorator(method):
+
+        @functools.wraps(method)
+        def wrapped_method(*args, **kwargs):
+            if condition(*condition_args, **condition_kwargs):
+                skip(reason)
+            return method(*args, **kwargs)
+
+        return wrapped_method
+
+    return decorator
+
+
+def skip_until(reason, condition, *condition_args, **condition_kwargs):
+
+    def decorator(method):
+
+        @functools.wraps(method)
+        def wrapped_method(*args, **kwargs):
+            if not condition(*condition_args, **condition_kwargs):
+                skip(reason)
+            return method(*args, **kwargs)
+
+        return wrapped_method
+
+    return decorator

tobiko/common/asserts.py

@@ -1,35 +0,0 @@
-# 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 testtools
-
-import tobiko.common.utils.network as net_utils
-from tobiko.common import exceptions
-
-
-class PingFailed(exceptions.TobikoException,
-                 testtools.TestCase.failureException):
-    message = "Failed pinging %(destination)r: %(reason)s"
-
-
-def assert_ping(ip, should_fail=False, fragmentation=True,
-                packet_size=None):
-    success = net_utils.ping_ip_address(ip, mtu=packet_size,
-                                        fragmentation=fragmentation)
-    if success:
-        if should_fail:
-            raise PingFailed(destination=ip, reason="IP address is reachable")
-    elif not should_fail:
-        raise PingFailed(destination=ip, reason="IP address is not reachable")

tobiko/common/managers/stack.py

@@ -21,8 +21,8 @@ from heatclient import exc
 from oslo_log import log
 import yaml
 
+import tobiko
 from tobiko.common import constants
-from tobiko.common import exceptions
 from tobiko.openstack import heat
 
 
@@ -186,15 +186,15 @@ def check_stack_status(stack, expected):
                           reason=stack.stack_status_reason)
 
 
-class InvalidOutputKey(exceptions.TobikoException):
+class InvalidOutputKey(tobiko.TobikoException):
     msg = ("Output key %(key)r not found in stack %(name).")
 
 
-class StackNotFound(exceptions.TobikoException):
+class StackNotFound(tobiko.TobikoException):
     msg = ("Stack %(name)r not found")
 
 
-class InvalidStackStatus(exceptions.TobikoException):
+class InvalidStackStatus(tobiko.TobikoException):
     msg = ("Stack %(name)r status %(observed)r not in %(expected)r "
            "(reason=%(status_reason)r)")
 

tobiko/common/utils/network.py

@@ -24,6 +24,7 @@ from oslo_log import log
 from tempest.common.utils import net_utils
 from tempest.lib.common.utils import test_utils
 
+import tobiko
 from tobiko import config
 
 LOG = log.getLogger(__name__)
@@ -126,3 +127,15 @@ def ping_ip_address(ip_address, should_succeed=True,
             return not should_succeed
 
     return test_utils.call_until_true(ping, timeout, 1)
+
+
+def assert_ping(ip, should_fail=False, fragmentation=True,
+                packet_size=None):
+    success = ping_ip_address(ip, mtu=packet_size,
+                              fragmentation=fragmentation)
+    if success:
+        if should_fail:
+            tobiko.fail("Host {!r} is reachable", ip)
+
+    elif not should_fail:
+        tobiko.fail("Host {!r} is not reachable", ip)

tobiko/tests/scenario/neutron/test_floating_ip.py

@@ -16,7 +16,7 @@ from __future__ import absolute_import
 
 import tobiko
 from tobiko import config
-from tobiko.common import asserts
+from tobiko.common.utils import network
 from tobiko.openstack import heat
 from tobiko.tests.scenario.neutron import base
 
@@ -68,23 +68,23 @@ class FloatingIPTest(base.NeutronTest):
     def test_ping_floating_ip(self, fixture_type=FloatingIPFixture):
         """Validates connectivity to a server post upgrade."""
         stack = self.setup_fixture(fixture_type)
-        asserts.assert_ping(stack.outputs.floating_ip_address)
+        network.assert_ping(stack.outputs.floating_ip_address)
 
     def test_ping_floating_ip_with_port_security(
             self, fixture_type=FloatingIPWithPortSecurityFixture):
         """Validates connectivity to a server post upgrade."""
         stack = self.setup_fixture(fixture_type)
-        asserts.assert_ping(stack.outputs.floating_ip_address,
+        network.assert_ping(stack.outputs.floating_ip_address,
                             should_fail=True)
 
     def test_ping_floating_ip_with_security_group(
             self, fixture_type=FloatingIPWithSecurityGroupFixture):
         """Validates connectivity to a server post upgrade."""
         stack = self.setup_fixture(fixture_type)
-        asserts.assert_ping(stack.outputs.floating_ip_address)
+        network.assert_ping(stack.outputs.floating_ip_address)
 
     def test_ping_with_oversize_packet(self, fixture_type=FloatingIPFixture):
         stack = self.setup_fixture(fixture_type)
-        asserts.assert_ping(stack.outputs.floating_ip_address,
+        network.assert_ping(stack.outputs.floating_ip_address,
                             packet_size=stack.internal_network.mtu + 1,
                             fragmentation=False, should_fail=True)

tobiko/tests/test_skip.py

@@ -0,0 +1,77 @@
+# Copyright (c) 2019 Red Hat
+# All Rights Reserved.
+#
+#    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 tobiko
+from tobiko.tests import unit
+
+
+def condition(value):
+    return value
+
+
+class PositiveSkipTest(unit.TobikoUnitTest):
+
+    @tobiko.skip_if('condition value was true',
+                    condition, True)
+    def test_skip_if_condition_called_with_args(self):
+        self.fail('Not skipped')
+
+    @tobiko.skip_if('condition value was true',
+                    condition, value=True)
+    def test_skip_if_condition_called_with_kwargs(self):
+        self.fail('Not skipped')
+
+    @tobiko.skip_until('condition value was false',
+                       condition, False)
+    def test_skip_until_condition_called_with_args(self):
+        self.fail('Not skipped')
+
+    @tobiko.skip_until('condition value was false',
+                       condition, value=False)
+    def test_skip_until_condition_called_with_kwargs(self):
+        self.fail('Not skipped')
+
+
+class NegativeSkipTest(unit.TobikoUnitTest):
+
+    test_method_called = False
+
+    def setUp(self):
+        super(NegativeSkipTest, self).setUp()
+        self.addCleanup(self.assert_test_method_called)
+
+    def assert_test_method_called(self):
+        self.assertTrue(self.test_method_called)
+
+    @tobiko.skip_if('condition value was false',
+                    condition, False)
+    def test_skip_if_condition_called_with_args(self):
+        self.test_method_called = True
+
+    @tobiko.skip_if('condition value was false',
+                    condition, value=False)
+    def test_skip_if_condition_called_with_kwargs(self):
+        self.test_method_called = True
+
+    @tobiko.skip_until('condition value was true',
+                       condition, True)
+    def test_skip_until_condition_called_with_args(self):
+        self.test_method_called = True
+
+    @tobiko.skip_until('condition value was true',
+                       condition, value=True)
+    def test_skip_until_condition_called_with_kwargs(self):
+        self.test_method_called = True