diff options
Diffstat (limited to 'trunk/qpid/python/tests_0-9/queue.py')
| -rw-r--r-- | trunk/qpid/python/tests_0-9/queue.py | 111 |
1 files changed, 0 insertions, 111 deletions
diff --git a/trunk/qpid/python/tests_0-9/queue.py b/trunk/qpid/python/tests_0-9/queue.py deleted file mode 100644 index de1153307c..0000000000 --- a/trunk/qpid/python/tests_0-9/queue.py +++ /dev/null @@ -1,111 +0,0 @@ -# -# 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]) |