You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ai...@apache.org on 2008/04/15 12:28:36 UTC
svn commit: r648207 [6/10] - in /incubator/qpid/branches/thegreatmerge: ./
qpid/java/ qpid/java/client/ qpid/java/client/example/bin/
qpid/java/client/example/src/main/java/org/apache/qpid/example/jmsexample/direct/
qpid/java/client/example/src/main/ja...
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/alternate_exchange.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/alternate_exchange.py?rev=648207&r1=648206&r2=648207&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/alternate_exchange.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/alternate_exchange.py Tue Apr 15 03:28:06 2008
@@ -16,12 +16,13 @@
# specific language governing permissions and limitations
# under the License.
#
-from qpid.client import Client, Closed
+import traceback
from qpid.queue import Empty
-from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.datatypes import Message
+from qpid.testlib import TestBase010
+from qpid.session import SessionException
-class AlternateExchangeTests(TestBase):
+class AlternateExchangeTests(TestBase010):
"""
Tests for the new mechanism for message returns introduced in 0-10
and available in 0-9 for preview
@@ -31,36 +32,42 @@
"""
Test that unroutable messages are delivered to the alternate-exchange if specified
"""
- channel = self.channel
+ session = self.session
#create an exchange with an alternate defined
- channel.exchange_declare(exchange="secondary", type="fanout")
- channel.exchange_declare(exchange="primary", type="direct", alternate_exchange="secondary")
+ 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
- channel.queue_declare(queue="returns", exclusive=True, auto_delete=True)
- channel.queue_bind(queue="returns", exchange="secondary")
- self.subscribe(destination="a", queue="returns")
- returned = self.client.queue("a")
+ 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=0, value=0xFFFFFFFF)
+ session.message_flow(destination="a", unit=1, value=0xFFFFFFFF)
+ returned = session.incoming("a")
#declare, bind (to the primary exchange) and consume from a queue for 'processed' messages
- channel.queue_declare(queue="processed", exclusive=True, auto_delete=True)
- channel.queue_bind(queue="processed", exchange="primary", routing_key="my-key")
- self.subscribe(destination="b", queue="processed")
- processed = self.client.queue("b")
+ 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=0, value=0xFFFFFFFF)
+ session.message_flow(destination="b", unit=1, value=0xFFFFFFFF)
+ processed = session.incoming("b")
#publish to the primary exchange
#...one message that makes it to the 'processed' queue:
- channel.message_transfer(destination="primary", content=Content("Good", properties={'routing_key':"my-key"}))
+ dp=self.session.delivery_properties(routing_key="my-key")
+ session.message_transfer(destination="primary", message=Message(dp, "Good"))
#...and one that does not:
- channel.message_transfer(destination="primary", content=Content("Bad", properties={'routing_key':"unused-key"}))
+ dp=self.session.delivery_properties(routing_key="unused-key")
+ session.message_transfer(destination="primary", message=Message(dp, "Bad"))
#delete the exchanges
- channel.exchange_delete(exchange="primary")
- channel.exchange_delete(exchange="secondary")
+ session.exchange_delete(exchange="primary")
+ session.exchange_delete(exchange="secondary")
#verify behaviour
- self.assertEqual("Good", processed.get(timeout=1).content.body)
- self.assertEqual("Bad", returned.get(timeout=1).content.body)
+ self.assertEqual("Good", processed.get(timeout=1).body)
+ self.assertEqual("Bad", returned.get(timeout=1).body)
self.assertEmpty(processed)
self.assertEmpty(returned)
@@ -68,81 +75,51 @@
"""
Test that messages in a queue being deleted are delivered to the alternate-exchange if specified
"""
- channel = self.channel
+ session = self.session
#set up a 'dead letter queue':
- channel.exchange_declare(exchange="dlq", type="fanout")
- channel.queue_declare(queue="deleted", exclusive=True, auto_delete=True)
- channel.queue_bind(exchange="dlq", queue="deleted")
- self.subscribe(destination="dlq", queue="deleted")
- dlq = self.client.queue("dlq")
+ 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=0, value=0xFFFFFFFF)
+ session.message_flow(destination="dlq", unit=1, value=0xFFFFFFFF)
+ dlq = session.incoming("dlq")
#create a queue using the dlq as its alternate exchange:
- channel.queue_declare(queue="delete-me", alternate_exchange="dlq")
+ session.queue_declare(queue="delete-me", alternate_exchange="dlq")
#send it some messages:
- channel.message_transfer(content=Content("One", properties={'routing_key':"delete-me"}))
- channel.message_transfer(content=Content("Two", properties={'routing_key':"delete-me"}))
- channel.message_transfer(content=Content("Three", properties={'routing_key':"delete-me"}))
+ 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:
- channel.queue_delete(queue="delete-me")
+ session.queue_delete(queue="delete-me")
#delete the dlq exchange:
- channel.exchange_delete(exchange="dlq")
+ session.exchange_delete(exchange="dlq")
#check the messages were delivered to the dlq:
- self.assertEqual("One", dlq.get(timeout=1).content.body)
- self.assertEqual("Two", dlq.get(timeout=1).content.body)
- self.assertEqual("Three", dlq.get(timeout=1).content.body)
+ 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_immediate(self):
- """
- Test that messages in a queue being deleted are delivered to the alternate-exchange if specified
- """
- channel = self.channel
- #set up a 'dead letter queue':
- channel.exchange_declare(exchange="dlq", type="fanout")
- channel.queue_declare(queue="immediate", exclusive=True, auto_delete=True)
- channel.queue_bind(exchange="dlq", queue="immediate")
- self.subscribe(destination="dlq", queue="immediate")
- dlq = self.client.queue("dlq")
-
- #create a queue using the dlq as its alternate exchange:
- channel.queue_declare(queue="no-consumers", alternate_exchange="dlq", exclusive=True, auto_delete=True)
- #send it some messages:
- #TODO: WE HAVE LOST THE IMMEDIATE FLAG; FIX THIS ONCE ITS BACK
- channel.message_transfer(content=Content("no one wants me", properties={'routing_key':"no-consumers"}))
-
- #check the messages were delivered to the dlq:
- self.assertEqual("no one wants me", dlq.get(timeout=1).content.body)
- self.assertEmpty(dlq)
-
- #cleanup:
- channel.queue_delete(queue="no-consumers")
- channel.exchange_delete(exchange="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
"""
- channel = self.channel
- channel.exchange_declare(exchange="alternate", type="fanout")
- channel.queue_declare(queue="q", exclusive=True, auto_delete=True, alternate_exchange="alternate")
+ session = self.session
+ session.exchange_declare(exchange="alternate", type="fanout")
+
+ session2 = self.conn.session("alternate", 2)
+ session2.queue_declare(queue="q", exclusive=True, auto_delete=True, alternate_exchange="alternate")
try:
- channel.exchange_delete(exchange="alternate")
+ session2.exchange_delete(exchange="alternate")
self.fail("Expected deletion of in-use alternate-exchange to fail")
- except Closed, e:
- #cleanup:
- other = self.connect()
- channel = other.channel(1)
- channel.session_open()
- channel.exchange_delete(exchange="alternate")
- channel.session_close()
- other.close()
-
- self.assertConnectionException(530, e.args[0])
-
+ except SessionException, e:
+ session = self.session
+ session.exchange_delete(exchange="alternate")
+ self.assertEquals(530, e.args[0].error_code)
def test_delete_while_used_by_exchange(self):
@@ -150,25 +127,19 @@
Ensure an exchange still in use as an alternate-exchange for
another exchange can't be deleted
"""
- channel = self.channel
- channel.exchange_declare(exchange="alternate", type="fanout")
- channel.exchange_declare(exchange="e", type="fanout", alternate_exchange="alternate")
+ 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:
- channel.exchange_delete(exchange="alternate")
- #cleanup:
- channel.exchange_delete(exchange="e")
+ session.exchange_delete(exchange="alternate")
self.fail("Expected deletion of in-use alternate-exchange to fail")
- except Closed, e:
- #cleanup:
- other = self.connect()
- channel = other.channel(1)
- channel.session_open()
- channel.exchange_delete(exchange="e")
- channel.exchange_delete(exchange="alternate")
- channel.session_close()
- other.close()
-
- self.assertConnectionException(530, e.args[0])
+ 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 assertEmpty(self, queue):
@@ -176,4 +147,3 @@
msg = queue.get(timeout=1)
self.fail("Queue not empty: " + msg)
except Empty: None
-
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/broker.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/broker.py?rev=648207&r1=648206&r2=648207&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/broker.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/broker.py Tue Apr 15 03:28:06 2008
@@ -18,10 +18,10 @@
#
from qpid.client import Closed
from qpid.queue import Empty
-from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.testlib import TestBase010
+from qpid.datatypes import Message, RangedSet
-class BrokerTests(TestBase):
+class BrokerTests(TestBase010):
"""Tests for basic Broker functionality"""
def test_ack_and_no_ack(self):
@@ -30,82 +30,64 @@
consumer. Second, this test tries to explicitly receive and
acknowledge a message with an acknowledging consumer.
"""
- ch = self.channel
- self.queue_declare(ch, queue = "myqueue")
+ session = self.session
+ session.queue_declare(queue = "myqueue", exclusive=True, auto_delete=True)
# No ack consumer
ctag = "tag1"
- self.subscribe(ch, queue = "myqueue", destination = ctag)
+ session.message_subscribe(queue = "myqueue", destination = ctag)
+ session.message_flow(destination=ctag, unit=0, value=0xFFFFFFFF)
+ session.message_flow(destination=ctag, unit=1, value=0xFFFFFFFF)
body = "test no-ack"
- ch.message_transfer(content = Content(body, properties = {"routing_key" : "myqueue"}))
- msg = self.client.queue(ctag).get(timeout = 5)
- self.assert_(msg.content.body == body)
+ session.message_transfer(message=Message(session.delivery_properties(routing_key="myqueue"), body))
+ msg = session.incoming(ctag).get(timeout = 5)
+ self.assert_(msg.body == body)
# Acknowledging consumer
- self.queue_declare(ch, queue = "otherqueue")
+ session.queue_declare(queue = "otherqueue", exclusive=True, auto_delete=True)
ctag = "tag2"
- self.subscribe(ch, queue = "otherqueue", destination = ctag, confirm_mode = 1)
- ch.message_flow(destination=ctag, unit=0, value=0xFFFFFFFF)
- ch.message_flow(destination=ctag, unit=1, value=0xFFFFFFFF)
+ session.message_subscribe(queue = "otherqueue", destination = ctag, accept_mode = 1)
+ session.message_flow(destination=ctag, unit=0, value=0xFFFFFFFF)
+ session.message_flow(destination=ctag, unit=1, value=0xFFFFFFFF)
body = "test ack"
- ch.message_transfer(content = Content(body, properties = {"routing_key" : "otherqueue"}))
- msg = self.client.queue(ctag).get(timeout = 5)
- msg.complete()
- self.assert_(msg.content.body == body)
+ session.message_transfer(message=Message(session.delivery_properties(routing_key="otherqueue"), body))
+ msg = session.incoming(ctag).get(timeout = 5)
+ session.message_accept(RangedSet(msg.id))
+ self.assert_(msg.body == body)
def test_simple_delivery_immediate(self):
"""
Test simple message delivery where consume is issued before publish
"""
- channel = self.channel
- self.exchange_declare(channel, exchange="test-exchange", type="direct")
- self.queue_declare(channel, queue="test-queue")
- channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
+ session = self.session
+ session.queue_declare(queue="test-queue", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="test-queue", exchange="amq.fanout")
consumer_tag = "tag1"
- self.subscribe(queue="test-queue", destination=consumer_tag)
- queue = self.client.queue(consumer_tag)
+ session.message_subscribe(queue="test-queue", destination=consumer_tag)
+ session.message_flow(unit = 0, value = 0xFFFFFFFF, destination = consumer_tag)
+ session.message_flow(unit = 1, value = 0xFFFFFFFF, destination = consumer_tag)
+ queue = session.incoming(consumer_tag)
body = "Immediate Delivery"
- channel.message_transfer(destination="test-exchange", content = Content(body, properties = {"routing_key" : "key"}))
+ session.message_transfer("amq.fanout", None, None, Message(body))
msg = queue.get(timeout=5)
- self.assert_(msg.content.body == body)
-
- # TODO: Ensure we fail if immediate=True and there's no consumer.
-
+ self.assert_(msg.body == body)
def test_simple_delivery_queued(self):
"""
Test basic message delivery where publish is issued before consume
(i.e. requires queueing of the message)
"""
- channel = self.channel
- self.exchange_declare(channel, exchange="test-exchange", type="direct")
- self.queue_declare(channel, queue="test-queue")
- channel.queue_bind(queue="test-queue", exchange="test-exchange", routing_key="key")
+ session = self.session
+ session.queue_declare(queue="test-queue", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="test-queue", exchange="amq.fanout")
body = "Queued Delivery"
- channel.message_transfer(destination="test-exchange", content = Content(body, properties = {"routing_key" : "key"}))
+ session.message_transfer("amq.fanout", None, None, Message(body))
consumer_tag = "tag1"
- self.subscribe(queue="test-queue", destination=consumer_tag)
- queue = self.client.queue(consumer_tag)
+ session.message_subscribe(queue="test-queue", destination=consumer_tag)
+ session.message_flow(unit = 0, value = 0xFFFFFFFF, destination = consumer_tag)
+ session.message_flow(unit = 1, value = 0xFFFFFFFF, destination = consumer_tag)
+ queue = session.incoming(consumer_tag)
msg = queue.get(timeout=5)
- self.assert_(msg.content.body == body)
-
- def test_invalid_channel(self):
- channel = self.client.channel(200)
- try:
- channel.queue_declare(exclusive=True)
- self.fail("Expected error on queue_declare for invalid channel")
- except Closed, e:
- self.assertConnectionException(504, e.args[0])
-
- def test_closed_channel(self):
- channel = self.client.channel(200)
- channel.session_open()
- channel.session_close()
- try:
- channel.queue_declare(exclusive=True)
- self.fail("Expected error on queue_declare for closed channel")
- except Closed, e:
- if isinstance(e.args[0], str): self.fail(e)
- self.assertConnectionException(504, e.args[0])
+ self.assert_(msg.body == body)
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/dtx.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/dtx.py?rev=648207&r1=648206&r2=648207&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/dtx.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/dtx.py Tue Apr 15 03:28:06 2008
@@ -18,12 +18,13 @@
#
from qpid.client import Client, Closed
from qpid.queue import Empty
-from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.datatypes import Message, RangedSet
+from qpid.session import SessionException
+from qpid.testlib import TestBase010
from struct import pack, unpack
from time import sleep
-class DtxTests(TestBase):
+class DtxTests(TestBase010):
"""
Tests for the amqp dtx related classes.
@@ -43,15 +44,15 @@
tx_counter = 0
def reset_channel(self):
- self.channel.session_close()
- self.channel = self.client.channel(self.channel.id + 1)
- self.channel.session_open()
+ self.session.close()
+ self.session = self.conn.session("dtx-session", 1)
def test_simple_commit(self):
"""
Test basic one-phase commit behaviour.
"""
- channel = self.channel
+ guard = self.keepQueuesAlive(["queue-a", "queue-b"])
+ session = self.session
tx = self.xid("my-xid")
self.txswap(tx, "commit")
@@ -60,9 +61,9 @@
self.assertMessageCount(0, "queue-b")
#commit
- self.assertEqual(self.XA_OK, channel.dtx_coordination_commit(xid=tx, one_phase=True).status)
+ self.assertEqual(self.XA_OK, session.dtx_commit(xid=tx, one_phase=True).status)
- #should close and reopen channel to ensure no unacked messages are held
+ #should close and reopen session to ensure no unacked messages are held
self.reset_channel()
#check result
@@ -74,19 +75,20 @@
"""
Test basic two-phase commit behaviour.
"""
- channel = self.channel
+ guard = self.keepQueuesAlive(["queue-a", "queue-b"])
+ session = self.session
tx = self.xid("my-xid")
self.txswap(tx, "prepare-commit")
#prepare
- self.assertEqual(self.XA_OK, channel.dtx_coordination_prepare(xid=tx).status)
+ self.assertEqual(self.XA_OK, session.dtx_prepare(xid=tx).status)
#neither queue should have any messages accessible
self.assertMessageCount(0, "queue-a")
self.assertMessageCount(0, "queue-b")
#commit
- self.assertEqual(self.XA_OK, channel.dtx_coordination_commit(xid=tx, one_phase=False).status)
+ self.assertEqual(self.XA_OK, session.dtx_commit(xid=tx, one_phase=False).status)
self.reset_channel()
@@ -100,7 +102,8 @@
"""
Test basic rollback behaviour.
"""
- channel = self.channel
+ guard = self.keepQueuesAlive(["queue-a", "queue-b"])
+ session = self.session
tx = self.xid("my-xid")
self.txswap(tx, "rollback")
@@ -109,7 +112,7 @@
self.assertMessageCount(0, "queue-b")
#rollback
- self.assertEqual(self.XA_OK, channel.dtx_coordination_rollback(xid=tx).status)
+ self.assertEqual(self.XA_OK, session.dtx_rollback(xid=tx).status)
self.reset_channel()
@@ -122,19 +125,20 @@
"""
Test basic rollback behaviour after the transaction has been prepared.
"""
- channel = self.channel
+ guard = self.keepQueuesAlive(["queue-a", "queue-b"])
+ session = self.session
tx = self.xid("my-xid")
self.txswap(tx, "prepare-rollback")
#prepare
- self.assertEqual(self.XA_OK, channel.dtx_coordination_prepare(xid=tx).status)
+ self.assertEqual(self.XA_OK, session.dtx_prepare(xid=tx).status)
#neither queue should have any messages accessible
self.assertMessageCount(0, "queue-a")
self.assertMessageCount(0, "queue-b")
#rollback
- self.assertEqual(self.XA_OK, channel.dtx_coordination_rollback(xid=tx).status)
+ self.assertEqual(self.XA_OK, session.dtx_rollback(xid=tx).status)
self.reset_channel()
@@ -148,17 +152,17 @@
check that an error is flagged if select is not issued before
start or end
"""
- channel = self.channel
+ session = self.session
tx = self.xid("dummy")
try:
- channel.dtx_demarcation_start(xid=tx)
+ session.dtx_start(xid=tx)
#if we get here we have failed, but need to do some cleanup:
- channel.dtx_demarcation_end(xid=tx)
- channel.dtx_coordination_rollback(xid=tx)
- self.fail("Channel not selected for use with dtx, expected exception!")
- except Closed, e:
- self.assertConnectionException(503, e.args[0])
+ session.dtx_end(xid=tx)
+ session.dtx_rollback(xid=tx)
+ self.fail("Session not selected for use with dtx, expected exception!")
+ except SessionException, e:
+ self.assertEquals(503, e.args[0].error_code)
def test_start_already_known(self):
"""
@@ -166,36 +170,35 @@
transaction that is already known is not allowed (unless the
join flag is set).
"""
- #create two channels on different connection & select them for use with dtx:
- channel1 = self.channel
- channel1.dtx_demarcation_select()
+ #create two sessions on different connection & select them for use with dtx:
+ session1 = self.session
+ session1.dtx_select()
other = self.connect()
- channel2 = other.channel(1)
- channel2.session_open()
- channel2.dtx_demarcation_select()
+ session2 = other.session("other", 0)
+ session2.dtx_select()
#create a xid
tx = self.xid("dummy")
- #start work on one channel under that xid:
- channel1.dtx_demarcation_start(xid=tx)
+ #start work on one session under that xid:
+ session1.dtx_start(xid=tx)
#then start on the other without the join set
failed = False
try:
- channel2.dtx_demarcation_start(xid=tx)
- except Closed, e:
+ session2.dtx_start(xid=tx)
+ except SessionException, e:
failed = True
error = e
#cleanup:
if not failed:
- channel2.dtx_demarcation_end(xid=tx)
+ session2.dtx_end(xid=tx)
other.close()
- channel1.dtx_demarcation_end(xid=tx)
- channel1.dtx_coordination_rollback(xid=tx)
+ session1.dtx_end(xid=tx)
+ session1.dtx_rollback(xid=tx)
#verification:
- if failed: self.assertConnectionException(503, e.args[0])
+ if failed: self.assertEquals(503, error.args[0].error_code)
else: self.fail("Xid already known, expected exception!")
def test_forget_xid_on_completion(self):
@@ -205,69 +208,69 @@
"""
#do some transactional work & complete the transaction
self.test_simple_commit()
- # channel has been reset, so reselect for use with dtx
- self.channel.dtx_demarcation_select()
+ # session has been reset, so reselect for use with dtx
+ self.session.dtx_select()
#start association for the same xid as the previously completed txn
tx = self.xid("my-xid")
- self.channel.dtx_demarcation_start(xid=tx)
- self.channel.dtx_demarcation_end(xid=tx)
- self.channel.dtx_coordination_rollback(xid=tx)
+ self.session.dtx_start(xid=tx)
+ self.session.dtx_end(xid=tx)
+ self.session.dtx_rollback(xid=tx)
def test_start_join_and_resume(self):
"""
Ensure the correct error is signalled when both the join and
resume flags are set on starting an association between a
- channel and a transcation.
+ session and a transcation.
"""
- channel = self.channel
- channel.dtx_demarcation_select()
+ session = self.session
+ session.dtx_select()
tx = self.xid("dummy")
try:
- channel.dtx_demarcation_start(xid=tx, join=True, resume=True)
+ session.dtx_start(xid=tx, join=True, resume=True)
#failed, but need some cleanup:
- channel.dtx_demarcation_end(xid=tx)
- channel.dtx_coordination_rollback(xid=tx)
+ session.dtx_end(xid=tx)
+ session.dtx_rollback(xid=tx)
self.fail("Join and resume both set, expected exception!")
- except Closed, e:
- self.assertConnectionException(503, e.args[0])
+ except SessionException, e:
+ self.assertEquals(503, e.args[0].error_code)
def test_start_join(self):
"""
- Verify 'join' behaviour, where a channel is associated with a
- transaction that is already associated with another channel.
+ Verify 'join' behaviour, where a session is associated with a
+ transaction that is already associated with another session.
"""
- #create two channels & select them for use with dtx:
- channel1 = self.channel
- channel1.dtx_demarcation_select()
-
- channel2 = self.client.channel(2)
- channel2.session_open()
- channel2.dtx_demarcation_select()
+ guard = self.keepQueuesAlive(["one", "two"])
+ #create two sessions & select them for use with dtx:
+ session1 = self.session
+ session1.dtx_select()
+
+ session2 = self.conn.session("second", 2)
+ session2.dtx_select()
#setup
- channel1.queue_declare(queue="one", exclusive=True, auto_delete=True)
- channel1.queue_declare(queue="two", exclusive=True, auto_delete=True)
- channel1.message_transfer(content=Content(properties={'routing_key':"one", 'message_id':"a"}, body="DtxMessage"))
- channel1.message_transfer(content=Content(properties={'routing_key':"two", 'message_id':"b"}, body="DtxMessage"))
+ session1.queue_declare(queue="one", auto_delete=True)
+ session1.queue_declare(queue="two", auto_delete=True)
+ session1.message_transfer(self.createMessage(session1, "one", "a", "DtxMessage"))
+ session1.message_transfer(self.createMessage(session1, "two", "b", "DtxMessage"))
#create a xid
tx = self.xid("dummy")
- #start work on one channel under that xid:
- channel1.dtx_demarcation_start(xid=tx)
+ #start work on one session under that xid:
+ session1.dtx_start(xid=tx)
#then start on the other with the join flag set
- channel2.dtx_demarcation_start(xid=tx, join=True)
+ session2.dtx_start(xid=tx, join=True)
- #do work through each channel
- self.swap(channel1, "one", "two")#swap 'a' from 'one' to 'two'
- self.swap(channel2, "two", "one")#swap 'b' from 'two' to 'one'
-
- #mark end on both channels
- channel1.dtx_demarcation_end(xid=tx)
- channel2.dtx_demarcation_end(xid=tx)
+ #do work through each session
+ self.swap(session1, "one", "two")#swap 'a' from 'one' to 'two'
+ self.swap(session2, "two", "one")#swap 'b' from 'two' to 'one'
+
+ #mark end on both sessions
+ session1.dtx_end(xid=tx)
+ session2.dtx_end(xid=tx)
#commit and check
- channel1.dtx_coordination_commit(xid=tx, one_phase=True)
+ session1.dtx_commit(xid=tx, one_phase=True)
self.assertMessageCount(1, "one")
self.assertMessageCount(1, "two")
self.assertMessageId("a", "two")
@@ -278,27 +281,27 @@
"""
Test suspension and resumption of an association
"""
- channel = self.channel
- channel.dtx_demarcation_select()
+ session = self.session
+ session.dtx_select()
#setup
- channel.queue_declare(queue="one", exclusive=True, auto_delete=True)
- channel.queue_declare(queue="two", exclusive=True, auto_delete=True)
- channel.message_transfer(content=Content(properties={'routing_key':"one", 'message_id':"a"}, body="DtxMessage"))
- channel.message_transfer(content=Content(properties={'routing_key':"two", 'message_id':"b"}, body="DtxMessage"))
+ session.queue_declare(queue="one", exclusive=True, auto_delete=True)
+ session.queue_declare(queue="two", exclusive=True, auto_delete=True)
+ session.message_transfer(self.createMessage(session, "one", "a", "DtxMessage"))
+ session.message_transfer(self.createMessage(session, "two", "b", "DtxMessage"))
tx = self.xid("dummy")
- channel.dtx_demarcation_start(xid=tx)
- self.swap(channel, "one", "two")#swap 'a' from 'one' to 'two'
- channel.dtx_demarcation_end(xid=tx, suspend=True)
+ session.dtx_start(xid=tx)
+ self.swap(session, "one", "two")#swap 'a' from 'one' to 'two'
+ session.dtx_end(xid=tx, suspend=True)
- channel.dtx_demarcation_start(xid=tx, resume=True)
- self.swap(channel, "two", "one")#swap 'b' from 'two' to 'one'
- channel.dtx_demarcation_end(xid=tx)
+ session.dtx_start(xid=tx, resume=True)
+ self.swap(session, "two", "one")#swap 'b' from 'two' to 'one'
+ session.dtx_end(xid=tx)
#commit and check
- channel.dtx_coordination_commit(xid=tx, one_phase=True)
+ session.dtx_commit(xid=tx, one_phase=True)
self.assertMessageCount(1, "one")
self.assertMessageCount(1, "two")
self.assertMessageId("a", "two")
@@ -310,27 +313,27 @@
done on another transaction when the first transaction is
suspended
"""
- channel = self.channel
- channel.dtx_demarcation_select()
+ session = self.session
+ session.dtx_select()
#setup
- channel.queue_declare(queue="one", exclusive=True, auto_delete=True)
- channel.queue_declare(queue="two", exclusive=True, auto_delete=True)
- channel.message_transfer(content=Content(properties={'routing_key':"one", 'message_id':"a"}, body="DtxMessage"))
- channel.message_transfer(content=Content(properties={'routing_key':"two", 'message_id':"b"}, body="DtxMessage"))
+ session.queue_declare(queue="one", exclusive=True, auto_delete=True)
+ session.queue_declare(queue="two", exclusive=True, auto_delete=True)
+ session.message_transfer(self.createMessage(session, "one", "a", "DtxMessage"))
+ session.message_transfer(self.createMessage(session, "two", "b", "DtxMessage"))
tx = self.xid("dummy")
- channel.dtx_demarcation_start(xid=tx)
- self.swap(channel, "one", "two")#swap 'a' from 'one' to 'two'
- channel.dtx_demarcation_end(xid=tx, suspend=True)
+ session.dtx_start(xid=tx)
+ self.swap(session, "one", "two")#swap 'a' from 'one' to 'two'
+ session.dtx_end(xid=tx, suspend=True)
- channel.dtx_demarcation_start(xid=tx, resume=True)
- self.swap(channel, "two", "one")#swap 'b' from 'two' to 'one'
- channel.dtx_demarcation_end(xid=tx)
+ session.dtx_start(xid=tx, resume=True)
+ self.swap(session, "two", "one")#swap 'b' from 'two' to 'one'
+ session.dtx_end(xid=tx)
#commit and check
- channel.dtx_coordination_commit(xid=tx, one_phase=True)
+ session.dtx_commit(xid=tx, one_phase=True)
self.assertMessageCount(1, "one")
self.assertMessageCount(1, "two")
self.assertMessageId("a", "two")
@@ -340,24 +343,23 @@
"""
Verify that the correct error is signalled if the suspend and
fail flag are both set when disassociating a transaction from
- the channel
+ the session
"""
- channel = self.channel
- channel.dtx_demarcation_select()
+ session = self.session
+ session.dtx_select()
tx = self.xid("suspend_and_fail")
- channel.dtx_demarcation_start(xid=tx)
+ session.dtx_start(xid=tx)
try:
- channel.dtx_demarcation_end(xid=tx, suspend=True, fail=True)
+ session.dtx_end(xid=tx, suspend=True, fail=True)
self.fail("Suspend and fail both set, expected exception!")
- except Closed, e:
- self.assertConnectionException(503, e.args[0])
+ except SessionException, e:
+ self.assertEquals(503, e.args[0].error_code)
#cleanup
other = self.connect()
- channel = other.channel(1)
- channel.session_open()
- channel.dtx_coordination_rollback(xid=tx)
- channel.session_close()
+ session = other.session("cleanup", 1)
+ session.dtx_rollback(xid=tx)
+ session.close()
other.close()
@@ -365,51 +367,51 @@
"""
Verifies that the correct exception is thrown when an attempt
is made to end the association for a xid not previously
- associated with the channel
+ associated with the session
"""
- channel = self.channel
- channel.dtx_demarcation_select()
+ session = self.session
+ session.dtx_select()
tx = self.xid("unknown-xid")
try:
- channel.dtx_demarcation_end(xid=tx)
+ session.dtx_end(xid=tx)
self.fail("Attempted to end association with unknown xid, expected exception!")
- except Closed, e:
+ except SessionException, e:
#FYI: this is currently *not* the exception specified, but I think the spec is wrong! Confirming...
- self.assertConnectionException(503, e.args[0])
+ self.assertEquals(503, e.args[0].error_code)
def test_end(self):
"""
Verify that the association is terminated by end and subsequent
operations are non-transactional
"""
- channel = self.client.channel(2)
- channel.session_open()
- channel.queue_declare(queue="tx-queue", exclusive=True, auto_delete=True)
+ guard = self.keepQueuesAlive(["tx-queue"])
+ session = self.conn.session("alternate", 1)
+ session.queue_declare(queue="tx-queue", exclusive=True, auto_delete=True)
#publish a message under a transaction
- channel.dtx_demarcation_select()
+ session.dtx_select()
tx = self.xid("dummy")
- channel.dtx_demarcation_start(xid=tx)
- channel.message_transfer(content=Content(properties={'routing_key':"tx-queue", 'message_id':"one"}, body="DtxMessage"))
- channel.dtx_demarcation_end(xid=tx)
+ session.dtx_start(xid=tx)
+ session.message_transfer(self.createMessage(session, "tx-queue", "one", "DtxMessage"))
+ session.dtx_end(xid=tx)
#now that association with txn is ended, publish another message
- channel.message_transfer(content=Content(properties={'routing_key':"tx-queue", 'message_id':"two"}, body="DtxMessage"))
+ session.message_transfer(self.createMessage(session, "tx-queue", "two", "DtxMessage"))
#check the second message is available, but not the first
self.assertMessageCount(1, "tx-queue")
- self.subscribe(channel, queue="tx-queue", destination="results", confirm_mode=1)
- msg = self.client.queue("results").get(timeout=1)
- self.assertEqual("two", msg.content['message_id'])
- channel.message_cancel(destination="results")
- #ack the message then close the channel
- msg.complete()
- channel.session_close()
+ self.subscribe(session, queue="tx-queue", destination="results")
+ msg = session.incoming("results").get(timeout=1)
+ self.assertEqual("two", self.getMessageProperty(msg, 'correlation_id'))
+ session.message_cancel(destination="results")
+ #ack the message then close the session
+ session.message_accept(RangedSet(msg.id))
+ session.close()
- channel = self.channel
+ session = self.session
#commit the transaction and check that the first message (and
#only the first message) is then delivered
- channel.dtx_coordination_commit(xid=tx, one_phase=True)
+ session.dtx_commit(xid=tx, one_phase=True)
self.assertMessageCount(1, "tx-queue")
self.assertMessageId("one", "tx-queue")
@@ -419,27 +421,26 @@
transaction in question has already been prepared.
"""
other = self.connect()
- tester = other.channel(1)
- tester.session_open()
+ tester = other.session("tester", 1)
tester.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
- tester.dtx_demarcation_select()
+ tester.dtx_select()
tx = self.xid("dummy")
- tester.dtx_demarcation_start(xid=tx)
- tester.message_transfer(content=Content(properties={'routing_key':"dummy"}, body="whatever"))
- tester.dtx_demarcation_end(xid=tx)
- tester.dtx_coordination_prepare(xid=tx)
+ tester.dtx_start(xid=tx)
+ tester.message_transfer(self.createMessage(tester, "dummy", "dummy", "whatever"))
+ tester.dtx_end(xid=tx)
+ tester.dtx_prepare(xid=tx)
failed = False
try:
- tester.dtx_coordination_commit(xid=tx, one_phase=True)
- except Closed, e:
+ tester.dtx_commit(xid=tx, one_phase=True)
+ except SessionException, e:
failed = True
error = e
if failed:
- self.channel.dtx_coordination_rollback(xid=tx)
- self.assertConnectionException(503, e.args[0])
+ self.session.dtx_rollback(xid=tx)
+ self.assertEquals(503, error.args[0].error_code)
else:
- tester.session_close()
+ tester.close()
other.close()
self.fail("Invalid use of one_phase=True, expected exception!")
@@ -453,99 +454,99 @@
transaction in question has already been prepared.
"""
other = self.connect()
- tester = other.channel(1)
- tester.session_open()
+ tester = other.session("tester", 1)
tester.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
- tester.dtx_demarcation_select()
+ tester.dtx_select()
tx = self.xid("dummy")
- tester.dtx_demarcation_start(xid=tx)
- tester.message_transfer(content=Content(properties={'routing_key':"dummy"}, body="whatever"))
- tester.dtx_demarcation_end(xid=tx)
+ tester.dtx_start(xid=tx)
+ tester.message_transfer(self.createMessage(tester, "dummy", "dummy", "whatever"))
+ tester.dtx_end(xid=tx)
failed = False
try:
- tester.dtx_coordination_commit(xid=tx, one_phase=False)
- except Closed, e:
+ tester.dtx_commit(xid=tx, one_phase=False)
+ except SessionException, e:
failed = True
error = e
if failed:
- self.channel.dtx_coordination_rollback(xid=tx)
- self.assertConnectionException(503, e.args[0])
+ self.session.dtx_rollback(xid=tx)
+ self.assertEquals(503, error.args[0].error_code)
else:
- tester.session_close()
+ tester.close()
other.close()
self.fail("Invalid use of one_phase=False, expected exception!")
def test_implicit_end(self):
"""
- Test that an association is implicitly ended when the channel
+ Test that an association is implicitly ended when the session
is closed (whether by exception or explicit client request)
and the transaction in question is marked as rollback only.
"""
- channel1 = self.channel
- channel2 = self.client.channel(2)
- channel2.session_open()
+ session1 = self.session
+ session2 = self.conn.session("other", 2)
#setup:
- channel2.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
- channel2.message_transfer(content=Content(properties={'routing_key':"dummy"}, body="whatever"))
+ session2.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
+ session2.message_transfer(self.createMessage(session2, "dummy", "a", "whatever"))
tx = self.xid("dummy")
- channel2.dtx_demarcation_select()
- channel2.dtx_demarcation_start(xid=tx)
- channel2.message_subscribe(queue="dummy", destination="dummy", confirm_mode=1)
- channel2.message_flow(destination="dummy", unit=0, value=1)
- channel2.message_flow(destination="dummy", unit=1, value=0xFFFFFFFF)
- self.client.queue("dummy").get(timeout=1).complete()
- channel2.message_cancel(destination="dummy")
- channel2.message_transfer(content=Content(properties={'routing_key':"dummy"}, body="whatever"))
- channel2.session_close()
+ session2.dtx_select()
+ session2.dtx_start(xid=tx)
+ session2.message_subscribe(queue="dummy", destination="dummy")
+ session2.message_flow(destination="dummy", unit=0, value=1)
+ session2.message_flow(destination="dummy", unit=1, value=0xFFFFFFFF)
+ msg = session2.incoming("dummy").get(timeout=1)
+ session2.message_accept(RangedSet(msg.id))
+ session2.message_cancel(destination="dummy")
+ session2.message_transfer(self.createMessage(session2, "dummy", "b", "whatever"))
+ session2.close()
- self.assertEqual(self.XA_RBROLLBACK, channel1.dtx_coordination_prepare(xid=tx).status)
- channel1.dtx_coordination_rollback(xid=tx)
+ self.assertEqual(self.XA_RBROLLBACK, session1.dtx_prepare(xid=tx).status)
+ session1.dtx_rollback(xid=tx)
def test_get_timeout(self):
"""
Check that get-timeout returns the correct value, (and that a
transaction with a timeout can complete normally)
"""
- channel = self.channel
+ session = self.session
tx = self.xid("dummy")
- channel.dtx_demarcation_select()
- channel.dtx_demarcation_start(xid=tx)
- self.assertEqual(0, channel.dtx_coordination_get_timeout(xid=tx).timeout)
- channel.dtx_coordination_set_timeout(xid=tx, timeout=60)
- self.assertEqual(60, channel.dtx_coordination_get_timeout(xid=tx).timeout)
- self.assertEqual(self.XA_OK, channel.dtx_demarcation_end(xid=tx).status)
- self.assertEqual(self.XA_OK, channel.dtx_coordination_rollback(xid=tx).status)
+ session.dtx_select()
+ session.dtx_start(xid=tx)
+ self.assertEqual(0, session.dtx_get_timeout(xid=tx).timeout)
+ session.dtx_set_timeout(xid=tx, timeout=60)
+ self.assertEqual(60, session.dtx_get_timeout(xid=tx).timeout)
+ self.assertEqual(self.XA_OK, session.dtx_end(xid=tx).status)
+ self.assertEqual(self.XA_OK, session.dtx_rollback(xid=tx).status)
def test_set_timeout(self):
"""
Test the timeout of a transaction results in the expected
behaviour
"""
- #open new channel to allow self.channel to be used in checking te queue
- channel = self.client.channel(2)
- channel.session_open()
+
+ guard = self.keepQueuesAlive(["queue-a", "queue-b"])
+ #open new session to allow self.session to be used in checking the queue
+ session = self.conn.session("worker", 1)
#setup:
tx = self.xid("dummy")
- channel.queue_declare(queue="queue-a", exclusive=True, auto_delete=True)
- channel.queue_declare(queue="queue-b", exclusive=True, auto_delete=True)
- channel.message_transfer(content=Content(properties={'routing_key':"queue-a", 'message_id':"timeout"}, body="DtxMessage"))
-
- channel.dtx_demarcation_select()
- channel.dtx_demarcation_start(xid=tx)
- self.swap(channel, "queue-a", "queue-b")
- channel.dtx_coordination_set_timeout(xid=tx, timeout=2)
+ session.queue_declare(queue="queue-a", auto_delete=True)
+ session.queue_declare(queue="queue-b", auto_delete=True)
+ session.message_transfer(self.createMessage(session, "queue-a", "timeout", "DtxMessage"))
+
+ session.dtx_select()
+ session.dtx_start(xid=tx)
+ self.swap(session, "queue-a", "queue-b")
+ session.dtx_set_timeout(xid=tx, timeout=2)
sleep(3)
#check that the work has been rolled back already
self.assertMessageCount(1, "queue-a")
self.assertMessageCount(0, "queue-b")
self.assertMessageId("timeout", "queue-a")
#check the correct codes are returned when we try to complete the txn
- self.assertEqual(self.XA_RBTIMEOUT, channel.dtx_demarcation_end(xid=tx).status)
- self.assertEqual(self.XA_RBTIMEOUT, channel.dtx_coordination_rollback(xid=tx).status)
+ self.assertEqual(self.XA_RBTIMEOUT, session.dtx_end(xid=tx).status)
+ self.assertEqual(self.XA_RBTIMEOUT, session.dtx_rollback(xid=tx).status)
@@ -553,28 +554,29 @@
"""
Test basic recover behaviour
"""
- channel = self.channel
+ session = self.session
- channel.dtx_demarcation_select()
- channel.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
+ session.dtx_select()
+ session.queue_declare(queue="dummy", exclusive=True, auto_delete=True)
prepared = []
for i in range(1, 10):
tx = self.xid("tx%s" % (i))
- channel.dtx_demarcation_start(xid=tx)
- channel.message_transfer(content=Content(properties={'routing_key':"dummy"}, body="message%s" % (i)))
- channel.dtx_demarcation_end(xid=tx)
+ session.dtx_start(xid=tx)
+ session.message_transfer(self.createMessage(session, "dummy", "message%s" % (i), "message%s" % (i)))
+ session.dtx_end(xid=tx)
if i in [2, 5, 6, 8]:
- channel.dtx_coordination_prepare(xid=tx)
+ session.dtx_prepare(xid=tx)
prepared.append(tx)
else:
- channel.dtx_coordination_rollback(xid=tx)
+ session.dtx_rollback(xid=tx)
- xids = channel.dtx_coordination_recover().in_doubt
+ xids = session.dtx_recover().in_doubt
+ print "xids=%s" % xids
#rollback the prepared transactions returned by recover
for x in xids:
- channel.dtx_coordination_rollback(xid=x)
+ session.dtx_rollback(xid=x)
#validate against the expected list of prepared transactions
actual = set(xids)
@@ -585,61 +587,83 @@
missing = expected.difference(actual)
extra = actual.difference(expected)
for x in missing:
- channel.dtx_coordination_rollback(xid=x)
+ session.dtx_rollback(xid=x)
self.fail("Recovered xids not as expected. missing: %s; extra: %s" % (missing, extra))
def test_bad_resume(self):
"""
Test that a resume on a session not selected for use with dtx fails
"""
- channel = self.channel
+ session = self.session
try:
- channel.dtx_demarcation_start(resume=True)
- except Closed, e:
- self.assertConnectionException(503, e.args[0])
+ session.dtx_start(resume=True)
+ except SessionException, e:
+ self.assertEquals(503, e.args[0].error_code)
def xid(self, txid):
DtxTests.tx_counter += 1
branchqual = "v%s" % DtxTests.tx_counter
- return pack('!LBB', 0, len(txid), len(branchqual)) + txid + branchqual
-
+ return self.session.xid(format=0, global_id=txid, branch_id=branchqual)
+
def txswap(self, tx, id):
- channel = self.channel
+ session = self.session
#declare two queues:
- channel.queue_declare(queue="queue-a", exclusive=True, auto_delete=True)
- channel.queue_declare(queue="queue-b", exclusive=True, auto_delete=True)
+ session.queue_declare(queue="queue-a", auto_delete=True)
+ session.queue_declare(queue="queue-b", auto_delete=True)
+
#put message with specified id on one queue:
- channel.message_transfer(content=Content(properties={'routing_key':"queue-a", 'message_id':id}, body="DtxMessage"))
+ dp=session.delivery_properties(routing_key="queue-a")
+ mp=session.message_properties(correlation_id=id)
+ session.message_transfer(message=Message(dp, mp, "DtxMessage"))
#start the transaction:
- channel.dtx_demarcation_select()
- self.assertEqual(self.XA_OK, self.channel.dtx_demarcation_start(xid=tx).status)
+ session.dtx_select()
+ self.assertEqual(self.XA_OK, self.session.dtx_start(xid=tx).status)
#'swap' the message from one queue to the other, under that transaction:
- self.swap(self.channel, "queue-a", "queue-b")
+ self.swap(self.session, "queue-a", "queue-b")
#mark the end of the transactional work:
- self.assertEqual(self.XA_OK, self.channel.dtx_demarcation_end(xid=tx).status)
+ self.assertEqual(self.XA_OK, self.session.dtx_end(xid=tx).status)
- def swap(self, channel, src, dest):
+ def swap(self, session, src, dest):
#consume from src:
- channel.message_subscribe(destination="temp-swap", queue=src, confirm_mode=1)
- channel.message_flow(destination="temp-swap", unit=0, value=1)
- channel.message_flow(destination="temp-swap", unit=1, value=0xFFFFFFFF)
- msg = self.client.queue("temp-swap").get(timeout=1)
- channel.message_cancel(destination="temp-swap")
- msg.complete();
-
- #re-publish to dest
- channel.message_transfer(content=Content(properties={'routing_key':dest, 'message_id':msg.content['message_id']},
- body=msg.content.body))
+ session.message_subscribe(destination="temp-swap", queue=src)
+ session.message_flow(destination="temp-swap", unit=0, value=1)
+ session.message_flow(destination="temp-swap", unit=1, value=0xFFFFFFFF)
+ msg = session.incoming("temp-swap").get(timeout=1)
+ session.message_cancel(destination="temp-swap")
+ session.message_accept(RangedSet(msg.id))
+ #todo: also complete at this point?
+
+ #re-publish to dest:
+ dp=session.delivery_properties(routing_key=dest)
+ mp=session.message_properties(correlation_id=self.getMessageProperty(msg, 'correlation_id'))
+ session.message_transfer(message=Message(dp, mp, msg.body))
def assertMessageCount(self, expected, queue):
- self.assertEqual(expected, self.channel.queue_query(queue=queue).message_count)
+ self.assertEqual(expected, self.session.queue_query(queue=queue).message_count)
def assertMessageId(self, expected, queue):
- self.channel.message_subscribe(queue=queue, destination="results")
- self.channel.message_flow(destination="results", unit=0, value=1)
- self.channel.message_flow(destination="results", unit=1, value=0xFFFFFFFF)
- self.assertEqual(expected, self.client.queue("results").get(timeout=1).content['message_id'])
- self.channel.message_cancel(destination="results")
+ self.session.message_subscribe(queue=queue, destination="results")
+ self.session.message_flow(destination="results", unit=0, value=1)
+ self.session.message_flow(destination="results", unit=1, value=0xFFFFFFFF)
+ self.assertEqual(expected, self.getMessageProperty(self.session.incoming("results").get(timeout=1), 'correlation_id'))
+ self.session.message_cancel(destination="results")
+
+ def getMessageProperty(self, msg, prop):
+ for h in msg.headers:
+ if hasattr(h, prop): return getattr(h, prop)
+ return None
+
+ def keepQueuesAlive(self, names):
+ session = self.conn.session("nasty", 99)
+ for n in names:
+ session.queue_declare(queue=n, auto_delete=True)
+ session.message_subscribe(destination=n, queue=n)
+ return session
+
+ def createMessage(self, session, key, id, body):
+ dp=session.delivery_properties(routing_key=key)
+ mp=session.message_properties(correlation_id=id)
+ session.message_transfer(message=Message(dp, mp, body))
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/example.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/example.py?rev=648207&r1=648206&r2=648207&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/example.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/example.py Tue Apr 15 03:28:06 2008
@@ -17,10 +17,10 @@
# under the License.
#
-from qpid.content import Content
-from qpid.testlib import testrunner, TestBase
+from qpid.datatypes import Message, RangedSet
+from qpid.testlib import TestBase010
-class ExampleTest (TestBase):
+class ExampleTest (TestBase010):
"""
An example Qpid test, illustrating the unittest framework and the
python Qpid client. The test class must inherit TestBase. The
@@ -35,9 +35,9 @@
"""
# By inheriting TestBase, self.client is automatically connected
- # and self.channel is automatically opened as channel(1)
- # Other channel methods mimic the protocol.
- channel = self.channel
+ # and self.session is automatically opened as session(1)
+ # Other session methods mimic the protocol.
+ session = self.session
# Now we can send regular commands. If you want to see what the method
# arguments mean or what other commands are available, you can use the
@@ -57,30 +57,30 @@
# interact with the server.
# Here we use ordinal arguments.
- self.exchange_declare(channel, 0, "test", "direct")
+ session.exchange_declare("test", "direct")
# Here we use keyword arguments.
- self.queue_declare(channel, queue="test-queue")
- channel.queue_bind(queue="test-queue", exchange="test", routing_key="key")
+ session.queue_declare(queue="test-queue", exclusive=True, auto_delete=True)
+ session.exchange_bind(queue="test-queue", exchange="test", binding_key="key")
- # Call Channel.basic_consume to register as a consumer.
+ # Call Session.subscribe to register as a consumer.
# All the protocol methods return a message object. The message object
# has fields corresponding to the reply method fields, plus a content
# field that is filled if the reply includes content. In this case the
# interesting field is the consumer_tag.
- channel.message_subscribe(queue="test-queue", destination="consumer_tag")
- channel.message_flow(destination="consumer_tag", unit=0, value=0xFFFFFFFF)
- channel.message_flow(destination="consumer_tag", unit=1, value=0xFFFFFFFF)
-
- # We can use the Client.queue(...) method to access the queue
- # corresponding to our consumer_tag.
- queue = self.client.queue("consumer_tag")
+ session.message_subscribe(queue="test-queue", destination="consumer_tag")
+ session.message_flow(destination="consumer_tag", unit=0, value=0xFFFFFFFF)
+ session.message_flow(destination="consumer_tag", unit=1, value=0xFFFFFFFF)
+
+ # We can use the session.incoming(...) method to access the messages
+ # delivered for our consumer_tag.
+ queue = session.incoming("consumer_tag")
# Now lets publish a message and see if our consumer gets it. To do
- # this we need to import the Content class.
- sent = Content("Hello World!")
- sent["routing_key"] = "key"
- channel.message_transfer(destination="test", content=sent)
+ # this we need to import the Message class.
+ delivery_properties = session.delivery_properties(routing_key="key")
+ sent = Message(delivery_properties, "Hello World!")
+ session.message_transfer(destination="test", message=sent)
# Now we'll wait for the message to arrive. We can use the timeout
# argument in case the server hangs. By default queue.get() will wait
@@ -88,8 +88,8 @@
msg = queue.get(timeout=10)
# And check that we got the right response with assertEqual
- self.assertEqual(sent.body, msg.content.body)
+ self.assertEqual(sent.body, msg.body)
# Now acknowledge the message.
- msg.complete()
+ session.message_accept(RangedSet(msg.id))
Modified: incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/exchange.py
URL: http://svn.apache.org/viewvc/incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/exchange.py?rev=648207&r1=648206&r2=648207&view=diff
==============================================================================
--- incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/exchange.py (original)
+++ incubator/qpid/branches/thegreatmerge/qpid/python/tests_0-10/exchange.py Tue Apr 15 03:28:06 2008
@@ -23,10 +23,94 @@
Test classes ending in 'RuleTests' are derived from rules in amqp.xml.
"""
-import Queue, logging
-from qpid.testlib import TestBase
-from qpid.content import Content
+import Queue, logging, traceback
+from qpid.testlib import TestBase010
+from qpid.datatypes import Message
from qpid.client import Closed
+from qpid.session import SessionException
+
+
+class TestHelper(TestBase010):
+ def setUp(self):
+ TestBase010.setUp(self)
+ self.queues = []
+ self.exchanges = []
+
+ def tearDown(self):
+ try:
+ for ssn, q in self.queues:
+ ssn.queue_delete(queue=q)
+ for ssn, ex in self.exchanges:
+ ssn.exchange_delete(exchange=ex)
+ except:
+ print "Error on tearDown:"
+ print traceback.print_exc()
+ TestBase010.tearDown(self)
+
+ def createMessage(self, key="", body=""):
+ return Message(self.session.delivery_properties(routing_key=key), body)
+
+ def getApplicationHeaders(self, msg):
+ for h in msg.headers:
+ if hasattr(h, 'application_headers'): return getattr(h, 'application_headers')
+ return None
+
+ def assertPublishGet(self, queue, exchange="", routing_key="", properties=None):
+ """
+ Publish to exchange and assert queue.get() returns the same message.
+ """
+ body = self.uniqueString()
+ dp=self.session.delivery_properties(routing_key=routing_key)
+ mp=self.session.message_properties(application_headers=properties)
+ self.session.message_transfer(destination=exchange, message=Message(dp, mp, body))
+ msg = queue.get(timeout=1)
+ self.assertEqual(body, msg.body)
+ if (properties):
+ self.assertEqual(properties, self.getApplicationHeaders(msg))
+
+ def assertPublishConsume(self, queue="", exchange="", routing_key="", properties=None):
+ """
+ Publish a message and consume it, assert it comes back intact.
+ Return the Queue object used to consume.
+ """
+ self.assertPublishGet(self.consume(queue), exchange, routing_key, properties)
+
+ def assertEmpty(self, queue):
+ """Assert that the queue is empty"""
+ try:
+ queue.get(timeout=1)
+ self.fail("Queue is not empty.")
+ except Queue.Empty: None # Ignore
+
+ def queue_declare(self, session=None, *args, **keys):
+ session = session or self.session
+ reply = session.queue_declare(*args, **keys)
+ self.queues.append((session, keys["queue"]))
+ return reply
+
+ def exchange_declare(self, session=None, ticket=0, exchange='',
+ type='', passive=False, durable=False,
+ auto_delete=False,
+ arguments={}):
+ session = session or self.session
+ reply = session.exchange_declare(exchange=exchange, type=type, passive=passive,durable=durable, auto_delete=auto_delete, arguments=arguments)
+ self.exchanges.append((session,exchange))
+ return reply
+
+ def uniqueString(self):
+ """Generate a unique string, unique for this TestBase instance"""
+ if not "uniqueCounter" in dir(self): self.uniqueCounter = 1;
+ return "Test Message " + str(self.uniqueCounter)
+
+ def consume(self, queueName):
+ """Consume from named queue returns the Queue object."""
+ if not "uniqueTag" in dir(self): self.uniqueTag = 1
+ else: self.uniqueTag += 1
+ consumer_tag = "tag" + str(self.uniqueTag)
+ self.session.message_subscribe(queue=queueName, destination=consumer_tag)
+ self.session.message_flow(destination=consumer_tag, unit=0, value=0xFFFFFFFF)
+ self.session.message_flow(destination=consumer_tag, unit=1, value=0xFFFFFFFF)
+ return self.session.incoming(consumer_tag)
class StandardExchangeVerifier:
@@ -37,7 +121,7 @@
def verifyDirectExchange(self, ex):
"""Verify that ex behaves like a direct exchange."""
self.queue_declare(queue="q")
- self.channel.queue_bind(queue="q", exchange=ex, routing_key="k")
+ self.session.exchange_bind(queue="q", exchange=ex, binding_key="k")
self.assertPublishConsume(exchange=ex, queue="q", routing_key="k")
try:
self.assertPublishConsume(exchange=ex, queue="q", routing_key="kk")
@@ -47,38 +131,38 @@
def verifyFanOutExchange(self, ex):
"""Verify that ex behaves like a fanout exchange."""
self.queue_declare(queue="q")
- self.channel.queue_bind(queue="q", exchange=ex)
+ self.session.exchange_bind(queue="q", exchange=ex)
self.queue_declare(queue="p")
- self.channel.queue_bind(queue="p", exchange=ex)
+ self.session.exchange_bind(queue="p", exchange=ex)
for qname in ["q", "p"]: self.assertPublishGet(self.consume(qname), ex)
def verifyTopicExchange(self, ex):
"""Verify that ex behaves like a topic exchange"""
self.queue_declare(queue="a")
- self.channel.queue_bind(queue="a", exchange=ex, routing_key="a.#.b.*")
+ self.session.exchange_bind(queue="a", exchange=ex, binding_key="a.#.b.*")
q = self.consume("a")
self.assertPublishGet(q, ex, "a.b.x")
self.assertPublishGet(q, ex, "a.x.b.x")
self.assertPublishGet(q, ex, "a.x.x.b.x")
# Shouldn't match
- self.channel.message_transfer(destination=ex, content=Content(properties={'routing_key':"a.b"}))
- self.channel.message_transfer(destination=ex, content=Content(properties={'routing_key':"a.b.x.y"}))
- self.channel.message_transfer(destination=ex, content=Content(properties={'routing_key':"x.a.b.x"}))
- self.channel.message_transfer(destination=ex, content=Content(properties={'routing_key':"a.b"}))
+ self.session.message_transfer(destination=ex, message=self.createMessage("a.b"))
+ self.session.message_transfer(destination=ex, message=self.createMessage("a.b.x.y"))
+ self.session.message_transfer(destination=ex, message=self.createMessage("x.a.b.x"))
+ self.session.message_transfer(destination=ex, message=self.createMessage("a.b"))
self.assert_(q.empty())
def verifyHeadersExchange(self, ex):
"""Verify that ex is a headers exchange"""
self.queue_declare(queue="q")
- self.channel.queue_bind(queue="q", exchange=ex, arguments={ "x-match":"all", "name":"fred" , "age":3} )
+ self.session.exchange_bind(queue="q", exchange=ex, arguments={ "x-match":"all", "name":"fred" , "age":3} )
q = self.consume("q")
headers = {"name":"fred", "age":3}
self.assertPublishGet(q, exchange=ex, properties=headers)
- self.channel.message_transfer(destination=ex) # No headers, won't deliver
+ self.session.message_transfer(destination=ex) # No headers, won't deliver
self.assertEmpty(q);
-class RecommendedTypesRuleTests(TestBase, StandardExchangeVerifier):
+class RecommendedTypesRuleTests(TestHelper, StandardExchangeVerifier):
"""
The server SHOULD implement these standard exchange types: topic, headers.
@@ -106,7 +190,7 @@
self.verifyHeadersExchange("h")
-class RequiredInstancesRuleTests(TestBase, StandardExchangeVerifier):
+class RequiredInstancesRuleTests(TestHelper, StandardExchangeVerifier):
"""
The server MUST, in each virtual host, pre-declare an exchange instance
for each standard exchange type that it implements, where the name of the
@@ -124,7 +208,7 @@
def testAmqMatch(self): self.verifyHeadersExchange("amq.match")
-class DefaultExchangeRuleTests(TestBase, StandardExchangeVerifier):
+class DefaultExchangeRuleTests(TestHelper, StandardExchangeVerifier):
"""
The server MUST predeclare a direct exchange to act as the default exchange
for content Publish methods and for default queue bindings.
@@ -144,20 +228,20 @@
# TODO aconway 2006-09-27: Fill in empty tests:
-class DefaultAccessRuleTests(TestBase):
+class DefaultAccessRuleTests(TestHelper):
"""
The server MUST NOT allow clients to access the default exchange except
by specifying an empty exchange name in the Queue.Bind and content Publish
methods.
"""
-class ExtensionsRuleTests(TestBase):
+class ExtensionsRuleTests(TestHelper):
"""
The server MAY implement other exchange types as wanted.
"""
-class DeclareMethodMinimumRuleTests(TestBase):
+class DeclareMethodMinimumRuleTests(TestHelper):
"""
The server SHOULD support a minimum of 16 exchanges per virtual host and
ideally, impose no limit except as defined by available resources.
@@ -168,7 +252,7 @@
"""
-class DeclareMethodTicketFieldValidityRuleTests(TestBase):
+class DeclareMethodTicketFieldValidityRuleTests(TestHelper):
"""
The client MUST provide a valid access ticket giving "active" access to
the realm in which the exchange exists or will be created, or "passive"
@@ -179,7 +263,7 @@
"""
-class DeclareMethodExchangeFieldReservedRuleTests(TestBase):
+class DeclareMethodExchangeFieldReservedRuleTests(TestHelper):
"""
Exchange names starting with "amq." are reserved for predeclared and
standardised exchanges. The client MUST NOT attempt to create an exchange
@@ -189,7 +273,7 @@
"""
-class DeclareMethodTypeFieldTypedRuleTests(TestBase):
+class DeclareMethodTypeFieldTypedRuleTests(TestHelper):
"""
Exchanges cannot be redeclared with different types. The client MUST not
attempt to redeclare an existing exchange with a different type than used
@@ -199,7 +283,7 @@
"""
-class DeclareMethodTypeFieldSupportRuleTests(TestBase):
+class DeclareMethodTypeFieldSupportRuleTests(TestHelper):
"""
The client MUST NOT attempt to create an exchange with a type that the
server does not support.
@@ -208,20 +292,20 @@
"""
-class DeclareMethodPassiveFieldNotFoundRuleTests(TestBase):
+class DeclareMethodPassiveFieldNotFoundRuleTests(TestHelper):
"""
If set, and the exchange does not already exist, the server MUST raise a
channel exception with reply code 404 (not found).
"""
def test(self):
try:
- self.channel.exchange_declare(exchange="humpty_dumpty", passive=True)
+ self.session.exchange_declare(exchange="humpty_dumpty", passive=True)
self.fail("Expected 404 for passive declaration of unknown exchange.")
- except Closed, e:
- self.assertChannelException(404, e.args[0])
+ except SessionException, e:
+ self.assertEquals(404, e.args[0].error_code)
-class DeclareMethodDurableFieldSupportRuleTests(TestBase):
+class DeclareMethodDurableFieldSupportRuleTests(TestHelper):
"""
The server MUST support both durable and transient exchanges.
@@ -229,7 +313,7 @@
"""
-class DeclareMethodDurableFieldStickyRuleTests(TestBase):
+class DeclareMethodDurableFieldStickyRuleTests(TestHelper):
"""
The server MUST ignore the durable field if the exchange already exists.
@@ -237,7 +321,7 @@
"""
-class DeclareMethodAutoDeleteFieldStickyRuleTests(TestBase):
+class DeclareMethodAutoDeleteFieldStickyRuleTests(TestHelper):
"""
The server MUST ignore the auto-delete field if the exchange already
exists.
@@ -246,7 +330,7 @@
"""
-class DeleteMethodTicketFieldValidityRuleTests(TestBase):
+class DeleteMethodTicketFieldValidityRuleTests(TestHelper):
"""
The client MUST provide a valid access ticket giving "active" access
rights to the exchange's access realm.
@@ -256,18 +340,18 @@
"""
-class DeleteMethodExchangeFieldExistsRuleTests(TestBase):
+class DeleteMethodExchangeFieldExistsRuleTests(TestHelper):
"""
The client MUST NOT attempt to delete an exchange that does not exist.
"""
-class HeadersExchangeTests(TestBase):
+class HeadersExchangeTests(TestHelper):
"""
Tests for headers exchange functionality.
"""
def setUp(self):
- TestBase.setUp(self)
+ TestHelper.setUp(self)
self.queue_declare(queue="q")
self.q = self.consume("q")
@@ -275,10 +359,11 @@
self.assertPublishGet(self.q, exchange="amq.match", properties=headers)
def myBasicPublish(self, headers):
- self.channel.message_transfer(destination="amq.match", content=Content("foobar", properties={'application_headers':headers}))
+ mp=self.session.message_properties(application_headers=headers)
+ self.session.message_transfer(destination="amq.match", message=Message(mp, "foobar"))
def testMatchAll(self):
- self.channel.queue_bind(queue="q", exchange="amq.match", arguments={ 'x-match':'all', "name":"fred", "age":3})
+ self.session.exchange_bind(queue="q", exchange="amq.match", arguments={ 'x-match':'all', "name":"fred", "age":3})
self.myAssertPublishGet({"name":"fred", "age":3})
self.myAssertPublishGet({"name":"fred", "age":3, "extra":"ignoreme"})
@@ -290,7 +375,7 @@
self.assertEmpty(self.q)
def testMatchAny(self):
- self.channel.queue_bind(queue="q", exchange="amq.match", arguments={ 'x-match':'any', "name":"fred", "age":3})
+ self.session.exchange_bind(queue="q", exchange="amq.match", arguments={ 'x-match':'any', "name":"fred", "age":3})
self.myAssertPublishGet({"name":"fred"})
self.myAssertPublishGet({"name":"fred", "ignoreme":10})
self.myAssertPublishGet({"ignoreme":10, "age":3})
@@ -301,35 +386,31 @@
self.assertEmpty(self.q)
-class MiscellaneousErrorsTests(TestBase):
+class MiscellaneousErrorsTests(TestHelper):
"""
Test some miscellaneous error conditions
"""
def testTypeNotKnown(self):
try:
- self.channel.exchange_declare(exchange="test_type_not_known_exchange", type="invalid_type")
+ self.session.exchange_declare(exchange="test_type_not_known_exchange", type="invalid_type")
self.fail("Expected 503 for declaration of unknown exchange type.")
- except Closed, e:
- self.assertConnectionException(503, e.args[0])
+ except SessionException, e:
+ self.assertEquals(503, e.args[0].error_code)
def testDifferentDeclaredType(self):
- self.channel.exchange_declare(exchange="test_different_declared_type_exchange", type="direct")
+ self.exchange_declare(exchange="test_different_declared_type_exchange", type="direct")
try:
- self.channel.exchange_declare(exchange="test_different_declared_type_exchange", type="topic")
+ session = self.conn.session("alternate", 2)
+ session.exchange_declare(exchange="test_different_declared_type_exchange", type="topic")
self.fail("Expected 530 for redeclaration of exchange with different type.")
- except Closed, e:
- self.assertConnectionException(530, e.args[0])
- #cleanup
- other = self.connect()
- c2 = other.channel(1)
- c2.session_open()
- c2.exchange_delete(exchange="test_different_declared_type_exchange")
+ except SessionException, e:
+ self.assertEquals(530, e.args[0].error_code)
-class ExchangeTests(TestBase):
+class ExchangeTests(TestHelper):
def testHeadersBindNoMatchArg(self):
- self.channel.queue_declare(queue="q", exclusive=True, auto_delete=True)
+ self.session.queue_declare(queue="q", exclusive=True, auto_delete=True)
try:
- self.channel.queue_bind(queue="q", exchange="amq.match", arguments={"name":"fred" , "age":3} )
+ self.session.exchange_bind(queue="q", exchange="amq.match", arguments={"name":"fred" , "age":3} )
self.fail("Expected failure for missing x-match arg.")
- except Closed, e:
- self.assertConnectionException(541, e.args[0])
+ except SessionException, e:
+ self.assertEquals(541, e.args[0].error_code)