diff options
Diffstat (limited to 'trunk/qpid/cpp/src/tests/federation.py')
| -rwxr-xr-x | trunk/qpid/cpp/src/tests/federation.py | 597 |
1 files changed, 0 insertions, 597 deletions
diff --git a/trunk/qpid/cpp/src/tests/federation.py b/trunk/qpid/cpp/src/tests/federation.py deleted file mode 100755 index aa68e8198b..0000000000 --- a/trunk/qpid/cpp/src/tests/federation.py +++ /dev/null @@ -1,597 +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 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 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 |