diff options
Diffstat (limited to 'cpp/src/tests/federation.py')
| -rwxr-xr-x | cpp/src/tests/federation.py | 2091 |
1 files changed, 0 insertions, 2091 deletions
diff --git a/cpp/src/tests/federation.py b/cpp/src/tests/federation.py deleted file mode 100755 index 201b06a4a2..0000000000 --- a/cpp/src/tests/federation.py +++ /dev/null @@ -1,2091 +0,0 @@ -#!/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 qpid.util import URL -from time import sleep, time - - -class _FedBroker(object): - """ - A proxy object for a remote broker. Contains connection and management - state. - """ - def __init__(self, host, port, - conn=None, session=None, qmf_broker=None): - self.host = host - self.port = port - self.url = "%s:%d" % (host, port) - self.client_conn = None - self.client_session = None - self.qmf_broker = None - self.qmf_object = None - if conn is not None: - self.client_conn = conn - if session is not None: - self.client_session = session - if qmf_broker is not None: - self.qmf_broker = qmf_broker - - -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 _setup_brokers(self): - ports = [self.remote_port()] - extra = self.defines.get("extra-brokers") - if extra: - for p in extra.split(): - ports.append(int(p)) - - # broker[0] has already been set up. - self._brokers = [_FedBroker(self.broker.host, - self.broker.port, - self.conn, - self.session, - self.qmf_broker)] - self._brokers[0].qmf_object = self.qmf.getObjects(_class="broker")[0] - - # setup remaining brokers - for _p in ports: - _b = _FedBroker(self.remote_host(), _p) - _b.client_conn = self.connect(host=self.remote_host(), port=_p) - _b.client_session = _b.client_conn.session("Fed_client_session_" + str(_p)) - _b.qmf_broker = self.qmf.addBroker(_b.url) - for _bo in self.qmf.getObjects(_class="broker"): - if _bo.getBroker().getUrl() == _b.qmf_broker.getUrl(): - _b.qmf_object = _bo - break - self._brokers.append(_b) - - def _teardown_brokers(self): - """ Un-does _setup_brokers() - """ - # broker[0] is configured at test setup, so it must remain configured - for _b in self._brokers[1:]: - self.qmf.delBroker(_b.qmf_broker) - if not _b.client_session.error(): - _b.client_session.close(timeout=10) - _b.client_conn.close(timeout=10) - - - 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", ttl=1000*60*5) - 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")) - assert(msg.get("delivery_properties").ttl > 0) - assert(msg.get("delivery_properties").ttl < 1000*60*50) - - 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_headers_xml(self): - session = self.session - r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) - r_session = r_conn.session("test_dynamic_headers_xml") - - session.exchange_declare(exchange="fed.xml", type="xml") - r_session.exchange_declare(exchange="fed.xml", type="xml") - - 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.xml", "fed.xml", "", "", "", 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.xml", binding_key="key1", arguments={'xquery':'true()'}) - self.subscribe(queue="fed1", destination="f1") - queue = session.incoming("f1") - - props = r_session.delivery_properties(routing_key="key1") - for i in range(1, 11): - r_session.message_transfer(destination="fed.xml", 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_xml(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_xml") - - session.exchange_declare(exchange="fed.xml_reorigin", type="xml") - r_session.exchange_declare(exchange="fed.xml_reorigin", type="xml") - - session.exchange_declare(exchange="fed.xml_reorigin_2", type="xml") - r_session.exchange_declare(exchange="fed.xml_reorigin_2", type="xml") - - 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.xml_reorigin_2", binding_key="key2", arguments={'xquery':'true()'}) - self.subscribe(queue="fed2", destination="f2") - queue2 = session.incoming("f2") - - link = qmf.getObjects(_class="link")[0] - result = link.bridge(False, "fed.xml_reorigin", "fed.xml_reorigin", "", "", "", False, False, True, 0) - - self.assertEqual(result.status, 0) - result = link.bridge(False, "fed.xml_reorigin_2", "fed.xml_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) - - foo=qmf.getObjects(_class="link") - session.queue_declare(queue="fed1", exclusive=True, auto_delete=True) - session.exchange_bind(queue="fed1", exchange="fed.xml_reorigin", binding_key="key1", arguments={'xquery':'true()'}) - self.subscribe(queue="fed1", destination="f1") - queue = session.incoming("f1") - - props = r_session.delivery_properties(routing_key="key1") - for i in range(1, 11): - r_session.message_transfer(destination="fed.xml_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_xml(self): - session = self.session - r_conn = self.connect(host=self.remote_host(), port=self.remote_port()) - r_session = r_conn.session("test_dynamic_xml_unbind") - - session.exchange_declare(exchange="fed.xml_unbind", type="xml") - r_session.exchange_declare(exchange="fed.xml_unbind", type="xml") - - 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.xml_unbind", "fed.xml_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.xml_unbind", binding_key="key1", arguments={'xquery':'true()'}) - queue.update() - self.assertEqual(queue.bindingCount, 2, - "bindings not accounted for (expected 2, got %d)" % queue.bindingCount) - - session.exchange_unbind(queue="fed1", exchange="fed.xml_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 test_dynamic_direct_route_prop(self): - """ Set up a tree of uni-directional routes across the direct exchange. - Bind the same key to the same queues on the leaf nodes. Verify a - message sent with the routing key transverses the tree an arrives at - each leaf. Remove one leaf's queue, and verify that messages still - reach the other leaf. - - Route Topology: - - +---> B2 queue:"test-queue", binding key:"spudboy" - B0 --> B1 --+ - +---> B3 queue:"test-queue", binding key:"spudboy" - """ - session = self.session - - # create the federation - - self.startQmf() - qmf = self.qmf - - self._setup_brokers() - - # create direct exchange on each broker, and retrieve the corresponding - # management object for that exchange - - exchanges=[] - for _b in self._brokers: - _b.client_session.exchange_declare(exchange="fedX.direct", type="direct") - self.assertEqual(_b.client_session.exchange_query(name="fedX.direct").type, - "direct", "exchange_declare failed!") - # pull the exchange out of qmf... - retries = 0 - my_exchange = None - while my_exchange is None: - objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange") - for ooo in objs: - if ooo.name == "fedX.direct": - my_exchange = ooo - break - if my_exchange is None: - retries += 1 - self.failIfEqual(retries, 10, - "QMF failed to find new exchange!") - sleep(1) - exchanges.append(my_exchange) - - self.assertEqual(len(exchanges), len(self._brokers), "Exchange creation failed!") - - # connect B0 --> B1 - result = self._brokers[1].qmf_object.connect(self._brokers[0].host, - self._brokers[0].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B2 - result = self._brokers[2].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B3 - result = self._brokers[3].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # for each link, bridge the "fedX.direct" exchanges: - - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker()))) - result = _l.bridge(False, # durable - "fedX.direct", # src - "fedX.direct", # dst - "", # key - "", # tag - "", # excludes - False, # srcIsQueue - False, # srcIsLocal - True, # dynamic - 0) # sync - self.assertEqual(result.status, 0) - - # wait for the inter-broker links to become operational - retries = 0 - operational = False - while not operational: - operational = True - for _l in qmf.getObjects(_class="link"): - #print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state))) - if _l.state != "Operational": - operational = False - if not operational: - retries += 1 - self.failIfEqual(retries, 10, - "inter-broker links failed to become operational.") - sleep(1) - - # @todo - There is no way to determine when the bridge objects become - # active. Hopefully, this is long enough! - sleep(6) - - # create a queue on B2, bound to "spudboy" - self._brokers[2].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[2].client_session.exchange_bind(queue="fedX1", exchange="fedX.direct", binding_key="spudboy") - - # create a queue on B3, bound to "spudboy" - self._brokers[3].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[3].client_session.exchange_bind(queue="fedX1", exchange="fedX.direct", binding_key="spudboy") - - # subscribe to messages arriving on B2's queue - self.subscribe(self._brokers[2].client_session, queue="fedX1", destination="f1") - queue_2 = self._brokers[2].client_session.incoming("f1") - - # subscribe to messages arriving on B3's queue - self.subscribe(self._brokers[3].client_session, queue="fedX1", destination="f1") - queue_3 = self._brokers[3].client_session.incoming("f1") - - # wait until the binding key has propagated to each broker (twice at - # broker B1). Work backwards from binding brokers. - - binding_counts = [1, 2, 1, 1] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(3,-1,-1): - retries = 0 - exchanges[i].update() - while exchanges[i].bindingCount < binding_counts[i]: - retries += 1 - self.failIfEqual(retries, 10, - "binding failed to propagate to broker %d" - % i) - sleep(3) - exchanges[i].update() - - # send 10 msgs from B0 - for i in range(1, 11): - dp = self._brokers[0].client_session.delivery_properties(routing_key="spudboy") - self._brokers[0].client_session.message_transfer(destination="fedX.direct", message=Message(dp, "Message_drp %d" % i)) - - # wait for 10 messages to be forwarded from B0->B1, - # 10 messages from B1->B2, - # and 10 messages from B1->B3 - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 10 or exchanges[0].msgRoutes != 10 or - exchanges[1].msgReceives != 10 or exchanges[1].msgRoutes != 20 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 10 or exchanges[3].msgRoutes != 10): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B2 and B3 - for i in range(1, 11): - msg = queue_2.get(timeout=5) - self.assertEqual("Message_drp %d" % i, msg.body) - msg = queue_3.get(timeout=5) - self.assertEqual("Message_drp %d" % i, msg.body) - - try: - extra = queue_2.get(timeout=1) - self.fail("Got unexpected message in queue_2: " + extra.body) - except Empty: None - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - - # tear down the queue on B2 - self._brokers[2].client_session.exchange_unbind(queue="fedX1", exchange="fedX.direct", binding_key="spudboy") - self._brokers[2].client_session.message_cancel(destination="f1") - self._brokers[2].client_session.queue_delete(queue="fedX1") - - # @todo - restore code when QPID-2499 fixed!! - sleep(6) - # wait for the binding count on B1 to drop from 2 to 1 - retries = 0 - exchanges[1].update() - while exchanges[1].bindingCount != 1: - retries += 1 - self.failIfEqual(retries, 10, - "unbinding failed to propagate to broker B1: %d" - % exchanges[1].bindingCount) - sleep(1) - exchanges[1].update() - - # send 10 msgs from B0 - for i in range(11, 21): - dp = self._brokers[0].client_session.delivery_properties(routing_key="spudboy") - self._brokers[0].client_session.message_transfer(destination="fedX.direct", message=Message(dp, "Message_drp %d" % i)) - - # verify messages are forwarded to B3 only - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 20 or exchanges[0].msgRoutes != 20 or - exchanges[1].msgReceives != 20 or exchanges[1].msgRoutes != 30 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 20 or exchanges[3].msgRoutes != 20): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route more msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B3 only - for i in range(11, 21): - msg = queue_3.get(timeout=5) - self.assertEqual("Message_drp %d" % i, msg.body) - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - # cleanup - - self._brokers[3].client_session.exchange_unbind(queue="fedX1", exchange="fedX.direct", binding_key="spudboy") - self._brokers[3].client_session.message_cancel(destination="f1") - self._brokers[3].client_session.queue_delete(queue="fedX1") - - for _b in qmf.getObjects(_class="bridge"): - result = _b.close() - self.assertEqual(result.status, 0) - - for _l in qmf.getObjects(_class="link"): - result = _l.close() - self.assertEqual(result.status, 0) - - for _b in self._brokers: - _b.client_session.exchange_delete(exchange="fedX.direct") - - self._teardown_brokers() - - self.verify_cleanup() - - def test_dynamic_topic_route_prop(self): - """ Set up a tree of uni-directional routes across a topic exchange. - Bind the same key to the same queues on the leaf nodes. Verify a - message sent with the routing key transverses the tree an arrives at - each leaf. Remove one leaf's queue, and verify that messages still - reach the other leaf. - - Route Topology: - - +---> B2 queue:"test-queue", binding key:"spud.*" - B0 --> B1 --+ - +---> B3 queue:"test-queue", binding key:"spud.*" - """ - session = self.session - - # create the federation - - self.startQmf() - qmf = self.qmf - - self._setup_brokers() - - # create exchange on each broker, and retrieve the corresponding - # management object for that exchange - - exchanges=[] - for _b in self._brokers: - _b.client_session.exchange_declare(exchange="fedX.topic", type="topic") - self.assertEqual(_b.client_session.exchange_query(name="fedX.topic").type, - "topic", "exchange_declare failed!") - # pull the exchange out of qmf... - retries = 0 - my_exchange = None - while my_exchange is None: - objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange") - for ooo in objs: - if ooo.name == "fedX.topic": - my_exchange = ooo - break - if my_exchange is None: - retries += 1 - self.failIfEqual(retries, 10, - "QMF failed to find new exchange!") - sleep(1) - exchanges.append(my_exchange) - - self.assertEqual(len(exchanges), len(self._brokers), "Exchange creation failed!") - - # connect B0 --> B1 - result = self._brokers[1].qmf_object.connect(self._brokers[0].host, - self._brokers[0].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B2 - result = self._brokers[2].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B3 - result = self._brokers[3].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # for each link, bridge the "fedX.topic" exchanges: - - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker()))) - result = _l.bridge(False, # durable - "fedX.topic", # src - "fedX.topic", # dst - "", # key - "", # tag - "", # excludes - False, # srcIsQueue - False, # srcIsLocal - True, # dynamic - 0) # sync - self.assertEqual(result.status, 0) - - # wait for the inter-broker links to become operational - retries = 0 - operational = False - while not operational: - operational = True - for _l in qmf.getObjects(_class="link"): - #print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state))) - if _l.state != "Operational": - operational = False - if not operational: - retries += 1 - self.failIfEqual(retries, 10, - "inter-broker links failed to become operational.") - sleep(1) - - # @todo - There is no way to determine when the bridge objects become - # active. - sleep(6) - - # create a queue on B2, bound to "spudboy" - self._brokers[2].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[2].client_session.exchange_bind(queue="fedX1", exchange="fedX.topic", binding_key="spud.*") - - # create a queue on B3, bound to "spudboy" - self._brokers[3].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[3].client_session.exchange_bind(queue="fedX1", exchange="fedX.topic", binding_key="spud.*") - - # subscribe to messages arriving on B2's queue - self.subscribe(self._brokers[2].client_session, queue="fedX1", destination="f1") - queue_2 = self._brokers[2].client_session.incoming("f1") - - # subscribe to messages arriving on B3's queue - self.subscribe(self._brokers[3].client_session, queue="fedX1", destination="f1") - queue_3 = self._brokers[3].client_session.incoming("f1") - - # wait until the binding key has propagated to each broker (twice at - # broker B1). Work backwards from binding brokers. - - binding_counts = [1, 2, 1, 1] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(3,-1,-1): - retries = 0 - exchanges[i].update() - while exchanges[i].bindingCount < binding_counts[i]: - retries += 1 - self.failIfEqual(retries, 10, - "binding failed to propagate to broker %d" - % i) - sleep(3) - exchanges[i].update() - - # send 10 msgs from B0 - for i in range(1, 11): - dp = self._brokers[0].client_session.delivery_properties(routing_key="spud.boy") - self._brokers[0].client_session.message_transfer(destination="fedX.topic", message=Message(dp, "Message_trp %d" % i)) - - # wait for 10 messages to be forwarded from B0->B1, - # 10 messages from B1->B2, - # and 10 messages from B1->B3 - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 10 or exchanges[0].msgRoutes != 10 or - exchanges[1].msgReceives != 10 or exchanges[1].msgRoutes != 20 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 10 or exchanges[3].msgRoutes != 10): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B2 and B3 - for i in range(1, 11): - msg = queue_2.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - msg = queue_3.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - - try: - extra = queue_2.get(timeout=1) - self.fail("Got unexpected message in queue_2: " + extra.body) - except Empty: None - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - # tear down the queue on B2 - self._brokers[2].client_session.exchange_unbind(queue="fedX1", exchange="fedX.topic", binding_key="spud.*") - self._brokers[2].client_session.message_cancel(destination="f1") - self._brokers[2].client_session.queue_delete(queue="fedX1") - - # wait for the binding count on B1 to drop from 2 to 1 - retries = 0 - exchanges[1].update() - while exchanges[1].bindingCount != 1: - retries += 1 - self.failIfEqual(retries, 10, - "unbinding failed to propagate to broker B1: %d" - % exchanges[1].bindingCount) - sleep(1) - exchanges[1].update() - - # send 10 msgs from B0 - for i in range(11, 21): - dp = self._brokers[0].client_session.delivery_properties(routing_key="spud.boy") - self._brokers[0].client_session.message_transfer(destination="fedX.topic", message=Message(dp, "Message_trp %d" % i)) - - # verify messages are forwarded to B3 only - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 20 or exchanges[0].msgRoutes != 20 or - exchanges[1].msgReceives != 20 or exchanges[1].msgRoutes != 30 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 20 or exchanges[3].msgRoutes != 20): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route more msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B3 only - for i in range(11, 21): - msg = queue_3.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - # cleanup - - self._brokers[3].client_session.exchange_unbind(queue="fedX1", exchange="fedX.topic", binding_key="spud.*") - self._brokers[3].client_session.message_cancel(destination="f1") - self._brokers[3].client_session.queue_delete(queue="fedX1") - - for _b in qmf.getObjects(_class="bridge"): - result = _b.close() - self.assertEqual(result.status, 0) - - for _l in qmf.getObjects(_class="link"): - result = _l.close() - self.assertEqual(result.status, 0) - - for _b in self._brokers: - _b.client_session.exchange_delete(exchange="fedX.topic") - - self._teardown_brokers() - - self.verify_cleanup() - - - def test_dynamic_fanout_route_prop(self): - """ Set up a tree of uni-directional routes across a fanout exchange. - Bind the same key to the same queues on the leaf nodes. Verify a - message sent with the routing key transverses the tree an arrives at - each leaf. Remove one leaf's queue, and verify that messages still - reach the other leaf. - - Route Topology: - - +---> B2 queue:"test-queue", binding key:"spud.*" - B0 --> B1 --+ - +---> B3 queue:"test-queue", binding key:"spud.*" - """ - session = self.session - - # create the federation - - self.startQmf() - qmf = self.qmf - - self._setup_brokers() - - # create fanout exchange on each broker, and retrieve the corresponding - # management object for that exchange - - exchanges=[] - for _b in self._brokers: - _b.client_session.exchange_declare(exchange="fedX.fanout", type="fanout") - self.assertEqual(_b.client_session.exchange_query(name="fedX.fanout").type, - "fanout", "exchange_declare failed!") - # pull the exchange out of qmf... - retries = 0 - my_exchange = None - while my_exchange is None: - objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange") - for ooo in objs: - if ooo.name == "fedX.fanout": - my_exchange = ooo - break - if my_exchange is None: - retries += 1 - self.failIfEqual(retries, 10, - "QMF failed to find new exchange!") - sleep(1) - exchanges.append(my_exchange) - - self.assertEqual(len(exchanges), len(self._brokers), "Exchange creation failed!") - - # connect B0 --> B1 - result = self._brokers[1].qmf_object.connect(self._brokers[0].host, - self._brokers[0].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B2 - result = self._brokers[2].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B3 - result = self._brokers[3].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # for each link, bridge the "fedX.fanout" exchanges: - - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker()))) - result = _l.bridge(False, # durable - "fedX.fanout", # src - "fedX.fanout", # dst - "", # key - "", # tag - "", # excludes - False, # srcIsQueue - False, # srcIsLocal - True, # dynamic - 0) # sync - self.assertEqual(result.status, 0) - - # wait for the inter-broker links to become operational - retries = 0 - operational = False - while not operational: - operational = True - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state))) - if _l.state != "Operational": - operational = False - if not operational: - retries += 1 - self.failIfEqual(retries, 10, - "inter-broker links failed to become operational.") - sleep(1) - - # @todo - There is no way to determine when the bridge objects become - # active. - sleep(6) - - # create a queue on B2, bound to the exchange - self._brokers[2].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[2].client_session.exchange_bind(queue="fedX1", exchange="fedX.fanout") - - # create a queue on B3, bound to the exchange - self._brokers[3].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - self._brokers[3].client_session.exchange_bind(queue="fedX1", exchange="fedX.fanout") - - # subscribe to messages arriving on B2's queue - self.subscribe(self._brokers[2].client_session, queue="fedX1", destination="f1") - queue_2 = self._brokers[2].client_session.incoming("f1") - - # subscribe to messages arriving on B3's queue - self.subscribe(self._brokers[3].client_session, queue="fedX1", destination="f1") - queue_3 = self._brokers[3].client_session.incoming("f1") - - # wait until the binding key has propagated to each broker (twice at - # broker B1). Work backwards from binding brokers. - - binding_counts = [1, 2, 1, 1] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(3,-1,-1): - retries = 0 - exchanges[i].update() - while exchanges[i].bindingCount < binding_counts[i]: - retries += 1 - self.failIfEqual(retries, 10, - "binding failed to propagate to broker %d" - % i) - sleep(3) - exchanges[i].update() - - # send 10 msgs from B0 - for i in range(1, 11): - dp = self._brokers[0].client_session.delivery_properties() - self._brokers[0].client_session.message_transfer(destination="fedX.fanout", message=Message(dp, "Message_frp %d" % i)) - - # wait for 10 messages to be forwarded from B0->B1, - # 10 messages from B1->B2, - # and 10 messages from B1->B3 - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 10 or exchanges[0].msgRoutes != 10 or - exchanges[1].msgReceives != 10 or exchanges[1].msgRoutes != 20 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 10 or exchanges[3].msgRoutes != 10): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B2 and B3 - for i in range(1, 11): - msg = queue_2.get(timeout=5) - self.assertEqual("Message_frp %d" % i, msg.body) - msg = queue_3.get(timeout=5) - self.assertEqual("Message_frp %d" % i, msg.body) - - try: - extra = queue_2.get(timeout=1) - self.fail("Got unexpected message in queue_2: " + extra.body) - except Empty: None - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - # tear down the queue on B2 - self._brokers[2].client_session.exchange_unbind(queue="fedX1", exchange="fedX.fanout") - self._brokers[2].client_session.message_cancel(destination="f1") - self._brokers[2].client_session.queue_delete(queue="fedX1") - - # wait for the binding count on B1 to drop from 2 to 1 - retries = 0 - exchanges[1].update() - while exchanges[1].bindingCount != 1: - retries += 1 - self.failIfEqual(retries, 10, - "unbinding failed to propagate to broker B1: %d" - % exchanges[1].bindingCount) - sleep(1) - exchanges[1].update() - - # send 10 msgs from B0 - for i in range(11, 21): - dp = self._brokers[0].client_session.delivery_properties() - self._brokers[0].client_session.message_transfer(destination="fedX.fanout", message=Message(dp, "Message_frp %d" % i)) - - # verify messages are forwarded to B3 only - retries = 0 - for ex in exchanges: - ex.update() - while (exchanges[0].msgReceives != 20 or exchanges[0].msgRoutes != 20 or - exchanges[1].msgReceives != 20 or exchanges[1].msgRoutes != 30 or - exchanges[2].msgReceives != 10 or exchanges[2].msgRoutes != 10 or - exchanges[3].msgReceives != 20 or exchanges[3].msgRoutes != 20): - retries += 1 - self.failIfEqual(retries, 10, - "federation failed to route more msgs %d:%d %d:%d %d:%d %d:%d" - % (exchanges[0].msgReceives, - exchanges[0].msgRoutes, - exchanges[1].msgReceives, - exchanges[1].msgRoutes, - exchanges[2].msgReceives, - exchanges[2].msgRoutes, - exchanges[3].msgReceives, - exchanges[3].msgRoutes)) - sleep(1) - for ex in exchanges: - ex.update() - - # get exactly 10 msgs on B3 only - for i in range(11, 21): - msg = queue_3.get(timeout=5) - self.assertEqual("Message_frp %d" % i, msg.body) - - try: - extra = queue_3.get(timeout=1) - self.fail("Got unexpected message in queue_3: " + extra.body) - except Empty: None - - # cleanup - - self._brokers[3].client_session.exchange_unbind(queue="fedX1", exchange="fedX.fanout") - self._brokers[3].client_session.message_cancel(destination="f1") - self._brokers[3].client_session.queue_delete(queue="fedX1") - - for _b in qmf.getObjects(_class="bridge"): - result = _b.close() - self.assertEqual(result.status, 0) - - for _l in qmf.getObjects(_class="link"): - result = _l.close() - self.assertEqual(result.status, 0) - - for _b in self._brokers: - _b.client_session.exchange_delete(exchange="fedX.fanout") - - self._teardown_brokers() - - 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 - - def test_dynamic_topic_bounce(self): - """ Bounce the connection between federated Topic Exchanges. - """ - class Params: - def exchange_type(self): return "topic" - def bind_queue(self, ssn, qname, ename): - ssn.exchange_bind(queue=qname, exchange=ename, - binding_key="spud.*") - def unbind_queue(self, ssn, qname, ename): - ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud.*") - def delivery_properties(self, ssn): - return ssn.delivery_properties(routing_key="spud.boy") - - self.generic_dynamic_bounce_test(Params()) - - def test_dynamic_direct_bounce(self): - """ Bounce the connection between federated Direct Exchanges. - """ - class Params: - def exchange_type(self): return "direct" - def bind_queue(self, ssn, qname, ename): - ssn.exchange_bind(queue=qname, exchange=ename, binding_key="spud") - def unbind_queue(self, ssn, qname, ename): - ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud") - def delivery_properties(self, ssn): - return ssn.delivery_properties(routing_key="spud") - self.generic_dynamic_bounce_test(Params()) - - def test_dynamic_fanout_bounce(self): - """ Bounce the connection between federated Fanout Exchanges. - """ - class Params: - def exchange_type(self): return "fanout" - def bind_queue(self, ssn, qname, ename): - ssn.exchange_bind(queue=qname, exchange=ename) - def unbind_queue(self, ssn, qname, ename): - ssn.exchange_unbind(queue=qname, exchange=ename) - def delivery_properties(self, ssn): - return ssn.delivery_properties(routing_key="spud") - self.generic_dynamic_bounce_test(Params()) - - def test_dynamic_headers_bounce(self): - """ Bounce the connection between federated Headers Exchanges. - """ - class Params: - def exchange_type(self): return "headers" - def bind_queue(self, ssn, qname, ename): - ssn.exchange_bind(queue=qname, exchange=ename, - binding_key="spud", arguments={'x-match':'any', 'class':'first'}) - def unbind_queue(self, ssn, qname, ename): - ssn.exchange_unbind(queue=qname, exchange=ename, binding_key="spud") - def delivery_properties(self, ssn): - return ssn.message_properties(application_headers={'class':'first'}) - ## @todo KAG - re-enable once federation bugs with headers exchanges - ## are fixed. - #self.generic_dynamic_bounce_test(Params()) - return - - - def generic_dynamic_bounce_test(self, params): - """ Verify that a federated broker can maintain a binding to a local - queue using the same key as a remote binding. Destroy and reconnect - the federation link, and verify routes are restored correctly. - See QPID-3170. - Topology: - - Queue1 <---"Key"---B0<==[Federated Exchange]==>B1---"Key"--->Queue2 - """ - session = self.session - - # create the federation - - self.startQmf() - qmf = self.qmf - - self._setup_brokers() - - # create exchange on each broker, and retrieve the corresponding - # management object for that exchange - - exchanges=[] - for _b in self._brokers[0:2]: - _b.client_session.exchange_declare(exchange="fedX", type=params.exchange_type()) - self.assertEqual(_b.client_session.exchange_query(name="fedX").type, - params.exchange_type(), "exchange_declare failed!") - # pull the exchange out of qmf... - retries = 0 - my_exchange = None - timeout = time() + 10 - while my_exchange is None and time() <= timeout: - objs = qmf.getObjects(_broker=_b.qmf_broker, _class="exchange") - for ooo in objs: - if ooo.name == "fedX": - my_exchange = ooo - break - if my_exchange is None: - self.fail("QMF failed to find new exchange!") - exchanges.append(my_exchange) - - # - # on each broker, create a local queue bound to the exchange with the - # same key value. - # - - self._brokers[0].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - params.bind_queue(self._brokers[0].client_session, "fedX1", "fedX") - self.subscribe(self._brokers[0].client_session, queue="fedX1", destination="f1") - queue_0 = self._brokers[0].client_session.incoming("f1") - - self._brokers[1].client_session.queue_declare(queue="fedX1", exclusive=True, auto_delete=True) - params.bind_queue(self._brokers[1].client_session, "fedX1", "fedX") - self.subscribe(self._brokers[1].client_session, queue="fedX1", destination="f1") - queue_1 = self._brokers[1].client_session.incoming("f1") - - # now federate the two brokers - - # connect B0 --> B1 - result = self._brokers[1].qmf_object.connect(self._brokers[0].host, - self._brokers[0].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # connect B1 --> B0 - result = self._brokers[0].qmf_object.connect(self._brokers[1].host, - self._brokers[1].port, - False, "PLAIN", "guest", "guest", "tcp") - self.assertEqual(result.status, 0) - - # for each link, bridge the "fedX" exchanges: - - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker()))) - result = _l.bridge(False, # durable - "fedX", # src - "fedX", # dst - "", # key - "", # tag - "", # excludes - False, # srcIsQueue - False, # srcIsLocal - True, # dynamic - 0) # sync - self.assertEqual(result.status, 0) - - # wait for all the inter-broker links to become operational - operational = False - timeout = time() + 10 - while not operational and time() <= timeout: - operational = True - for _l in qmf.getObjects(_class="link"): - #print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.state))) - if _l.state != "Operational": - operational = False - self.failUnless(operational, "inter-broker links failed to become operational.") - - # @todo - There is no way to determine when the bridge objects become - # active. - - # wait until the binding key has propagated to each broker - each - # broker should see 2 bindings (1 local, 1 remote) - - binding_counts = [2, 2] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(2): - exchanges[i].update() - timeout = time() + 10 - while exchanges[i].bindingCount < binding_counts[i] and time() <= timeout: - exchanges[i].update() - self.failUnless(exchanges[i].bindingCount == binding_counts[i]) - - # send 10 msgs to B0 - for i in range(1, 11): - # dp = self._brokers[0].client_session.delivery_properties(routing_key=params.routing_key()) - dp = params.delivery_properties(self._brokers[0].client_session) - self._brokers[0].client_session.message_transfer(destination="fedX", message=Message(dp, "Message_trp %d" % i)) - - # get exactly 10 msgs on B0's local queue and B1's queue - for i in range(1, 11): - try: - msg = queue_0.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - msg = queue_1.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - except Empty: - self.fail("Only got %d msgs - expected 10" % i) - try: - extra = queue_0.get(timeout=1) - self.fail("Got unexpected message in queue_0: " + extra.body) - except Empty: None - - try: - extra = queue_1.get(timeout=1) - self.fail("Got unexpected message in queue_1: " + extra.body) - except Empty: None - - # - # Tear down the bridges between the two exchanges, then wait - # for the bindings to be cleaned up - # - - for _b in qmf.getObjects(_class="bridge"): - result = _b.close() - self.assertEqual(result.status, 0) - - binding_counts = [1, 1] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(2): - exchanges[i].update() - timeout = time() + 10 - while exchanges[i].bindingCount != binding_counts[i] and time() <= timeout: - exchanges[i].update() - self.failUnless(exchanges[i].bindingCount == binding_counts[i]) - - # - # restore the bridges between the two exchanges, and wait for the - # bindings to propagate. - # - - for _l in qmf.getObjects(_class="link"): - # print("Link=%s:%s %s" % (_l.host, _l.port, str(_l.getBroker()))) - result = _l.bridge(False, # durable - "fedX", # src - "fedX", # dst - "", # key - "", # tag - "", # excludes - False, # srcIsQueue - False, # srcIsLocal - True, # dynamic - 0) # sync - self.assertEqual(result.status, 0) - - binding_counts = [2, 2] - self.assertEqual(len(binding_counts), len(exchanges), "Update Test!") - for i in range(2): - exchanges[i].update() - timeout = time() + 10 - while exchanges[i].bindingCount != binding_counts[i] and time() <= timeout: - exchanges[i].update() - self.failUnless(exchanges[i].bindingCount == binding_counts[i]) - - # - # verify traffic flows correctly - # - - for i in range(1, 11): - #dp = self._brokers[1].client_session.delivery_properties(routing_key=params.routing_key()) - dp = params.delivery_properties(self._brokers[1].client_session) - self._brokers[1].client_session.message_transfer(destination="fedX", message=Message(dp, "Message_trp %d" % i)) - - # get exactly 10 msgs on B0's queue and B1's queue - for i in range(1, 11): - try: - msg = queue_0.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - msg = queue_1.get(timeout=5) - self.assertEqual("Message_trp %d" % i, msg.body) - except Empty: - self.fail("Only got %d msgs - expected 10" % i) - try: - extra = queue_0.get(timeout=1) - self.fail("Got unexpected message in queue_0: " + extra.body) - except Empty: None - - try: - extra = queue_1.get(timeout=1) - self.fail("Got unexpected message in queue_1: " + extra.body) - except Empty: None - - - # - # cleanup - # - params.unbind_queue(self._brokers[0].client_session, "fedX1", "fedX") - self._brokers[0].client_session.message_cancel(destination="f1") - self._brokers[0].client_session.queue_delete(queue="fedX1") - - params.unbind_queue(self._brokers[1].client_session, "fedX1", "fedX") - self._brokers[1].client_session.message_cancel(destination="f1") - self._brokers[1].client_session.queue_delete(queue="fedX1") - - for _b in qmf.getObjects(_class="bridge"): - result = _b.close() - self.assertEqual(result.status, 0) - - for _l in qmf.getObjects(_class="link"): - result = _l.close() - self.assertEqual(result.status, 0) - - for _b in self._brokers[0:2]: - _b.client_session.exchange_delete(exchange="fedX") - - self._teardown_brokers() - - self.verify_cleanup() - - |