112 lines
4.3 KiB
Python
112 lines
4.3 KiB
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.
|
||
|
|
#
|
||
|
|
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])
|