Modified: qpid/proton/trunk/tests/proton_tests/message.py URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/message.py?rev=1392286&r1=1392285&r2=1392286&view=diff ============================================================================== --- qpid/proton/trunk/tests/proton_tests/message.py (original) +++ qpid/proton/trunk/tests/proton_tests/message.py Mon Oct 1 11:33:09 2012 @@ -17,29 +17,27 @@ # under the License. # -import os, common, xproton -from xproton import * +import os, common +from proton import * class Test(common.Test): def setup(self): - self.msg = pn_message() + self.msg = Message() def teardown(self): - pn_message_free(self.msg) self.msg = None class AccessorsTest(Test): def _test(self, name, default, values): - getter = getattr(xproton, "pn_message_get_%s" % name) - setter = getattr(xproton, "pn_message_set_%s" % name) - d = getter(self.msg) + d = getattr(self.msg, name) assert d == default, d for v in values: - assert setter(self.msg, v) == 0 - assert getter(self.msg) == v + setattr(self.msg, name, v) + gotten = getattr(self.msg, name) + assert gotten == v, gotten def _test_str(self, name): self._test(name, None, ("asdf", "fdsa", "")) @@ -48,22 +46,16 @@ class AccessorsTest(Test): self._test(name, 0, (0, 123456789, 987654321)) def testDurable(self): - assert pn_message_is_durable(self.msg) == False - for v in (True, False): - assert pn_message_set_durable(self.msg, v) == 0 - assert pn_message_is_durable(self.msg) == v + self._test("durable", False, (True, False)) def testPriority(self): - self._test("priority", PN_DEFAULT_PRIORITY, range(0, 255)) + self._test("priority", Message.DEFAULT_PRIORITY, range(0, 255)) def testTtl(self): self._test("ttl", 0, range(12345, 54321)) def testFirstAquirer(self): - assert pn_message_is_first_acquirer(self.msg) == False - for v in (True, False): - assert pn_message_set_first_acquirer(self.msg, v) == 0 - assert pn_message_is_first_acquirer(self.msg) == v + self._test("first_acquirer", False, (True, False)) def testDeliveryCount(self): self._test("delivery_count", 0, range(0, 1024)) @@ -104,84 +96,71 @@ class AccessorsTest(Test): class CodecTest(Test): def testRoundTrip(self): - assert not pn_message_set_ttl(self.msg, 3) - assert not pn_message_set_priority(self.msg, 100) - assert not pn_message_set_address(self.msg, "address") - assert not pn_message_set_subject(self.msg, "subject") + self.msg.ttl = 3 + self.msg.priority = 100 + self.msg.address = "address" + self.msg.subject = "subject" body = 'Hello World!' - rc = pn_message_load(self.msg, body) - assert not rc, rc + self.msg.load(body) - cd, data = pn_message_encode(self.msg, 1024) - assert cd == 0, cd + data = self.msg.encode() - msg2 = pn_message() - cd = pn_message_decode(msg2, data, len(data)) - assert cd == 0, (cd, data) - - assert pn_message_get_ttl(self.msg) == pn_message_get_ttl(msg2), \ - (pn_message_get_ttl(self.msg), pn_message_get_ttl(msg2)) - assert pn_message_get_priority(self.msg) == pn_message_get_priority(msg2) - assert pn_message_get_address(self.msg) == pn_message_get_address(msg2), \ - (pn_message_get_address(self.msg), pn_message_get_address(msg2)) - assert pn_message_get_subject(self.msg) == pn_message_get_subject(msg2) - cd, saved = pn_message_save(self.msg, 1024) - assert not cd, cd - assert saved == body, (body, saved) - - pn_message_free(msg2) + msg2 = Message() + msg2.decode(data) + assert self.msg.ttl == msg2.ttl, (self.msg.ttl, msg2.ttl) + assert self.msg.priority == msg2.priority, (self.msg.priority, msg2.priority) + assert self.msg.address == msg2.address, (self.msg.address, msg2.address) + assert self.msg.subject == msg2.subject, (self.msg.subject, msg2.subject) + saved = self.msg.save() + assert saved == body, (body, saved) class LoadSaveTest(Test): def _test(self, fmt, *bodies): - pn_message_set_format(self.msg, fmt) + self.msg.format = fmt for body in bodies: - pn_message_clear(self.msg) - cd, saved = pn_message_save(self.msg, 1024) - assert (cd, saved) == (0, ""), (cd, saved) - err = pn_message_load(self.msg, body) - assert err == 0, (pn_message_error(self.msg), repr(body)) - cd, saved = pn_message_save(self.msg, 1024) - assert cd >= 0, (cd, pn_message_error(self.msg)) + self.msg.clear() + saved = self.msg.save() + assert saved == "", saved + self.msg.load(body) + saved = self.msg.save() assert saved == body, (body, saved) def testIntegral(self): - self._test(PN_AMQP, "0", "1", "-1", "9223372036854775807") + self._test(Message.AMQP, "0", "1", "-1", "9223372036854775807") def testFloating(self): - self._test(PN_AMQP, "1.1", "3.14159", "-3.14159", "-1.1") + self._test(Message.AMQP, "1.1", "3.14159", "-3.14159", "-1.1") def testSymbol(self): - self._test(PN_AMQP, ':symbol', ':"quoted symbol"') + self._test(Message.AMQP, ':symbol', ':"quoted symbol"') def testString(self): - self._test(PN_AMQP, '"string"', '"string with spaces"') + self._test(Message.AMQP, '"string"', '"string with spaces"') def testBinary(self): - self._test(PN_AMQP, 'b"binary"', 'b"binary with spaces and special values: \\x00\\x01\\x02"') + self._test(Message.AMQP, 'b"binary"', 'b"binary with spaces and special values: \\x00\\x01\\x02"') def testMap(self): - self._test(PN_AMQP, '{"one"=1, :two=2, :pi=3.14159}', '{[1, 2, 3]=[3, 2, 1], {1=2}={3=4}}') + self._test(Message.AMQP, '{"one"=1, :two=2, :pi=3.14159}', '{[1, 2, 3]=[3, 2, 1], {1=2}={3=4}}') def testList(self): - self._test(PN_AMQP, '[1, 2, 3]', '["one", "two", "three"]', '[:one, 2, 3.14159]', + self._test(Message.AMQP, '[1, 2, 3]', '["one", "two", "three"]', '[:one, 2, 3.14159]', '[{1=2}, {3=4}, {5=6}]') def testDescriptor(self): - self._test(PN_AMQP, '@21 ["one", 2, "three", @:url "http://example.org";]') + self._test(Message.AMQP, '@21 ["one", 2, "three", @:url "http://example.org";]') def testData(self): - self._test(PN_DATA, "this is data\x00\x01\x02 blah blah") + self._test(Message.DATA, "this is data\x00\x01\x02 blah blah") def testText(self): - self._test(PN_TEXT, "this is a text string") + self._test(Message.TEXT, "this is a text string") def testTextLoadNone(self): - pn_message_set_format(self.msg, PN_TEXT) - pn_message_clear(self.msg) - cd = pn_message_load(self.msg, None) - assert cd == 0 - cd, saved = pn_message_save(self.msg, 1024) - assert cd >= 0 + self.msg.format = Message.TEXT + self.msg.clear() + self.msg.load(None) + saved = self.msg.save() assert saved == ""
Modified: qpid/proton/trunk/tests/proton_tests/messenger.py URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/messenger.py?rev=1392286&r1=1392285&r2=1392286&view=diff ============================================================================== --- qpid/proton/trunk/tests/proton_tests/messenger.py (original) +++ qpid/proton/trunk/tests/proton_tests/messenger.py Mon Oct 1 11:33:09 2012 @@ -17,82 +17,77 @@ # under the License. # -import os, common, xproton -from xproton import * +import os, common +from proton import * from threading import Thread class Test(common.Test): def setup(self): - self.server = pn_messenger("server") - pn_messenger_set_timeout(self.server, 10000) - pn_messenger_start(self.server) - pn_messenger_subscribe(self.server, "amqp://~0.0.0.0:12345") - self.thread = Thread(target=self.run) + self.server = Messenger("server") + self.server.timeout=10000 + self.server.start() + self.server.subscribe("amqp://~0.0.0.0:12345") + self.thread = Thread(name="server-thread", target=self.run) self.running = True self.thread.start() - self.client = pn_messenger("client") - pn_messenger_set_timeout(self.client, 10000) - pn_messenger_start(self.client) + self.client = Messenger("client") + self.client.timeout=1000 + self.client.start() def teardown(self): - self.running = False - msg = pn_message() - pn_message_set_address(msg, "amqp://0.0.0.0:12345") - pn_messenger_put(self.client, msg) - pn_messenger_send(self.client) - pn_messenger_stop(self.client) + if self.running: + self.running = False + msg = Message() + msg.address="amqp://0.0.0.0:12345" + self.client.put(msg) + self.client.send() + self.client.stop() self.thread.join() - pn_messenger_free(self.client) - pn_messenger_free(self.server) self.client = None self.server = None - pn_message_free(msg) class MessengerTest(Test): def run(self): - msg = pn_message() - while self.running: - pn_messenger_recv(self.server, 10) - while pn_messenger_incoming(self.server): - if pn_messenger_get(self.server, msg): - print pn_messenger_error(self.server) - else: - reply_to = pn_message_get_reply_to(msg) - if reply_to: - pn_message_set_address(msg, reply_to) - pn_messenger_put(self.server, msg) - pn_messenger_stop(self.server) - pn_message_free(msg) + msg = Message() + try: + while self.running: + self.server.recv(10) + while self.server.incoming: + self.server.get(msg) + if msg.reply_to: + msg.address = msg.reply_to + self.server.put(msg) + except Timeout: + print "server timed out" + self.server.stop() + self.running = False def testSendReceive(self): - msg = pn_message() - pn_message_set_address(msg, "amqp://0.0.0.0:12345") - pn_message_set_subject(msg, "Hello World!") + msg = Message() + msg.address="amqp://0.0.0.0:12345" + msg.subject="Hello World!" body = "First the world, then the galaxy!" - pn_message_load(msg, body) - pn_messenger_put(self.client, msg) - pn_messenger_send(self.client) - - reply = pn_message() - assert not pn_messenger_recv(self.client, 1) - assert pn_messenger_incoming(self.client) == 1 - assert not pn_messenger_get(self.client, reply) - - assert pn_message_get_subject(reply) == "Hello World!" - cd, rbod = pn_message_save(reply, 1024) - assert not cd - assert rbod == body, (rbod, body) + msg.load(body) + self.client.put(msg) + self.client.send() + + reply = Message() + self.client.recv(1) + assert self.client.incoming == 1 + self.client.get(reply) - pn_message_free(msg) - pn_message_free(reply) + assert reply.subject == "Hello World!" + rbod = reply.save() + assert rbod == body, (rbod, body) def testSendBogus(self): - msg = pn_message() - pn_message_set_address(msg, "totally-bogus-address") - assert pn_messenger_put(self.client, msg) == PN_ERR - err = pn_messenger_error(self.client) - assert "unable to send to address: totally-bogus-address (" in err, err - pn_message_free(msg) + msg = Message() + msg.address="totally-bogus-address" + try: + self.client.put(msg) + except MessengerException, exc: + err = str(exc) + assert "unable to send to address: totally-bogus-address (" in err, err Modified: qpid/proton/trunk/tests/proton_tests/sasl.py URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/sasl.py?rev=1392286&r1=1392285&r2=1392286&view=diff ============================================================================== --- qpid/proton/trunk/tests/proton_tests/sasl.py (original) +++ qpid/proton/trunk/tests/proton_tests/sasl.py Mon Oct 1 11:33:09 2012 @@ -17,8 +17,8 @@ # under the License. # -import os, common, xproton -from xproton import * +import os, common +from proton import * class Test(common.Test): pass @@ -26,34 +26,28 @@ class Test(common.Test): class SaslTest(Test): def testPipelined(self): - cli = pn_transport() - cli_auth = pn_sasl(cli) - pn_sasl_mechanisms(cli_auth, "ANONYMOUS") - pn_sasl_client(cli_auth) + cli = Transport() + cli_auth = SASL(cli) + cli_auth.mechanisms("ANONYMOUS") + cli_auth.client() - assert pn_sasl_outcome(cli_auth) == PN_SASL_NONE + assert cli_auth.outcome is None - srv = pn_transport() - srv_auth = pn_sasl(srv) - pn_sasl_mechanisms(srv_auth, "ANONYMOUS") - pn_sasl_server(srv_auth) - pn_sasl_done(srv_auth, PN_SASL_OK) + srv = Transport() + srv_auth = SASL(srv) + srv_auth.mechanisms("ANONYMOUS") + srv_auth.server() + srv_auth.done(SASL.OK) - cli_code, cli_out = pn_output(cli, 1024) - srv_code, srv_out = pn_output(srv, 1024) + cli_out = cli.output(1024) + srv_out = srv.output(1024) - assert cli_code > 0, cli_code - assert srv_code > 0, srv_code + n = srv.input(cli_out) + assert n == len(cli_out), (n, cli_out) - n = pn_input(srv, cli_out) - assert n == len(cli_out), "(%s) %s" % (n, pn_error_text(pn_transport_error(srv))) + assert cli_auth.outcome is None - assert pn_sasl_outcome(cli_auth) == PN_SASL_NONE - - n = pn_input(cli, srv_out) + n = cli.input(srv_out) assert n == len(srv_out), n - assert pn_sasl_outcome(cli_auth) == PN_SASL_OK - - pn_transport_free(cli) - pn_transport_free(srv) + assert cli_auth.outcome == SASL.OK Modified: qpid/proton/trunk/tests/proton_tests/transport.py URL: http://svn.apache.org/viewvc/qpid/proton/trunk/tests/proton_tests/transport.py?rev=1392286&r1=1392285&r2=1392286&view=diff ============================================================================== --- qpid/proton/trunk/tests/proton_tests/transport.py (original) +++ qpid/proton/trunk/tests/proton_tests/transport.py Mon Oct 1 11:33:09 2012 @@ -17,8 +17,8 @@ # under the License. # -import os, common, xproton -from xproton import * +import os, common +from proton import * class Test(common.Test): pass @@ -26,80 +26,104 @@ class Test(common.Test): class TransportTest(Test): def setup(self): - self.transport = pn_transport() + self.transport = Transport() def teardown(self): - pn_transport_free(self.transport) self.transport = None def testEOS(self): - n = pn_input(self.transport, "") - assert n == PN_ERR, pn_error_text(pn_transport_error(self.transport)) + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) def testPartial(self): - n = pn_input(self.transport, "AMQ") + n = self.transport.input("AMQ") assert n == 3, n - n = pn_input(self.transport, "") - assert n == PN_ERR, n + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) def testGarbage(self): - n = pn_input(self.transport, "GARBAGE_") - assert n == PN_ERR, pn_error_text(pn_transport_error(self.transport)) - n = pn_input(self.transport, "") - assert n == PN_ERR, n + try: + n = self.transport.input("GARBAGE_") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) def testSmallGarbage(self): - n = pn_input(self.transport, "XXX") - assert n == PN_ERR, pn_error_text(pn_transport_error(self.transport)) - n = pn_input(self.transport, "") - assert n == PN_ERR, n + try: + n = self.transport.input("XXX") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) def testBigGarbage(self): - n = pn_input(self.transport, "GARBAGE_XXX") - assert n == PN_ERR, pn_error_text(pn_transport_error(self.transport)) - n = pn_input(self.transport, "") - assert n == PN_ERR, n + try: + n = self.transport.input("GARBAGE_XXX") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "AMQP header missmatch" in str(e) def testHeader(self): - n = pn_input(self.transport, "AMQP\x00\x01\x00\x00") + n = self.transport.input("AMQP\x00\x01\x00\x00") assert n == 8, n - n = pn_input(self.transport, "") - assert n == PN_ERR, n + try: + n = self.transport.input("") + assert False, n + except TransportException, e: + assert "connection aborted" in str(e) def testOutput(self): - n, out = pn_output(self.transport, 1024) - assert n == len(out) + out = self.transport.output(1024) + assert out is not None def testBindAfterOpen(self): - conn = pn_connection() - ssn = pn_session(conn) - pn_connection_open(conn) - pn_session_open(ssn) - pn_connection_set_container(conn, "test-container") - pn_connection_set_hostname(conn, "test-hostname") - trn = pn_transport() - pn_transport_bind(trn, conn) - n, out = pn_output(trn, 1024) - assert n == len(out), n + conn = Connection() + ssn = conn.session() + conn.open() + ssn.open() + conn.container = "test-container" + conn.hostname = "test-hostname" + trn = Transport() + trn.bind(conn) + out = trn.output(1024) assert "test-container" in out, repr(out) - n = pn_input(self.transport, out) - assert n > 0 and n < len(out) + assert "test-hostname" in out, repr(out) + n = self.transport.input(out) + assert n > 0, n out = out[n:] - n = pn_input(self.transport, out) + n = self.transport.input(out) assert n == 0 - c = pn_connection() - assert pn_connection_remote_container(c) == None - assert pn_connection_remote_hostname(c) == None - pn_transport_bind(self.transport, c) - assert pn_connection_remote_container(c) == "test-container" - assert pn_connection_remote_hostname(c) == "test-hostname" - assert pn_session_head(c, 0) == None - n = pn_input(self.transport, out) - assert n == len(out), (n, out) - assert pn_session_head(c, 0) != None - - pn_transport_free(trn) - pn_connection_free(conn) - pn_connection_free(c) + c = Connection() + assert c.remote_container == None + assert c.remote_hostname == None + assert c.session_head(0) == None + self.transport.bind(c) + assert c.remote_container == "test-container" + assert c.remote_hostname == "test-hostname" + if out: + assert c.session_head(0) == None + n = self.transport.input(out) + assert n == len(out), (n, out) + assert c.session_head(0) != None --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
