diff --git a/LCS/Messaging/python/messaging/messagebus.py b/LCS/Messaging/python/messaging/messagebus.py
index 2f6160dac32afda18689f2b8576e9139bf576ad6..827cf0857b51947907b7aeb56ba34ed80225e41a 100644
--- a/LCS/Messaging/python/messaging/messagebus.py
+++ b/LCS/Messaging/python/messaging/messagebus.py
@@ -1402,7 +1402,8 @@ class BusListener:
create_bound_queue(exchange=exchange, queue=self.address, routing_key=routing_key,
broker=self.broker, log_level=logging.INFO)
- def designated_queue_name(self, exchange: str, routing_key: str) -> str:
+ @staticmethod
+ def designated_queue_name(exchange: str, routing_key: str="#") -> str:
"""
create a designated queue name based on the given exchange name, routing_key, and the current running program name.
Like so: <exchange>.for.<program_name>.<listener_type_name>.on.<sanitzed_routing_key>
@@ -1416,7 +1417,7 @@ class BusListener:
sanitized_routing_key = "all"
return "%s.queue.for.%s.%s.on.%s" % (exchange,
program_name(include_extension=False),
- self.__class__.__name__,
+ __class__.__name__,
sanitized_routing_key)
def is_running(self) -> bool:
diff --git a/LCS/Messaging/python/messaging/test/t_RPC.py b/LCS/Messaging/python/messaging/test/t_RPC.py
index e51786b0b8d592d2f3f998e852caa9c1fffbbd9a..cf1faa219764e6088172c2a8d6fa3c564d895d5f 100644
--- a/LCS/Messaging/python/messaging/test/t_RPC.py
+++ b/LCS/Messaging/python/messaging/test/t_RPC.py
@@ -11,7 +11,7 @@ import unittest
import uuid
from time import sleep
-from lofar.messaging.messagebus import TemporaryExchange, can_connect_to_broker
+from lofar.messaging.messagebus import TemporaryExchange, can_connect_to_broker, exchange_exists, queue_exists
from lofar.messaging.rpc import RPCClient, RPCService, RPCException, RPCTimeoutException, ServiceMessageHandler
TEST_SERVICE_NAME = "%s.%s" % (__name__, uuid.uuid4())
@@ -46,12 +46,14 @@ class TestRPC(unittest.TestCase):
def test_rpc_client_to_service_call(self):
with TemporaryExchange(__name__) as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
with RPCService(TEST_SERVICE_NAME,
handler_type=MyServiceMessageHandler,
handler_kwargs={'my_arg1': "foo",
'my_arg2': "bar"},
exchange=tmp_exchange.address,
num_threads=1) as service:
+ service_queue_address = service.address
self.assertTrue(service.is_listening())
self.assertTrue(service.is_running())
@@ -68,6 +70,9 @@ class TestRPC(unittest.TestCase):
with self.assertRaises(RPCTimeoutException):
rpc_client.execute("my_public_slow_method")
+ self.assertFalse(queue_exists(service_queue_address))
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+
if __name__ == '__main__':
if not can_connect_to_broker():
logger.error("Cannot connect to default rabbitmq broker. Skipping test.")
diff --git a/LCS/Messaging/python/messaging/test/t_messagebus.py b/LCS/Messaging/python/messaging/test/t_messagebus.py
index 3e08e839fdf7539573d56db1551b98735161b8e1..c13961d73bfccaddedd83fc7bbd8ce4b60cb0b78 100644
--- a/LCS/Messaging/python/messaging/test/t_messagebus.py
+++ b/LCS/Messaging/python/messaging/test/t_messagebus.py
@@ -127,6 +127,8 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
tmp_exchange_address = tmp_exchange.address
self.assertTrue("MyTestExchange" in tmp_exchange_address)
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+
# test if the temporary exchange has been deleted when leaving scope
# We should not be able to connect to it anymore
with self.assertRaisesRegex(MessageBusError, '.*NOT_FOUND.*'):
@@ -142,6 +144,8 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
tmp_queue_address = tmp_queue.address
self.assertTrue("MyTestQueue" in tmp_queue_address)
+ self.assertFalse(queue_exists(tmp_queue_address))
+
# test if the temporary queue has been deleted when leaving scope
# We should not be able to connect to it anymore
with self.assertRaisesRegex(MessageBusError, '.*NOT_FOUND.*'):
@@ -153,10 +157,12 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
test the usage of the TemporaryExchange and TemporaryQueue in conjunction with normal ToBus and Frombus usage
"""
with TemporaryExchange("MyTestExchange") as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
# create a normal ToBus on this tmp_exchange
with tmp_exchange.create_tobus() as tobus_on_exchange:
# create a TemporaryQueue, bound to the tmp_exchange
with TemporaryQueue("MyTestQueue", exchange=tmp_exchange.address) as tmp_queue:
+ tmp_queue_address = tmp_queue.address
# create a normal FromBus on this tmp_queue
with tmp_queue.create_frombus() as frombus:
# and let's see if the tmp_queue can also create a tobus which then points to the bound_exchange
@@ -178,6 +184,9 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
self.assertEqual(original_msg.id, received_msg.id)
self.assertEqual(original_msg.content, received_msg.content)
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(tmp_queue_address))
+
def test_send_receive_over_temporary_queue_with_subject_filtering(self):
"""
test the usage of the TemporaryQueue in conjunction with normal ToBus and Frombus usage with additional filtering on subject
@@ -186,6 +195,7 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
SUBJECT2 = "FAKE_SUBJECT"
with TemporaryQueue("MyTestQueue", routing_key=SUBJECT) as tmp_queue:
+ tmp_queue_address = tmp_queue.address
# create a normal To/FromBus on this tmp_queue
NUM_MESSAGES_TO_SEND = 3
with tmp_queue.create_tobus() as tobus:
@@ -217,6 +227,8 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
logger.info("received message: %s", received_msg)
self.assertEqual(None, received_msg)
+ self.assertFalse(queue_exists(tmp_queue_address))
+
def test_send_receive_over_temporary_exchange_with_queue_with_subject_filtering(self):
"""
test the usage of the TemporaryQueue in conjunction with normal ToBus and Frombus usage with additional filtering on subject
@@ -225,9 +237,11 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
SUBJECT2 = "FAKE_SUBJECT"
NUM_MESSAGES_TO_SEND = 3
with TemporaryExchange("MyTestExchange") as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
with tmp_exchange.create_tobus() as tobus:
# create a TemporaryQueue, which listens for/receives only the messages with the given SUBJECT
with TemporaryQueue("MyTestQueue", exchange=tmp_exchange.address, routing_key=SUBJECT) as tmp_queue:
+ tmp_queue_address = tmp_queue.address
with tmp_queue.create_frombus() as frombus:
for i in range(NUM_MESSAGES_TO_SEND):
# send a message...
@@ -256,17 +270,23 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
logger.info("received message: %s", received_msg)
self.assertEqual(None, received_msg)
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(tmp_queue_address))
+
def test_send_receive_over_temporary_exchange_with_multiple_bound_queues_with_subject_filtering(
self):
"""
test the usage of the TemporaryQueue in conjunction with normal ToBus and Frombus usage with additional filtering on subject
"""
with TemporaryExchange("MyTestExchange") as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
with tmp_exchange.create_tobus() as tobus:
SUBJECT1 = "FooBarSubject"
SUBJECT2 = "FAKE_SUBJECT"
with TemporaryQueue("MyTestQueue", exchange=tmp_exchange.address, routing_key=SUBJECT1) as tmp_queue1, \
TemporaryQueue("MyTestQueue", exchange=tmp_exchange.address, routing_key=SUBJECT2) as tmp_queue2:
+ tmp_queue1_address = tmp_queue1.address
+ tmp_queue2_address = tmp_queue2.address
# create a normal To/FromBus on this tmp_queue
NUM_MESSAGES_TO_SEND = 3
@@ -309,6 +329,9 @@ class TestTemporaryExchangeAndQueue(unittest.TestCase):
self.assertEqual(original_msg.content, received_msg2.content)
self.assertEqual(original_msg.subject, received_msg2.subject)
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(tmp_queue1_address))
+ self.assertFalse(queue_exists(tmp_queue2_address))
# ======== FromBus unit tests ======== #
@@ -320,7 +343,11 @@ class FromBusInitFailed(unittest.TestCase):
def setUp(self):
self.test_queue = TemporaryQueue(__class__.__name__)
self.test_queue.open()
- self.addCleanup(self.test_queue.close)
+
+ def tearDown(self):
+ tmp_queue_address = self.test_queue.address
+ self.test_queue.close()
+ self.assertFalse(queue_exists(tmp_queue_address))
def test_no_broker_address(self):
"""
@@ -347,8 +374,11 @@ class FromBusInContext(unittest.TestCase):
def setUp(self):
self.test_queue = TemporaryQueue(__class__.__name__)
self.test_queue.open()
- self.addCleanup(self.test_queue.close)
- self.error = "[FromBus] Failed to create receiver for source"
+
+ def tearDown(self):
+ tmp_queue_address = self.test_queue.address
+ self.test_queue.close()
+ self.assertFalse(queue_exists(tmp_queue_address))
def test_receiver_exists(self):
with FromBus(self.test_queue.address) as frombus:
@@ -377,16 +407,20 @@ class ToBusInitFailed(unittest.TestCase):
"""
def setUp(self):
- self.test_queue = TemporaryQueue(__class__.__name__)
- self.test_queue.open()
- self.addCleanup(self.test_queue.close)
+ self.test_exchange = TemporaryExchange(__class__.__name__)
+ self.test_exchange.open()
+
+ def tearDown(self):
+ tmp_exchange_address = self.test_exchange.address
+ self.test_exchange.close()
+ self.assertFalse(exchange_exists(tmp_exchange_address))
def test_no_broker_address(self):
"""
Connecting to non-existent broker address must raise MessageBusError
"""
with self.assertRaisesRegex(MessageBusError, ".*failed to resolve broker hostname"):
- with ToBus(self.test_queue.address, broker="foo.bar"):
+ with ToBus(self.test_exchange.address, broker="foo.bar"):
pass
def test_connection_refused(self):
@@ -394,7 +428,7 @@ class ToBusInitFailed(unittest.TestCase):
Connecting to broker on wrong port must raise MessageBusError
"""
with self.assertRaisesRegex(MessageBusError, ".*failed to resolve broker hostname"):
- with ToBus(self.test_queue.address, broker="localhost:4"):
+ with ToBus(self.test_exchange.address, broker="localhost:4"):
pass
@@ -404,12 +438,23 @@ class SendReceiveMessage(unittest.TestCase):
"""
def setUp(self):
- self.test_queue = TemporaryQueue(__class__.__name__)
+ self.test_exchange = TemporaryExchange(__class__.__name__)
+ self.test_exchange.open()
+
+ self.test_queue = TemporaryQueue(__class__.__name__, exchange=self.test_exchange.address)
self.test_queue.open()
- self.addCleanup(self.test_queue.close)
self.frombus = self.test_queue.create_frombus()
- self.tobus = self.test_queue.create_tobus()
+ self.tobus = self.test_exchange.create_tobus()
+
+ def tearDown(self):
+ tmp_queue_address = self.test_queue.address
+ self.test_queue.close()
+ self.assertFalse(queue_exists(tmp_queue_address))
+
+ tmp_exchange_address = self.test_exchange.address
+ self.test_exchange.close()
+ self.assertFalse(exchange_exists(tmp_exchange_address))
def _test_sendrecv(self, send_msg):
"""
@@ -505,7 +550,9 @@ class SendReceiveMessage(unittest.TestCase):
class PriorityTest(unittest.TestCase):
def test_priority(self):
with TemporaryExchange(self.__class__.__name__) as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
with tmp_exchange.create_temporary_queue() as tmp_queue:
+ tmp_queue_address = tmp_queue.address
msg1 = EventMessage(priority=4, subject="some.event", content=1)
msg2 = EventMessage(priority=5, subject="some.event", content=2)
@@ -521,6 +568,9 @@ class PriorityTest(unittest.TestCase):
self.assertEqual(msg1.id, result_msg2.id)
self.assertEqual(msg2.id, result_msg1.id)
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(tmp_queue_address))
+
class Rejector(BusListener):
handled_messages = 0
@@ -545,7 +595,9 @@ class RejectorTester(unittest.TestCase):
number_of_messages = 1000
with TemporaryExchange("Rejection") as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
with BusListenerJanitor(Rejector(tmp_exchange.address)) as rejector:
+ rejector_address = Rejector.designated_queue_name(tmp_exchange_address)
with tmp_exchange.create_tobus() as spammer:
for _ in range(number_of_messages):
msg = EventMessage(content="ping", subject="spam")
@@ -559,6 +611,9 @@ class RejectorTester(unittest.TestCase):
logger.info("Number of messages on queue: {}".format(frombus.nr_of_messages_in_queue()))
self.assertEqual(0, frombus.nr_of_messages_in_queue())
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(rejector_address))
+
class PingPongPlayer(BusListener):
@@ -643,10 +698,14 @@ class PingPongTester(unittest.TestCase):
# setup temporary exchange, on which the player can publish their messages (ping/pong balls)
with TemporaryExchange("PingPongTable") as tmp_exchange:
+ tmp_exchange_address = tmp_exchange.address
+
# create two players, on "both sides of the table"
# i.e.: they each play on the tmp_exchange, but have the auto-generated designated listen queues for incoming balls
with BusListenerJanitor(PingPongPlayer("Player1", "Player2", tmp_exchange.address, num_threads_per_player)) as player1:
- with BusListenerJanitor(PingPongPlayer("Player2", "Player1", tmp_exchange.address, num_threads_per_player)) as player2:
+ player1_address = player1.address
+ with BusListenerJanitor(PingPongPlayer("Player2", "Player1", tmp_exchange.address, num_threads_per_player)) as player2:
+ player2_address = player2.address
start_timestamp = datetime.utcnow()
# first serve, referee throws a ping ball on the table in the direction of player1
@@ -682,6 +741,10 @@ class PingPongTester(unittest.TestCase):
player1_num_turns, NUM_TURNS, player2_num_turns, NUM_TURNS,
num_threads_per_player, 2*NUM_TURNS/(datetime.utcnow() - start_timestamp).total_seconds())
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+ self.assertFalse(queue_exists(player1_address))
+ self.assertFalse(queue_exists(player2_address))
+
class MessageHandlerTester(unittest.TestCase):
def test_handler_init_raises(self):
@@ -692,11 +755,15 @@ class MessageHandlerTester(unittest.TestCase):
# try to start a BusListener using this handler. Should fail and raise a MessagingRuntimeError
with TemporaryExchange(self.__class__.__name__) as tmp_exchange:
+ tmp_exchange_name = tmp_exchange.address
with self.assertRaises(MessagingRuntimeError):
with BusListenerJanitor(BusListener(handler_type=RaisingHandler,
- exchange=tmp_exchange.address)) as listener:
+ exchange=tmp_exchange_name)) as listener:
pass
+ self.assertFalse(exchange_exists(tmp_exchange_name))
+ self.assertFalse(queue_exists(BusListener.designated_queue_name(tmp_exchange_name)))
+
def test_empty_template_handler(self):
# define a MessageHandler with a template for callback on<something> methods
class BaseTemplateHandler(AbstractMessageHandler):
@@ -757,22 +824,34 @@ class MessageHandlerTester(unittest.TestCase):
# try to start a BusListener using a BaseTemplateHandler. Should fail and raise a TypeError
with TemporaryExchange(self.__class__.__name__) as tmp_exchange:
+ tmp_exchange_name = tmp_exchange.address
with self.assertRaises(RuntimeError):
with BusListenerJanitor(BusListener(handler_type=BaseTemplateHandler,
- exchange=tmp_exchange.address)) as listener:
+ exchange=tmp_exchange_name)) as listener:
pass
+ self.assertFalse(exchange_exists(tmp_exchange_name))
+ self.assertFalse(queue_exists(BusListener.designated_queue_name(tmp_exchange_name)))
+
+
class ReconnectOnConnectionLossTests(unittest.TestCase):
def setUp(self):
self.tmp_exchange = TemporaryExchange()
- self.addCleanup(self.tmp_exchange.close)
-
self.tmp_queue = self.tmp_exchange.create_temporary_queue()
- self.addCleanup(self.tmp_queue.close)
self.tmp_exchange.open()
self.tmp_queue.open()
+ def tearDown(self):
+ tmp_queue_address = self.tmp_queue.address
+ self.tmp_queue.close()
+ self.assertFalse(queue_exists(tmp_queue_address))
+
+ tmp_exchange_address = self.tmp_exchange.address
+ self.tmp_exchange.close()
+ self.assertFalse(exchange_exists(tmp_exchange_address))
+
+
def _close_connection_of_bus_on_broker(self, bus: _AbstractBus):
# use the http REST API using request to forcefully close the connection on the broker-side
url = "http://%s:15672/api/connections/%s" % (bus.broker, bus.connection_name)