x/gantt
Code Issues Proposed changes

Make kombu support optional for running unit tests.

Browse Source 
Fix bug 838386.

Previously, the unit tests required kombu to be installed to run.  This
patch makes it so the unit tests will still complete successfully, even
if kombu is not installed.

Change-Id: I06831a37f8c857bbfd0bce1b1c3aeda7d341a589
This commit is contained in:
Russell Bryant
2012-05-02 12:14:56 -04:00
parent e24f416726
commit e5370131c4
3 changed files with 77 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
nova/tests/rpc/common.py
View File

@@ -40,24 +40,32 @@ LOG = logging.getLogger(__name__)
class BaseRpcTestCase(test.TestCase):
def setUp(self, supports_timeouts=True):
super(BaseRpcTestCase, self).setUp()
self.conn = self.rpc.create_connection(FLAGS, True)
self.receiver = TestReceiver()
self.conn.create_consumer('test', self.receiver, False)
self.conn.consume_in_thread()
self.context = context.get_admin_context()
self.supports_timeouts = supports_timeouts
self.context = context.get_admin_context()
if self.rpc:
self.conn = self.rpc.create_connection(FLAGS, True)
self.receiver = TestReceiver()
self.conn.create_consumer('test', self.receiver, False)
self.conn.consume_in_thread()
def tearDown(self):
self.conn.close()
if self.rpc:
self.conn.close()
super(BaseRpcTestCase, self).tearDown()
def test_call_succeed(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
value = 42
result = self.rpc.call(FLAGS, self.context, 'test',
{"method": "echo", "args": {"value": value}})
self.assertEqual(value, result)
def test_call_succeed_despite_multiple_returns_yield(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
value = 42
result = self.rpc.call(FLAGS, self.context, 'test',
{"method": "echo_three_times_yield",
@@ -65,6 +73,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(value + 2, result)
def test_multicall_succeed_once(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
value = 42
result = self.rpc.multicall(FLAGS, self.context,
'test',
@@ -76,6 +87,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(value + i, x)
def test_multicall_three_nones(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
value = 42
result = self.rpc.multicall(FLAGS, self.context,
'test',
@@ -87,6 +101,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(i, 2)
def test_multicall_succeed_three_times_yield(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
value = 42
result = self.rpc.multicall(FLAGS, self.context,
'test',
@@ -96,6 +113,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(value + i, x)
def test_context_passed(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
"""Makes sure a context is passed through rpc call."""
value = 42
result = self.rpc.call(FLAGS, self.context,
@@ -104,6 +124,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(self.context.to_dict(), result)
def test_nested_calls(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
"""Test that we can do an rpc.call inside another call."""
class Nested(object):
@staticmethod
@@ -134,6 +157,9 @@ class BaseRpcTestCase(test.TestCase):
self.assertEqual(value, result)
def test_call_timeout(self):
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
"""Make sure rpc.call will time out"""
if not self.supports_timeouts:
raise nose.SkipTest(_("RPC backend does not support timeouts"))
@@ -160,6 +186,9 @@ class BaseRpcAMQPTestCase(BaseRpcTestCase):
"""Base test class for all AMQP-based RPC tests"""
def test_proxycallback_handles_exceptions(self):
"""Make sure exceptions unpacking messages don't cause hangs."""
if not self.rpc:
raise nose.SkipTest('rpc driver not available.')
orig_unpack = rpc_amqp.unpack_context
info = {'unpacked': False}

30
nova/tests/rpc/test_kombu.py
View File

@@ -25,9 +25,16 @@ from nova import flags
from nova import log as logging
from nova import test
from nova.rpc import amqp as rpc_amqp
from nova.rpc import impl_kombu
from nova.tests.rpc import common
try:
import kombu
from nova.rpc import impl_kombu
except ImportError:
kombu = None
impl_kombu = None
FLAGS = flags.FLAGS
LOG = logging.getLogger(__name__)
@@ -52,14 +59,19 @@ def _raise_exc_stub(stubs, times, obj, method, exc_msg,
class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
def setUp(self):
self.rpc = impl_kombu
impl_kombu.register_opts(FLAGS)
if kombu:
self.rpc = impl_kombu
impl_kombu.register_opts(FLAGS)
else:
self.rpc = None
super(RpcKombuTestCase, self).setUp()
def tearDown(self):
impl_kombu.cleanup()
if kombu:
impl_kombu.cleanup()
super(RpcKombuTestCase, self).tearDown()
@test.skip_if(kombu is None, "Test requires kombu")
def test_reusing_connection(self):
"""Test that reusing a connection returns same one."""
conn_context = self.rpc.create_connection(FLAGS, new=False)
@@ -70,6 +82,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
conn_context.close()
self.assertEqual(conn1, conn2)
@test.skip_if(kombu is None, "Test requires kombu")
def test_topic_send_receive(self):
"""Test sending to a topic exchange/queue"""
@@ -88,6 +101,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(self.received_message, message)
@test.skip_if(kombu is None, "Test requires kombu")
def test_direct_send_receive(self):
"""Test sending to a direct exchange/queue"""
conn = self.rpc.create_connection(FLAGS)
@@ -105,6 +119,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(self.received_message, message)
@test.skip_if(kombu is None, "Test requires kombu")
def test_cast_interface_uses_default_options(self):
"""Test kombu rpc.cast"""
@@ -129,6 +144,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
impl_kombu.cast(FLAGS, ctxt, 'fake_topic', {'msg': 'fake'})
@test.skip_if(kombu is None, "Test requires kombu")
def test_cast_to_server_uses_server_params(self):
"""Test kombu rpc.cast"""
@@ -187,6 +203,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
conn2.close()
self.assertEqual(self.received_message, message)
@test.skip_if(kombu is None, "Test requires kombu")
def test_declare_consumer_errors_will_reconnect(self):
# Test that any exception with 'timeout' in it causes a
# reconnection
@@ -216,6 +233,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(info['called'], 2)
self.assertTrue(isinstance(result, self.rpc.DirectConsumer))
@test.skip_if(kombu is None, "Test requires kombu")
def test_declare_consumer_ioerrors_will_reconnect(self):
"""Test that an IOError exception causes a reconnection"""
info = _raise_exc_stub(self.stubs, 2, self.rpc.DirectConsumer,
@@ -228,6 +246,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(info['called'], 3)
self.assertTrue(isinstance(result, self.rpc.DirectConsumer))
@test.skip_if(kombu is None, "Test requires kombu")
def test_publishing_errors_will_reconnect(self):
# Test that any exception with 'timeout' in it causes a
# reconnection when declaring the publisher class and when
@@ -275,6 +294,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(info['called'], 2)
@test.skip_if(kombu is None, "Test requires kombu")
def test_iterconsume_errors_will_reconnect(self):
conn = self.rpc.Connection(FLAGS)
message = 'reconnect test message'
@@ -295,6 +315,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
self.assertEqual(self.received_message, message)
# Only called once, because our stub goes away during reconnection
@test.skip_if(kombu is None, "Test requires kombu")
def test_call_exception(self):
"""Test that exception gets passed back properly.
@@ -322,6 +343,7 @@ class RpcKombuTestCase(common.BaseRpcAMQPTestCase):
#Traceback should be included in exception message
self.assertTrue('raise NotImplementedError(value)' in unicode(exc))
@test.skip_if(kombu is None, "Test requires kombu")
def test_call_converted_exception(self):
"""Test that exception gets passed back properly.

23
nova/tests/rpc/test_kombu_ssl.py
View File

@@ -21,7 +21,14 @@ Unit Tests for remote procedure calls using kombu + ssl
from nova import flags
from nova import test
from nova.rpc import impl_kombu
try:
import kombu
from nova.rpc import impl_kombu
except ImportError:
kombu = None
impl_kombu = None
# Flag settings we will ensure get passed to amqplib
SSL_VERSION = "SSLv2"
@@ -36,13 +43,15 @@ class RpcKombuSslTestCase(test.TestCase):
def setUp(self):
super(RpcKombuSslTestCase, self).setUp()
impl_kombu.register_opts(FLAGS)
self.flags(kombu_ssl_keyfile=SSL_KEYFILE,
kombu_ssl_ca_certs=SSL_CA_CERT,
kombu_ssl_certfile=SSL_CERT,
kombu_ssl_version=SSL_VERSION,
rabbit_use_ssl=True)
if kombu:
impl_kombu.register_opts(FLAGS)
self.flags(kombu_ssl_keyfile=SSL_KEYFILE,
kombu_ssl_ca_certs=SSL_CA_CERT,
kombu_ssl_certfile=SSL_CERT,
kombu_ssl_version=SSL_VERSION,
rabbit_use_ssl=True)
@test.skip_if(kombu is None, "Test requires kombu")
def test_ssl_on_extended(self):
rpc = impl_kombu
conn = rpc.create_connection(FLAGS, True)