Implement failure replies in the fake driver

Change-Id: Ifd9ede7cb17a471ae2f9024b49ef6bbdc645476a

oslo/messaging/_drivers/impl_fake.py

@@ -32,9 +32,8 @@ class FakeIncomingMessage(base.IncomingMessage):
         self._reply_q = reply_q
 
     def reply(self, reply=None, failure=None, log_failure=True):
-        # FIXME: handle failure
         if self._reply_q:
-            self._reply_q.put(reply)
+            self._reply_q.put((reply, failure))
 
 
 class FakeListener(base.Listener):
@@ -129,7 +128,11 @@ class FakeDriver(base.BaseDriver):
 
         if wait_for_reply:
             try:
-                return reply_q.get(timeout=timeout)
+                reply, failure = reply_q.get(timeout=timeout)
+                if failure:
+                    raise failure
+                else:
+                    return reply
             except Queue.Empty:
                 raise messaging.MessagingTimeout(
                     'No reply on topic %s' % target.topic)

tests/test_rpc_server.py

@@ -258,6 +258,35 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
 
         self._stop_server(client, server_thread)
 
+    def test_failure(self):
+        transport = messaging.get_transport(self.conf, url='fake:')
+
+        class TestEndpoint(object):
+            def ping(self, ctxt, arg):
+                raise ValueError
+
+        server_thread = self._setup_server(transport, TestEndpoint())
+        client = self._setup_client(transport)
+
+        self.assertRaises(ValueError, client.call, {}, 'ping', arg='foo')
+
+        self._stop_server(client, server_thread)
+
+    def test_expected_failure(self):
+        transport = messaging.get_transport(self.conf, url='fake:')
+
+        class TestEndpoint(object):
+            @messaging.expected_exceptions(ValueError)
+            def ping(self, ctxt, arg):
+                raise ValueError
+
+        server_thread = self._setup_server(transport, TestEndpoint())
+        client = self._setup_client(transport)
+
+        self.assertRaises(ValueError, client.call, {}, 'ping', arg='foo')
+
+        self._stop_server(client, server_thread)
+
 
 class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):