Merge "Don't block forever for rpc.(multi)call response."

3 files changed, 42 insertions, 7 deletions

diff --git a/nova/tests/rpc/common.py b/nova/tests/rpc/common.py
index dc8aafcfe..c41375ace 100644
--- a/nova/tests/rpc/common.py
+++ b/nova/tests/rpc/common.py
@@ -19,9 +19,13 @@
 Unit Tests for remote procedure calls shared between all implementations
 """
 
+import time
+
+import nose
+
 from nova import context
 from nova import log as logging
-from nova.rpc.common import RemoteError
+from nova.rpc.common import RemoteError, Timeout
 from nova import test
 
 
@@ -29,13 +33,14 @@ LOG = logging.getLogger('nova.tests.rpc')
 
 
 class _BaseRpcTestCase(test.TestCase):
-    def setUp(self):
+    def setUp(self, supports_timeouts=True):
         super(_BaseRpcTestCase, self).setUp()
         self.conn = self.rpc.create_connection(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
 
     def tearDown(self):
         self.conn.close()
@@ -162,6 +167,28 @@ class _BaseRpcTestCase(test.TestCase):
         conn.close()
         self.assertEqual(value, result)
 
+    def test_call_timeout(self):
+        """Make sure rpc.call will time out"""
+        if not self.supports_timeouts:
+            raise nose.SkipTest(_("RPC backend does not support timeouts"))
+
+        value = 42
+        self.assertRaises(Timeout,
+                          self.rpc.call,
+                          self.context,
+                          'test',
+                          {"method": "block",
+                           "args": {"value": value}}, timeout=1)
+        try:
+            self.rpc.call(self.context,
+                     'test',
+                     {"method": "block",
+                      "args": {"value": value}},
+                     timeout=1)
+            self.fail("should have thrown Timeout")
+        except Timeout as exc:
+            pass
+
 
 class TestReceiver(object):
     """Simple Proxy class so the consumer has methods to call.
@@ -205,3 +232,7 @@ class TestReceiver(object):
     def fail(context, value):
         """Raises an exception with the value sent in."""
         raise Exception(value)
+
+    @staticmethod
+    def block(context, value):
+        time.sleep(2)
diff --git a/nova/tests/rpc/test_carrot.py b/nova/tests/rpc/test_carrot.py
index 2523810d8..153747da2 100644
--- a/nova/tests/rpc/test_carrot.py
+++ b/nova/tests/rpc/test_carrot.py
@@ -30,7 +30,7 @@ LOG = logging.getLogger('nova.tests.rpc')
 class RpcCarrotTestCase(common._BaseRpcTestCase):
     def setUp(self):
         self.rpc = impl_carrot
-        super(RpcCarrotTestCase, self).setUp()
+        super(RpcCarrotTestCase, self).setUp(supports_timeouts=False)
 
     def tearDown(self):
         super(RpcCarrotTestCase, self).tearDown()
diff --git a/nova/tests/rpc/test_qpid.py b/nova/tests/rpc/test_qpid.py
index 0417674b8..9e318fbfd 100644
--- a/nova/tests/rpc/test_qpid.py
+++ b/nova/tests/rpc/test_qpid.py
@@ -221,21 +221,25 @@ class RpcQpidTestCase(test.TestCase):
         self.mock_session.sender(send_addr).AndReturn(self.mock_sender)
         self.mock_sender.send(mox.IgnoreArg())
 
-        self.mock_session.next_receiver().AndReturn(self.mock_receiver)
+        self.mock_session.next_receiver(timeout=mox.IsA(int)).AndReturn(
+                                                        self.mock_receiver)
         self.mock_receiver.fetch().AndReturn(qpid.messaging.Message(
                         {"result": "foo", "failure": False, "ending": False}))
         if multi:
-            self.mock_session.next_receiver().AndReturn(self.mock_receiver)
+            self.mock_session.next_receiver(timeout=mox.IsA(int)).AndReturn(
+                                                        self.mock_receiver)
             self.mock_receiver.fetch().AndReturn(
                             qpid.messaging.Message(
                                 {"result": "bar", "failure": False,
                                  "ending": False}))
-            self.mock_session.next_receiver().AndReturn(self.mock_receiver)
+            self.mock_session.next_receiver(timeout=mox.IsA(int)).AndReturn(
+                                                        self.mock_receiver)
             self.mock_receiver.fetch().AndReturn(
                             qpid.messaging.Message(
                                 {"result": "baz", "failure": False,
                                  "ending": False}))
-        self.mock_session.next_receiver().AndReturn(self.mock_receiver)
+        self.mock_session.next_receiver(timeout=mox.IsA(int)).AndReturn(
+                                                        self.mock_receiver)
         self.mock_receiver.fetch().AndReturn(qpid.messaging.Message(
                         {"failure": False, "ending": True}))
         self.mock_session.close()