diff options
Diffstat (limited to 'qpid/python/tests_0-9/tx.py')
| -rw-r--r-- | qpid/python/tests_0-9/tx.py | 188 |
1 files changed, 0 insertions, 188 deletions
diff --git a/qpid/python/tests_0-9/tx.py b/qpid/python/tests_0-9/tx.py deleted file mode 100644 index 0f6b4f5cd1..0000000000 --- a/qpid/python/tests_0-9/tx.py +++ /dev/null @@ -1,188 +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 testrunner, TestBase - -class TxTests(TestBase): - """ - 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 - """ - channel = self.channel - queue_a, queue_b, queue_c = self.perform_txn_work(channel, "tx-commit-a", "tx-commit-b", "tx-commit-c") - channel.tx_commit() - - #check results - for i in range(1, 5): - msg = queue_c.get(timeout=1) - self.assertEqual("TxMessage %d" % i, msg.body) - msg.ok() - - msg = queue_b.get(timeout=1) - self.assertEqual("TxMessage 6", msg.body) - msg.ok() - - msg = queue_a.get(timeout=1) - self.assertEqual("TxMessage 7", msg.body) - msg.ok() - - 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 - channel.tx_commit() - - def test_auto_rollback(self): - """ - Test that a channel closed with an open transaction is effectively rolled back - """ - channel = self.channel - queue_a, queue_b, queue_c = self.perform_txn_work(channel, "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 - - channel.tx_rollback() - - #check results - for i in range(1, 5): - msg = queue_a.get(timeout=1) - self.assertEqual("Message %d" % i, msg.body) - msg.ok() - - msg = queue_b.get(timeout=1) - self.assertEqual("Message 6", msg.body) - msg.ok() - - msg = queue_c.get(timeout=1) - self.assertEqual("Message 7", msg.body) - msg.ok() - - 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 - channel.tx_commit() - - def test_rollback(self): - """ - Test that rolled back publishes are not delivered and rolled back acks are re-delivered - """ - channel = self.channel - queue_a, queue_b, queue_c = self.perform_txn_work(channel, "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 - - channel.tx_rollback() - - #check results - for i in range(1, 5): - msg = queue_a.get(timeout=1) - self.assertEqual("Message %d" % i, msg.body) - msg.ok() - - msg = queue_b.get(timeout=1) - self.assertEqual("Message 6", msg.body) - msg.ok() - - msg = queue_c.get(timeout=1) - self.assertEqual("Message 7", msg.body) - msg.ok() - - 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 - channel.tx_commit() - - def perform_txn_work(self, channel, 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: - channel.queue_declare(queue=name_a, exclusive=True) - channel.queue_declare(queue=name_b, exclusive=True) - channel.queue_declare(queue=name_c, exclusive=True) - - key = "my_key_" + name_b - topic = "my_topic_" + name_c - - channel.queue_bind(queue=name_b, exchange="amq.direct", routing_key=key) - channel.queue_bind(queue=name_c, exchange="amq.topic", routing_key=topic) - - for i in range(1, 5): - channel.message_transfer(routing_key=name_a, body="Message %d" % i) - - channel.message_transfer(routing_key=key, destination="amq.direct", body="Message 6") - channel.message_transfer(routing_key=topic, destination="amq.topic", body="Message 7") - - channel.tx_select() - - #consume and ack messages - channel.message_consume(queue=name_a, destination="sub_a", no_ack=False) - queue_a = self.client.queue("sub_a") - for i in range(1, 5): - msg = queue_a.get(timeout=1) - self.assertEqual("Message %d" % i, msg.body) - - msg.ok(batchoffset=-3) - - channel.message_consume(queue=name_b, destination="sub_b", no_ack=False) - queue_b = self.client.queue("sub_b") - msg = queue_b.get(timeout=1) - self.assertEqual("Message 6", msg.body) - msg.ok() - - sub_c = channel.message_consume(queue=name_c, destination="sub_c", no_ack=False) - queue_c = self.client.queue("sub_c") - msg = queue_c.get(timeout=1) - self.assertEqual("Message 7", msg.body) - msg.ok() - - #publish messages - for i in range(1, 5): - channel.message_transfer(routing_key=topic, destination="amq.topic", body="TxMessage %d" % i) - - channel.message_transfer(routing_key=key, destination="amq.direct", body="TxMessage 6") - channel.message_transfer(routing_key=name_a, body="TxMessage 7") - - return queue_a, queue_b, queue_c |