204 lines
9.1 KiB
Python
204 lines
9.1 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.
|
|
#
|
|
import traceback
|
|
from qpid.queue import Empty
|
|
from qpid.datatypes import Message
|
|
from qpid.testlib import TestBase010
|
|
from qpid.session import SessionException
|
|
|
|
class AlternateExchangeTests(TestBase010):
|
|
"""
|
|
Tests for the new mechanism for message returns introduced in 0-10
|
|
and available in 0-9 for preview
|
|
"""
|
|
|
|
def test_unroutable(self):
|
|
"""
|
|
Test that unroutable messages are delivered to the alternate-exchange if specified
|
|
"""
|
|
session = self.session
|
|
#create an exchange with an alternate defined
|
|
session.exchange_declare(exchange="secondary", type="fanout")
|
|
session.exchange_declare(exchange="primary", type="direct", alternate_exchange="secondary")
|
|
|
|
#declare, bind (to the alternate exchange) and consume from a queue for 'returned' messages
|
|
session.queue_declare(queue="returns", exclusive=True, auto_delete=True)
|
|
session.exchange_bind(queue="returns", exchange="secondary")
|
|
session.message_subscribe(destination="a", queue="returns")
|
|
session.message_flow(destination="a", unit=session.credit_unit.message, value=0xFFFFFFFFL)
|
|
session.message_flow(destination="a", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
|
|
returned = session.incoming("a")
|
|
|
|
#declare, bind (to the primary exchange) and consume from a queue for 'processed' messages
|
|
session.queue_declare(queue="processed", exclusive=True, auto_delete=True)
|
|
session.exchange_bind(queue="processed", exchange="primary", binding_key="my-key")
|
|
session.message_subscribe(destination="b", queue="processed")
|
|
session.message_flow(destination="b", unit=session.credit_unit.message, value=0xFFFFFFFFL)
|
|
session.message_flow(destination="b", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
|
|
processed = session.incoming("b")
|
|
|
|
#publish to the primary exchange
|
|
#...one message that makes it to the 'processed' queue:
|
|
dp=self.session.delivery_properties(routing_key="my-key")
|
|
session.message_transfer(destination="primary", message=Message(dp, "Good"))
|
|
#...and one that does not:
|
|
dp=self.session.delivery_properties(routing_key="unused-key")
|
|
session.message_transfer(destination="primary", message=Message(dp, "Bad"))
|
|
|
|
#delete the exchanges
|
|
session.exchange_delete(exchange="primary")
|
|
session.exchange_delete(exchange="secondary")
|
|
|
|
#verify behaviour
|
|
self.assertEqual("Good", processed.get(timeout=1).body)
|
|
self.assertEqual("Bad", returned.get(timeout=1).body)
|
|
self.assertEmpty(processed)
|
|
self.assertEmpty(returned)
|
|
|
|
def test_queue_delete(self):
|
|
"""
|
|
Test that messages in a queue being deleted are delivered to the alternate-exchange if specified
|
|
"""
|
|
session = self.session
|
|
#set up a 'dead letter queue':
|
|
session.exchange_declare(exchange="dlq", type="fanout")
|
|
session.queue_declare(queue="deleted", exclusive=True, auto_delete=True)
|
|
session.exchange_bind(exchange="dlq", queue="deleted")
|
|
session.message_subscribe(destination="dlq", queue="deleted")
|
|
session.message_flow(destination="dlq", unit=session.credit_unit.message, value=0xFFFFFFFFL)
|
|
session.message_flow(destination="dlq", unit=session.credit_unit.byte, value=0xFFFFFFFFL)
|
|
dlq = session.incoming("dlq")
|
|
|
|
#create a queue using the dlq as its alternate exchange:
|
|
session.queue_declare(queue="delete-me", alternate_exchange="dlq")
|
|
#send it some messages:
|
|
dp=self.session.delivery_properties(routing_key="delete-me")
|
|
session.message_transfer(message=Message(dp, "One"))
|
|
session.message_transfer(message=Message(dp, "Two"))
|
|
session.message_transfer(message=Message(dp, "Three"))
|
|
#delete it:
|
|
session.queue_delete(queue="delete-me")
|
|
#delete the dlq exchange:
|
|
session.exchange_delete(exchange="dlq")
|
|
|
|
#check the messages were delivered to the dlq:
|
|
self.assertEqual("One", dlq.get(timeout=1).body)
|
|
self.assertEqual("Two", dlq.get(timeout=1).body)
|
|
self.assertEqual("Three", dlq.get(timeout=1).body)
|
|
self.assertEmpty(dlq)
|
|
|
|
def test_delete_while_used_by_queue(self):
|
|
"""
|
|
Ensure an exchange still in use as an alternate-exchange for a
|
|
queue can't be deleted
|
|
"""
|
|
session = self.session
|
|
session.exchange_declare(exchange="alternate", type="fanout")
|
|
|
|
session2 = self.conn.session("alternate", 2)
|
|
session2.queue_declare(queue="q", alternate_exchange="alternate")
|
|
try:
|
|
session2.exchange_delete(exchange="alternate")
|
|
self.fail("Expected deletion of in-use alternate-exchange to fail")
|
|
except SessionException, e:
|
|
session = self.session
|
|
session.queue_delete(queue="q")
|
|
session.exchange_delete(exchange="alternate")
|
|
self.assertEquals(530, e.args[0].error_code)
|
|
|
|
|
|
def test_delete_while_used_by_exchange(self):
|
|
"""
|
|
Ensure an exchange still in use as an alternate-exchange for
|
|
another exchange can't be deleted
|
|
"""
|
|
session = self.session
|
|
session.exchange_declare(exchange="alternate", type="fanout")
|
|
|
|
session = self.conn.session("alternate", 2)
|
|
session.exchange_declare(exchange="e", type="fanout", alternate_exchange="alternate")
|
|
try:
|
|
session.exchange_delete(exchange="alternate")
|
|
self.fail("Expected deletion of in-use alternate-exchange to fail")
|
|
except SessionException, e:
|
|
session = self.session
|
|
session.exchange_delete(exchange="e")
|
|
session.exchange_delete(exchange="alternate")
|
|
self.assertEquals(530, e.args[0].error_code)
|
|
|
|
|
|
def test_modify_existing_exchange_alternate(self):
|
|
"""
|
|
Ensure that attempting to modify an exhange to change
|
|
the alternate throws an exception
|
|
"""
|
|
session = self.session
|
|
session.exchange_declare(exchange="alt1", type="direct")
|
|
session.exchange_declare(exchange="alt2", type="direct")
|
|
session.exchange_declare(exchange="onealternate", type="fanout", alternate_exchange="alt1")
|
|
try:
|
|
# attempt to change the alternate on an already existing exchange
|
|
session.exchange_declare(exchange="onealternate", type="fanout", alternate_exchange="alt2")
|
|
self.fail("Expected changing an alternate on an existing exchange to fail")
|
|
except SessionException, e:
|
|
self.assertEquals(530, e.args[0].error_code)
|
|
session = self.conn.session("alternate", 2)
|
|
session.exchange_delete(exchange="onealternate")
|
|
session.exchange_delete(exchange="alt2")
|
|
session.exchange_delete(exchange="alt1")
|
|
|
|
|
|
def test_add_alternate_to_exchange(self):
|
|
"""
|
|
Ensure that attempting to modify an exhange by adding
|
|
an alternate throws an exception
|
|
"""
|
|
session = self.session
|
|
session.exchange_declare(exchange="alt1", type="direct")
|
|
session.exchange_declare(exchange="noalternate", type="fanout")
|
|
try:
|
|
# attempt to add an alternate on an already existing exchange
|
|
session.exchange_declare(exchange="noalternate", type="fanout", alternate_exchange="alt1")
|
|
self.fail("Expected adding an alternate on an existing exchange to fail")
|
|
except SessionException, e:
|
|
self.assertEquals(530, e.args[0].error_code)
|
|
session = self.conn.session("alternate", 2)
|
|
session.exchange_delete(exchange="noalternate")
|
|
session.exchange_delete(exchange="alt1")
|
|
|
|
|
|
def test_del_alternate_to_exchange(self):
|
|
"""
|
|
Ensure that attempting to modify an exhange by declaring
|
|
it again without an alternate does nothing
|
|
"""
|
|
session = self.session
|
|
session.exchange_declare(exchange="alt1", type="direct")
|
|
session.exchange_declare(exchange="onealternate", type="fanout", alternate_exchange="alt1")
|
|
# attempt to re-declare without an alternate - silently ignore
|
|
session.exchange_declare(exchange="onealternate", type="fanout" )
|
|
session.exchange_delete(exchange="onealternate")
|
|
session.exchange_delete(exchange="alt1")
|
|
|
|
|
|
def assertEmpty(self, queue):
|
|
try:
|
|
msg = queue.get(timeout=1)
|
|
self.fail("Queue not empty: " + msg)
|
|
except Empty: None
|