#!/usr/bin/env python # # Licensed to the Apache Software Foundation (ASF) under one # or more contributor license agreements. See the NOTICE file # distributed with this work for additional information # regarding copyright ownership. The ASF licenses this file # to you under the Apache License, Version 2.0 (the # "License"); you may not use this file except in compliance # with the License. You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. # import sys from qpid.testlib import TestBase010 from qpid.datatypes import Message from qpid.queue import Empty from time import sleep class FederationTests(TestBase010): def remote_host(self): return self.defines.get("remote-host", "localhost") def remote_port(self): return int(self.defines["remote-port"]) def verify_cleanup(self): attempts = 0 total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link")) while total > 0: attempts += 1 if attempts >= 10: self.fail("Bridges and links didn't clean up") return sleep(1) total = len(self.qmf.getObjects(_class="bridge")) + len(self.qmf.getObjects(_class="link")) def test_bridge_create_and_close(self): self.startQmf(); qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "amq.direct", "amq.direct", "my-key", "", "", False, False, False, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_pull_from_exchange(self): session = self.session self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "amq.direct", "amq.fanout", "my-key", "", "", False, False, False, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] #setup queue to receive messages from local broker session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="amq.fanout") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") sleep(6) #send messages to remote broker and confirm it is routed to local broker r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_pull_from_exchange") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="my-key") r_session.message_transfer(destination="amq.direct", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_push_to_exchange(self): session = self.session self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "amq.direct", "amq.fanout", "my-key", "", "", False, True, False, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] #setup queue to receive messages from remote broker r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_push_to_exchange") r_session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) r_session.exchange_bind(queue="fed1", exchange="amq.fanout") self.subscribe(session=r_session, queue="fed1", destination="f1") queue = r_session.incoming("f1") sleep(6) #send messages to local broker and confirm it is routed to remote broker for i in range(1, 11): dp = session.delivery_properties(routing_key="my-key") session.message_transfer(destination="amq.direct", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_pull_from_queue(self): session = self.session #setup queue on remote broker and add some messages r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_pull_from_queue") r_session.queue_declare(queue="my-bridge-queue", auto_delete=True) for i in range(1, 6): dp = r_session.delivery_properties(routing_key="my-bridge-queue") r_session.message_transfer(message=Message(dp, "Message %d" % i)) #setup queue to receive messages from local broker session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="amq.fanout") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "my-bridge-queue", "amq.fanout", "my-key", "", "", True, False, False, 1) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(3) #add some more messages (i.e. after bridge was created) for i in range(6, 11): dp = r_session.delivery_properties(routing_key="my-bridge-queue") r_session.message_transfer(message=Message(dp, "Message %d" % i)) for i in range(1, 11): try: msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) except Empty: self.fail("Failed to find expected message containing 'Message %d'" % i) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_tracing_automatic(self): remoteUrl = "%s:%d" % (self.remote_host(), self.remote_port()) self.startQmf() l_broker = self.qmf_broker r_broker = self.qmf.addBroker(remoteUrl) l_brokerObj = self.qmf.getObjects(_class="broker", _broker=l_broker)[0] r_brokerObj = self.qmf.getObjects(_class="broker", _broker=r_broker)[0] l_res = l_brokerObj.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") r_res = r_brokerObj.connect(self.broker.host, self.broker.port, False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(l_res.status, 0) self.assertEqual(r_res.status, 0) l_link = self.qmf.getObjects(_class="link", _broker=l_broker)[0] r_link = self.qmf.getObjects(_class="link", _broker=r_broker)[0] l_res = l_link.bridge(False, "amq.direct", "amq.direct", "key", "", "", False, False, False, 0) r_res = r_link.bridge(False, "amq.direct", "amq.direct", "key", "", "", False, False, False, 0) self.assertEqual(l_res.status, 0) self.assertEqual(r_res.status, 0) count = 0 while l_link.state != "Operational" or r_link.state != "Operational": count += 1 if count > 10: self.fail("Fed links didn't become operational after 10 seconds") sleep(1) l_link = self.qmf.getObjects(_class="link", _broker=l_broker)[0] r_link = self.qmf.getObjects(_class="link", _broker=r_broker)[0] sleep(3) #setup queue to receive messages from local broker session = self.session session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="amq.direct", binding_key="key") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") #setup queue on remote broker and add some messages r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_trace") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="key") r_session.message_transfer(destination="amq.direct", message=Message(dp, "Message %d" % i)) for i in range(1, 11): try: msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) except Empty: self.fail("Failed to find expected message containing 'Message %d'" % i) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None def test_tracing(self): session = self.session self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "amq.direct", "amq.fanout", "my-key", "my-bridge-id", "exclude-me,also-exclude-me", False, False, False, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] #setup queue to receive messages from local broker session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="amq.fanout") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") sleep(6) #send messages to remote broker and confirm it is routed to local broker r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_tracing") trace = [None, "exclude-me", "a,exclude-me,b", "also-exclude-me,c", "dont-exclude-me"] body = ["yes", "first-bad", "second-bad", "third-bad", "yes"] for b, t in zip(body, trace): headers = {} if (t): headers["x-qpid.trace"]=t dp = r_session.delivery_properties(routing_key="my-key") mp = r_session.message_properties(application_headers=headers) r_session.message_transfer(destination="amq.direct", message=Message(dp, mp, b)) for e in ["my-bridge-id", "dont-exclude-me,my-bridge-id"]: msg = queue.get(timeout=5) self.assertEqual("yes", msg.body) self.assertEqual(e, self.getAppHeader(msg, "x-qpid.trace")) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_fanout(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_fanout") session.exchange_declare(exchange="fed.fanout", type="fanout") r_session.exchange_declare(exchange="fed.fanout", type="fanout") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.fanout", "fed.fanout", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.fanout") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties() r_session.message_transfer(destination="fed.fanout", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_direct(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_direct") session.exchange_declare(exchange="fed.direct", type="direct") r_session.exchange_declare(exchange="fed.direct", type="direct") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.direct", "fed.direct", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.direct", binding_key="fd-key") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="fd-key") r_session.message_transfer(destination="fed.direct", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_topic(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_topic") session.exchange_declare(exchange="fed.topic", type="topic") r_session.exchange_declare(exchange="fed.topic", type="topic") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.topic", "fed.topic", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.topic", binding_key="ft-key.#") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="ft-key.one.two") r_session.message_transfer(destination="fed.topic", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_topic_reorigin(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_topic_reorigin") session.exchange_declare(exchange="fed.topic_reorigin", type="topic") r_session.exchange_declare(exchange="fed.topic_reorigin", type="topic") session.exchange_declare(exchange="fed.topic_reorigin_2", type="topic") r_session.exchange_declare(exchange="fed.topic_reorigin_2", type="topic") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) session.queue_declare(queue="fed2", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed2", exchange="fed.topic_reorigin_2", binding_key="ft-key.one.#") self.subscribe(queue="fed2", destination="f2") queue2 = session.incoming("f2") link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.topic_reorigin", "fed.topic_reorigin", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) result = link.bridge(False, "fed.topic_reorigin_2", "fed.topic_reorigin_2", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] bridge2 = qmf.getObjects(_class="bridge")[1] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.topic_reorigin", binding_key="ft-key.#") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="ft-key.one.two") r_session.message_transfer(destination="fed.topic_reorigin", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = bridge2.close() self.assertEqual(result.status, 0) # extra check: verify we don't leak bridge objects - keep the link # around and verify the bridge count has gone to zero attempts = 0 bridgeCount = len(qmf.getObjects(_class="bridge")) while bridgeCount > 0: attempts += 1 if attempts >= 5: self.fail("Bridges didn't clean up") return sleep(1) bridgeCount = len(qmf.getObjects(_class="bridge")) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_direct_reorigin(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_direct_reorigin") session.exchange_declare(exchange="fed.direct_reorigin", type="direct") r_session.exchange_declare(exchange="fed.direct_reorigin", type="direct") session.exchange_declare(exchange="fed.direct_reorigin_2", type="direct") r_session.exchange_declare(exchange="fed.direct_reorigin_2", type="direct") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) session.queue_declare(queue="fed2", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed2", exchange="fed.direct_reorigin_2", binding_key="ft-key.two") self.subscribe(queue="fed2", destination="f2") queue2 = session.incoming("f2") link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.direct_reorigin", "fed.direct_reorigin", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) result = link.bridge(False, "fed.direct_reorigin_2", "fed.direct_reorigin_2", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] bridge2 = qmf.getObjects(_class="bridge")[1] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.direct_reorigin", binding_key="ft-key.one") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="ft-key.one") r_session.message_transfer(destination="fed.direct_reorigin", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) # Extra test: don't explicitly close() bridge2. When the link is closed, # it should clean up bridge2 automagically. verify_cleanup() will detect # if bridge2 isn't cleaned up and will fail the test. # #result = bridge2.close() #self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_headers(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_headers") session.exchange_declare(exchange="fed.headers", type="headers") r_session.exchange_declare(exchange="fed.headers", type="headers") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.headers", "fed.headers", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.headers", binding_key="key1", arguments={'x-match':'any', 'class':'first'}) self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") props = r_session.message_properties(application_headers={'class':'first'}) for i in range(1, 11): r_session.message_transfer(destination="fed.headers", message=Message(props, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) content = msg.body self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_headers_reorigin(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_headers_reorigin") session.exchange_declare(exchange="fed.headers_reorigin", type="headers") r_session.exchange_declare(exchange="fed.headers_reorigin", type="headers") session.exchange_declare(exchange="fed.headers_reorigin_2", type="headers") r_session.exchange_declare(exchange="fed.headers_reorigin_2", type="headers") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) session.queue_declare(queue="fed2", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed2", exchange="fed.headers_reorigin_2", binding_key="key2", arguments={'x-match':'any', 'class':'second'}) self.subscribe(queue="fed2", destination="f2") queue2 = session.incoming("f2") link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.headers_reorigin", "fed.headers_reorigin", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) result = link.bridge(False, "fed.headers_reorigin_2", "fed.headers_reorigin_2", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] bridge2 = qmf.getObjects(_class="bridge")[1] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.headers_reorigin", binding_key="key1", arguments={'x-match':'any', 'class':'first'}) self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") props = r_session.message_properties(application_headers={'class':'first'}) for i in range(1, 11): r_session.message_transfer(destination="fed.headers_reorigin", message=Message(props, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) # Extra test: don't explicitly close() bridge2. When the link is closed, # it should clean up bridge2 automagically. verify_cleanup() will detect # if bridge2 isn't cleaned up and will fail the test. # #result = bridge2.close() #self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_headers_unbind(self): session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_headers_unbind") session.exchange_declare(exchange="fed.headers_unbind", type="headers") r_session.exchange_declare(exchange="fed.headers_unbind", type="headers") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.headers_unbind", "fed.headers_unbind", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) queue = qmf.getObjects(_class="queue", name="fed1")[0] queue.update() self.assertEqual(queue.bindingCount, 1, "bindings not accounted for (expected 1, got %d)" % queue.bindingCount) session.exchange_bind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1", arguments={'x-match':'any', 'class':'first'}) queue.update() self.assertEqual(queue.bindingCount, 2, "bindings not accounted for (expected 2, got %d)" % queue.bindingCount) session.exchange_unbind(queue="fed1", exchange="fed.headers_unbind", binding_key="key1") queue.update() self.assertEqual(queue.bindingCount, 1, "bindings not accounted for (expected 1, got %d)" % queue.bindingCount) result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def test_dynamic_topic_nodup(self): """Verify that a message whose routing key matches more than one binding does not get duplicated to the same queue. """ session = self.session r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) r_session = r_conn.session("test_dynamic_topic_nodup") session.exchange_declare(exchange="fed.topic", type="topic") r_session.exchange_declare(exchange="fed.topic", type="topic") self.startQmf() qmf = self.qmf broker = qmf.getObjects(_class="broker")[0] result = broker.connect(self.remote_host(), self.remote_port(), False, "PLAIN", "guest", "guest", "tcp") self.assertEqual(result.status, 0) link = qmf.getObjects(_class="link")[0] result = link.bridge(False, "fed.topic", "fed.topic", "", "", "", False, False, True, 0) self.assertEqual(result.status, 0) bridge = qmf.getObjects(_class="bridge")[0] sleep(5) session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) session.exchange_bind(queue="fed1", exchange="fed.topic", binding_key="red.*") session.exchange_bind(queue="fed1", exchange="fed.topic", binding_key="*.herring") self.subscribe(queue="fed1", destination="f1") queue = session.incoming("f1") for i in range(1, 11): dp = r_session.delivery_properties(routing_key="red.herring") r_session.message_transfer(destination="fed.topic", message=Message(dp, "Message %d" % i)) for i in range(1, 11): msg = queue.get(timeout=5) self.assertEqual("Message %d" % i, msg.body) try: extra = queue.get(timeout=1) self.fail("Got unexpected message in queue: " + extra.body) except Empty: None result = bridge.close() self.assertEqual(result.status, 0) result = link.close() self.assertEqual(result.status, 0) self.verify_cleanup() def getProperty(self, msg, name): for h in msg.headers: if hasattr(h, name): return getattr(h, name) return None def getAppHeader(self, msg, name): headers = self.getProperty(msg, "application_headers") if headers: return headers[name] return None