From 6577298076dbacbcc2d3280057b9129282ebf162 Mon Sep 17 00:00:00 2001
From: "Rafael H. Schloming"
Date: Thu, 18 Feb 2010 20:22:23 +0000
Subject: split messaging.py into multiple files and made qpid.messaging a
package
git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk/qpid@911550 13f79535-47bb-0310-9956-ffa450edef68
---
python/qpid/tests/__init__.py | 32 +-
python/qpid/tests/address.py | 309 ---------
python/qpid/tests/messaging.py | 1079 ------------------------------
python/qpid/tests/messaging/__init__.py | 106 +++
python/qpid/tests/messaging/address.py | 309 +++++++++
python/qpid/tests/messaging/endpoints.py | 878 ++++++++++++++++++++++++
python/qpid/tests/messaging/message.py | 116 ++++
7 files changed, 1439 insertions(+), 1390 deletions(-)
delete mode 100644 python/qpid/tests/address.py
delete mode 100644 python/qpid/tests/messaging.py
create mode 100644 python/qpid/tests/messaging/__init__.py
create mode 100644 python/qpid/tests/messaging/address.py
create mode 100644 python/qpid/tests/messaging/endpoints.py
create mode 100644 python/qpid/tests/messaging/message.py
(limited to 'python/qpid/tests')
diff --git a/python/qpid/tests/__init__.py b/python/qpid/tests/__init__.py
index 039214ca42..101a0c3759 100644
--- a/python/qpid/tests/__init__.py
+++ b/python/qpid/tests/__init__.py
@@ -26,7 +26,35 @@ class Test:
self.config = config
# API Tests
-import address, framing, mimetype, messaging
+import qpid.tests.framing
+import qpid.tests.mimetype
+import qpid.tests.messaging
# Legacy Tests
-import codec, queue, datatypes, connection, spec010, codec010
+import qpid.tests.codec
+import qpid.tests.queue
+import qpid.tests.datatypes
+import qpid.tests.connection
+import qpid.tests.spec010
+import qpid.tests.codec010
+
+class TestTestsXXX(Test):
+
+ def testFoo(self):
+ print "this test has output"
+
+ def testBar(self):
+ print "this test "*8
+ print "has"*10
+ print "a"*75
+ print "lot of"*10
+ print "output"*10
+
+ def testQux(self):
+ import sys
+ sys.stdout.write("this test has output with no newline")
+
+ def testQuxFail(self):
+ import sys
+ sys.stdout.write("this test has output with no newline")
+ fdsa
diff --git a/python/qpid/tests/address.py b/python/qpid/tests/address.py
deleted file mode 100644
index 7e6c6a5ee5..0000000000
--- a/python/qpid/tests/address.py
+++ /dev/null
@@ -1,309 +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.tests import Test
-from qpid.address import lex, parse, ParseError, EOF, ID, NUMBER, SYM, WSPACE, \
- LEXER
-from qpid.lexer import Token
-from qpid.harness import Skipped
-from parser import ParserBase
-
-def indent(st):
- return " " + st.replace("\n", "\n ")
-
-def pprint_address(name, subject, options):
- return "NAME: %s\nSUBJECT: %s\nOPTIONS: %s" % \
- (pprint(name), pprint(subject), pprint(options))
-
-def pprint(o):
- if isinstance(o, dict):
- return pprint_map(o)
- elif isinstance(o, list):
- return pprint_list(o)
- elif isinstance(o, basestring):
- return pprint_string(o)
- else:
- return repr(o)
-
-def pprint_map(m):
- items = ["%s: %s" % (pprint(k), pprint(v)) for k, v in m.items()]
- items.sort()
- return pprint_items("{", items, "}")
-
-def pprint_list(l):
- return pprint_items("[", [pprint(x) for x in l], "]")
-
-def pprint_items(start, items, end):
- if items:
- return "%s\n%s\n%s" % (start, ",\n".join([indent(i) for i in items]), end)
- else:
- return "%s%s" % (start, end)
-
-def pprint_string(s):
- result = "'"
- for c in s:
- if c == "'":
- result += "\\'"
- elif c == "\n":
- result += "\\n"
- elif ord(c) >= 0x80:
- result += "\\u%04x" % ord(c)
- else:
- result += c
- result += "'"
- return result
-
-class AddressTests(ParserBase, Test):
-
- EXCLUDE = (WSPACE, EOF)
-
- def fields(self, line, n):
- result = line.split(":", n - 1)
- result.extend([None]*(n - len(result)))
- return result
-
- def call(self, parser, mode, input):
- try:
- from subprocess import Popen, PIPE, STDOUT
- po = Popen([parser, mode], stdin=PIPE, stdout=PIPE, stderr=STDOUT)
- except ImportError, e:
- raise Skipped("%s" % e)
- except OSError, e:
- raise Skipped("%s: %s" % (e, parser))
- out, _ = po.communicate(input=input)
- return out
-
- def parser(self):
- return self.config.defines.get("address.parser")
-
- def do_lex(self, st):
- parser = self.parser()
- if parser:
- out = self.call(parser, "lex", st)
- lines = out.split("\n")
- toks = []
- for line in lines:
- if line.strip():
- name, position, value = self.fields(line, 3)
- toks.append(Token(LEXER.type(name), value, position, st))
- return toks
- else:
- return lex(st)
-
- def do_parse(self, st):
- return parse(st)
-
- def valid(self, addr, name=None, subject=None, options=None):
- parser = self.parser()
- if parser:
- got = self.call(parser, "parse", addr)
- expected = "%s\n" % pprint_address(name, subject, options)
- assert expected == got, "expected\n%s\ngot\n%s" % (expected, got)
- else:
- ParserBase.valid(self, addr, (name, subject, options))
-
- def invalid(self, addr, error=None):
- parser = self.parser()
- if parser:
- got = self.call(parser, "parse", addr)
- expected = "ERROR: %s\n" % error
- assert expected == got, "expected %r, got %r" % (expected, got)
- else:
- ParserBase.invalid(self, addr, error)
-
- def testDashInId1(self):
- self.lex("foo-bar", ID)
-
- def testDashInId2(self):
- self.lex("foo-3", ID)
-
- def testDashAlone1(self):
- self.lex("foo - bar", ID, SYM, ID)
-
- def testDashAlone2(self):
- self.lex("foo - 3", ID, SYM, NUMBER)
-
- def testLeadingDash(self):
- self.lex("-foo", SYM, ID)
-
- def testTrailingDash(self):
- self.lex("foo-", ID, SYM)
-
- def testNegativeNum(self):
- self.lex("-3", NUMBER)
-
- def testHash(self):
- self.valid("foo/bar.#", "foo", "bar.#")
-
- def testStar(self):
- self.valid("foo/bar.*", "foo", "bar.*")
-
- def testColon(self):
- self.valid("foo.bar/baz.qux:moo:arf", "foo.bar", "baz.qux:moo:arf")
-
- def testOptions(self):
- self.valid("foo.bar/baz.qux:moo:arf; {key: value}",
- "foo.bar", "baz.qux:moo:arf", {"key": "value"})
-
- def testOptionsTrailingComma(self):
- self.valid("name/subject; {key: value,}", "name", "subject",
- {"key": "value"})
-
- def testSemiSubject(self):
- self.valid("foo.bar/'baz.qux;moo:arf'; {key: value}",
- "foo.bar", "baz.qux;moo:arf", {"key": "value"})
-
- def testCommaSubject(self):
- self.valid("foo.bar/baz.qux.{moo,arf}", "foo.bar", "baz.qux.{moo,arf}")
-
- def testCommaSubjectOptions(self):
- self.valid("foo.bar/baz.qux.{moo,arf}; {key: value}", "foo.bar",
- "baz.qux.{moo,arf}", {"key": "value"})
-
- def testUnbalanced(self):
- self.valid("foo.bar/baz.qux.{moo,arf; {key: value}", "foo.bar",
- "baz.qux.{moo,arf", {"key": "value"})
-
- def testSlashQuote(self):
- self.valid("foo.bar\\/baz.qux.{moo,arf; {key: value}",
- "foo.bar/baz.qux.{moo,arf",
- None, {"key": "value"})
-
- def testSlashHexEsc1(self):
- self.valid("foo.bar\\x00baz.qux.{moo,arf; {key: value}",
- "foo.bar\x00baz.qux.{moo,arf",
- None, {"key": "value"})
-
- def testSlashHexEsc2(self):
- self.valid("foo.bar\\xffbaz.qux.{moo,arf; {key: value}",
- "foo.bar\xffbaz.qux.{moo,arf",
- None, {"key": "value"})
-
- def testSlashHexEsc3(self):
- self.valid("foo.bar\\xFFbaz.qux.{moo,arf; {key: value}",
- "foo.bar\xFFbaz.qux.{moo,arf",
- None, {"key": "value"})
-
- def testSlashUnicode1(self):
- self.valid("foo.bar\\u1234baz.qux.{moo,arf; {key: value}",
- u"foo.bar\u1234baz.qux.{moo,arf", None, {"key": "value"})
-
- def testSlashUnicode2(self):
- self.valid("foo.bar\\u0000baz.qux.{moo,arf; {key: value}",
- u"foo.bar\u0000baz.qux.{moo,arf", None, {"key": "value"})
-
- def testSlashUnicode3(self):
- self.valid("foo.bar\\uffffbaz.qux.{moo,arf; {key: value}",
- u"foo.bar\uffffbaz.qux.{moo,arf", None, {"key": "value"})
-
- def testSlashUnicode4(self):
- self.valid("foo.bar\\uFFFFbaz.qux.{moo,arf; {key: value}",
- u"foo.bar\uFFFFbaz.qux.{moo,arf", None, {"key": "value"})
-
- def testNoName(self):
- self.invalid("; {key: value}",
- "unexpected token SEMI(;) line:1,0:; {key: value}")
-
- def testEmpty(self):
- self.invalid("", "unexpected token EOF line:1,0:")
-
- def testNoNameSlash(self):
- self.invalid("/asdf; {key: value}",
- "unexpected token SLASH(/) line:1,0:/asdf; {key: value}")
-
- def testBadOptions1(self):
- self.invalid("name/subject; {",
- "expecting (NUMBER, STRING, ID, LBRACE, LBRACK, RBRACE), "
- "got EOF line:1,15:name/subject; {")
-
- def testBadOptions2(self):
- self.invalid("name/subject; { 3",
- "expecting COLON, got EOF "
- "line:1,17:name/subject; { 3")
-
- def testBadOptions3(self):
- self.invalid("name/subject; { key:",
- "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), got EOF "
- "line:1,20:name/subject; { key:")
-
- def testBadOptions4(self):
- self.invalid("name/subject; { key: value",
- "expecting (COMMA, RBRACE), got EOF "
- "line:1,26:name/subject; { key: value")
-
- def testBadOptions5(self):
- self.invalid("name/subject; { key: value asdf",
- "expecting (COMMA, RBRACE), got ID(asdf) "
- "line:1,27:name/subject; { key: value asdf")
-
- def testBadOptions6(self):
- self.invalid("name/subject; { key: value,",
- "expecting (NUMBER, STRING, ID, LBRACE, LBRACK, RBRACE), got EOF "
- "line:1,27:name/subject; { key: value,")
-
- def testBadOptions7(self):
- self.invalid("name/subject; { key: value } asdf",
- "expecting EOF, got ID(asdf) "
- "line:1,29:name/subject; { key: value } asdf")
-
- def testList1(self):
- self.valid("name/subject; { key: [] }", "name", "subject", {"key": []})
-
- def testList2(self):
- self.valid("name/subject; { key: ['one'] }", "name", "subject", {"key": ['one']})
-
- def testList3(self):
- self.valid("name/subject; { key: [1, 2, 3] }", "name", "subject",
- {"key": [1, 2, 3]})
-
- def testList4(self):
- self.valid("name/subject; { key: [1, [2, 3], 4] }", "name", "subject",
- {"key": [1, [2, 3], 4]})
-
- def testBadList1(self):
- self.invalid("name/subject; { key: [ }", "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), "
- "got RBRACE(}) line:1,23:name/subject; { key: [ }")
-
- def testBadList2(self):
- self.invalid("name/subject; { key: [ 1 }", "expecting (COMMA, RBRACK), "
- "got RBRACE(}) line:1,25:name/subject; { key: [ 1 }")
-
- def testBadList3(self):
- self.invalid("name/subject; { key: [ 1 2 }", "expecting (COMMA, RBRACK), "
- "got NUMBER(2) line:1,25:name/subject; { key: [ 1 2 }")
-
- def testBadList4(self):
- self.invalid("name/subject; { key: [ 1 2 ] }", "expecting (COMMA, RBRACK), "
- "got NUMBER(2) line:1,25:name/subject; { key: [ 1 2 ] }")
-
- def testMap1(self):
- self.valid("name/subject; { 'key': value }",
- "name", "subject", {"key": "value"})
-
- def testMap2(self):
- self.valid("name/subject; { 1: value }", "name", "subject", {1: "value"})
-
- def testMap3(self):
- self.valid('name/subject; { "foo.bar": value }',
- "name", "subject", {"foo.bar": "value"})
-
- def testBoolean(self):
- self.valid("name/subject; { true: True, false: False }",
- "name", "subject", {"true": True, "false": False})
diff --git a/python/qpid/tests/messaging.py b/python/qpid/tests/messaging.py
deleted file mode 100644
index 125f1b7157..0000000000
--- a/python/qpid/tests/messaging.py
+++ /dev/null
@@ -1,1079 +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.
-#
-
-# setup, usage, teardown, errors(sync), errors(async), stress, soak,
-# boundary-conditions, config
-
-import time
-from qpid import compat
-from qpid.tests import Test
-from qpid.harness import Skipped
-from qpid.messaging import Connection, ConnectError, Disconnected, Empty, \
- InsufficientCapacity, Message, ReceiveError, SendError, SessionError, \
- UNLIMITED, uuid4
-from Queue import Queue, Empty as QueueEmpty
-
-class Base(Test):
-
- def setup_connection(self):
- return None
-
- def setup_session(self):
- return None
-
- def setup_sender(self):
- return None
-
- def setup_receiver(self):
- return None
-
- def setup(self):
- self.test_id = uuid4()
- self.broker = self.config.broker
- try:
- self.conn = self.setup_connection()
- except ConnectError, e:
- raise Skipped(e)
- self.ssn = self.setup_session()
- self.snd = self.setup_sender()
- if self.snd is not None:
- self.snd.durable = self.durable()
- self.rcv = self.setup_receiver()
-
- def teardown(self):
- if self.conn is not None and self.conn.connected():
- self.conn.close()
-
- def content(self, base, count = None):
- if count is None:
- return "%s[%s]" % (base, self.test_id)
- else:
- return "%s[%s, %s]" % (base, count, self.test_id)
-
- def ping(self, ssn):
- PING_Q = 'ping-queue; {create: always, delete: always}'
- # send a message
- sender = ssn.sender(PING_Q, durable=self.durable())
- content = self.content("ping")
- sender.send(content)
- receiver = ssn.receiver(PING_Q)
- msg = receiver.fetch(0)
- ssn.acknowledge()
- assert msg.content == content, "expected %r, got %r" % (content, msg.content)
-
- def drain(self, rcv, limit=None, timeout=0, expected=None):
- contents = []
- try:
- while limit is None or len(contents) < limit:
- contents.append(rcv.fetch(timeout=timeout).content)
- except Empty:
- pass
- if expected is not None:
- assert expected == contents, "expected %s, got %s" % (expected, contents)
- return contents
-
- def assertEmpty(self, rcv):
- contents = self.drain(rcv)
- assert len(contents) == 0, "%s is supposed to be empty: %s" % (rcv, contents)
-
- def assertPending(self, rcv, expected):
- p = rcv.pending()
- assert p == expected, "expected %s, got %s" % (expected, p)
-
- def sleep(self):
- time.sleep(self.delay())
-
- def delay(self):
- return float(self.config.defines.get("delay", "2"))
-
- def get_bool(self, name):
- return self.config.defines.get(name, "false").lower() in ("true", "yes", "1")
-
- def durable(self):
- return self.get_bool("durable")
-
- def reconnect(self):
- return self.get_bool("reconnect")
-
-class SetupTests(Base):
-
- def testOpen(self):
- # XXX: need to flesh out URL support/syntax
- self.conn = Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
- self.ping(self.conn.session())
-
- def testConnect(self):
- # XXX: need to flesh out URL support/syntax
- self.conn = Connection(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
- self.conn.connect()
- self.ping(self.conn.session())
-
- def testConnectError(self):
- try:
- self.conn = Connection.open("localhost", 0)
- assert False, "connect succeeded"
- except ConnectError, e:
- # XXX: should verify that e includes appropriate diagnostic info
- pass
-
-class ConnectionTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def testSessionAnon(self):
- ssn1 = self.conn.session()
- ssn2 = self.conn.session()
- self.ping(ssn1)
- self.ping(ssn2)
- assert ssn1 is not ssn2
-
- def testSessionNamed(self):
- ssn1 = self.conn.session("one")
- ssn2 = self.conn.session("two")
- self.ping(ssn1)
- self.ping(ssn2)
- assert ssn1 is not ssn2
- assert ssn1 is self.conn.session("one")
- assert ssn2 is self.conn.session("two")
-
- def testDisconnect(self):
- ssn = self.conn.session()
- self.ping(ssn)
- self.conn.disconnect()
- try:
- self.ping(ssn)
- assert False, "ping succeeded"
- except Disconnected:
- # this is the expected failure when pinging on a disconnected
- # connection
- pass
- self.conn.connect()
- self.ping(ssn)
-
- def testClose(self):
- self.conn.close()
- assert not self.conn.connected()
-
-ACK_QC = 'test-ack-queue; {create: always}'
-ACK_QD = 'test-ack-queue; {delete: always}'
-
-class SessionTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def testSender(self):
- snd = self.ssn.sender('test-snd-queue; {create: sender, delete: receiver}',
- durable=self.durable())
- snd2 = self.ssn.sender(snd.target, durable=self.durable())
- assert snd is not snd2
- snd2.close()
-
- content = self.content("testSender")
- snd.send(content)
- rcv = self.ssn.receiver(snd.target)
- msg = rcv.fetch(0)
- assert msg.content == content
- self.ssn.acknowledge(msg)
-
- def testReceiver(self):
- rcv = self.ssn.receiver('test-rcv-queue; {create: always}')
- rcv2 = self.ssn.receiver(rcv.source)
- assert rcv is not rcv2
- rcv2.close()
-
- content = self.content("testReceiver")
- snd = self.ssn.sender(rcv.source, durable=self.durable())
- snd.send(content)
- msg = rcv.fetch(0)
- assert msg.content == content
- self.ssn.acknowledge(msg)
- snd2 = self.ssn.receiver('test-rcv-queue; {delete: always}')
-
- def testDisconnectedReceiver(self):
- self.conn.disconnect()
- rcv = self.ssn.receiver("test-dis-rcv-queue; {create: always, delete: always}")
- m = self.content("testDisconnectedReceiver")
- self.conn.connect()
- snd = self.ssn.sender("test-dis-rcv-queue")
- snd.send(m)
- self.drain(rcv, expected=[m])
-
- def testNextReceiver(self):
- ADDR = 'test-next-rcv-queue; {create: always, delete: always}'
- rcv1 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
- rcv2 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
- rcv3 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
-
- snd = self.ssn.sender(ADDR)
-
- msgs = []
- for i in range(10):
- content = self.content("testNextReceiver", i)
- snd.send(content)
- msgs.append(content)
-
- fetched = []
- try:
- while True:
- rcv = self.ssn.next_receiver(timeout=self.delay())
- assert rcv in (rcv1, rcv2, rcv3)
- assert rcv.pending() > 0
- fetched.append(rcv.fetch().content)
- except Empty:
- pass
- assert msgs == fetched, "expecting %s, got %s" % (msgs, fetched)
- self.ssn.acknowledge()
- #we set the capacity to 0 to prevent the deletion of the queue -
- #triggered the deletion policy when the first receiver is closed -
- #resulting in session exceptions being issued for the remaining
- #active subscriptions:
- for r in [rcv1, rcv2, rcv3]:
- r.capacity = 0
-
- # XXX, we need a convenient way to assert that required queues are
- # empty on setup, and possibly also to drain queues on teardown
- def ackTest(self, acker, ack_capacity=None):
- # send a bunch of messages
- snd = self.ssn.sender(ACK_QC, durable=self.durable())
- contents = [self.content("ackTest", i) for i in range(15)]
- for c in contents:
- snd.send(c)
-
- # drain the queue, verify the messages are there and then close
- # without acking
- rcv = self.ssn.receiver(ACK_QC)
- self.drain(rcv, expected=contents)
- self.ssn.close()
-
- # drain the queue again, verify that they are all the messages
- # were requeued, and ack this time before closing
- self.ssn = self.conn.session()
- if ack_capacity is not None:
- self.ssn.ack_capacity = ack_capacity
- rcv = self.ssn.receiver(ACK_QC)
- self.drain(rcv, expected=contents)
- acker(self.ssn)
- self.ssn.close()
-
- # drain the queue a final time and verify that the messages were
- # dequeued
- self.ssn = self.conn.session()
- rcv = self.ssn.receiver(ACK_QD)
- self.assertEmpty(rcv)
-
- def testAcknowledge(self):
- self.ackTest(lambda ssn: ssn.acknowledge())
-
- def testAcknowledgeAsync(self):
- self.ackTest(lambda ssn: ssn.acknowledge(sync=False))
-
- def testAcknowledgeAsyncAckCap0(self):
- try:
- try:
- self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 0)
- assert False, "acknowledge shouldn't succeed with ack_capacity of zero"
- except InsufficientCapacity:
- pass
- finally:
- self.ssn.ack_capacity = UNLIMITED
- self.drain(self.ssn.receiver(ACK_QD))
- self.ssn.acknowledge()
-
- def testAcknowledgeAsyncAckCap1(self):
- self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 1)
-
- def testAcknowledgeAsyncAckCap5(self):
- self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 5)
-
- def testAcknowledgeAsyncAckCapUNLIMITED(self):
- self.ackTest(lambda ssn: ssn.acknowledge(sync=False), UNLIMITED)
-
- def send(self, ssn, queue, base, count=1):
- snd = ssn.sender(queue, durable=self.durable())
- contents = []
- for i in range(count):
- c = self.content(base, i)
- snd.send(c)
- contents.append(c)
- snd.close()
- return contents
-
- def txTest(self, commit):
- TX_Q = 'test-tx-queue; {create: sender, delete: receiver}'
- TX_Q_COPY = 'test-tx-queue-copy; {create: always, delete: always}'
- txssn = self.conn.session(transactional=True)
- contents = self.send(self.ssn, TX_Q, "txTest", 3)
- txrcv = txssn.receiver(TX_Q)
- txsnd = txssn.sender(TX_Q_COPY, durable=self.durable())
- rcv = self.ssn.receiver(txrcv.source)
- copy_rcv = self.ssn.receiver(txsnd.target)
- self.assertEmpty(copy_rcv)
- for i in range(3):
- m = txrcv.fetch(0)
- txsnd.send(m)
- self.assertEmpty(copy_rcv)
- txssn.acknowledge()
- if commit:
- txssn.commit()
- self.assertEmpty(rcv)
- assert contents == self.drain(copy_rcv)
- else:
- txssn.rollback()
- assert contents == self.drain(rcv)
- self.assertEmpty(copy_rcv)
- self.ssn.acknowledge()
-
- def testCommit(self):
- self.txTest(True)
-
- def testRollback(self):
- self.txTest(False)
-
- def txTestSend(self, commit):
- TX_SEND_Q = 'test-tx-send-queue; {create: sender, delete: receiver}'
- txssn = self.conn.session(transactional=True)
- contents = self.send(txssn, TX_SEND_Q, "txTestSend", 3)
- rcv = self.ssn.receiver(TX_SEND_Q)
- self.assertEmpty(rcv)
-
- if commit:
- txssn.commit()
- assert contents == self.drain(rcv)
- self.ssn.acknowledge()
- else:
- txssn.rollback()
- self.assertEmpty(rcv)
- txssn.commit()
- self.assertEmpty(rcv)
-
- def testCommitSend(self):
- self.txTestSend(True)
-
- def testRollbackSend(self):
- self.txTestSend(False)
-
- def txTestAck(self, commit):
- TX_ACK_QC = 'test-tx-ack-queue; {create: always}'
- TX_ACK_QD = 'test-tx-ack-queue; {delete: always}'
- txssn = self.conn.session(transactional=True)
- txrcv = txssn.receiver(TX_ACK_QC)
- self.assertEmpty(txrcv)
- contents = self.send(self.ssn, TX_ACK_QC, "txTestAck", 3)
- assert contents == self.drain(txrcv)
-
- if commit:
- txssn.acknowledge()
- else:
- txssn.rollback()
- drained = self.drain(txrcv)
- assert contents == drained, "expected %s, got %s" % (contents, drained)
- txssn.acknowledge()
- txssn.rollback()
- assert contents == self.drain(txrcv)
- txssn.commit() # commit without ack
- self.assertEmpty(txrcv)
-
- txssn.close()
-
- txssn = self.conn.session(transactional=True)
- txrcv = txssn.receiver(TX_ACK_QC)
- assert contents == self.drain(txrcv)
- txssn.acknowledge()
- txssn.commit()
- rcv = self.ssn.receiver(TX_ACK_QD)
- self.assertEmpty(rcv)
- txssn.close()
- self.assertEmpty(rcv)
-
- def testCommitAck(self):
- self.txTestAck(True)
-
- def testRollbackAck(self):
- self.txTestAck(False)
-
- def testClose(self):
- self.ssn.close()
- try:
- self.ping(self.ssn)
- assert False, "ping succeeded"
- except Disconnected:
- pass
-
-RECEIVER_Q = 'test-receiver-queue; {create: always, delete: always}'
-
-class ReceiverTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def setup_sender(self):
- return self.ssn.sender(RECEIVER_Q)
-
- def setup_receiver(self):
- return self.ssn.receiver(RECEIVER_Q)
-
- def send(self, base, count = None):
- content = self.content(base, count)
- self.snd.send(content)
- return content
-
- def testFetch(self):
- try:
- msg = self.rcv.fetch(0)
- assert False, "unexpected message: %s" % msg
- except Empty:
- pass
- try:
- start = time.time()
- msg = self.rcv.fetch(self.delay())
- assert False, "unexpected message: %s" % msg
- except Empty:
- elapsed = time.time() - start
- assert elapsed >= self.delay()
-
- one = self.send("testFetch", 1)
- two = self.send("testFetch", 2)
- three = self.send("testFetch", 3)
- msg = self.rcv.fetch(0)
- assert msg.content == one
- msg = self.rcv.fetch(self.delay())
- assert msg.content == two
- msg = self.rcv.fetch()
- assert msg.content == three
- self.ssn.acknowledge()
-
- def testCapacityIncrease(self):
- content = self.send("testCapacityIncrease")
- self.sleep()
- assert self.rcv.pending() == 0
- self.rcv.capacity = UNLIMITED
- self.sleep()
- assert self.rcv.pending() == 1
- msg = self.rcv.fetch(0)
- assert msg.content == content
- assert self.rcv.pending() == 0
- self.ssn.acknowledge()
-
- def testCapacityDecrease(self):
- self.rcv.capacity = UNLIMITED
- one = self.send("testCapacityDecrease", 1)
- self.sleep()
- assert self.rcv.pending() == 1
- msg = self.rcv.fetch(0)
- assert msg.content == one
-
- self.rcv.capacity = 0
-
- two = self.send("testCapacityDecrease", 2)
- self.sleep()
- assert self.rcv.pending() == 0
- msg = self.rcv.fetch(0)
- assert msg.content == two
-
- self.ssn.acknowledge()
-
- def testCapacity(self):
- self.rcv.capacity = 5
- self.assertPending(self.rcv, 0)
-
- for i in range(15):
- self.send("testCapacity", i)
- self.sleep()
- self.assertPending(self.rcv, 5)
-
- self.drain(self.rcv, limit = 5)
- self.sleep()
- self.assertPending(self.rcv, 5)
-
- drained = self.drain(self.rcv)
- assert len(drained) == 10, "%s, %s" % (len(drained), drained)
- self.assertPending(self.rcv, 0)
-
- self.ssn.acknowledge()
-
- def testCapacityUNLIMITED(self):
- self.rcv.capacity = UNLIMITED
- self.assertPending(self.rcv, 0)
-
- for i in range(10):
- self.send("testCapacityUNLIMITED", i)
- self.sleep()
- self.assertPending(self.rcv, 10)
-
- self.drain(self.rcv)
- self.assertPending(self.rcv, 0)
-
- self.ssn.acknowledge()
-
- def testPending(self):
- self.rcv.capacity = UNLIMITED
- assert self.rcv.pending() == 0
-
- for i in range(3):
- self.send("testPending", i)
- self.sleep()
- assert self.rcv.pending() == 3
-
- for i in range(3, 10):
- self.send("testPending", i)
- self.sleep()
- assert self.rcv.pending() == 10
-
- self.drain(self.rcv, limit=3)
- assert self.rcv.pending() == 7
-
- self.drain(self.rcv)
- assert self.rcv.pending() == 0
-
- self.ssn.acknowledge()
-
- def testDoubleClose(self):
- m1 = self.content("testDoubleClose", 1)
- m2 = self.content("testDoubleClose", 2)
-
- snd = self.ssn.sender("""test-double-close; {
- create: always,
- delete: sender,
- node-properties: {
- type: topic
- }
-}
-""")
- r1 = self.ssn.receiver(snd.target)
- r2 = self.ssn.receiver(snd.target)
- snd.send(m1)
- self.drain(r1, expected=[m1])
- self.drain(r2, expected=[m1])
- r1.close()
- snd.send(m2)
- self.drain(r2, expected=[m2])
- r2.close()
-
- # XXX: need testClose
-
- def testMode(self):
- msgs = [self.content("testMode", 1),
- self.content("testMode", 2),
- self.content("testMode", 3)]
-
- for m in msgs:
- self.snd.send(m)
-
- rb = self.ssn.receiver('test-receiver-queue; {mode: browse}')
- rc = self.ssn.receiver('test-receiver-queue; {mode: consume}')
- self.drain(rb, expected=msgs)
- self.drain(rc, expected=msgs)
- rb2 = self.ssn.receiver(rb.source)
- self.assertEmpty(rb2)
- self.drain(self.rcv, expected=[])
-
-class AddressTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def badOption(self, options, error):
- try:
- self.ssn.sender("test-bad-options-snd; %s" % options)
- assert False
- except SendError, e:
- assert "error in options: %s" % error == str(e), e
-
- try:
- self.ssn.receiver("test-bad-options-rcv; %s" % options)
- assert False
- except ReceiveError, e:
- assert "error in options: %s" % error == str(e), e
-
- def testIllegalKey(self):
- self.badOption("{create: always, node-properties: "
- "{this-property-does-not-exist: 3}}",
- "node-properties: this-property-does-not-exist: "
- "illegal key")
-
- def testWrongValue(self):
- self.badOption("{create: asdf}", "create: asdf not in "
- "('always', 'sender', 'receiver', 'never')")
-
- def testWrongType1(self):
- self.badOption("{node-properties: asdf}",
- "node-properties: asdf is not a map")
-
- def testWrongType2(self):
- self.badOption("{node-properties: {durable: []}}",
- "node-properties: durable: [] is not a bool")
-
- def testNonQueueBindings(self):
- self.badOption("{node-properties: {type: topic, x-properties: "
- "{bindings: []}}}",
- "node-properties: x-properties: bindings: "
- "bindings are only permitted on nodes of type queue")
-
- def testCreateQueue(self):
- snd = self.ssn.sender("test-create-queue; {create: always, delete: always, "
- "node-properties: {type: queue, durable: False, "
- "x-properties: {auto_delete: true}}}")
- content = self.content("testCreateQueue")
- snd.send(content)
- rcv = self.ssn.receiver("test-create-queue")
- self.drain(rcv, expected=[content])
-
- def createExchangeTest(self, props=""):
- addr = """test-create-exchange; {
- create: always,
- delete: always,
- node-properties: {
- type: topic,
- durable: False,
- x-properties: {auto_delete: true, %s}
- }
- }""" % props
- snd = self.ssn.sender(addr)
- snd.send("ping")
- rcv1 = self.ssn.receiver("test-create-exchange/first")
- rcv2 = self.ssn.receiver("test-create-exchange/first")
- rcv3 = self.ssn.receiver("test-create-exchange/second")
- for r in (rcv1, rcv2, rcv3):
- try:
- r.fetch(0)
- assert False
- except Empty:
- pass
- msg1 = Message(self.content("testCreateExchange", 1), subject="first")
- msg2 = Message(self.content("testCreateExchange", 2), subject="second")
- snd.send(msg1)
- snd.send(msg2)
- self.drain(rcv1, expected=[msg1.content])
- self.drain(rcv2, expected=[msg1.content])
- self.drain(rcv3, expected=[msg2.content])
-
- def testCreateExchange(self):
- self.createExchangeTest()
-
- def testCreateExchangeDirect(self):
- self.createExchangeTest("type: direct")
-
- def testCreateExchangeTopic(self):
- self.createExchangeTest("type: topic")
-
- def testDeleteBySender(self):
- snd = self.ssn.sender("test-delete; {create: always}")
- snd.send("ping")
- snd.close()
- snd = self.ssn.sender("test-delete; {delete: always}")
- snd.send("ping")
- snd.close()
- try:
- self.ssn.sender("test-delete")
- except SendError, e:
- assert "no such queue" in str(e)
-
- def testDeleteByReceiver(self):
- rcv = self.ssn.receiver("test-delete; {create: always, delete: always}")
- try:
- rcv.fetch(0)
- except Empty:
- pass
- rcv.close()
-
- try:
- self.ssn.receiver("test-delete")
- assert False
- except ReceiveError, e:
- assert "no such queue" in str(e)
-
- def testDeleteSpecial(self):
- snd = self.ssn.sender("amq.topic; {delete: always}")
- snd.send("asdf")
- try:
- snd.close()
- except SessionError, e:
- assert "Cannot delete default exchange" in str(e)
- # XXX: need to figure out close after error
- self.conn._remove_session(self.ssn)
-
- def testBindings(self):
- snd = self.ssn.sender("""
-test-bindings-queue; {
- create: always,
- delete: always,
- node-properties: {
- x-properties: {
- bindings: ["amq.topic/a.#", "amq.direct/b", "amq.topic/c.*"]
- }
- }
-}
-""")
- snd.send("one")
- snd_a = self.ssn.sender("amq.topic/a.foo")
- snd_b = self.ssn.sender("amq.direct/b")
- snd_c = self.ssn.sender("amq.topic/c.bar")
- snd_a.send("two")
- snd_b.send("three")
- snd_c.send("four")
- rcv = self.ssn.receiver("test-bindings-queue")
- self.drain(rcv, expected=["one", "two", "three", "four"])
-
- def testBindingsAdditive(self):
- m1 = self.content("testBindingsAdditive", 1)
- m2 = self.content("testBindingsAdditive", 2)
- m3 = self.content("testBindingsAdditive", 3)
- m4 = self.content("testBindingsAdditive", 4)
-
- snd = self.ssn.sender("""
-test-bindings-additive-queue; {
- create: always,
- delete: always,
- node-properties: {
- x-properties: {
- bindings: ["amq.topic/a"]
- }
- }
-}
-""")
-
- snd_a = self.ssn.sender("amq.topic/a")
- snd_b = self.ssn.sender("amq.topic/b")
-
- snd_a.send(m1)
- snd_b.send(m2)
-
- rcv = self.ssn.receiver("test-bindings-additive-queue")
- self.drain(rcv, expected=[m1])
-
- new_snd = self.ssn.sender("""
-test-bindings-additive-queue; {
- node-properties: {
- x-properties: {
- bindings: ["amq.topic/b"]
- }
- }
-}
-""")
-
- new_snd.send(m3)
- snd_b.send(m4)
- self.drain(rcv, expected=[m3, m4])
-
- def testSubjectOverride(self):
- snd = self.ssn.sender("amq.topic/a")
- rcv_a = self.ssn.receiver("amq.topic/a")
- rcv_b = self.ssn.receiver("amq.topic/b")
- m1 = self.content("testSubjectOverride", 1)
- m2 = self.content("testSubjectOverride", 2)
- snd.send(m1)
- snd.send(Message(subject="b", content=m2))
- self.drain(rcv_a, expected=[m1])
- self.drain(rcv_b, expected=[m2])
-
- def testSubjectDefault(self):
- m1 = self.content("testSubjectDefault", 1)
- m2 = self.content("testSubjectDefault", 2)
- snd = self.ssn.sender("amq.topic/a")
- rcv = self.ssn.receiver("amq.topic")
- snd.send(m1)
- snd.send(Message(subject="b", content=m2))
- e1 = rcv.fetch(timeout=0)
- e2 = rcv.fetch(timeout=0)
- assert e1.subject == "a", "subject: %s" % e1.subject
- assert e2.subject == "b", "subject: %s" % e2.subject
- self.assertEmpty(rcv)
-
-NOSUCH_Q = "this-queue-should-not-exist"
-UNPARSEABLE_ADDR = "name/subject; {bad options"
-UNLEXABLE_ADDR = "\0x0\0x1\0x2\0x3"
-
-class AddressErrorTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def senderErrorTest(self, addr, exc, check=lambda e: True):
- try:
- self.ssn.sender(addr, durable=self.durable())
- assert False, "sender creation succeeded"
- except exc, e:
- assert check(e), "unexpected error: %s" % compat.format_exc(e)
-
- def receiverErrorTest(self, addr, exc, check=lambda e: True):
- try:
- self.ssn.receiver(addr)
- assert False, "receiver creation succeeded"
- except exc, e:
- assert check(e), "unexpected error: %s" % compat.format_exc(e)
-
- def testNoneTarget(self):
- # XXX: should have specific exception for this
- self.senderErrorTest(None, SendError)
-
- def testNoneSource(self):
- # XXX: should have specific exception for this
- self.receiverErrorTest(None, ReceiveError)
-
- def testNoTarget(self):
- # XXX: should have specific exception for this
- self.senderErrorTest(NOSUCH_Q, SendError, lambda e: NOSUCH_Q in str(e))
-
- def testNoSource(self):
- # XXX: should have specific exception for this
- self.receiverErrorTest(NOSUCH_Q, ReceiveError, lambda e: NOSUCH_Q in str(e))
-
- def testUnparseableTarget(self):
- # XXX: should have specific exception for this
- self.senderErrorTest(UNPARSEABLE_ADDR, SendError,
- lambda e: "expecting COLON" in str(e))
-
- def testUnparseableSource(self):
- # XXX: should have specific exception for this
- self.receiverErrorTest(UNPARSEABLE_ADDR, ReceiveError,
- lambda e: "expecting COLON" in str(e))
-
- def testUnlexableTarget(self):
- # XXX: should have specific exception for this
- self.senderErrorTest(UNLEXABLE_ADDR, SendError,
- lambda e: "unrecognized characters" in str(e))
-
- def testUnlexableSource(self):
- # XXX: should have specific exception for this
- self.receiverErrorTest(UNLEXABLE_ADDR, ReceiveError,
- lambda e: "unrecognized characters" in str(e))
-
- def testInvalidMode(self):
- # XXX: should have specific exception for this
- self.receiverErrorTest('name; {mode: "this-is-a-bad-receiver-mode"}',
- ReceiveError,
- lambda e: "not in ('browse', 'consume')" in str(e))
-
-SENDER_Q = 'test-sender-q; {create: always, delete: always}'
-
-class SenderTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def setup_sender(self):
- return self.ssn.sender(SENDER_Q)
-
- def setup_receiver(self):
- return self.ssn.receiver(SENDER_Q)
-
- def checkContent(self, content):
- self.snd.send(content)
- msg = self.rcv.fetch(0)
- assert msg.content == content
-
- out = Message(content)
- self.snd.send(out)
- echo = self.rcv.fetch(0)
- assert out.content == echo.content
- assert echo.content == msg.content
- self.ssn.acknowledge()
-
- def testSendString(self):
- self.checkContent(self.content("testSendString"))
-
- def testSendList(self):
- self.checkContent(["testSendList", 1, 3.14, self.test_id])
-
- def testSendMap(self):
- self.checkContent({"testSendMap": self.test_id, "pie": "blueberry", "pi": 3.14})
-
- def asyncTest(self, capacity):
- self.snd.capacity = capacity
- msgs = [self.content("asyncTest", i) for i in range(15)]
- for m in msgs:
- self.snd.send(m, sync=False)
- drained = self.drain(self.rcv, timeout=self.delay())
- assert msgs == drained, "expected %s, got %s" % (msgs, drained)
- self.ssn.acknowledge()
-
- def testSendAsyncCapacity0(self):
- try:
- self.asyncTest(0)
- assert False, "send shouldn't succeed with zero capacity"
- except InsufficientCapacity:
- # this is expected
- pass
-
- def testSendAsyncCapacity1(self):
- self.asyncTest(1)
-
- def testSendAsyncCapacity5(self):
- self.asyncTest(5)
-
- def testSendAsyncCapacityUNLIMITED(self):
- self.asyncTest(UNLIMITED)
-
- def testCapacityTimeout(self):
- self.snd.capacity = 1
- msgs = []
- caught = False
- while len(msgs) < 100:
- m = self.content("testCapacity", len(msgs))
- try:
- self.snd.send(m, sync=False, timeout=0)
- msgs.append(m)
- except InsufficientCapacity:
- caught = True
- break
- self.snd.sync()
- self.drain(self.rcv, expected=msgs)
- self.ssn.acknowledge()
- assert caught, "did not exceed capacity"
-
-class MessageTests(Base):
-
- def testCreateString(self):
- m = Message("string")
- assert m.content == "string"
- assert m.content_type is None
-
- def testCreateUnicode(self):
- m = Message(u"unicode")
- assert m.content == u"unicode"
- assert m.content_type == "text/plain"
-
- def testCreateMap(self):
- m = Message({})
- assert m.content == {}
- assert m.content_type == "amqp/map"
-
- def testCreateList(self):
- m = Message([])
- assert m.content == []
- assert m.content_type == "amqp/list"
-
- def testContentTypeOverride(self):
- m = Message()
- m.content_type = "text/html; charset=utf8"
- m.content = u""
- assert m.content_type == "text/html; charset=utf8"
-
-ECHO_Q = 'test-message-echo-queue; {create: always, delete: always}'
-
-class MessageEchoTests(Base):
-
- def setup_connection(self):
- return Connection.open(self.broker.host, self.broker.port,
- reconnect=self.reconnect())
-
- def setup_session(self):
- return self.conn.session()
-
- def setup_sender(self):
- return self.ssn.sender(ECHO_Q)
-
- def setup_receiver(self):
- return self.ssn.receiver(ECHO_Q)
-
- def check(self, msg):
- self.snd.send(msg)
- echo = self.rcv.fetch(0)
-
- assert msg.id == echo.id
- assert msg.subject == echo.subject
- assert msg.user_id == echo.user_id
- assert msg.to == echo.to
- assert msg.reply_to == echo.reply_to
- assert msg.correlation_id == echo.correlation_id
- assert msg.properties == echo.properties
- assert msg.content_type == echo.content_type
- assert msg.content == echo.content, "%s, %s" % (msg, echo)
-
- self.ssn.acknowledge(echo)
-
- def testStringContent(self):
- self.check(Message("string"))
-
- def testUnicodeContent(self):
- self.check(Message(u"unicode"))
-
-
- TEST_MAP = {"key1": "string",
- "key2": u"unicode",
- "key3": 3,
- "key4": -3,
- "key5": 3.14,
- "key6": -3.14,
- "key7": ["one", 2, 3.14],
- "key8": [],
- "key9": {"sub-key0": 3}}
-
- def testMapContent(self):
- self.check(Message(MessageEchoTests.TEST_MAP))
-
- def testListContent(self):
- self.check(Message([]))
- self.check(Message([1, 2, 3]))
- self.check(Message(["one", 2, 3.14, {"four": 4}]))
-
- def testProperties(self):
- msg = Message()
- msg.to = "to-address"
- msg.subject = "subject"
- msg.correlation_id = str(self.test_id)
- msg.properties = MessageEchoTests.TEST_MAP
- msg.reply_to = "reply-address"
- self.check(msg)
-
-class TestTestsXXX(Test):
-
- def testFoo(self):
- print "this test has output"
-
- def testBar(self):
- print "this test "*8
- print "has"*10
- print "a"*75
- print "lot of"*10
- print "output"*10
-
- def testQux(self):
- import sys
- sys.stdout.write("this test has output with no newline")
-
- def testQuxFail(self):
- import sys
- sys.stdout.write("this test has output with no newline")
- fdsa
diff --git a/python/qpid/tests/messaging/__init__.py b/python/qpid/tests/messaging/__init__.py
new file mode 100644
index 0000000000..6785614f41
--- /dev/null
+++ b/python/qpid/tests/messaging/__init__.py
@@ -0,0 +1,106 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import time
+from qpid.messaging import *
+from qpid.tests import Test
+
+class Base(Test):
+
+ def setup_connection(self):
+ return None
+
+ def setup_session(self):
+ return None
+
+ def setup_sender(self):
+ return None
+
+ def setup_receiver(self):
+ return None
+
+ def setup(self):
+ self.test_id = uuid4()
+ self.broker = self.config.broker
+ try:
+ self.conn = self.setup_connection()
+ except ConnectError, e:
+ raise Skipped(e)
+ self.ssn = self.setup_session()
+ self.snd = self.setup_sender()
+ if self.snd is not None:
+ self.snd.durable = self.durable()
+ self.rcv = self.setup_receiver()
+
+ def teardown(self):
+ if self.conn is not None and self.conn.connected():
+ self.conn.close()
+
+ def content(self, base, count = None):
+ if count is None:
+ return "%s[%s]" % (base, self.test_id)
+ else:
+ return "%s[%s, %s]" % (base, count, self.test_id)
+
+ def ping(self, ssn):
+ PING_Q = 'ping-queue; {create: always, delete: always}'
+ # send a message
+ sender = ssn.sender(PING_Q, durable=self.durable())
+ content = self.content("ping")
+ sender.send(content)
+ receiver = ssn.receiver(PING_Q)
+ msg = receiver.fetch(0)
+ ssn.acknowledge()
+ assert msg.content == content, "expected %r, got %r" % (content, msg.content)
+
+ def drain(self, rcv, limit=None, timeout=0, expected=None):
+ contents = []
+ try:
+ while limit is None or len(contents) < limit:
+ contents.append(rcv.fetch(timeout=timeout).content)
+ except Empty:
+ pass
+ if expected is not None:
+ assert expected == contents, "expected %s, got %s" % (expected, contents)
+ return contents
+
+ def assertEmpty(self, rcv):
+ contents = self.drain(rcv)
+ assert len(contents) == 0, "%s is supposed to be empty: %s" % (rcv, contents)
+
+ def assertPending(self, rcv, expected):
+ p = rcv.pending()
+ assert p == expected, "expected %s, got %s" % (expected, p)
+
+ def sleep(self):
+ time.sleep(self.delay())
+
+ def delay(self):
+ return float(self.config.defines.get("delay", "2"))
+
+ def get_bool(self, name):
+ return self.config.defines.get(name, "false").lower() in ("true", "yes", "1")
+
+ def durable(self):
+ return self.get_bool("durable")
+
+ def reconnect(self):
+ return self.get_bool("reconnect")
+
+import address, endpoints, message
diff --git a/python/qpid/tests/messaging/address.py b/python/qpid/tests/messaging/address.py
new file mode 100644
index 0000000000..7adbc0c6f7
--- /dev/null
+++ b/python/qpid/tests/messaging/address.py
@@ -0,0 +1,309 @@
+#
+# 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.tests import Test
+from qpid.messaging.address import lex, parse, ParseError, EOF, ID, NUMBER, SYM, WSPACE, \
+ LEXER
+from qpid.lexer import Token
+from qpid.harness import Skipped
+from qpid.tests.parser import ParserBase
+
+def indent(st):
+ return " " + st.replace("\n", "\n ")
+
+def pprint_address(name, subject, options):
+ return "NAME: %s\nSUBJECT: %s\nOPTIONS: %s" % \
+ (pprint(name), pprint(subject), pprint(options))
+
+def pprint(o):
+ if isinstance(o, dict):
+ return pprint_map(o)
+ elif isinstance(o, list):
+ return pprint_list(o)
+ elif isinstance(o, basestring):
+ return pprint_string(o)
+ else:
+ return repr(o)
+
+def pprint_map(m):
+ items = ["%s: %s" % (pprint(k), pprint(v)) for k, v in m.items()]
+ items.sort()
+ return pprint_items("{", items, "}")
+
+def pprint_list(l):
+ return pprint_items("[", [pprint(x) for x in l], "]")
+
+def pprint_items(start, items, end):
+ if items:
+ return "%s\n%s\n%s" % (start, ",\n".join([indent(i) for i in items]), end)
+ else:
+ return "%s%s" % (start, end)
+
+def pprint_string(s):
+ result = "'"
+ for c in s:
+ if c == "'":
+ result += "\\'"
+ elif c == "\n":
+ result += "\\n"
+ elif ord(c) >= 0x80:
+ result += "\\u%04x" % ord(c)
+ else:
+ result += c
+ result += "'"
+ return result
+
+class AddressTests(ParserBase, Test):
+
+ EXCLUDE = (WSPACE, EOF)
+
+ def fields(self, line, n):
+ result = line.split(":", n - 1)
+ result.extend([None]*(n - len(result)))
+ return result
+
+ def call(self, parser, mode, input):
+ try:
+ from subprocess import Popen, PIPE, STDOUT
+ po = Popen([parser, mode], stdin=PIPE, stdout=PIPE, stderr=STDOUT)
+ except ImportError, e:
+ raise Skipped("%s" % e)
+ except OSError, e:
+ raise Skipped("%s: %s" % (e, parser))
+ out, _ = po.communicate(input=input)
+ return out
+
+ def parser(self):
+ return self.config.defines.get("address.parser")
+
+ def do_lex(self, st):
+ parser = self.parser()
+ if parser:
+ out = self.call(parser, "lex", st)
+ lines = out.split("\n")
+ toks = []
+ for line in lines:
+ if line.strip():
+ name, position, value = self.fields(line, 3)
+ toks.append(Token(LEXER.type(name), value, position, st))
+ return toks
+ else:
+ return lex(st)
+
+ def do_parse(self, st):
+ return parse(st)
+
+ def valid(self, addr, name=None, subject=None, options=None):
+ parser = self.parser()
+ if parser:
+ got = self.call(parser, "parse", addr)
+ expected = "%s\n" % pprint_address(name, subject, options)
+ assert expected == got, "expected\n%s\ngot\n%s" % (expected, got)
+ else:
+ ParserBase.valid(self, addr, (name, subject, options))
+
+ def invalid(self, addr, error=None):
+ parser = self.parser()
+ if parser:
+ got = self.call(parser, "parse", addr)
+ expected = "ERROR: %s\n" % error
+ assert expected == got, "expected %r, got %r" % (expected, got)
+ else:
+ ParserBase.invalid(self, addr, error)
+
+ def testDashInId1(self):
+ self.lex("foo-bar", ID)
+
+ def testDashInId2(self):
+ self.lex("foo-3", ID)
+
+ def testDashAlone1(self):
+ self.lex("foo - bar", ID, SYM, ID)
+
+ def testDashAlone2(self):
+ self.lex("foo - 3", ID, SYM, NUMBER)
+
+ def testLeadingDash(self):
+ self.lex("-foo", SYM, ID)
+
+ def testTrailingDash(self):
+ self.lex("foo-", ID, SYM)
+
+ def testNegativeNum(self):
+ self.lex("-3", NUMBER)
+
+ def testHash(self):
+ self.valid("foo/bar.#", "foo", "bar.#")
+
+ def testStar(self):
+ self.valid("foo/bar.*", "foo", "bar.*")
+
+ def testColon(self):
+ self.valid("foo.bar/baz.qux:moo:arf", "foo.bar", "baz.qux:moo:arf")
+
+ def testOptions(self):
+ self.valid("foo.bar/baz.qux:moo:arf; {key: value}",
+ "foo.bar", "baz.qux:moo:arf", {"key": "value"})
+
+ def testOptionsTrailingComma(self):
+ self.valid("name/subject; {key: value,}", "name", "subject",
+ {"key": "value"})
+
+ def testSemiSubject(self):
+ self.valid("foo.bar/'baz.qux;moo:arf'; {key: value}",
+ "foo.bar", "baz.qux;moo:arf", {"key": "value"})
+
+ def testCommaSubject(self):
+ self.valid("foo.bar/baz.qux.{moo,arf}", "foo.bar", "baz.qux.{moo,arf}")
+
+ def testCommaSubjectOptions(self):
+ self.valid("foo.bar/baz.qux.{moo,arf}; {key: value}", "foo.bar",
+ "baz.qux.{moo,arf}", {"key": "value"})
+
+ def testUnbalanced(self):
+ self.valid("foo.bar/baz.qux.{moo,arf; {key: value}", "foo.bar",
+ "baz.qux.{moo,arf", {"key": "value"})
+
+ def testSlashQuote(self):
+ self.valid("foo.bar\\/baz.qux.{moo,arf; {key: value}",
+ "foo.bar/baz.qux.{moo,arf",
+ None, {"key": "value"})
+
+ def testSlashHexEsc1(self):
+ self.valid("foo.bar\\x00baz.qux.{moo,arf; {key: value}",
+ "foo.bar\x00baz.qux.{moo,arf",
+ None, {"key": "value"})
+
+ def testSlashHexEsc2(self):
+ self.valid("foo.bar\\xffbaz.qux.{moo,arf; {key: value}",
+ "foo.bar\xffbaz.qux.{moo,arf",
+ None, {"key": "value"})
+
+ def testSlashHexEsc3(self):
+ self.valid("foo.bar\\xFFbaz.qux.{moo,arf; {key: value}",
+ "foo.bar\xFFbaz.qux.{moo,arf",
+ None, {"key": "value"})
+
+ def testSlashUnicode1(self):
+ self.valid("foo.bar\\u1234baz.qux.{moo,arf; {key: value}",
+ u"foo.bar\u1234baz.qux.{moo,arf", None, {"key": "value"})
+
+ def testSlashUnicode2(self):
+ self.valid("foo.bar\\u0000baz.qux.{moo,arf; {key: value}",
+ u"foo.bar\u0000baz.qux.{moo,arf", None, {"key": "value"})
+
+ def testSlashUnicode3(self):
+ self.valid("foo.bar\\uffffbaz.qux.{moo,arf; {key: value}",
+ u"foo.bar\uffffbaz.qux.{moo,arf", None, {"key": "value"})
+
+ def testSlashUnicode4(self):
+ self.valid("foo.bar\\uFFFFbaz.qux.{moo,arf; {key: value}",
+ u"foo.bar\uFFFFbaz.qux.{moo,arf", None, {"key": "value"})
+
+ def testNoName(self):
+ self.invalid("; {key: value}",
+ "unexpected token SEMI(;) line:1,0:; {key: value}")
+
+ def testEmpty(self):
+ self.invalid("", "unexpected token EOF line:1,0:")
+
+ def testNoNameSlash(self):
+ self.invalid("/asdf; {key: value}",
+ "unexpected token SLASH(/) line:1,0:/asdf; {key: value}")
+
+ def testBadOptions1(self):
+ self.invalid("name/subject; {",
+ "expecting (NUMBER, STRING, ID, LBRACE, LBRACK, RBRACE), "
+ "got EOF line:1,15:name/subject; {")
+
+ def testBadOptions2(self):
+ self.invalid("name/subject; { 3",
+ "expecting COLON, got EOF "
+ "line:1,17:name/subject; { 3")
+
+ def testBadOptions3(self):
+ self.invalid("name/subject; { key:",
+ "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), got EOF "
+ "line:1,20:name/subject; { key:")
+
+ def testBadOptions4(self):
+ self.invalid("name/subject; { key: value",
+ "expecting (COMMA, RBRACE), got EOF "
+ "line:1,26:name/subject; { key: value")
+
+ def testBadOptions5(self):
+ self.invalid("name/subject; { key: value asdf",
+ "expecting (COMMA, RBRACE), got ID(asdf) "
+ "line:1,27:name/subject; { key: value asdf")
+
+ def testBadOptions6(self):
+ self.invalid("name/subject; { key: value,",
+ "expecting (NUMBER, STRING, ID, LBRACE, LBRACK, RBRACE), got EOF "
+ "line:1,27:name/subject; { key: value,")
+
+ def testBadOptions7(self):
+ self.invalid("name/subject; { key: value } asdf",
+ "expecting EOF, got ID(asdf) "
+ "line:1,29:name/subject; { key: value } asdf")
+
+ def testList1(self):
+ self.valid("name/subject; { key: [] }", "name", "subject", {"key": []})
+
+ def testList2(self):
+ self.valid("name/subject; { key: ['one'] }", "name", "subject", {"key": ['one']})
+
+ def testList3(self):
+ self.valid("name/subject; { key: [1, 2, 3] }", "name", "subject",
+ {"key": [1, 2, 3]})
+
+ def testList4(self):
+ self.valid("name/subject; { key: [1, [2, 3], 4] }", "name", "subject",
+ {"key": [1, [2, 3], 4]})
+
+ def testBadList1(self):
+ self.invalid("name/subject; { key: [ }", "expecting (NUMBER, STRING, ID, LBRACE, LBRACK), "
+ "got RBRACE(}) line:1,23:name/subject; { key: [ }")
+
+ def testBadList2(self):
+ self.invalid("name/subject; { key: [ 1 }", "expecting (COMMA, RBRACK), "
+ "got RBRACE(}) line:1,25:name/subject; { key: [ 1 }")
+
+ def testBadList3(self):
+ self.invalid("name/subject; { key: [ 1 2 }", "expecting (COMMA, RBRACK), "
+ "got NUMBER(2) line:1,25:name/subject; { key: [ 1 2 }")
+
+ def testBadList4(self):
+ self.invalid("name/subject; { key: [ 1 2 ] }", "expecting (COMMA, RBRACK), "
+ "got NUMBER(2) line:1,25:name/subject; { key: [ 1 2 ] }")
+
+ def testMap1(self):
+ self.valid("name/subject; { 'key': value }",
+ "name", "subject", {"key": "value"})
+
+ def testMap2(self):
+ self.valid("name/subject; { 1: value }", "name", "subject", {1: "value"})
+
+ def testMap3(self):
+ self.valid('name/subject; { "foo.bar": value }',
+ "name", "subject", {"foo.bar": "value"})
+
+ def testBoolean(self):
+ self.valid("name/subject; { true: True, false: False }",
+ "name", "subject", {"true": True, "false": False})
diff --git a/python/qpid/tests/messaging/endpoints.py b/python/qpid/tests/messaging/endpoints.py
new file mode 100644
index 0000000000..2e70f13f3a
--- /dev/null
+++ b/python/qpid/tests/messaging/endpoints.py
@@ -0,0 +1,878 @@
+#
+# 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.
+#
+
+# setup, usage, teardown, errors(sync), errors(async), stress, soak,
+# boundary-conditions, config
+
+import time
+from qpid import compat
+from qpid.harness import Skipped
+from qpid.messaging import *
+from qpid.tests.messaging import Base
+
+class SetupTests(Base):
+
+ def testOpen(self):
+ # XXX: need to flesh out URL support/syntax
+ self.conn = Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+ self.ping(self.conn.session())
+
+ def testConnect(self):
+ # XXX: need to flesh out URL support/syntax
+ self.conn = Connection(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+ self.conn.connect()
+ self.ping(self.conn.session())
+
+ def testConnectError(self):
+ try:
+ self.conn = Connection.open("localhost", 0)
+ assert False, "connect succeeded"
+ except ConnectError, e:
+ # XXX: should verify that e includes appropriate diagnostic info
+ pass
+
+class ConnectionTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def testSessionAnon(self):
+ ssn1 = self.conn.session()
+ ssn2 = self.conn.session()
+ self.ping(ssn1)
+ self.ping(ssn2)
+ assert ssn1 is not ssn2
+
+ def testSessionNamed(self):
+ ssn1 = self.conn.session("one")
+ ssn2 = self.conn.session("two")
+ self.ping(ssn1)
+ self.ping(ssn2)
+ assert ssn1 is not ssn2
+ assert ssn1 is self.conn.session("one")
+ assert ssn2 is self.conn.session("two")
+
+ def testDisconnect(self):
+ ssn = self.conn.session()
+ self.ping(ssn)
+ self.conn.disconnect()
+ try:
+ self.ping(ssn)
+ assert False, "ping succeeded"
+ except Disconnected:
+ # this is the expected failure when pinging on a disconnected
+ # connection
+ pass
+ self.conn.connect()
+ self.ping(ssn)
+
+ def testClose(self):
+ self.conn.close()
+ assert not self.conn.connected()
+
+ACK_QC = 'test-ack-queue; {create: always}'
+ACK_QD = 'test-ack-queue; {delete: always}'
+
+class SessionTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def testSender(self):
+ snd = self.ssn.sender('test-snd-queue; {create: sender, delete: receiver}',
+ durable=self.durable())
+ snd2 = self.ssn.sender(snd.target, durable=self.durable())
+ assert snd is not snd2
+ snd2.close()
+
+ content = self.content("testSender")
+ snd.send(content)
+ rcv = self.ssn.receiver(snd.target)
+ msg = rcv.fetch(0)
+ assert msg.content == content
+ self.ssn.acknowledge(msg)
+
+ def testReceiver(self):
+ rcv = self.ssn.receiver('test-rcv-queue; {create: always}')
+ rcv2 = self.ssn.receiver(rcv.source)
+ assert rcv is not rcv2
+ rcv2.close()
+
+ content = self.content("testReceiver")
+ snd = self.ssn.sender(rcv.source, durable=self.durable())
+ snd.send(content)
+ msg = rcv.fetch(0)
+ assert msg.content == content
+ self.ssn.acknowledge(msg)
+ snd2 = self.ssn.receiver('test-rcv-queue; {delete: always}')
+
+ def testDisconnectedReceiver(self):
+ self.conn.disconnect()
+ rcv = self.ssn.receiver("test-dis-rcv-queue; {create: always, delete: always}")
+ m = self.content("testDisconnectedReceiver")
+ self.conn.connect()
+ snd = self.ssn.sender("test-dis-rcv-queue")
+ snd.send(m)
+ self.drain(rcv, expected=[m])
+
+ def testNextReceiver(self):
+ ADDR = 'test-next-rcv-queue; {create: always, delete: always}'
+ rcv1 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
+ rcv2 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
+ rcv3 = self.ssn.receiver(ADDR, capacity=UNLIMITED)
+
+ snd = self.ssn.sender(ADDR)
+
+ msgs = []
+ for i in range(10):
+ content = self.content("testNextReceiver", i)
+ snd.send(content)
+ msgs.append(content)
+
+ fetched = []
+ try:
+ while True:
+ rcv = self.ssn.next_receiver(timeout=self.delay())
+ assert rcv in (rcv1, rcv2, rcv3)
+ assert rcv.pending() > 0
+ fetched.append(rcv.fetch().content)
+ except Empty:
+ pass
+ assert msgs == fetched, "expecting %s, got %s" % (msgs, fetched)
+ self.ssn.acknowledge()
+ #we set the capacity to 0 to prevent the deletion of the queue -
+ #triggered the deletion policy when the first receiver is closed -
+ #resulting in session exceptions being issued for the remaining
+ #active subscriptions:
+ for r in [rcv1, rcv2, rcv3]:
+ r.capacity = 0
+
+ # XXX, we need a convenient way to assert that required queues are
+ # empty on setup, and possibly also to drain queues on teardown
+ def ackTest(self, acker, ack_capacity=None):
+ # send a bunch of messages
+ snd = self.ssn.sender(ACK_QC, durable=self.durable())
+ contents = [self.content("ackTest", i) for i in range(15)]
+ for c in contents:
+ snd.send(c)
+
+ # drain the queue, verify the messages are there and then close
+ # without acking
+ rcv = self.ssn.receiver(ACK_QC)
+ self.drain(rcv, expected=contents)
+ self.ssn.close()
+
+ # drain the queue again, verify that they are all the messages
+ # were requeued, and ack this time before closing
+ self.ssn = self.conn.session()
+ if ack_capacity is not None:
+ self.ssn.ack_capacity = ack_capacity
+ rcv = self.ssn.receiver(ACK_QC)
+ self.drain(rcv, expected=contents)
+ acker(self.ssn)
+ self.ssn.close()
+
+ # drain the queue a final time and verify that the messages were
+ # dequeued
+ self.ssn = self.conn.session()
+ rcv = self.ssn.receiver(ACK_QD)
+ self.assertEmpty(rcv)
+
+ def testAcknowledge(self):
+ self.ackTest(lambda ssn: ssn.acknowledge())
+
+ def testAcknowledgeAsync(self):
+ self.ackTest(lambda ssn: ssn.acknowledge(sync=False))
+
+ def testAcknowledgeAsyncAckCap0(self):
+ try:
+ try:
+ self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 0)
+ assert False, "acknowledge shouldn't succeed with ack_capacity of zero"
+ except InsufficientCapacity:
+ pass
+ finally:
+ self.ssn.ack_capacity = UNLIMITED
+ self.drain(self.ssn.receiver(ACK_QD))
+ self.ssn.acknowledge()
+
+ def testAcknowledgeAsyncAckCap1(self):
+ self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 1)
+
+ def testAcknowledgeAsyncAckCap5(self):
+ self.ackTest(lambda ssn: ssn.acknowledge(sync=False), 5)
+
+ def testAcknowledgeAsyncAckCapUNLIMITED(self):
+ self.ackTest(lambda ssn: ssn.acknowledge(sync=False), UNLIMITED)
+
+ def send(self, ssn, queue, base, count=1):
+ snd = ssn.sender(queue, durable=self.durable())
+ contents = []
+ for i in range(count):
+ c = self.content(base, i)
+ snd.send(c)
+ contents.append(c)
+ snd.close()
+ return contents
+
+ def txTest(self, commit):
+ TX_Q = 'test-tx-queue; {create: sender, delete: receiver}'
+ TX_Q_COPY = 'test-tx-queue-copy; {create: always, delete: always}'
+ txssn = self.conn.session(transactional=True)
+ contents = self.send(self.ssn, TX_Q, "txTest", 3)
+ txrcv = txssn.receiver(TX_Q)
+ txsnd = txssn.sender(TX_Q_COPY, durable=self.durable())
+ rcv = self.ssn.receiver(txrcv.source)
+ copy_rcv = self.ssn.receiver(txsnd.target)
+ self.assertEmpty(copy_rcv)
+ for i in range(3):
+ m = txrcv.fetch(0)
+ txsnd.send(m)
+ self.assertEmpty(copy_rcv)
+ txssn.acknowledge()
+ if commit:
+ txssn.commit()
+ self.assertEmpty(rcv)
+ assert contents == self.drain(copy_rcv)
+ else:
+ txssn.rollback()
+ assert contents == self.drain(rcv)
+ self.assertEmpty(copy_rcv)
+ self.ssn.acknowledge()
+
+ def testCommit(self):
+ self.txTest(True)
+
+ def testRollback(self):
+ self.txTest(False)
+
+ def txTestSend(self, commit):
+ TX_SEND_Q = 'test-tx-send-queue; {create: sender, delete: receiver}'
+ txssn = self.conn.session(transactional=True)
+ contents = self.send(txssn, TX_SEND_Q, "txTestSend", 3)
+ rcv = self.ssn.receiver(TX_SEND_Q)
+ self.assertEmpty(rcv)
+
+ if commit:
+ txssn.commit()
+ assert contents == self.drain(rcv)
+ self.ssn.acknowledge()
+ else:
+ txssn.rollback()
+ self.assertEmpty(rcv)
+ txssn.commit()
+ self.assertEmpty(rcv)
+
+ def testCommitSend(self):
+ self.txTestSend(True)
+
+ def testRollbackSend(self):
+ self.txTestSend(False)
+
+ def txTestAck(self, commit):
+ TX_ACK_QC = 'test-tx-ack-queue; {create: always}'
+ TX_ACK_QD = 'test-tx-ack-queue; {delete: always}'
+ txssn = self.conn.session(transactional=True)
+ txrcv = txssn.receiver(TX_ACK_QC)
+ self.assertEmpty(txrcv)
+ contents = self.send(self.ssn, TX_ACK_QC, "txTestAck", 3)
+ assert contents == self.drain(txrcv)
+
+ if commit:
+ txssn.acknowledge()
+ else:
+ txssn.rollback()
+ drained = self.drain(txrcv)
+ assert contents == drained, "expected %s, got %s" % (contents, drained)
+ txssn.acknowledge()
+ txssn.rollback()
+ assert contents == self.drain(txrcv)
+ txssn.commit() # commit without ack
+ self.assertEmpty(txrcv)
+
+ txssn.close()
+
+ txssn = self.conn.session(transactional=True)
+ txrcv = txssn.receiver(TX_ACK_QC)
+ assert contents == self.drain(txrcv)
+ txssn.acknowledge()
+ txssn.commit()
+ rcv = self.ssn.receiver(TX_ACK_QD)
+ self.assertEmpty(rcv)
+ txssn.close()
+ self.assertEmpty(rcv)
+
+ def testCommitAck(self):
+ self.txTestAck(True)
+
+ def testRollbackAck(self):
+ self.txTestAck(False)
+
+ def testClose(self):
+ self.ssn.close()
+ try:
+ self.ping(self.ssn)
+ assert False, "ping succeeded"
+ except Disconnected:
+ pass
+
+RECEIVER_Q = 'test-receiver-queue; {create: always, delete: always}'
+
+class ReceiverTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def setup_sender(self):
+ return self.ssn.sender(RECEIVER_Q)
+
+ def setup_receiver(self):
+ return self.ssn.receiver(RECEIVER_Q)
+
+ def send(self, base, count = None):
+ content = self.content(base, count)
+ self.snd.send(content)
+ return content
+
+ def testFetch(self):
+ try:
+ msg = self.rcv.fetch(0)
+ assert False, "unexpected message: %s" % msg
+ except Empty:
+ pass
+ try:
+ start = time.time()
+ msg = self.rcv.fetch(self.delay())
+ assert False, "unexpected message: %s" % msg
+ except Empty:
+ elapsed = time.time() - start
+ assert elapsed >= self.delay()
+
+ one = self.send("testFetch", 1)
+ two = self.send("testFetch", 2)
+ three = self.send("testFetch", 3)
+ msg = self.rcv.fetch(0)
+ assert msg.content == one
+ msg = self.rcv.fetch(self.delay())
+ assert msg.content == two
+ msg = self.rcv.fetch()
+ assert msg.content == three
+ self.ssn.acknowledge()
+
+ def testCapacityIncrease(self):
+ content = self.send("testCapacityIncrease")
+ self.sleep()
+ assert self.rcv.pending() == 0
+ self.rcv.capacity = UNLIMITED
+ self.sleep()
+ assert self.rcv.pending() == 1
+ msg = self.rcv.fetch(0)
+ assert msg.content == content
+ assert self.rcv.pending() == 0
+ self.ssn.acknowledge()
+
+ def testCapacityDecrease(self):
+ self.rcv.capacity = UNLIMITED
+ one = self.send("testCapacityDecrease", 1)
+ self.sleep()
+ assert self.rcv.pending() == 1
+ msg = self.rcv.fetch(0)
+ assert msg.content == one
+
+ self.rcv.capacity = 0
+
+ two = self.send("testCapacityDecrease", 2)
+ self.sleep()
+ assert self.rcv.pending() == 0
+ msg = self.rcv.fetch(0)
+ assert msg.content == two
+
+ self.ssn.acknowledge()
+
+ def testCapacity(self):
+ self.rcv.capacity = 5
+ self.assertPending(self.rcv, 0)
+
+ for i in range(15):
+ self.send("testCapacity", i)
+ self.sleep()
+ self.assertPending(self.rcv, 5)
+
+ self.drain(self.rcv, limit = 5)
+ self.sleep()
+ self.assertPending(self.rcv, 5)
+
+ drained = self.drain(self.rcv)
+ assert len(drained) == 10, "%s, %s" % (len(drained), drained)
+ self.assertPending(self.rcv, 0)
+
+ self.ssn.acknowledge()
+
+ def testCapacityUNLIMITED(self):
+ self.rcv.capacity = UNLIMITED
+ self.assertPending(self.rcv, 0)
+
+ for i in range(10):
+ self.send("testCapacityUNLIMITED", i)
+ self.sleep()
+ self.assertPending(self.rcv, 10)
+
+ self.drain(self.rcv)
+ self.assertPending(self.rcv, 0)
+
+ self.ssn.acknowledge()
+
+ def testPending(self):
+ self.rcv.capacity = UNLIMITED
+ assert self.rcv.pending() == 0
+
+ for i in range(3):
+ self.send("testPending", i)
+ self.sleep()
+ assert self.rcv.pending() == 3
+
+ for i in range(3, 10):
+ self.send("testPending", i)
+ self.sleep()
+ assert self.rcv.pending() == 10
+
+ self.drain(self.rcv, limit=3)
+ assert self.rcv.pending() == 7
+
+ self.drain(self.rcv)
+ assert self.rcv.pending() == 0
+
+ self.ssn.acknowledge()
+
+ def testDoubleClose(self):
+ m1 = self.content("testDoubleClose", 1)
+ m2 = self.content("testDoubleClose", 2)
+
+ snd = self.ssn.sender("""test-double-close; {
+ create: always,
+ delete: sender,
+ node-properties: {
+ type: topic
+ }
+}
+""")
+ r1 = self.ssn.receiver(snd.target)
+ r2 = self.ssn.receiver(snd.target)
+ snd.send(m1)
+ self.drain(r1, expected=[m1])
+ self.drain(r2, expected=[m1])
+ r1.close()
+ snd.send(m2)
+ self.drain(r2, expected=[m2])
+ r2.close()
+
+ # XXX: need testClose
+
+ def testMode(self):
+ msgs = [self.content("testMode", 1),
+ self.content("testMode", 2),
+ self.content("testMode", 3)]
+
+ for m in msgs:
+ self.snd.send(m)
+
+ rb = self.ssn.receiver('test-receiver-queue; {mode: browse}')
+ rc = self.ssn.receiver('test-receiver-queue; {mode: consume}')
+ self.drain(rb, expected=msgs)
+ self.drain(rc, expected=msgs)
+ rb2 = self.ssn.receiver(rb.source)
+ self.assertEmpty(rb2)
+ self.drain(self.rcv, expected=[])
+
+class AddressTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def badOption(self, options, error):
+ try:
+ self.ssn.sender("test-bad-options-snd; %s" % options)
+ assert False
+ except SendError, e:
+ assert "error in options: %s" % error == str(e), e
+
+ try:
+ self.ssn.receiver("test-bad-options-rcv; %s" % options)
+ assert False
+ except ReceiveError, e:
+ assert "error in options: %s" % error == str(e), e
+
+ def testIllegalKey(self):
+ self.badOption("{create: always, node-properties: "
+ "{this-property-does-not-exist: 3}}",
+ "node-properties: this-property-does-not-exist: "
+ "illegal key")
+
+ def testWrongValue(self):
+ self.badOption("{create: asdf}", "create: asdf not in "
+ "('always', 'sender', 'receiver', 'never')")
+
+ def testWrongType1(self):
+ self.badOption("{node-properties: asdf}",
+ "node-properties: asdf is not a map")
+
+ def testWrongType2(self):
+ self.badOption("{node-properties: {durable: []}}",
+ "node-properties: durable: [] is not a bool")
+
+ def testNonQueueBindings(self):
+ self.badOption("{node-properties: {type: topic, x-properties: "
+ "{bindings: []}}}",
+ "node-properties: x-properties: bindings: "
+ "bindings are only permitted on nodes of type queue")
+
+ def testCreateQueue(self):
+ snd = self.ssn.sender("test-create-queue; {create: always, delete: always, "
+ "node-properties: {type: queue, durable: False, "
+ "x-properties: {auto_delete: true}}}")
+ content = self.content("testCreateQueue")
+ snd.send(content)
+ rcv = self.ssn.receiver("test-create-queue")
+ self.drain(rcv, expected=[content])
+
+ def createExchangeTest(self, props=""):
+ addr = """test-create-exchange; {
+ create: always,
+ delete: always,
+ node-properties: {
+ type: topic,
+ durable: False,
+ x-properties: {auto_delete: true, %s}
+ }
+ }""" % props
+ snd = self.ssn.sender(addr)
+ snd.send("ping")
+ rcv1 = self.ssn.receiver("test-create-exchange/first")
+ rcv2 = self.ssn.receiver("test-create-exchange/first")
+ rcv3 = self.ssn.receiver("test-create-exchange/second")
+ for r in (rcv1, rcv2, rcv3):
+ try:
+ r.fetch(0)
+ assert False
+ except Empty:
+ pass
+ msg1 = Message(self.content("testCreateExchange", 1), subject="first")
+ msg2 = Message(self.content("testCreateExchange", 2), subject="second")
+ snd.send(msg1)
+ snd.send(msg2)
+ self.drain(rcv1, expected=[msg1.content])
+ self.drain(rcv2, expected=[msg1.content])
+ self.drain(rcv3, expected=[msg2.content])
+
+ def testCreateExchange(self):
+ self.createExchangeTest()
+
+ def testCreateExchangeDirect(self):
+ self.createExchangeTest("type: direct")
+
+ def testCreateExchangeTopic(self):
+ self.createExchangeTest("type: topic")
+
+ def testDeleteBySender(self):
+ snd = self.ssn.sender("test-delete; {create: always}")
+ snd.send("ping")
+ snd.close()
+ snd = self.ssn.sender("test-delete; {delete: always}")
+ snd.send("ping")
+ snd.close()
+ try:
+ self.ssn.sender("test-delete")
+ except SendError, e:
+ assert "no such queue" in str(e)
+
+ def testDeleteByReceiver(self):
+ rcv = self.ssn.receiver("test-delete; {create: always, delete: always}")
+ try:
+ rcv.fetch(0)
+ except Empty:
+ pass
+ rcv.close()
+
+ try:
+ self.ssn.receiver("test-delete")
+ assert False
+ except ReceiveError, e:
+ assert "no such queue" in str(e)
+
+ def testDeleteSpecial(self):
+ snd = self.ssn.sender("amq.topic; {delete: always}")
+ snd.send("asdf")
+ try:
+ snd.close()
+ except SessionError, e:
+ assert "Cannot delete default exchange" in str(e)
+ # XXX: need to figure out close after error
+ self.conn._remove_session(self.ssn)
+
+ def testBindings(self):
+ snd = self.ssn.sender("""
+test-bindings-queue; {
+ create: always,
+ delete: always,
+ node-properties: {
+ x-properties: {
+ bindings: ["amq.topic/a.#", "amq.direct/b", "amq.topic/c.*"]
+ }
+ }
+}
+""")
+ snd.send("one")
+ snd_a = self.ssn.sender("amq.topic/a.foo")
+ snd_b = self.ssn.sender("amq.direct/b")
+ snd_c = self.ssn.sender("amq.topic/c.bar")
+ snd_a.send("two")
+ snd_b.send("three")
+ snd_c.send("four")
+ rcv = self.ssn.receiver("test-bindings-queue")
+ self.drain(rcv, expected=["one", "two", "three", "four"])
+
+ def testBindingsAdditive(self):
+ m1 = self.content("testBindingsAdditive", 1)
+ m2 = self.content("testBindingsAdditive", 2)
+ m3 = self.content("testBindingsAdditive", 3)
+ m4 = self.content("testBindingsAdditive", 4)
+
+ snd = self.ssn.sender("""
+test-bindings-additive-queue; {
+ create: always,
+ delete: always,
+ node-properties: {
+ x-properties: {
+ bindings: ["amq.topic/a"]
+ }
+ }
+}
+""")
+
+ snd_a = self.ssn.sender("amq.topic/a")
+ snd_b = self.ssn.sender("amq.topic/b")
+
+ snd_a.send(m1)
+ snd_b.send(m2)
+
+ rcv = self.ssn.receiver("test-bindings-additive-queue")
+ self.drain(rcv, expected=[m1])
+
+ new_snd = self.ssn.sender("""
+test-bindings-additive-queue; {
+ node-properties: {
+ x-properties: {
+ bindings: ["amq.topic/b"]
+ }
+ }
+}
+""")
+
+ new_snd.send(m3)
+ snd_b.send(m4)
+ self.drain(rcv, expected=[m3, m4])
+
+ def testSubjectOverride(self):
+ snd = self.ssn.sender("amq.topic/a")
+ rcv_a = self.ssn.receiver("amq.topic/a")
+ rcv_b = self.ssn.receiver("amq.topic/b")
+ m1 = self.content("testSubjectOverride", 1)
+ m2 = self.content("testSubjectOverride", 2)
+ snd.send(m1)
+ snd.send(Message(subject="b", content=m2))
+ self.drain(rcv_a, expected=[m1])
+ self.drain(rcv_b, expected=[m2])
+
+ def testSubjectDefault(self):
+ m1 = self.content("testSubjectDefault", 1)
+ m2 = self.content("testSubjectDefault", 2)
+ snd = self.ssn.sender("amq.topic/a")
+ rcv = self.ssn.receiver("amq.topic")
+ snd.send(m1)
+ snd.send(Message(subject="b", content=m2))
+ e1 = rcv.fetch(timeout=0)
+ e2 = rcv.fetch(timeout=0)
+ assert e1.subject == "a", "subject: %s" % e1.subject
+ assert e2.subject == "b", "subject: %s" % e2.subject
+ self.assertEmpty(rcv)
+
+NOSUCH_Q = "this-queue-should-not-exist"
+UNPARSEABLE_ADDR = "name/subject; {bad options"
+UNLEXABLE_ADDR = "\0x0\0x1\0x2\0x3"
+
+class AddressErrorTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def senderErrorTest(self, addr, exc, check=lambda e: True):
+ try:
+ self.ssn.sender(addr, durable=self.durable())
+ assert False, "sender creation succeeded"
+ except exc, e:
+ assert check(e), "unexpected error: %s" % compat.format_exc(e)
+
+ def receiverErrorTest(self, addr, exc, check=lambda e: True):
+ try:
+ self.ssn.receiver(addr)
+ assert False, "receiver creation succeeded"
+ except exc, e:
+ assert check(e), "unexpected error: %s" % compat.format_exc(e)
+
+ def testNoneTarget(self):
+ # XXX: should have specific exception for this
+ self.senderErrorTest(None, SendError)
+
+ def testNoneSource(self):
+ # XXX: should have specific exception for this
+ self.receiverErrorTest(None, ReceiveError)
+
+ def testNoTarget(self):
+ # XXX: should have specific exception for this
+ self.senderErrorTest(NOSUCH_Q, SendError, lambda e: NOSUCH_Q in str(e))
+
+ def testNoSource(self):
+ # XXX: should have specific exception for this
+ self.receiverErrorTest(NOSUCH_Q, ReceiveError, lambda e: NOSUCH_Q in str(e))
+
+ def testUnparseableTarget(self):
+ # XXX: should have specific exception for this
+ self.senderErrorTest(UNPARSEABLE_ADDR, SendError,
+ lambda e: "expecting COLON" in str(e))
+
+ def testUnparseableSource(self):
+ # XXX: should have specific exception for this
+ self.receiverErrorTest(UNPARSEABLE_ADDR, ReceiveError,
+ lambda e: "expecting COLON" in str(e))
+
+ def testUnlexableTarget(self):
+ # XXX: should have specific exception for this
+ self.senderErrorTest(UNLEXABLE_ADDR, SendError,
+ lambda e: "unrecognized characters" in str(e))
+
+ def testUnlexableSource(self):
+ # XXX: should have specific exception for this
+ self.receiverErrorTest(UNLEXABLE_ADDR, ReceiveError,
+ lambda e: "unrecognized characters" in str(e))
+
+ def testInvalidMode(self):
+ # XXX: should have specific exception for this
+ self.receiverErrorTest('name; {mode: "this-is-a-bad-receiver-mode"}',
+ ReceiveError,
+ lambda e: "not in ('browse', 'consume')" in str(e))
+
+SENDER_Q = 'test-sender-q; {create: always, delete: always}'
+
+class SenderTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def setup_sender(self):
+ return self.ssn.sender(SENDER_Q)
+
+ def setup_receiver(self):
+ return self.ssn.receiver(SENDER_Q)
+
+ def checkContent(self, content):
+ self.snd.send(content)
+ msg = self.rcv.fetch(0)
+ assert msg.content == content
+
+ out = Message(content)
+ self.snd.send(out)
+ echo = self.rcv.fetch(0)
+ assert out.content == echo.content
+ assert echo.content == msg.content
+ self.ssn.acknowledge()
+
+ def testSendString(self):
+ self.checkContent(self.content("testSendString"))
+
+ def testSendList(self):
+ self.checkContent(["testSendList", 1, 3.14, self.test_id])
+
+ def testSendMap(self):
+ self.checkContent({"testSendMap": self.test_id, "pie": "blueberry", "pi": 3.14})
+
+ def asyncTest(self, capacity):
+ self.snd.capacity = capacity
+ msgs = [self.content("asyncTest", i) for i in range(15)]
+ for m in msgs:
+ self.snd.send(m, sync=False)
+ drained = self.drain(self.rcv, timeout=self.delay())
+ assert msgs == drained, "expected %s, got %s" % (msgs, drained)
+ self.ssn.acknowledge()
+
+ def testSendAsyncCapacity0(self):
+ try:
+ self.asyncTest(0)
+ assert False, "send shouldn't succeed with zero capacity"
+ except InsufficientCapacity:
+ # this is expected
+ pass
+
+ def testSendAsyncCapacity1(self):
+ self.asyncTest(1)
+
+ def testSendAsyncCapacity5(self):
+ self.asyncTest(5)
+
+ def testSendAsyncCapacityUNLIMITED(self):
+ self.asyncTest(UNLIMITED)
+
+ def testCapacityTimeout(self):
+ self.snd.capacity = 1
+ msgs = []
+ caught = False
+ while len(msgs) < 100:
+ m = self.content("testCapacity", len(msgs))
+ try:
+ self.snd.send(m, sync=False, timeout=0)
+ msgs.append(m)
+ except InsufficientCapacity:
+ caught = True
+ break
+ self.snd.sync()
+ self.drain(self.rcv, expected=msgs)
+ self.ssn.acknowledge()
+ assert caught, "did not exceed capacity"
diff --git a/python/qpid/tests/messaging/message.py b/python/qpid/tests/messaging/message.py
new file mode 100644
index 0000000000..ef2ec1aac4
--- /dev/null
+++ b/python/qpid/tests/messaging/message.py
@@ -0,0 +1,116 @@
+#
+# 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.messaging import *
+from qpid.tests.messaging import Base
+
+class MessageTests(Base):
+
+ def testCreateString(self):
+ m = Message("string")
+ assert m.content == "string"
+ assert m.content_type is None
+
+ def testCreateUnicode(self):
+ m = Message(u"unicode")
+ assert m.content == u"unicode"
+ assert m.content_type == "text/plain"
+
+ def testCreateMap(self):
+ m = Message({})
+ assert m.content == {}
+ assert m.content_type == "amqp/map"
+
+ def testCreateList(self):
+ m = Message([])
+ assert m.content == []
+ assert m.content_type == "amqp/list"
+
+ def testContentTypeOverride(self):
+ m = Message()
+ m.content_type = "text/html; charset=utf8"
+ m.content = u""
+ assert m.content_type == "text/html; charset=utf8"
+
+ECHO_Q = 'test-message-echo-queue; {create: always, delete: always}'
+
+class MessageEchoTests(Base):
+
+ def setup_connection(self):
+ return Connection.open(self.broker.host, self.broker.port,
+ reconnect=self.reconnect())
+
+ def setup_session(self):
+ return self.conn.session()
+
+ def setup_sender(self):
+ return self.ssn.sender(ECHO_Q)
+
+ def setup_receiver(self):
+ return self.ssn.receiver(ECHO_Q)
+
+ def check(self, msg):
+ self.snd.send(msg)
+ echo = self.rcv.fetch(0)
+
+ assert msg.id == echo.id
+ assert msg.subject == echo.subject
+ assert msg.user_id == echo.user_id
+ assert msg.to == echo.to
+ assert msg.reply_to == echo.reply_to
+ assert msg.correlation_id == echo.correlation_id
+ assert msg.properties == echo.properties
+ assert msg.content_type == echo.content_type
+ assert msg.content == echo.content, "%s, %s" % (msg, echo)
+
+ self.ssn.acknowledge(echo)
+
+ def testStringContent(self):
+ self.check(Message("string"))
+
+ def testUnicodeContent(self):
+ self.check(Message(u"unicode"))
+
+
+ TEST_MAP = {"key1": "string",
+ "key2": u"unicode",
+ "key3": 3,
+ "key4": -3,
+ "key5": 3.14,
+ "key6": -3.14,
+ "key7": ["one", 2, 3.14],
+ "key8": [],
+ "key9": {"sub-key0": 3}}
+
+ def testMapContent(self):
+ self.check(Message(MessageEchoTests.TEST_MAP))
+
+ def testListContent(self):
+ self.check(Message([]))
+ self.check(Message([1, 2, 3]))
+ self.check(Message(["one", 2, 3.14, {"four": 4}]))
+
+ def testProperties(self):
+ msg = Message()
+ msg.to = "to-address"
+ msg.subject = "subject"
+ msg.correlation_id = str(self.test_id)
+ msg.properties = MessageEchoTests.TEST_MAP
+ msg.reply_to = "reply-address"
+ self.check(msg)
--
cgit v1.2.1