diff options
Diffstat (limited to 'qpid/ruby/tests/qmf.rb')
| -rw-r--r-- | qpid/ruby/tests/qmf.rb | 187 |
1 files changed, 187 insertions, 0 deletions
diff --git a/qpid/ruby/tests/qmf.rb b/qpid/ruby/tests/qmf.rb new file mode 100644 index 0000000000..75250a7938 --- /dev/null +++ b/qpid/ruby/tests/qmf.rb @@ -0,0 +1,187 @@ +# +# 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. +# + +require "test/unit" +require "qpid" +require "tests/util" +require "socket" + +class QmfTest < Test::Unit::TestCase + + def setup() + # Make sure errors in threads lead to a noisy death of the test + Thread.abort_on_exception = true + + sock = TCPSocket.new('localhost', 5672) + + @conn = Qpid::Connection.new(sock) + @conn.start() + + @session = @conn.session("test-session") + + # It's a bit odd that we're using two connections but that's the way + # the python one works afaict. + @qmf = Qpid::Qmf::Session.new() + @qmf_broker = @qmf.add_broker("amqp://localhost:5672") + + brokers = @qmf.objects(:class => "broker") + assert_equal(1, brokers.length) + @broker = brokers[0] + end + + def teardown + unless @session.error? + @session.close(10) + end + @conn.close(10) + if @qmf + @qmf.del_broker(@qmf_broker) + end + end + + def test_broker_connectivity() + body = "Echo Message Body" + for seq in 1..10 + res = @broker.echo(seq, body) + assert_equal(0, res.status) + assert_equal("OK", res.text) + assert_equal(seq, res.sequence) + assert_equal(body, res.body) + end + end + + def test_move_queued_messages() + """ + Test ability to move messages from the head of one queue to another. + Need to test moveing all and N messages. + """ + + "Set up source queue" + @session.queue_declare(:queue => "src-queue", :exclusive => true, :auto_delete => true) + @session.exchange_bind(:queue => "src-queue", :exchange => "amq.direct", :binding_key => "routing_key") + + props = @session.delivery_properties(:routing_key => "routing_key") + for count in 1..20 + body = "Move Message %d" % count + src_msg = Qpid::Message.new(props, body) + @session.message_transfer(:destination => "amq.direct", :message => src_msg) + end + + "Set up destination queue" + @session.queue_declare(:queue => "dest-queue", :exclusive => true, :auto_delete => true) + @session.exchange_bind(:queue => "dest-queue", :exchange => "amq.direct") + + queues = @qmf.objects(:class => "queue") + + "Move 10 messages from src-queue to dest-queue" + result = @qmf.objects(:class => "broker")[0].queueMoveMessages("src-queue", "dest-queue", 10) + assert_equal(0, result.status) + + sq = @qmf.objects(:class => "queue", "name" => "src-queue")[0] + dq = @qmf.objects(:class => "queue", "name" => "dest-queue")[0] + + assert_equal(10, sq.msgDepth) + assert_equal(10, dq.msgDepth) + + "Move all remaining messages to destination" + result = @qmf.objects(:class => "broker")[0].queueMoveMessages("src-queue", "dest-queue", 0) + assert_equal(0, result.status) + + sq = @qmf.objects(:class => "queue", 'name' => "src-queue")[0] + dq = @qmf.objects(:class => "queue", 'name' => "dest-queue")[0] + + assert_equal(0, sq.msgDepth) + assert_equal(20, dq.msgDepth) + + "Use a bad source queue name" + result = @qmf.objects(:class => "broker")[0].queueMoveMessages("bad-src-queue", "dest-queue", 0) + assert_equal(4, result.status) + + "Use a bad destination queue name" + result = @qmf.objects(:class => "broker")[0].queueMoveMessages("src-queue", "bad-dest-queue", 0) + assert_equal(4, result.status) + + " Use a large qty (40) to move from dest-queue back to " + " src-queue- should move all " + result = @qmf.objects(:class => "broker")[0].queueMoveMessages("dest-queue", "src-queue", 40) + assert_equal(0, result.status) + + sq = @qmf.objects(:class => "queue", 'name' => "src-queue")[0] + dq = @qmf.objects(:class => "queue", 'name' => "dest-queue")[0] + + assert_equal(20, sq.msgDepth) + assert_equal(0, dq.msgDepth) + + "Consume the messages of the queue and check they are all there in order" + @session.message_subscribe(:queue => "src-queue", + :destination => "tag") + @session.message_flow(:destination => "tag", + :unit => @session.message_credit_unit.message, + :value => 0xFFFFFFFF) + @session.message_flow(:destination => "tag", + :unit => @session.message_credit_unit.byte, + :value => 0xFFFFFFFF) + queue = @session.incoming("tag") + for count in 1..20 + consumed_msg = queue.get(timeout=1) + body = "Move Message %d" % count + assert_equal(body, consumed_msg.body) + end + end + + # Test ability to purge messages from the head of a queue. Need to test + # moveing all, 1 (top message) and N messages. + def test_purge_queue + # Set up purge queue" + @session.queue_declare(:queue => "purge-queue", + :exclusive => true, + :auto_delete => true) + @session.exchange_bind(:queue => "purge-queue", + :exchange => "amq.direct", + :binding_key => "routing_key") + + props = @session.delivery_properties(:routing_key => "routing_key") + 20.times do |count| + body = "Purge Message %d" % count + msg = Qpid::Message.new(props, body) + @session.message_transfer(:destination => "amq.direct", + :message => msg) + end + + pq = @qmf.objects(:class => "queue", 'name' => "purge-queue")[0] + + "Purge top message from purge-queue" + result = pq.purge(1) + assert_equal(0, result.status) + pq = @qmf.objects(:class => "queue", 'name' => "purge-queue")[0] + assert_equal(19, pq.msgDepth) + + "Purge top 9 messages from purge-queue" + result = pq.purge(9) + assert_equal(0, result.status) + pq = @qmf.objects(:class => "queue", 'name' => "purge-queue")[0] + assert_equal(10, pq.msgDepth) + + "Purge all messages from purge-queue" + result = pq.purge(0) + assert_equal(0, result.status) + pq = @qmf.objects(:class => "queue", 'name' => "purge-queue")[0] + assert_equal(0, pq.msgDepth) + end +end |