[rhmessaging-commits] rhmessaging commits: r3751 - store/trunk/cpp/tests/new_python_tests.

Author: kpvdr Date: 2009-12-18 13:07:38 -0500 (Fri, 18 Dec 2009) New Revision: 3751 Added: store/trunk/cpp/tests/new_python_tests/client_persistence.py Removed: store/trunk/cpp/tests/new_python_tests/client_persistence_tests.py Modified: store/trunk/cpp/tests/new_python_tests/__init__.py Log: Updated tests for AlternateExchange; added test for redelivered flag. Modified: store/trunk/cpp/tests/new_python_tests/__init__.py =================================================================== --- store/trunk/cpp/tests/new_python_tests/__init__.py 2009-12-16 19:39:16 UTC (rev 3750) +++ store/trunk/cpp/tests/new_python_tests/__init__.py 2009-12-18 18:07:38 UTC (rev 3751) @@ -21,4 +21,4 @@ # # The GNU Lesser General Public License is available in the file COPYING. -from client_persistence_tests import * +from client_persistence import * Copied: store/trunk/cpp/tests/new_python_tests/client_persistence.py (from rev 3750, store/trunk/cpp/tests/new_python_tests/client_persistence_tests.py) =================================================================== --- store/trunk/cpp/tests/new_python_tests/client_persistence.py (rev 0) +++ store/trunk/cpp/tests/new_python_tests/client_persistence.py 2009-12-18 18:07:38 UTC (rev 3751) @@ -0,0 +1,136 @@ +# Copyright (c) 2008 Red Hat, Inc. +# +# This file is part of the Qpid async store library msgstore.so. +# +# This library is free software; you can redistribute it and/or +# modify it under the terms of the GNU Lesser General Public +# License as published by the Free Software Foundation; either +# version 2.1 of the License, or (at your option) any later version. +# +# This library is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU +# Lesser General Public License for more details. +# +# You should have received a copy of the GNU Lesser General Public +# License along with this library; if not, write to the Free Software +# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 +# USA +# +# The GNU Lesser General Public License is available in the file COPYING. + +from qpid.brokertest import * +from qpid.messaging import Message +from qmf.console import Session + +class Qmf: + """ + QMF functions not yet available in the new QMF API. Remove this and replace with new API when it becomes available. + """ + def __init__(self, broker): + self.__session = Session() + self.__broker = self.__session.addBroker("amqp://localhost:%d"%broker.port()) + + def addExchange(self, exchangeName, exchangeType, altExchangeName=None, passive=False, durable=False, arguments = {}): + """Add a new exchange""" + amqpSession = self.__broker.getAmqpSession() + if altExchangeName: + amqpSession.exchange_declare(exchange=exchangeName, type=exchangeType, alternate_exchange=altExchangeName, passive=passive, durable=durable, arguments=arguments) + else: + amqpSession.exchange_declare(exchange=exchangeName, type=exchangeType, passive=passive, durable=durable, arguments=arguments) + + def addQueue(self, queueName, altExchangeName=None, passive=False, durable=False, arguments = {}): + """Add a new queue""" + amqpSession = self.__broker.getAmqpSession() + if altExchangeName: + amqpSession = amqpSession.queue_declare(queueName, alternate_exchange=altExchangeName, passive=passive, durable=durable, arguments=arguments) + else: + amqpSession = amqpSession.queue_declare(queueName, passive=passive, durable=durable, arguments=arguments) + + def __query(self, name, _class, package, altExchangeName=None): + try: + objList = self.__session.getObjects(_class=_class, _package=package) + found = False + for o in objList: + if o.name == name: + found = True + if altExchangeName != None: + altExchList = self.__session.getObjects(_objectId=o.altExchange) + if len(altExchList) == 0 or altExchList[0].name != altExchangeName: return False + break + return found + except: return False + + + def queryExchange(self, exchangeName, altExchangeName=None): + """Test for the presence of an exchange, and optionally whether it has an alternate exchange set to a known value.""" + return self.__query(exchangeName, "exchange", "org.apache.qpid.broker", altExchangeName) + + def queryQueue(self, queueName, altExchangeName=None): + """Test for the presence of an exchange, and optionally whether it has an alternate exchange set to a known value.""" + return self.__query(queueName, "queue", "org.apache.qpid.broker", altExchangeName) + + +class AlternateExchagePropertyTests(BrokerTest): + """ + Test the persistence of the Alternate Exchange property for exchanges and queues. + """ + + def __args(self): + assert BrokerTest.store_lib + return ["--load-module", BrokerTest.store_lib] + + def testExchange(self): + """Exchange alternate exchange property persistence test""" + broker = self.broker(self.__args(), name="testBroker", expect=EXPECT_EXIT_FAIL) + qmf = Qmf(broker) + qmf.addExchange("altExch", "direct", durable=True) # Serves as alternate exchange instance + qmf.addExchange("testExch", "direct", durable=True, altExchangeName="altExch") + broker.kill() + + broker = self.broker(self.__args(), name="testBroker") + qmf = Qmf(broker) + try: qmf.addExchange("altExch", "direct", passive=True) + except Exception, e: self.fail("Alternate exchange (\"altExch\") instance not recovered: %s" % e) + try: qmf.addExchange("testExch", "direct", passive=True) + except Exception, e: self.fail("Test exchange (\"testExch\") instance not recovered: %s" % e) + self.assertTrue(qmf.queryExchange("testExch", altExchangeName = "altExch"), "Alternate exchange property not found or is incorrect on exchange \"testExch\".") + + def testQueue(self): + """Queue alternate exchange property persistexchangeNamece test""" + broker = self.broker(self.__args(), name="testBroker", expect=EXPECT_EXIT_FAIL) + qmf = Qmf(broker) + qmf.addExchange("altExch", "direct", durable=True) # Serves as alternate exchange instance + qmf.addQueue("testQueue", durable=True, altExchangeName="altExch") + broker.kill() + + broker = self.broker(self.__args(), name="testBroker") + qmf = Qmf(broker) + try: qmf.addExchange("altExch", "direct", passive=True) + except Exception, e: self.fail("Alternate exchange (\"altExch\") instance not recovered: %s" % e) + try: qmf.addQueue("testQueue", passive=True) + except Exception, e: self.fail("Test queue (\"testQueue\") instance not recovered: %s" % e) + self.assertTrue(qmf.queryQueue("testQueue", altExchangeName = "altExch"), "Alternate exchange property not found or is incorrect on queue \"testQueue\".") + + +class RedeliveredTests(BrokerTest): + """ + Test the behavior of the redelivered flag in the context of persistence + """ + + def __args(self): + assert BrokerTest.store_lib + return ["--load-module", BrokerTest.store_lib] + + def testBrokerRecovery(self): + """Test that the redelivered flag is set on messages after recovery of broker""" + broker = self.broker(self.__args(), name="testAfterRecover", expect=EXPECT_EXIT_FAIL) + mc = "xyz"*100 + m = Message(mc, durable=True) + broker.send_message("testQueue", m) + broker.kill() + broker = self.broker(self.__args(), name="testAfterRecover") + rm = broker.get_message("testQueue") + self.assertEqual(mc, rm.content) + self.assertTrue(rm.redelivered) + \ No newline at end of file Deleted: store/trunk/cpp/tests/new_python_tests/client_persistence_tests.py =================================================================== --- store/trunk/cpp/tests/new_python_tests/client_persistence_tests.py 2009-12-16 19:39:16 UTC (rev 3750) +++ store/trunk/cpp/tests/new_python_tests/client_persistence_tests.py 2009-12-18 18:07:38 UTC (rev 3751) @@ -1,94 +0,0 @@ -# Copyright (c) 2008 Red Hat, Inc. -# -# This file is part of the Qpid async store library msgstore.so. -# -# This library is free software; you can redistribute it and/or -# modify it under the terms of the GNU Lesser General Public -# License as published by the Free Software Foundation; either -# version 2.1 of the License, or (at your option) any later version. -# -# This library is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public -# License along with this library; if not, write to the Free Software -# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 -# USA -# -# The GNU Lesser General Public License is available in the file COPYING. - -from qpid.brokertest import * -from qpid.messaging import Message -from qmf.console import Session - - - -class AlternateExchageTests(BrokerTest): - """ - Test the persistence of the Alternate Exchange property for exchanges and queues. - """ - - def args(self): - assert BrokerTest.store_lib - return ["--load-module", BrokerTest.store_lib] - - def testExchangeAlternateExchangeProperty(self): - """Exchange alternate exchange property persistence test""" - - brokerName = "TestQueueAlternateExchange" - exchangeName = "TestExchange" - alternateExchangeName = "TestAlternateExchange" - - broker = self.broker(self.args(), name=brokerName, expect=EXPECT_EXIT_FAIL) - # Use old API until new one can handle these operations - connection = broker.connect_old() - session = connection.session(str(qpid.datatypes.uuid4())) - session.exchange_declare(exchange=alternateExchangeName, type="direct", durable=True) - session.exchange_declare(exchange=exchangeName, type="direct", alternate_exchange=alternateExchangeName, durable=True) - connection.close() - broker.kill() - - broker = self.broker(self.args(), name=brokerName) - connection = broker.connect_old() - session = connection.session(str(qpid.datatypes.uuid4())) - session.exchange_declare(exchange=alternateExchangeName, type="direct", durable=True, passive=True) - session.exchange_declare(exchange=exchangeName, type="direct", durable=True, passive=True) - # There is no AMQP support for directly querying exchanges for alternate-exchange (unlike queues)!!, - # so use QMF instead. - qmfSession = Session() - qmfBroker = qmfSession.addBroker("amqp://localhost:%d"%broker.port()) - exchanges = qmfSession.getObjects(_class="exchange", _package="org.apache.qpid.broker") - for e in exchanges: - if e.name == exchangeName: - altExch = qmfSession.getObjects(_objectId=e.altExchange) - alternateExchangeNameProps = altExch[0].getProperties() - self.assertEqual(altExch[0].name, alternateExchangeName) - - connection.close() - - def testQueueAlternateExchangeProperty(self): - """Queue alternate exchange property persistexchangeNamece test""" - - brokerName = "TestQueueAlternateExchange" - queueName = "TestQueue" - alternateExchangeName = "TestAlternateExchange" - - broker = self.broker(self.args(), name=brokerName, expect=EXPECT_EXIT_FAIL) - # Use old API until new one can handle these operations - connection = broker.connect_old() - session = connection.session(str(qpid.datatypes.uuid4())) - session.exchange_declare(exchange=alternateExchangeName, type="direct", durable=True) - session.queue_declare(queue=queueName, alternate_exchange=alternateExchangeName, durable=True) - connection.close() - broker.kill() - - broker = self.broker(self.args(), name=brokerName) - connection = broker.connect_old() - session = connection.session(str(qpid.datatypes.uuid4())) - session.exchange_declare(exchange=alternateExchangeName, type="direct", durable=True, passive=True) - session.queue_declare(queue=queueName, alternate_exchange=alternateExchangeName, durable=True, passive=True) - self.assertEqual(session.queue_query(queue=queueName)["alternate_exchange"], alternateExchangeName) - - connection.close()