Add rabbit unit test for sending and receiving replies

Change-Id: I9574940904673257317a0caa86c585459e066ff7

oslo/messaging/_drivers/amqpdriver.py

@@ -204,7 +204,7 @@ class ReplyWaiter(object):
         # thread, it wakes up the other thread
         final_reply = None
         while True:
-            if self.conn_lock.acquire(blocking=False):
+            if self.conn_lock.acquire(False):
                 try:
                     reply, ending = self._poll_connection(msg_id)
                     if reply:

tests/test_rabbit.py

@@ -15,6 +15,7 @@
 #    under the License.
 
 import datetime
+import threading
 import uuid
 
 import fixtures
@@ -54,15 +55,28 @@ class TestRabbitDriver(test_utils.BaseTestCase):
 
         listener = driver.listen(target)
 
-        ctxt = {}
-        message = {'foo': 'bar'}
+        replies = []
 
-        driver.send(target, ctxt, message)
+        def send_and_wait_for_reply():
+            replies.append(driver.send(target,
+                                       {},
+                                       {'foo': 'bar'},
+                                       wait_for_reply=True))
+
+        sender = threading.Thread(target=send_and_wait_for_reply)
+        sender.start()
 
         received = listener.poll()
         self.assertTrue(received is not None)
-        self.assertEquals(received.ctxt, {})
-        self.assertEquals(received.message, {'foo': 'bar'})
+        self.assertEqual(received.ctxt, {})
+        self.assertEqual(received.message, {'foo': 'bar'})
+
+        received.reply({'bar': 'foo'})
+
+        sender.join()
+
+        self.assertEqual(len(replies), 1)
+        self.assertEqual(replies[0], {'bar': 'foo'})
 
 
 def _declare_queue(target):