diff options
Diffstat (limited to 'trunk/qpid/cpp/src/tests/federation.py')
| -rwxr-xr-x | trunk/qpid/cpp/src/tests/federation.py | 276 |
1 files changed, 0 insertions, 276 deletions
diff --git a/trunk/qpid/cpp/src/tests/federation.py b/trunk/qpid/cpp/src/tests/federation.py deleted file mode 100755 index b92df89839..0000000000 --- a/trunk/qpid/cpp/src/tests/federation.py +++ /dev/null @@ -1,276 +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, testrunner -from qpid.management import managementChannel, managementClient -from qpid.datatypes import Message -from qpid.queue import Empty -from time import sleep - -def add_module(args=sys.argv[1:]): - for a in args: - if a.startswith("federation"): - return False - return True - -def scan_args(name, default=None, args=sys.argv[1:]): - if (name in args): - pos = args.index(name) - return args[pos + 1] - elif default: - return default - else: - print "Please specify extra argument: %s" % name - sys.exit(2) - -def extract_args(name, args): - if (name in args): - pos = args.index(name) - del args[pos:pos+2] - else: - return None - -def remote_host(): - return scan_args("--remote-host", "localhost") - -def remote_port(): - return int(scan_args("--remote-port")) - -class Helper: - def __init__(self, parent): - self.parent = parent - self.session = parent.conn.session("Helper") - self.mc = managementClient(self.session.spec) - self.mch = self.mc.addChannel(self.session) - self.mc.syncWaitForStable(self.mch) - - def shutdown (self): - self.mc.removeChannel (self.mch) - - def get_objects(self, type): - return self.mc.syncGetObjects(self.mch, type) - - def get_object(self, type, position = 1, expected = None): - objects = self.get_objects(type) - if not expected: expected = position - self.assertEqual(len(objects), expected) - return objects[(position - 1)] - - - def call_method(self, object, method, args=None): - res = self.mc.syncCallMethod(self.mch, object.id, object.classKey, method, args) - self.assertEqual(res.status, 0) - self.assertEqual(res.statusText, "OK") - return res - - def assertEqual(self, a, b): - self.parent.assertEqual(a, b) - -class FederationTests(TestBase010): - - def test_bridge_create_and_close(self): - mgmt = Helper(self) - broker = mgmt.get_object("broker") - - mgmt.call_method(broker, "connect", {"host":remote_host(), "port":remote_port()}) - link = mgmt.get_object("link") - - mgmt.call_method(link, "bridge", {"durable":0, "src":"amq.direct", "dest":"amq.direct", "key":"my-key"}) - bridge = mgmt.get_object("bridge") - - mgmt.call_method(bridge, "close") - mgmt.call_method(link, "close") - - sleep(6) - self.assertEqual(len(mgmt.get_objects("bridge")), 0) - self.assertEqual(len(mgmt.get_objects("link")), 0) - - mgmt.shutdown () - - def test_pull_from_exchange(self): - session = self.session - - mgmt = Helper(self) - broker = mgmt.get_object("broker") - - mgmt.call_method(broker, "connect", {"host":remote_host(), "port":remote_port()}) - link = mgmt.get_object("link") - - mgmt.call_method(link, "bridge", {"durable":0, "src":"amq.direct", "dest":"amq.fanout", "key":"my-key"}) - bridge = mgmt.get_object("bridge") - - #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=remote_host(), port=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 - - mgmt.call_method(bridge, "close") - mgmt.call_method(link, "close") - sleep(6) - self.assertEqual(len(mgmt.get_objects("bridge")), 0) - self.assertEqual(len(mgmt.get_objects("link")), 0) - - mgmt.shutdown() - - def test_pull_from_queue(self): - session = self.session - - #setup queue on remote broker and add some messages - r_conn = self.connect(host=remote_host(), port=remote_port()) - r_session = r_conn.session("test_pull_from_queue") - r_session.queue_declare(queue="my-bridge-queue", exclusive=True, 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") - - mgmt = Helper(self) - broker = mgmt.get_object("broker") - - mgmt.call_method(broker, "connect", {"host":remote_host(), "port":remote_port()}) - link = mgmt.get_object("link") - - mgmt.call_method(link, "bridge", {"durable":0, "src":"my-bridge-queue", "dest":"amq.fanout", - "key":"", "tag":"", "excludes":"", "srcIsQueue":1}) - sleep(6) - bridge = mgmt.get_object("bridge") - - #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 - - - mgmt.call_method(bridge, "close") - mgmt.call_method(link, "close") - sleep(6) - self.assertEqual(len(mgmt.get_objects("bridge")), 0) - self.assertEqual(len(mgmt.get_objects("link")), 0) - - mgmt.shutdown () - - def test_tracing(self): - session = self.session - - mgmt = Helper(self) - broker = mgmt.get_object("broker") - - mgmt.call_method(broker, "connect", {"host":remote_host(), "port":remote_port()}) - link = mgmt.get_object("link") - - mgmt.call_method(link, "bridge", {"durable":0, "src":"amq.direct", "dest":"amq.fanout", "key":"my-key", - "tag":"my-bridge-id", "excludes":"exclude-me,also-exclude-me"}) - sleep(6) - bridge = mgmt.get_object("bridge") - - #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") - - #send messages to remote broker and confirm it is routed to local broker - r_conn = self.connect(host=remote_host(), port=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 - - mgmt.call_method(bridge, "close") - mgmt.call_method(link, "close") - sleep(6) - self.assertEqual(len(mgmt.get_objects("bridge")), 0) - self.assertEqual(len(mgmt.get_objects("link")), 0) - - mgmt.shutdown () - - 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 - -if __name__ == '__main__': - args = sys.argv[1:] - #need to remove the extra options from args as test runner doesn't recognise them - extract_args("--remote-port", args) - extract_args("--remote-host", args) - - if add_module(): - #add module(s) to run to testrunners args - args.append("federation") - - if not testrunner.run(args): sys.exit(1) |