diff options
Diffstat (limited to 'cpp/src/tests/federation.py')
| -rwxr-xr-x | cpp/src/tests/federation.py | 191 |
1 files changed, 191 insertions, 0 deletions
diff --git a/cpp/src/tests/federation.py b/cpp/src/tests/federation.py new file mode 100755 index 0000000000..28ebc4a24d --- /dev/null +++ b/cpp/src/tests/federation.py @@ -0,0 +1,191 @@ +#!/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 TestBase, testrunner +from qpid.management import managementChannel, managementClient +from qpid.queue import Empty +from qpid.content import Content + + +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.channel = parent.client.channel(2) + self.mc = managementClient(self.channel.spec) + self.mch = self.mc.addChannel(self.channel) + self.mc.syncWaitForStable(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(TestBase): + + def test_bridge_create_and_close(self): + mgmt = Helper(self) + broker = mgmt.get_object("broker") + + for i in range(10): + mgmt.call_method(broker, "connect", {"host":remote_host(), "port":remote_port()}) + link = mgmt.get_object("link") + + mgmt.call_method(link, "bridge", {"src":"amq.direct", "dest":"amq.direct", "key":"my-key"}) + bridge = mgmt.get_object("bridge") + + mgmt.call_method(bridge, "close") + self.assertEqual(len(mgmt.get_objects("bridge")), 0) + + mgmt.call_method(link, "close") + self.assertEqual(len(mgmt.get_objects("link")), 0) + + def test_pull_from_exchange(self): + channel = self.channel + + 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", {"src":"amq.direct", "dest":"amq.fanout", "key":"my-key"}) + bridge = mgmt.get_object("bridge") + + #setup queue to receive messages from local broker + channel.queue_declare(queue="fed1", exclusive=True, auto_delete=True) + channel.queue_bind(queue="fed1", exchange="amq.fanout") + self.subscribe(queue="fed1", destination="f1") + queue = self.client.queue("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_channel = r_conn.channel(1) + r_channel.session_open() + + for i in range(1, 11): + r_channel.message_transfer(destination="amq.direct", content=Content(properties={'routing_key' : "my-key"}, body="Message %d" % i)) + + for i in range(1, 11): + msg = queue.get(timeout=5) + self.assertEqual("Message %d" % i, msg.content.body) + try: + extra = queue.get(timeout=1) + self.fail("Got unexpected message in queue: " + extra.content.body) + except Empty: None + + + mgmt.call_method(bridge, "close") + self.assertEqual(len(mgmt.get_objects("bridge")), 0) + + mgmt.call_method(link, "close") + self.assertEqual(len(mgmt.get_objects("link")), 0) + + def test_pull_from_queue(self): + channel = self.channel + + #setup queue on remote broker and add some messages + r_conn = self.connect(host=remote_host(), port=remote_port()) + r_channel = r_conn.channel(1) + r_channel.session_open() + r_channel.queue_declare(queue="my-bridge-queue", exclusive=True, auto_delete=True) + for i in range(1, 6): + r_channel.message_transfer(content=Content(properties={'routing_key' : "my-bridge-queue"}, body="Message %d" % i)) + + #setup queue to receive messages from local broker + channel.queue_declare(queue="fed1", exclusive=True, auto_delete=True) + channel.queue_bind(queue="fed1", exchange="amq.fanout") + self.subscribe(queue="fed1", destination="f1") + queue = self.client.queue("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", {"src":"my-bridge-queue", "dest":"amq.fanout", "key":"", "src_is_queue":1}) + bridge = mgmt.get_object("bridge") + + #add some more messages (i.e. after bridge was created) + for i in range(6, 11): + r_channel.message_transfer(content=Content(properties={'routing_key' : "my-bridge-queue"}, body="Message %d" % i)) + + for i in range(1, 11): + msg = queue.get(timeout=5) + self.assertEqual("Message %d" % i, msg.content.body) + try: + extra = queue.get(timeout=1) + self.fail("Got unexpected message in queue: " + extra.content.body) + except Empty: None + + + mgmt.call_method(bridge, "close") + self.assertEqual(len(mgmt.get_objects("bridge")), 0) + + mgmt.call_method(link, "close") + self.assertEqual(len(mgmt.get_objects("link")), 0) + +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) + #add module(s) to run to testrunners args + args.append("federation") + + if not testrunner.run(args): sys.exit(1) |