Fixes and automated tests for federation function.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@646505 13f79535-47bb-0310-9956-ffa450edef68

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)