From 950c37c595f6f8a58008ac9d5c29735fed1d9295 Mon Sep 17 00:00:00 2001
From: Mark McLoughlin <markmc@redhat.com>
Date: Thu, 1 Aug 2013 08:04:57 +0100
Subject: [PATCH] Add rabbit unit test for sending and receiving replies
Change-Id: I9574940904673257317a0caa86c585459e066ff7
---
oslo/messaging/_drivers/amqpdriver.py | 2 +-
tests/test_rabbit.py | 24 +++++++++++++++++++-----
2 files changed, 20 insertions(+), 6 deletions(-)
diff --git a/oslo/messaging/_drivers/amqpdriver.py b/oslo/messaging/_drivers/amqpdriver.py
index 5a84c1a89..6cd35bb03 100644
--- a/oslo/messaging/_drivers/amqpdriver.py
+++ b/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:
diff --git a/tests/test_rabbit.py b/tests/test_rabbit.py
index 8266e6b0c..9ea084547 100644
--- a/tests/test_rabbit.py
+++ b/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):