diff options
Diffstat (limited to 'qpid/tests/src/py/qpid_tests/broker_0_10/tx.py')
| -rw-r--r-- | qpid/tests/src/py/qpid_tests/broker_0_10/tx.py | 265 |
1 files changed, 265 insertions, 0 deletions
diff --git a/qpid/tests/src/py/qpid_tests/broker_0_10/tx.py b/qpid/tests/src/py/qpid_tests/broker_0_10/tx.py new file mode 100644 index 0000000000..8cdc539a08 --- /dev/null +++ b/qpid/tests/src/py/qpid_tests/broker_0_10/tx.py @@ -0,0 +1,265 @@ +# +# 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.datatypes import Message, RangedSet +from qpid.testlib import TestBase010 + +class TxTests(TestBase010): + """ + Tests for 'methods' on the amqp tx 'class' + """ + + def test_commit(self): + """ + Test that commited publishes are delivered and commited acks are not re-delivered + """ + session = self.session + + #declare queues and create subscribers in the checking session + #to ensure that the queues are not auto-deleted too early: + self.declare_queues(["tx-commit-a", "tx-commit-b", "tx-commit-c"]) + session.message_subscribe(queue="tx-commit-a", destination="qa") + session.message_subscribe(queue="tx-commit-b", destination="qb") + session.message_subscribe(queue="tx-commit-c", destination="qc") + + #use a separate session for actual work + session2 = self.conn.session("worker", 2) + self.perform_txn_work(session2, "tx-commit-a", "tx-commit-b", "tx-commit-c") + session2.tx_commit() + session2.close() + + session.tx_select() + + self.enable_flow("qa") + queue_a = session.incoming("qa") + + self.enable_flow("qb") + queue_b = session.incoming("qb") + + self.enable_flow("qc") + queue_c = session.incoming("qc") + + #check results + for i in range(1, 5): + msg = queue_c.get(timeout=1) + self.assertEqual("TxMessage %d" % i, msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_b.get(timeout=1) + self.assertEqual("TxMessage 6", msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_a.get(timeout=1) + self.assertEqual("TxMessage 7", msg.body) + session.message_accept(RangedSet(msg.id)) + + for q in [queue_a, queue_b, queue_c]: + try: + extra = q.get(timeout=1) + self.fail("Got unexpected message: " + extra.body) + except Empty: None + + #cleanup + session.tx_commit() + + def test_auto_rollback(self): + """ + Test that a session closed with an open transaction is effectively rolled back + """ + session = self.session + self.declare_queues(["tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c"]) + session.message_subscribe(queue="tx-autorollback-a", destination="qa") + session.message_subscribe(queue="tx-autorollback-b", destination="qb") + session.message_subscribe(queue="tx-autorollback-c", destination="qc") + + session2 = self.conn.session("worker", 2) + queue_a, queue_b, queue_c, ignore = self.perform_txn_work(session2, "tx-autorollback-a", "tx-autorollback-b", "tx-autorollback-c") + + for q in [queue_a, queue_b, queue_c]: + try: + extra = q.get(timeout=1) + self.fail("Got unexpected message: " + extra.body) + except Empty: None + + session2.close() + + session.tx_select() + + self.enable_flow("qa") + queue_a = session.incoming("qa") + + self.enable_flow("qb") + queue_b = session.incoming("qb") + + self.enable_flow("qc") + queue_c = session.incoming("qc") + + #check results + for i in range(1, 5): + msg = queue_a.get(timeout=1) + self.assertEqual("Message %d" % i, msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_b.get(timeout=1) + self.assertEqual("Message 6", msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_c.get(timeout=1) + self.assertEqual("Message 7", msg.body) + session.message_accept(RangedSet(msg.id)) + + for q in [queue_a, queue_b, queue_c]: + try: + extra = q.get(timeout=1) + self.fail("Got unexpected message: " + extra.body) + except Empty: None + + #cleanup + session.tx_commit() + + def test_rollback(self): + """ + Test that rolled back publishes are not delivered and rolled back acks are re-delivered + """ + session = self.session + queue_a, queue_b, queue_c, consumed = self.perform_txn_work(session, "tx-rollback-a", "tx-rollback-b", "tx-rollback-c") + + for q in [queue_a, queue_b, queue_c]: + try: + extra = q.get(timeout=1) + self.fail("Got unexpected message: " + extra.body) + except Empty: None + + session.tx_rollback() + + #need to release messages to get them redelivered now: + session.message_release(consumed) + + #check results + for i in range(1, 5): + msg = queue_a.get(timeout=1) + self.assertEqual("Message %d" % i, msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_b.get(timeout=1) + self.assertEqual("Message 6", msg.body) + session.message_accept(RangedSet(msg.id)) + + msg = queue_c.get(timeout=1) + self.assertEqual("Message 7", msg.body) + session.message_accept(RangedSet(msg.id)) + + for q in [queue_a, queue_b, queue_c]: + try: + extra = q.get(timeout=1) + self.fail("Got unexpected message: " + extra.body) + except Empty: None + + #cleanup + session.tx_commit() + + def perform_txn_work(self, session, name_a, name_b, name_c): + """ + Utility method that does some setup and some work under a transaction. Used for testing both + commit and rollback + """ + #setup: + self.declare_queues([name_a, name_b, name_c]) + + key = "my_key_" + name_b + topic = "my_topic_" + name_c + + session.exchange_bind(queue=name_b, exchange="amq.direct", binding_key=key) + session.exchange_bind(queue=name_c, exchange="amq.topic", binding_key=topic) + + dp = session.delivery_properties(routing_key=name_a) + for i in range(1, 5): + mp = session.message_properties(message_id="msg%d" % i) + session.message_transfer(message=Message(dp, mp, "Message %d" % i)) + + dp = session.delivery_properties(routing_key=key) + mp = session.message_properties(message_id="msg6") + session.message_transfer(destination="amq.direct", message=Message(dp, mp, "Message 6")) + + dp = session.delivery_properties(routing_key=topic) + mp = session.message_properties(message_id="msg7") + session.message_transfer(destination="amq.topic", message=Message(dp, mp, "Message 7")) + + session.tx_select() + + #consume and ack messages + acked = RangedSet() + self.subscribe(session, queue=name_a, destination="sub_a") + queue_a = session.incoming("sub_a") + for i in range(1, 5): + msg = queue_a.get(timeout=1) + acked.add(msg.id) + self.assertEqual("Message %d" % i, msg.body) + + self.subscribe(session, queue=name_b, destination="sub_b") + queue_b = session.incoming("sub_b") + msg = queue_b.get(timeout=1) + self.assertEqual("Message 6", msg.body) + acked.add(msg.id) + + sub_c = self.subscribe(session, queue=name_c, destination="sub_c") + queue_c = session.incoming("sub_c") + msg = queue_c.get(timeout=1) + self.assertEqual("Message 7", msg.body) + acked.add(msg.id) + + session.message_accept(acked) + + dp = session.delivery_properties(routing_key=topic) + #publish messages + for i in range(1, 5): + mp = session.message_properties(message_id="tx-msg%d" % i) + session.message_transfer(destination="amq.topic", message=Message(dp, mp, "TxMessage %d" % i)) + + dp = session.delivery_properties(routing_key=key) + mp = session.message_properties(message_id="tx-msg6") + session.message_transfer(destination="amq.direct", message=Message(dp, mp, "TxMessage 6")) + + dp = session.delivery_properties(routing_key=name_a) + mp = session.message_properties(message_id="tx-msg7") + session.message_transfer(message=Message(dp, mp, "TxMessage 7")) + return queue_a, queue_b, queue_c, acked + + def declare_queues(self, names, session=None): + session = session or self.session + for n in names: + session.queue_declare(queue=n, auto_delete=True) + + def subscribe(self, session=None, **keys): + session = session or self.session + consumer_tag = keys["destination"] + session.message_subscribe(**keys) + session.message_flow(destination=consumer_tag, unit=session.credit_unit.message, value=0xFFFFFFFFL) + session.message_flow(destination=consumer_tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL) + + def enable_flow(self, tag, session=None): + session = session or self.session + session.message_flow(destination=tag, unit=session.credit_unit.message, value=0xFFFFFFFFL) + session.message_flow(destination=tag, unit=session.credit_unit.byte, value=0xFFFFFFFFL) + + def complete(self, session, msg): + session.receiver._completed.add(msg.id)#TODO: this may be done automatically + session.channel.session_completed(session.receiver._completed) + |