Browse Source

Fix Python 3 testing

Raising SkipTest at import time fails in recent version of the test
suite. We change that by being more picky at one should or should not be
tested based on what we managed to import; the upside is that we test a
few more things under Python 3 now.

Change-Id: I8971c7291cb261faf00fec73cf53b6132bdaf948
changes/06/116606/6
Julien Danjou 7 years ago
parent
commit
48a9ba4e27
  1. 7
      tests/drivers/test_impl_qpid.py
  2. 23
      tests/executors/test_executor.py
  3. 11
      tests/test_opts.py

7
tests/drivers/test_impl_qpid.py

@ -16,15 +16,15 @@ import operator
import random
import threading
import time
import unittest
import mock
try:
import qpid
except ImportError:
raise unittest.SkipTest("qpid not available")
qpid = None
from six.moves import _thread
import testscenarios
import testtools
from oslo import messaging
from oslo.messaging._drivers import impl_qpid as qpid_driver
@ -67,6 +67,7 @@ def _is_qpidd_service_running():
class _QpidBaseTestCase(test_utils.BaseTestCase):
@testtools.skipIf(qpid is None, "qpid not available")
def setUp(self):
super(_QpidBaseTestCase, self).setUp()
self.messaging_conf.transport_driver = 'qpid'
@ -548,6 +549,7 @@ class TestQpidReconnectOrder(test_utils.BaseTestCase):
"""Unit Test cases to test reconnection
"""
@testtools.skipIf(qpid is None, "qpid not available")
def test_reconnect_order(self):
brokers = ['host1', 'host2', 'host3', 'host4', 'host5']
brokers_count = len(brokers)
@ -768,6 +770,7 @@ def get_fake_qpid_session():
class QPidHATestCase(test_utils.BaseTestCase):
@testtools.skipIf(qpid is None, "qpid not available")
def setUp(self):
super(QPidHATestCase, self).setUp()
self.brokers = ['host1', 'host2', 'host3', 'host4', 'host5']

23
tests/executors/test_executor.py

@ -16,17 +16,20 @@
import contextlib
import threading
import unittest
try:
import eventlet
except ImportError:
raise unittest.SkipTest("Eventlet not available")
eventlet = None
import mock
import testscenarios
import testtools
from oslo.messaging._executors import impl_blocking
from oslo.messaging._executors import impl_eventlet
try:
from oslo.messaging._executors import impl_eventlet
except ImportError:
impl_eventlet = None
from tests import utils as test_utils
load_tests = testscenarios.load_tests_apply_scenarios
@ -34,14 +37,15 @@ load_tests = testscenarios.load_tests_apply_scenarios
class TestExecutor(test_utils.BaseTestCase):
_impl = [('blocking', dict(executor=impl_blocking.BlockingExecutor,
stop_before_return=True)),
('eventlet', dict(executor=impl_eventlet.EventletExecutor,
stop_before_return=False))]
@classmethod
def generate_scenarios(cls):
cls.scenarios = testscenarios.multiply_scenarios(cls._impl)
impl = [('blocking', dict(executor=impl_blocking.BlockingExecutor,
stop_before_return=True))]
if impl_eventlet is not None:
impl.append(
('eventlet', dict(executor=impl_eventlet.EventletExecutor,
stop_before_return=False)))
cls.scenarios = testscenarios.multiply_scenarios(impl)
@staticmethod
def _run_in_thread(executor):
@ -90,6 +94,7 @@ class ExceptedException(Exception):
class EventletContextManagerSpawnTest(test_utils.BaseTestCase):
@testtools.skipIf(impl_eventlet is None, "Eventlet not available")
def setUp(self):
super(EventletContextManagerSpawnTest, self).setUp()
self.before = mock.Mock()

11
tests/test_opts.py

@ -12,21 +12,22 @@
# 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 unittest
import pkg_resources
import testtools
try:
from oslo.messaging import opts
except ImportError:
import six
if six.PY3:
raise unittest.SkipTest
opts = None
from tests import utils as test_utils
class OptsTestCase(test_utils.BaseTestCase):
@testtools.skipIf(opts is None, "Options not importable")
def setUp(self):
super(OptsTestCase, self).setUp()
def _test_list_opts(self, result):
self.assertEqual(3, len(result))

Loading…
Cancel
Save