diff options
Diffstat (limited to 'python/qpid/tests/messaging')
| -rw-r--r-- | python/qpid/tests/messaging/__init__.py | 185 | ||||
| -rw-r--r-- | python/qpid/tests/messaging/address.py | 321 | ||||
| -rw-r--r-- | python/qpid/tests/messaging/endpoints.py | 1335 | ||||
| -rw-r--r-- | python/qpid/tests/messaging/message.py | 155 |
4 files changed, 0 insertions, 1996 deletions
diff --git a/python/qpid/tests/messaging/__init__.py b/python/qpid/tests/messaging/__init__.py deleted file mode 100644 index 8f6680d5e3..0000000000 --- a/python/qpid/tests/messaging/__init__.py +++ /dev/null @@ -1,185 +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. -# - -import time -from math import ceil -from qpid.harness import Skipped -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.attached(): - self.teardown_connection(self.conn) - self.conn = None - - def teardown_connection(self, conn): - conn.close(timeout=self.timeout()) - - 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 message(self, base, count = None, **kwargs): - return Message(content=self.content(base, count), **kwargs) - - 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, redelivered=False): - messages = [] - try: - while limit is None or len(messages) < limit: - messages.append(rcv.fetch(timeout=timeout)) - except Empty: - pass - if expected is not None: - self.assertEchos(expected, messages, redelivered) - return messages - - def diff(self, m1, m2, excluded_properties=()): - result = {} - for attr in ("id", "subject", "user_id", "reply_to", - "correlation_id", "durable", "priority", "ttl", - "redelivered", "content_type", "content"): - a1 = getattr(m1, attr) - a2 = getattr(m2, attr) - if a1 != a2: - result[attr] = (a1, a2) - p1 = dict(m1.properties) - p2 = dict(m2.properties) - for ep in excluded_properties: - p1.pop(ep, None) - p2.pop(ep, None) - if p1 != p2: - result["properties"] = (p1, p2) - return result - - def assertEcho(self, msg, echo, redelivered=False): - if not isinstance(msg, Message) or not isinstance(echo, Message): - if isinstance(msg, Message): - msg = msg.content - if isinstance(echo, Message): - echo = echo.content - assert msg == echo, "expected %s, got %s" % (msg, echo) - else: - delta = self.diff(msg, echo, ("x-amqp-0-10.routing-key",)) - mttl, ettl = delta.pop("ttl", (0, 0)) - if redelivered: - assert echo.redelivered, \ - "expected %s to be redelivered: %s" % (msg, echo) - if delta.has_key("redelivered"): - del delta["redelivered"] - assert mttl is not None and ettl is not None, "%s, %s" % (mttl, ettl) - assert mttl >= ettl, "%s, %s" % (mttl, ettl) - assert not delta, "expected %s, got %s, delta %s" % (msg, echo, delta) - - def assertEchos(self, msgs, echoes, redelivered=False): - assert len(msgs) == len(echoes), "%s, %s" % (msgs, echoes) - for m, e in zip(msgs, echoes): - self.assertEcho(m, e, redelivered) - - def assertEmpty(self, rcv): - contents = self.drain(rcv) - assert len(contents) == 0, "%s is supposed to be empty: %s" % (rcv, contents) - - def assertAvailable(self, rcv, expected=None, lower=None, upper=None): - if expected is not None: - if lower is not None or upper is not None: - raise ValueError("cannot specify lower or upper when specifying expected") - lower = expected - upper = expected - else: - if lower is None: - lower = int(ceil(rcv.threshold*rcv.capacity)) - if upper is None: - upper = rcv.capacity - - p = rcv.available() - if upper == lower: - assert p == lower, "expected %s, got %s" % (lower, p) - else: - assert lower <= p <= upper, "expected %s to be in range [%s, %s]" % (p, lower, upper) - - def sleep(self): - time.sleep(self.delay()) - - def delay(self): - return float(self.config.defines.get("delay", "2")) - - def timeout(self): - return float(self.config.defines.get("timeout", "60")) - - 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") - - - def transport(self): - if self.broker.scheme == self.broker.AMQPS: - return "ssl" - else: - return "tcp" - - def connection_options(self): - return {"reconnect": self.reconnect(), - "transport": self.transport()} - -import address, endpoints, message diff --git a/python/qpid/tests/messaging/address.py b/python/qpid/tests/messaging/address.py deleted file mode 100644 index aa9562a717..0000000000 --- a/python/qpid/tests/messaging/address.py +++ /dev/null @@ -1,321 +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.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<EXP>%s</EXP>\ngot\n<GOT>%s</GOT>" % (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 testIdNum(self): - self.lex("id1", ID) - - def testIdSpaceNum(self): - self.lex("id 1", ID, 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 testOptionsNone(self): - self.valid("name/subject; {key: None}", "name", "subject", - {"key": None}) - - 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; { true1: True, true2: true, " - "false1: False, false2: false }", - "name", "subject", {"true1": True, "true2": True, - "false1": False, "false2": False}) diff --git a/python/qpid/tests/messaging/endpoints.py b/python/qpid/tests/messaging/endpoints.py deleted file mode 100644 index db5ec03df2..0000000000 --- a/python/qpid/tests/messaging/endpoints.py +++ /dev/null @@ -1,1335 +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 errno, os, socket, sys, time -from qpid import compat -from qpid.compat import set -from qpid.messaging import * -from qpid.messaging.transports import TRANSPORTS -from qpid.tests.messaging import Base -from threading import Thread - -class SetupTests(Base): - - def testEstablish(self): - self.conn = Connection.establish(self.broker, **self.connection_options()) - self.ping(self.conn.session()) - - def testOpen(self): - self.conn = Connection(self.broker, **self.connection_options()) - self.conn.open() - self.ping(self.conn.session()) - - def testOpenReconnectURLs(self): - options = self.connection_options() - options["reconnect_urls"] = [self.broker, self.broker] - self.conn = Connection(self.broker, **options) - self.conn.open() - self.ping(self.conn.session()) - - def testTcpNodelay(self): - self.conn = Connection.establish(self.broker, tcp_nodelay=True) - assert self.conn._driver._transport.socket.getsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY) - - def testConnectError(self): - try: - # Specifying port 0 yields a bad address on Windows; port 4 is unassigned - self.conn = Connection.establish("localhost:4") - assert False, "connect succeeded" - except ConnectError, e: - assert "refused" in str(e) - - def testGetError(self): - self.conn = Connection("localhost:0") - try: - self.conn.open() - assert False, "connect succeeded" - except ConnectError, e: - assert self.conn.get_error() == e - - def use_fds(self): - fds = [] - try: - while True: - fds.append(os.open(getattr(os, "devnull", "/dev/null"), os.O_RDONLY)) - except OSError, e: - if e.errno != errno.EMFILE: - raise e - else: - return fds - - def testOpenCloseResourceLeaks(self): - fds = self.use_fds() - try: - for i in range(32): - if fds: os.close(fds.pop()) - for i in xrange(64): - conn = Connection.establish(self.broker, **self.connection_options()) - conn.close() - finally: - while fds: - os.close(fds.pop()) - - def testOpenFailResourceLeaks(self): - fds = self.use_fds() - try: - for i in range(32): - if fds: os.close(fds.pop()) - for i in xrange(64): - conn = Connection("localhost:0", **self.connection_options()) - # XXX: we need to force a waiter to be created for this test - # to work - conn._lock.acquire() - conn._wait(lambda: False, timeout=0.001) - conn._lock.release() - try: - conn.open() - except ConnectError, e: - pass - finally: - while fds: - os.close(fds.pop()) - - def testReconnect(self): - options = self.connection_options() - real = TRANSPORTS["tcp"] - - class flaky: - - def __init__(self, conn, host, port): - self.real = real(conn, host, port) - self.sent_count = 0 - self.recv_count = 0 - - def fileno(self): - return self.real.fileno() - - def reading(self, reading): - return self.real.reading(reading) - - def writing(self, writing): - return self.real.writing(writing) - - def send(self, bytes): - if self.sent_count > 2048: - raise socket.error("fake error") - n = self.real.send(bytes) - self.sent_count += n - return n - - def recv(self, n): - if self.recv_count > 2048: - return "" - bytes = self.real.recv(n) - self.recv_count += len(bytes) - return bytes - - def close(self): - self.real.close() - - TRANSPORTS["flaky"] = flaky - - options["reconnect"] = True - options["reconnect_interval"] = 0 - options["reconnect_limit"] = 100 - options["reconnect_log"] = False - options["transport"] = "flaky" - - self.conn = Connection.establish(self.broker, **options) - ssn = self.conn.session() - snd = ssn.sender("test-reconnect-queue; {create: always, delete: always}") - rcv = ssn.receiver(snd.target) - - msgs = [self.message("testReconnect", i) for i in range(20)] - for m in msgs: - snd.send(m) - - content = set() - drained = [] - duplicates = [] - try: - while True: - m = rcv.fetch(timeout=0) - if m.content not in content: - content.add(m.content) - drained.append(m) - else: - duplicates.append(m) - ssn.acknowledge(m) - except Empty: - pass - # XXX: apparently we don't always get duplicates, should figure out why - #assert duplicates, "no duplicates" - assert len(drained) == len(msgs) - for m, d in zip(msgs, drained): - # XXX: we should figure out how to provide proper end to end - # redelivered - self.assertEcho(m, d, d.redelivered) - -class ConnectionTests(Base): - - def setup_connection(self): - return Connection.establish(self.broker, **self.connection_options()) - - def testCheckClosed(self): - assert not self.conn.check_closed() - - 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 testDetach(self): - ssn = self.conn.session() - self.ping(ssn) - self.conn.detach() - try: - self.ping(ssn) - assert False, "ping succeeded" - except Detached: - # this is the expected failure when pinging on a detached - # connection - pass - self.conn.attach() - self.ping(ssn) - - def testClose(self): - self.conn.close() - assert not self.conn.attached() - - def testSimultaneousClose(self): - ssns = [self.conn.session() for i in range(3)] - for s in ssns: - for i in range(3): - s.receiver("amq.topic") - s.sender("amq.topic") - - def closer(errors): - try: - self.conn.close() - except: - _, e, _ = sys.exc_info() - errors.append(compat.format_exc(e)) - - t1_errors = [] - t2_errors = [] - t1 = Thread(target=lambda: closer(t1_errors)) - t2 = Thread(target=lambda: closer(t2_errors)) - t1.start() - t2.start() - t1.join(self.delay()) - t2.join(self.delay()) - - assert not t1_errors, t1_errors[0] - assert not t2_errors, t2_errors[0] - -class hangable: - - def __init__(self, conn, host, port): - self.tcp = TRANSPORTS["tcp"](conn, host, port) - self.hung = False - - def hang(self): - self.hung = True - - def fileno(self): - return self.tcp.fileno() - - def reading(self, reading): - if self.hung: - return True - else: - return self.tcp.reading(reading) - - def writing(self, writing): - if self.hung: - return False - else: - return self.tcp.writing(writing) - - def send(self, bytes): - if self.hung: - return 0 - else: - return self.tcp.send(bytes) - - def recv(self, n): - if self.hung: - return "" - else: - return self.tcp.recv(n) - - def close(self): - self.tcp.close() - -TRANSPORTS["hangable"] = hangable - -class TimeoutTests(Base): - - def setup_connection(self): - options = self.connection_options() - options["transport"] = "hangable" - return Connection.establish(self.broker, **options) - - def setup_session(self): - return self.conn.session() - - def setup_sender(self): - return self.ssn.sender("amq.topic") - - def setup_receiver(self): - return self.ssn.receiver("amq.topic; {link: {reliability: unreliable}}") - - def teardown_connection(self, conn): - try: - conn.detach(timeout=0) - except Timeout: - pass - - def hang(self): - self.conn._driver._transport.hang() - - def timeoutTest(self, method): - self.hang() - try: - method(timeout=self.delay()) - assert False, "did not time out" - except Timeout: - pass - - def testSenderSync(self): - self.snd.send(self.content("testSenderSync"), sync=False) - self.timeoutTest(self.snd.sync) - - def testSenderClose(self): - self.snd.send(self.content("testSenderClose"), sync=False) - self.timeoutTest(self.snd.close) - - def testReceiverClose(self): - self.timeoutTest(self.rcv.close) - - def testSessionSync(self): - self.snd.send(self.content("testSessionSync"), sync=False) - self.timeoutTest(self.ssn.sync) - - def testSessionClose(self): - self.timeoutTest(self.ssn.close) - - def testConnectionDetach(self): - self.timeoutTest(self.conn.detach) - - def testConnectionClose(self): - self.timeoutTest(self.conn.close) - -ACK_QC = 'test-ack-queue; {create: always}' -ACK_QD = 'test-ack-queue; {delete: always}' - -class SessionTests(Base): - - def setup_connection(self): - return Connection.establish(self.broker, **self.connection_options()) - - 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 testDetachedReceiver(self): - self.conn.detach() - rcv = self.ssn.receiver("test-dis-rcv-queue; {create: always, delete: always}") - m = self.content("testDetachedReceiver") - self.conn.attach() - 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.available() > 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 testRelease(self): - msgs = [self.message("testRelease", i) for i in range(3)] - snd = self.ssn.sender("test-release-queue; {create: always, delete: always}") - for m in msgs: - snd.send(m) - rcv = self.ssn.receiver(snd.target) - echos = self.drain(rcv, expected=msgs) - self.ssn.acknowledge(echos[0]) - self.ssn.acknowledge(echos[1], Disposition(RELEASED, set_redelivered=True)) - self.ssn.acknowledge(echos[2], Disposition(RELEASED)) - self.drain(rcv, limit=1, expected=msgs[1:2], redelivered=True) - self.drain(rcv, expected=msgs[2:3]) - self.ssn.acknowledge() - - def testReject(self): - msgs = [self.message("testReject", i) for i in range(3)] - snd = self.ssn.sender(""" - test-reject-queue; { - create: always, - delete: always, - node: { - x-declare: { - alternate-exchange: 'amq.topic' - } - } - } -""") - for m in msgs: - snd.send(m) - rcv = self.ssn.receiver(snd.target) - rej = self.ssn.receiver("amq.topic") - echos = self.drain(rcv, expected=msgs) - self.ssn.acknowledge(echos[0]) - self.ssn.acknowledge(echos[1], Disposition(REJECTED)) - self.ssn.acknowledge(echos[2], - Disposition(REJECTED, code=3, text="test-reject")) - self.drain(rej, expected=msgs[1:]) - self.ssn.acknowledge() - - def send(self, ssn, target, base, count=1): - snd = ssn.sender(target, durable=self.durable()) - messages = [] - for i in range(count): - c = self.message(base, i) - snd.send(c) - messages.append(c) - snd.close() - return messages - - 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) - messages = 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) - self.drain(copy_rcv, expected=messages) - else: - txssn.rollback() - self.drain(rcv, expected=messages, redelivered=True) - 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) - messages = self.send(txssn, TX_SEND_Q, "txTestSend", 3) - rcv = self.ssn.receiver(TX_SEND_Q) - self.assertEmpty(rcv) - - if commit: - txssn.commit() - self.drain(rcv, expected=messages) - 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) - messages = self.send(self.ssn, TX_ACK_QC, "txTestAck", 3) - self.drain(txrcv, expected=messages) - - if commit: - txssn.acknowledge() - else: - txssn.rollback() - self.drain(txrcv, expected=messages, redelivered=True) - txssn.acknowledge() - txssn.rollback() - self.drain(txrcv, expected=messages, redelivered=True) - txssn.commit() # commit without ack - self.assertEmpty(txrcv) - - txssn.close() - - txssn = self.conn.session(transactional=True) - txrcv = txssn.receiver(TX_ACK_QC) - self.drain(txrcv, expected=messages, redelivered=True) - 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 testDoubleCommit(self): - ssn = self.conn.session(transactional=True) - snd = ssn.sender("amq.direct") - rcv = ssn.receiver("amq.direct") - msgs = [self.message("testDoubleCommit", i) for i in range(3)] - for m in msgs: - snd.send(m) - ssn.commit() - self.drain(rcv, expected=msgs) - ssn.acknowledge() - ssn.commit() - - def testClose(self): - self.ssn.close() - try: - self.ping(self.ssn) - assert False, "ping succeeded" - except Detached: - pass - -RECEIVER_Q = 'test-receiver-queue; {create: always, delete: always}' - -class ReceiverTests(Base): - - def setup_connection(self): - return Connection.establish(self.broker, **self.connection_options()) - - 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, sync=True): - content = self.content(base, count) - self.snd.send(content, sync=sync) - 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 fetchFromClosedTest(self, entry): - entry.close() - try: - msg = self.rcv.fetch(0) - assert False, "unexpected result: %s" % msg - except Empty, e: - assert False, "unexpected exception: %s" % e - except LinkClosed, e: - pass - - def testFetchFromClosedReceiver(self): - self.fetchFromClosedTest(self.rcv) - - def testFetchFromClosedSession(self): - self.fetchFromClosedTest(self.ssn) - - def testFetchFromClosedConnection(self): - self.fetchFromClosedTest(self.conn) - - def fetchFromConcurrentCloseTest(self, entry): - def closer(): - self.sleep() - entry.close() - t = Thread(target=closer) - t.start() - try: - msg = self.rcv.fetch() - assert False, "unexpected result: %s" % msg - except Empty, e: - assert False, "unexpected exception: %s" % e - except LinkClosed, e: - pass - t.join() - - def testFetchFromConcurrentCloseReceiver(self): - self.fetchFromConcurrentCloseTest(self.rcv) - - def testFetchFromConcurrentCloseSession(self): - self.fetchFromConcurrentCloseTest(self.ssn) - - def testFetchFromConcurrentCloseConnection(self): - self.fetchFromConcurrentCloseTest(self.conn) - - def testCapacityIncrease(self): - content = self.send("testCapacityIncrease") - self.sleep() - assert self.rcv.available() == 0 - self.rcv.capacity = UNLIMITED - self.sleep() - assert self.rcv.available() == 1 - msg = self.rcv.fetch(0) - assert msg.content == content - assert self.rcv.available() == 0 - self.ssn.acknowledge() - - def testCapacityDecrease(self): - self.rcv.capacity = UNLIMITED - one = self.send("testCapacityDecrease", 1) - self.sleep() - assert self.rcv.available() == 1 - msg = self.rcv.fetch(0) - assert msg.content == one - - self.rcv.capacity = 0 - - two = self.send("testCapacityDecrease", 2) - self.sleep() - assert self.rcv.available() == 0 - msg = self.rcv.fetch(0) - assert msg.content == two - - self.ssn.acknowledge() - - def capacityTest(self, capacity, threshold=None): - if threshold is not None: - self.rcv.threshold = threshold - self.rcv.capacity = capacity - self.assertAvailable(self.rcv, 0) - - for i in range(2*capacity): - self.send("capacityTest(%s, %s)" % (capacity, threshold), i, sync=False) - self.snd.sync() - self.sleep() - self.assertAvailable(self.rcv) - - first = capacity/2 - second = capacity - first - self.drain(self.rcv, limit = first) - self.sleep() - self.assertAvailable(self.rcv) - self.drain(self.rcv, limit = second) - self.sleep() - self.assertAvailable(self.rcv) - - drained = self.drain(self.rcv) - assert len(drained) == capacity, "%s, %s" % (len(drained), drained) - self.assertAvailable(self.rcv, 0) - - self.ssn.acknowledge() - - def testCapacity5(self): - self.capacityTest(5) - - def testCapacity5Threshold1(self): - self.capacityTest(5, 1) - - def testCapacity10(self): - self.capacityTest(10) - - def testCapacity10Threshold1(self): - self.capacityTest(10, 1) - - def testCapacity100(self): - self.capacityTest(100) - - def testCapacity100Threshold1(self): - self.capacityTest(100, 1) - - def testCapacityUNLIMITED(self): - self.rcv.capacity = UNLIMITED - self.assertAvailable(self.rcv, 0) - - for i in range(10): - self.send("testCapacityUNLIMITED", i) - self.sleep() - self.assertAvailable(self.rcv, 10) - - self.drain(self.rcv) - self.assertAvailable(self.rcv, 0) - - self.ssn.acknowledge() - - def testAvailable(self): - self.rcv.capacity = UNLIMITED - assert self.rcv.available() == 0 - - for i in range(3): - self.send("testAvailable", i) - self.sleep() - assert self.rcv.available() == 3 - - for i in range(3, 10): - self.send("testAvailable", i) - self.sleep() - assert self.rcv.available() == 10 - - self.drain(self.rcv, limit=3) - assert self.rcv.available() == 7 - - self.drain(self.rcv) - assert self.rcv.available() == 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: { - 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=[]) - - # XXX: need testUnsettled() - - def unreliabilityTest(self, mode="unreliable"): - msgs = [self.message("testUnreliable", i) for i in range(3)] - snd = self.ssn.sender("test-unreliability-queue; {create: sender, delete: receiver}") - rcv = self.ssn.receiver(snd.target) - for m in msgs: - snd.send(m) - - # close without ack on reliable receiver, messages should be requeued - ssn = self.conn.session() - rrcv = ssn.receiver("test-unreliability-queue") - self.drain(rrcv, expected=msgs) - ssn.close() - - # close without ack on unreliable receiver, messages should not be requeued - ssn = self.conn.session() - urcv = ssn.receiver("test-unreliability-queue; {link: {reliability: %s}}" % mode) - self.drain(urcv, expected=msgs, redelivered=True) - ssn.close() - - self.assertEmpty(rcv) - - def testUnreliable(self): - self.unreliabilityTest(mode="unreliable") - - def testAtMostOnce(self): - self.unreliabilityTest(mode="at-most-once") - -class AddressTests(Base): - - def setup_connection(self): - return Connection.establish(self.broker, **self.connection_options()) - - 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 InvalidOption, e: - assert "error in options: %s" % error == str(e), e - - try: - self.ssn.receiver("test-bad-options-rcv; %s" % options) - assert False - except InvalidOption, e: - assert "error in options: %s" % error == str(e), e - - def testIllegalKey(self): - self.badOption("{create: always, node: " - "{this-property-does-not-exist: 3}}", - "node: 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: asdf}", - "node: asdf is not a map") - - def testWrongType2(self): - self.badOption("{node: {durable: []}}", - "node: durable: [] is not a bool") - - def testCreateQueue(self): - snd = self.ssn.sender("test-create-queue; {create: always, delete: always, " - "node: {type: queue, durable: False, " - "x-declare: {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: { - type: topic, - durable: False, - x-declare: {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 NotFound, 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 NotFound, 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() - assert False, "successfully deleted amq.topic" - 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 testNodeBindingsQueue(self): - snd = self.ssn.sender(""" -test-node-bindings-queue; { - create: always, - delete: always, - node: { - x-bindings: [{exchange: "amq.topic", key: "a.#"}, - {exchange: "amq.direct", key: "b"}, - {exchange: "amq.topic", key: "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-node-bindings-queue") - self.drain(rcv, expected=["one", "two", "three", "four"]) - - def testNodeBindingsTopic(self): - rcv = self.ssn.receiver("test-node-bindings-topic-queue; {create: always, delete: always}") - rcv_a = self.ssn.receiver("test-node-bindings-topic-queue-a; {create: always, delete: always}") - rcv_b = self.ssn.receiver("test-node-bindings-topic-queue-b; {create: always, delete: always}") - rcv_c = self.ssn.receiver("test-node-bindings-topic-queue-c; {create: always, delete: always}") - snd = self.ssn.sender(""" -test-node-bindings-topic; { - create: always, - delete: always, - node: { - type: topic, - x-bindings: [{queue: test-node-bindings-topic-queue, key: "#"}, - {queue: test-node-bindings-topic-queue-a, key: "a.#"}, - {queue: test-node-bindings-topic-queue-b, key: "b"}, - {queue: test-node-bindings-topic-queue-c, key: "c.*"}] - } -} -""") - m1 = Message("one") - m2 = Message(subject="a.foo", content="two") - m3 = Message(subject="b", content="three") - m4 = Message(subject="c.bar", content="four") - snd.send(m1) - snd.send(m2) - snd.send(m3) - snd.send(m4) - self.drain(rcv, expected=[m1, m2, m3, m4]) - self.drain(rcv_a, expected=[m2]) - self.drain(rcv_b, expected=[m3]) - self.drain(rcv_c, expected=[m4]) - - def testLinkBindings(self): - m_a = self.message("testLinkBindings", 1, subject="a") - m_b = self.message("testLinkBindings", 2, subject="b") - - self.ssn.sender("test-link-bindings-queue; {create: always, delete: always}") - snd = self.ssn.sender("amq.topic") - - snd.send(m_a) - snd.send(m_b) - snd.close() - - rcv = self.ssn.receiver("test-link-bindings-queue") - self.assertEmpty(rcv) - - snd = self.ssn.sender(""" -amq.topic; { - link: { - x-bindings: [{queue: test-link-bindings-queue, key: a}] - } -} -""") - - snd.send(m_a) - snd.send(m_b) - - self.drain(rcv, expected=[m_a]) - rcv.close() - - rcv = self.ssn.receiver(""" -test-link-bindings-queue; { - link: { - x-bindings: [{exchange: "amq.topic", key: b}] - } -} -""") - - snd.send(m_a) - snd.send(m_b) - - self.drain(rcv, expected=[m_a, m_b]) - - 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) - - def doReliabilityTest(self, reliability, messages, expected): - snd = self.ssn.sender("amq.topic") - rcv = self.ssn.receiver("amq.topic; {link: {reliability: %s}}" % reliability) - for m in messages: - snd.send(m) - self.conn.detach() - self.conn.attach() - self.drain(rcv, expected=expected) - - def testReliabilityUnreliable(self): - msgs = [self.message("testReliabilityUnreliable", i) for i in range(3)] - self.doReliabilityTest("unreliable", msgs, []) - - def testReliabilityAtLeastOnce(self): - msgs = [self.message("testReliabilityAtLeastOnce", i) for i in range(3)] - self.doReliabilityTest("at-least-once", msgs, msgs) - - def testLinkName(self): - msgs = [self.message("testLinkName", i) for i in range(3)] - snd = self.ssn.sender("amq.topic") - trcv = self.ssn.receiver("amq.topic; {link: {name: test-link-name}}") - qrcv = self.ssn.receiver("test-link-name") - for m in msgs: - snd.send(m) - self.drain(qrcv, expected=msgs) - - def testAssert1(self): - try: - snd = self.ssn.sender("amq.topic; {assert: always, node: {type: queue}}") - assert 0, "assertion failed to trigger" - except AssertionFailed, e: - pass - - def testAssert2(self): - snd = self.ssn.sender("amq.topic; {assert: always}") - -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.establish(self.broker, **self.connection_options()) - - 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): - self.senderErrorTest(None, MalformedAddress) - - def testNoneSource(self): - self.receiverErrorTest(None, MalformedAddress) - - def testNoTarget(self): - self.senderErrorTest(NOSUCH_Q, NotFound, lambda e: NOSUCH_Q in str(e)) - - def testNoSource(self): - self.receiverErrorTest(NOSUCH_Q, NotFound, lambda e: NOSUCH_Q in str(e)) - - def testUnparseableTarget(self): - self.senderErrorTest(UNPARSEABLE_ADDR, MalformedAddress, - lambda e: "expecting COLON" in str(e)) - - def testUnparseableSource(self): - self.receiverErrorTest(UNPARSEABLE_ADDR, MalformedAddress, - lambda e: "expecting COLON" in str(e)) - - def testUnlexableTarget(self): - self.senderErrorTest(UNLEXABLE_ADDR, MalformedAddress, - lambda e: "unrecognized characters" in str(e)) - - def testUnlexableSource(self): - self.receiverErrorTest(UNLEXABLE_ADDR, MalformedAddress, - lambda e: "unrecognized characters" in str(e)) - - def testInvalidMode(self): - self.receiverErrorTest('name; {mode: "this-is-a-bad-receiver-mode"}', - InvalidOption, - 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.establish(self.broker, **self.connection_options()) - - 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) - self.drain(self.rcv, timeout=self.delay(), expected=msgs) - 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 deleted file mode 100644 index 297374b82b..0000000000 --- a/python/qpid/tests/messaging/message.py +++ /dev/null @@ -1,155 +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.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"<html/>" - 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.establish(self.broker, **self.connection_options()) - - 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) - self.assertEcho(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}, - "key10": True, - "key11": False, - "x-amqp-0-10.app-id": "test-app-id", - "x-amqp-0-10.content-encoding": "test-content-encoding"} - - 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.subject = "subject" - msg.correlation_id = str(self.test_id) - msg.durable = True - msg.priority = 7 - msg.ttl = 60 - msg.properties = MessageEchoTests.TEST_MAP - msg.reply_to = "reply-address" - self.check(msg) - - def testContentTypeUnknown(self): - msg = Message(content_type = "this-content-type-does-not-exist") - self.check(msg) - - def testTextPlain(self): - self.check(Message(content_type="text/plain", content="asdf")) - - def testTextPlainEmpty(self): - self.check(Message(content_type="text/plain")) - - def check_rt(self, addr, expected=None): - if expected is None: - expected = addr - msg = Message(reply_to=addr) - self.snd.send(msg) - echo = self.rcv.fetch(0) - assert echo.reply_to == expected, echo.reply_to - self.ssn.acknowledge(echo) - - def testReplyTo(self): - self.check_rt("name") - - def testReplyToQueue(self): - self.check_rt("name; {node: {type: queue}}", "name") - - def testReplyToQueueSubject(self): - self.check_rt("name/subject; {node: {type: queue}}", "name") - - def testReplyToTopic(self): - self.check_rt("name; {node: {type: topic}}") - - def testReplyToTopicSubject(self): - self.check_rt("name/subject; {node: {type: topic}}") - - def testBooleanEncoding(self): - msg = Message({"true": True, "false": False}) - self.snd.send(msg) - echo = self.rcv.fetch(0) - self.assertEcho(msg, echo) - t = echo.content["true"] - f = echo.content["false"] - assert isinstance(t, bool), t - assert isinstance(f, bool), f |