diff options
author | Rajith Muditha Attapattu <rajith@apache.org> | 2011-05-27 15:44:23 +0000 |
---|---|---|
committer | Rajith Muditha Attapattu <rajith@apache.org> | 2011-05-27 15:44:23 +0000 |
commit | 66765100f4257159622cefe57bed50125a5ad017 (patch) | |
tree | a88ee23bb194eb91f0ebb2d9b23ff423e3ea8e37 /qpid/tests/src/py/qpid_tests/broker_0_9/queue.py | |
parent | 1aeaa7b16e5ce54f10c901d75c4d40f9f88b9db6 (diff) | |
parent | 88b98b2f4152ef59a671fad55a0d08338b6b78ca (diff) | |
download | qpid-python-rajith_jms_client.tar.gz |
Creating a branch for experimenting with some ideas for JMS client.rajith_jms_client
git-svn-id: https://svn.apache.org/repos/asf/qpid/branches/rajith_jms_client@1128369 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'qpid/tests/src/py/qpid_tests/broker_0_9/queue.py')
-rw-r--r-- | qpid/tests/src/py/qpid_tests/broker_0_9/queue.py | 111 |
1 files changed, 111 insertions, 0 deletions
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_9/queue.py b/qpid/tests/src/py/qpid_tests/broker_0_9/queue.py new file mode 100644 index 0000000000..de1153307c --- /dev/null +++ b/qpid/tests/src/py/qpid_tests/broker_0_9/queue.py @@ -0,0 +1,111 @@ +# +# 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. +# +from qpid.client import Client, Closed +from qpid.queue import Empty +from qpid.content import Content +from qpid.testlib import TestBase + +class QueueTests(TestBase): + """Tests for 'methods' on the amqp queue 'class'""" + + def test_unbind_direct(self): + self.unbind_test(exchange="amq.direct", routing_key="key") + + def test_unbind_topic(self): + self.unbind_test(exchange="amq.topic", routing_key="key") + + def test_unbind_fanout(self): + self.unbind_test(exchange="amq.fanout") + + def test_unbind_headers(self): + self.unbind_test(exchange="amq.match", args={ "x-match":"all", "a":"b"}, headers={"a":"b"}) + + def unbind_test(self, exchange, routing_key="", args=None, headers={}): + #bind two queues and consume from them + channel = self.channel + + channel.queue_declare(queue="queue-1", exclusive="True") + channel.queue_declare(queue="queue-2", exclusive="True") + + channel.basic_consume(queue="queue-1", consumer_tag="queue-1", no_ack=True) + channel.basic_consume(queue="queue-2", consumer_tag="queue-2", no_ack=True) + + queue1 = self.client.queue("queue-1") + queue2 = self.client.queue("queue-2") + + channel.queue_bind(exchange=exchange, queue="queue-1", routing_key=routing_key, arguments=args) + channel.queue_bind(exchange=exchange, queue="queue-2", routing_key=routing_key, arguments=args) + + #send a message that will match both bindings + channel.basic_publish(exchange=exchange, routing_key=routing_key, + content=Content("one", properties={"headers": headers})) + + #unbind first queue + channel.queue_unbind(exchange=exchange, queue="queue-1", routing_key=routing_key, arguments=args) + + #send another message + channel.basic_publish(exchange=exchange, routing_key=routing_key, + content=Content("two", properties={"headers": headers})) + + #check one queue has both messages and the other has only one + self.assertEquals("one", queue1.get(timeout=1).content.body) + try: + msg = queue1.get(timeout=1) + self.fail("Got extra message: %s" % msg.body) + except Empty: pass + + self.assertEquals("one", queue2.get(timeout=1).content.body) + self.assertEquals("two", queue2.get(timeout=1).content.body) + try: + msg = queue2.get(timeout=1) + self.fail("Got extra message: " + msg) + except Empty: pass + + def test_autodelete_shared(self): + """ + Test auto-deletion (of non-exclusive queues) + """ + channel = self.channel + other = self.connect() + channel2 = other.channel(1) + channel2.channel_open() + + channel.queue_declare(queue="auto-delete-me", auto_delete=True) + + #consume from both channels + reply = channel.basic_consume(queue="auto-delete-me", no_ack=True) + channel2.basic_consume(queue="auto-delete-me", no_ack=True) + + #implicit cancel + channel2.channel_close() + + #check it is still there + channel.queue_declare(queue="auto-delete-me", passive=True) + + #explicit cancel => queue is now unused again: + channel.basic_cancel(consumer_tag=reply.consumer_tag) + + #NOTE: this assumes there is no timeout in use + + #check that it has gone be declaring passively + try: + channel.queue_declare(queue="auto-delete-me", passive=True) + self.fail("Expected queue to have been deleted") + except Closed, e: + self.assertChannelException(404, e.args[0]) |