# # 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 . import common # import logging import time from proton import Condition from proton import Message from proton import symbol import pyngus class APITest(common.Test): def setup(self, conn1_props=None, conn2_props=None): super(APITest, self).setup() # logging.getLogger("pyngus").setLevel(logging.DEBUG) self.container1 = pyngus.Container("test-container-1") self.conn1_handler = common.ConnCallback() if conn1_props is None: # props = {"x-trace-protocol": True} conn1_props = {"x-trace-protocol": False} self.conn1 = self.container1.create_connection("conn1", self.conn1_handler, conn1_props) self.conn1.open() self.container2 = pyngus.Container("test-container-2") self.conn2_handler = common.ConnCallback() self.conn2 = self.container2.create_connection("conn2", self.conn2_handler, conn2_props) self.conn2.open() def process_connections(self, timestamp=None): common.process_connections(self.conn1, self.conn2, timestamp) def teardown(self): if self.conn1: self.conn1.destroy() if self.container1: self.container1.destroy() if self.conn2: self.conn2.destroy() if self.container2: self.container2.destroy() super(APITest, self).teardown() def _setup_sender_sync(self): """Create links, initiated by sender.""" sl_handler = common.SenderCallback() sender = self.conn1.create_sender("src", "tgt", sl_handler) sender.user_context = sl_handler sender.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] rl_handler = common.ReceiverCallback() receiver = self.conn2.accept_receiver(args.link_handle, event_handler=rl_handler) receiver.user_context = rl_handler receiver.open() self.process_connections() assert receiver.active and rl_handler.active_ct > 0 assert sender.active and sl_handler.active_ct > 0 return (sender, receiver) def _setup_receiver_sync(self): """Create links, initiated by receiver.""" rl_handler = common.ReceiverCallback() receiver = self.conn2.create_receiver("tgt", "src", rl_handler) receiver.user_context = rl_handler receiver.open() self.process_connections() assert self.conn1_handler.sender_requested_ct == 1 args = self.conn1_handler.sender_requested_args[0] sl_handler = common.SenderCallback() sender = self.conn1.accept_sender(args.link_handle, event_handler=sl_handler) sender.user_context = sl_handler sender.open() self.process_connections() assert sender.active and sl_handler.active_ct > 0 assert receiver.active and rl_handler.active_ct > 0 return (sender, receiver) def test_create_destroy(self): sender = self.conn1.create_sender("source-addr", "target-addr", name="my-name") sender.user_context = "whatever" assert sender.name == "my-name" assert sender.source_address == "source-addr" assert sender.target_address is None receiver = self.conn2.create_receiver("target-addr", "source-addr", name="other-name") receiver.user_context = "meh" assert receiver.name == "other-name" assert receiver.target_address == "target-addr" assert receiver.source_address is None sender.destroy() receiver.destroy() def test_send_abort(self): sl_handler = common.SenderCallback() cb = common.DeliveryCallback() sender = self.conn1.create_sender("saddr", "taddr", sl_handler) msg = Message() msg.body = "Hi" sender.send(msg, cb, None) sender.open() sender.close() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] rl_handler = common.ReceiverCallback() receiver1 = self.conn2.accept_receiver(args.link_handle, event_handler=rl_handler) receiver1.open() self.process_connections() assert rl_handler.remote_closed_ct receiver1.close() self.process_connections() assert sl_handler.closed_ct assert sl_handler.active_ct == 0 assert cb.count assert cb.status == pyngus.SenderLink.ABORTED def test_receiver_abort(self): rl_handler = common.ReceiverCallback() receiver = self.conn1.create_receiver("taddr", "saddr", rl_handler) receiver.add_capacity(1) receiver.open() receiver.close() self.process_connections() assert self.conn2_handler.sender_requested_ct == 1 args = self.conn2_handler.sender_requested_args[0] sl_handler = common.SenderCallback() sender = self.conn2.accept_sender(args.link_handle, event_handler=sl_handler) sender.open() self.process_connections() assert sl_handler.remote_closed_ct sender.close() self.process_connections() assert rl_handler.closed_ct assert rl_handler.active_ct == 0 def test_sender_setup_sync(self): sender, receiver = self._setup_sender_sync() sl_handler = sender.user_context rl_handler = receiver.user_context sender.close() self.process_connections() assert sl_handler.closed_ct == 0 assert rl_handler.remote_closed_ct == 1 assert rl_handler.remote_closed_error is None receiver.close() self.process_connections() assert sl_handler.closed_ct == 1 assert sl_handler.remote_closed_ct == 0 assert rl_handler.closed_ct == 1 def test_pipeline_close(self): sender, receiver = self._setup_sender_sync() sl_handler = sender.user_context rl_handler = receiver.user_context sender.close() self.conn1.close() self.process_connections() receiver.close() self.conn2.close() self.process_connections() assert sl_handler.closed_ct == 1 assert rl_handler.closed_ct == 1 def test_sender_close_cond_sync(self): sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context cond = Condition("bad", "hate you", {"yo-mama": "wears army boots"}) sender.close(cond) self.process_connections() assert rl_handler.remote_closed_ct == 1 assert rl_handler.remote_closed_error r_cond = rl_handler.remote_closed_error assert r_cond.name == "bad" assert r_cond.description == "hate you" assert r_cond.info.get("yo-mama") == "wears army boots" def test_receiver_setup_sync(self): sender, receiver = self._setup_receiver_sync() sl_handler = sender.user_context rl_handler = receiver.user_context receiver.close() self.process_connections() assert sl_handler.remote_closed_ct == 1 assert sl_handler.remote_closed_error is None sender.close() self.process_connections() assert rl_handler.closed_ct == 1 assert rl_handler.remote_closed_ct == 0 assert sl_handler.closed_ct == 1 def test_receiver_close_cond_sync(self): sender, receiver = self._setup_receiver_sync() sl_handler = sender.user_context cond = Condition("meh", "blah", {"dog": "cat"}) receiver.close(cond) self.process_connections() assert sl_handler.remote_closed_ct == 1 assert sl_handler.remote_closed_error r_cond = sl_handler.remote_closed_error assert r_cond.name == "meh" assert r_cond.description == "blah" assert r_cond.info.get("dog") == "cat" def test_credit_sync(self): sender, receiver = self._setup_sender_sync() sl_handler = sender.user_context rl_handler = receiver.user_context assert receiver.capacity == 0 receiver.add_capacity(3) assert receiver.capacity == 3 assert sender.credit == 0 assert sl_handler.credit_granted_ct == 0 self.process_connections() # verify credit is sent to sender: assert sender.credit == 3 assert sl_handler.credit_granted_ct == 1 receiver.add_capacity(1) self.process_connections() assert receiver.capacity == 4 assert sender.credit == 4 # callback only occurs when credit is no longer zero: assert sl_handler.credit_granted_ct == 1 assert sender.pending == 0 msg = Message() msg.body = "Hi" sender.send(msg) # none pending because credit was consumed assert sender.credit == 3 assert sender.pending == 0 self.process_connections() # verify receiver's capacity decreases on send: assert receiver.capacity == 3 assert rl_handler.message_received_ct == 1 assert sender.credit == 3 assert sender.pending == 0 while sender.credit != 0: sender.send(msg) self.process_connections() assert receiver.capacity == 0 assert rl_handler.message_received_ct == 4 # verify no msgs sent if no credit: sender.send(msg) sender.send(msg) self.process_connections() assert sender.pending == 2 assert sl_handler.credit_granted_ct == 1 receiver.add_capacity(1) self.process_connections() assert receiver.capacity == 0 assert rl_handler.message_received_ct == 5 assert sender.credit == 0 assert sender.pending == 1 assert sl_handler.credit_granted_ct == 1 receiver.add_capacity(1) self.process_connections() assert sender.credit == 0 assert sender.pending == 0 assert sl_handler.credit_granted_ct == 1 # verify new credit becomes available: receiver.add_capacity(1) self.process_connections() assert sender.credit == 1 assert sl_handler.credit_granted_ct == 2 def test_send_presettled(self): sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg) receiver.add_capacity(1) self.process_connections() assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] assert msg2.body == "Hi" receiver.message_accepted(handle) def test_send_accepted(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle") self.process_connections() assert rl_handler.message_received_ct == 0 receiver.add_capacity(1) self.process_connections() assert cb.link is None # not acknowledged yet assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] receiver.message_accepted(handle) self.process_connections() assert cb.link == sender assert cb.handle == "my-handle" assert cb.status == pyngus.SenderLink.ACCEPTED def test_send_released(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle") receiver.add_capacity(1) self.process_connections() assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] receiver.message_released(handle) self.process_connections() assert cb.link == sender assert cb.handle == "my-handle" assert cb.status == pyngus.SenderLink.RELEASED def test_send_rejected(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle") receiver.add_capacity(1) self.process_connections() assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] cond = Condition("itchy", "Needs scratching", {"bath": True}) receiver.message_rejected(handle, cond) self.process_connections() assert cb.link == sender assert cb.handle == "my-handle" assert cb.status == pyngus.SenderLink.REJECTED r_cond = cb.info.get("condition") assert r_cond and r_cond.name == "itchy" def test_send_modified(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle") receiver.add_capacity(1) self.process_connections() assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] annotations = {"dog": 1, "cat": False} receiver.message_modified(handle, False, True, annotations) self.process_connections() assert cb.link == sender assert cb.handle == "my-handle" assert cb.status == pyngus.SenderLink.MODIFIED assert cb.info.get("delivery-failed") is False assert cb.info.get("undeliverable-here") is True info = cb.info.get("message-annotations") assert info and info["dog"] == 1 def test_send_expired_no_credit(self): cb = common.DeliveryCallback() sender, receiver = self._setup_receiver_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle", deadline=10) # receiver.add_capacity(1) self.process_connections(timestamp=9) assert rl_handler.message_received_ct == 0 assert sender.pending == 1 assert cb.status is None self.process_connections(timestamp=10) assert sender.pending == 0 assert cb.status == pyngus.SenderLink.TIMED_OUT def test_send_expired_late_reply(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context receiver.add_capacity(1) self.process_connections(timestamp=1) msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle", deadline=10) self.process_connections(timestamp=9) assert rl_handler.message_received_ct == 1 assert sender.pending == 1 assert sender.credit == 0 assert cb.status is None self.process_connections(timestamp=10) assert rl_handler.message_received_ct == 1 assert sender.pending == 0 assert cb.status == pyngus.SenderLink.TIMED_OUT # late reply: assert cb.count == 1 msg2, handle = rl_handler.received_messages[0] receiver.message_accepted(handle) self.process_connections(timestamp=15) assert cb.count == 1 def test_send_expired_no_reply(self): cb = common.DeliveryCallback() sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, cb, "my-handle", deadline=10) self.process_connections(timestamp=1) assert rl_handler.message_received_ct == 0 assert sender.pending == 1 assert sender.credit == 0 assert cb.count == 0 receiver.add_capacity(1) self.process_connections(timestamp=2) assert rl_handler.message_received_ct == 1 assert sender.pending == 1 assert sender.credit == 0 assert cb.count == 0 self.process_connections(timestamp=12) assert sender.pending == 0 assert cb.count == 1 assert cb.status == pyngus.SenderLink.TIMED_OUT def test_send_expired_no_callback(self): sender, receiver = self._setup_receiver_sync() rl_handler = receiver.user_context msg = Message() msg.body = "Hi" sender.send(msg, deadline=10) assert sender.pending == 1 self.process_connections(timestamp=12) assert rl_handler.message_received_ct == 0 assert sender.pending == 0 def test_send_deadline_idle(self): """Validate the connection's deadline processing.""" self.teardown() self.setup(conn1_props={"idle-time-out": 99}) sender1 = self.conn1.create_sender("src1", "tgt1") sender1.open() self.process_connections(timestamp=1) assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] receiver1 = self.conn2.accept_receiver(args.link_handle) receiver1.open() sender2 = self.conn1.create_sender("src2", "tgt2") sender2.open() self.process_connections(timestamp=1) assert self.conn2_handler.receiver_requested_ct == 2 args = self.conn2_handler.receiver_requested_args[1] receiver2 = self.conn2.accept_receiver(args.link_handle) receiver2.open() self.process_connections(timestamp=1) assert self.conn1.deadline == 100.0 msg = Message() msg.body = "Hi" sender1.send(msg, deadline=11) assert self.conn1.deadline == 11 self.process_connections(timestamp=2) assert self.conn1.deadline == 11 sender2.send(msg, deadline=7) assert self.conn1.deadline == 7 self.process_connections(timestamp=7) assert self.conn1.deadline == 11 self.process_connections(timestamp=11) assert self.conn1.deadline == 100 # next send timeout after the idle keepalive: sender1.send(msg, deadline=101) self.process_connections(timestamp=11) assert self.conn1.deadline == 100 # have remote send idle, next deadline should # be the pending send: self.process_connections(timestamp=self.conn2.deadline) assert self.conn1.deadline == 101 def test_send_close_on_ack(self): """Verify that the sender can close itself when delivery complete.""" class SendDoneCallback(common.DeliveryCallback): def __call__(self, link, handle, status, info): super(SendDoneCallback, self).__call__(link, handle, status, info) if self.count == 1: # verify that we can safely close ourself, even if there is # a send that has not completed: assert status == pyngus.SenderLink.ACCEPTED cond = Condition("indigestion", "old sushi", {"smoked eel": "yummy"}) link.close(cond) else: # the unsent message is aborted prior # to invoking closed callback: assert status == pyngus.SenderLink.ABORTED sl_handler = link.user_context assert sl_handler.closed_ct == 0 sender, receiver = self._setup_sender_sync() rl_handler = receiver.user_context sl_handler = sender.user_context receiver.add_capacity(1) msg = Message() msg.body = "Hi" cb = SendDoneCallback() sender.send(msg, cb, "my-handle") # no credit - this one won't get sent: sender.send(msg, cb, "my-handle") self.process_connections() assert sender.active assert rl_handler.message_received_ct == 1 msg2, handle = rl_handler.received_messages[0] receiver.message_accepted(handle) self.process_connections() assert not sender.active assert not sender.closed # pending messages are aborted when close completes: assert cb.count == 1 assert sl_handler.closed_ct == 0 receiver.close() self.process_connections() assert sender.closed assert sl_handler.closed_ct assert cb.count == 2 # last callback should be abort: cond = cb.info.get('condition') assert cond assert cond.name == "indigestion" assert cb.status == pyngus.SenderLink.ABORTED def test_multi_frame_message(self): """Verify multi-frame message send/receive.""" self.teardown() props = {"max-frame-size": 512} self.setup(conn1_props=props, conn2_props=props) sender1 = self.conn1.create_sender("src1", "tgt1") sender1.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] rl_handler = common.ReceiverCallback() receiver1 = self.conn2.accept_receiver(args.link_handle, event_handler=rl_handler) receiver1.add_capacity(1) receiver1.open() self.process_connections() msg = Message() msg.body = "Hi!" * 512 # > max frame size cb = common.DeliveryCallback() sender1.send(msg, cb) # manually transfer output from conn1 to conn2 in small batches, # forcing conn2 to process a partial delivery: self.conn1.process(time.time()) while self.conn1.has_output: count = min(self.conn1.has_output, 512) part = self.conn1.output_data()[:count] count = self.conn2.process_input(part) self.conn2.process(time.time()) self.conn1.output_written(count) assert rl_handler.message_received_ct == 1 rmsg, handle = rl_handler.received_messages[0] assert rmsg.body == msg.body receiver1.message_accepted(handle) self.process_connections() assert cb.count assert cb.status == pyngus.SenderLink.ACCEPTED def test_dynamic_receiver_props(self): """Verify dynamic-node-properties can be requested.""" # dynamic receive link: DELETE_ON_CLOSE = 0x000000000000002b desired_props = {symbol("lifetime-policy"): DELETE_ON_CLOSE} props = {"dynamic-node-properties": desired_props} sender = self.conn1.create_sender("saddr", target_address=None, properties=props) sender.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] dnp = args.properties.get('dynamic-node-properties') assert dnp and dnp == desired_props # dynamic send link: desired_props = {symbol("supported-dist-modes"): [symbol("move"), symbol("copy")]} props = {"dynamic-node-properties": desired_props} receiver = self.conn1.create_receiver("taddr", source_address=None, properties=props) receiver.open() self.process_connections() assert self.conn2_handler.sender_requested_ct == 1 args = self.conn2_handler.sender_requested_args[0] dnp = args.properties.get('dynamic-node-properties') assert dnp and dnp == desired_props def test_use_after_free(self): """Causes proton library to segfault!!!""" if self.PROTON_VERSION < (0, 8): raise common.Skipped("Skipping test - causes segfault in proton!") sender = self.conn1.create_sender("src1", "tgt1") sender.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] assert args.properties.get('snd-settle-mode') is None self.conn2.reject_receiver(args.link_handle) self.process_connections() sender.destroy() props = {"snd-settle-mode": "settled"} sender = self.conn1.create_sender("src1", "tgt1", properties=props) sender.open() self.process_connections() def test_settle_modes(self): """Verify the configured settlement modes are visible to the peer. """ # defaults (none given) sender = self.conn1.create_sender("src1", "tgt1") sender.open() receiver = self.conn1.create_receiver("tgta", "srca") receiver.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 1 args = self.conn2_handler.receiver_requested_args[0] assert args.properties.get('snd-settle-mode') is None assert args.properties.get('rcv-settle-mode') is None assert self.conn2_handler.sender_requested_ct == 1 args = self.conn2_handler.sender_requested_args[0] assert args.properties.get('snd-settle-mode') is None assert args.properties.get('rcv-settle-mode') is None # settled props = {"snd-settle-mode": "settled", "rcv-settle-mode": "second"} sender = self.conn1.create_sender("src2", "tgt2", properties=props) sender.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct == 2 args = self.conn2_handler.receiver_requested_args[1] mode = args.properties.get('snd-settle-mode') assert mode and mode == 'settled' mode = args.properties.get('rcv-settle-mode') assert mode and mode == 'second' # unsettled props = {"snd-settle-mode": "unsettled", "rcv-settle-mode": "second"} recver = self.conn1.create_receiver("tgt3", "src3", properties=props) recver.open() self.process_connections() assert self.conn2_handler.sender_requested_ct == 2 args = self.conn2_handler.sender_requested_args[1] mode = args.properties.get('snd-settle-mode') assert mode and mode == 'unsettled' mode = args.properties.get('rcv-settle-mode') assert mode and mode == 'second' def test_non_reentrant_callback(self): class SenderBadCallback(common.SenderCallback): def sender_active(self, sender): # Illegal: sender.destroy() class ReceiverBadCallback(common.ReceiverCallback): def receiver_active(self, receiver): # Illegal: receiver.destroy() class ReceiverBadCallback2(common.ReceiverCallback): def receiver_active(self, receiver): # Illegal: receiver.connection.destroy() sc = SenderBadCallback() sender = self.conn1.create_sender("src1", "tgt1", event_handler=sc) sender.open() self.process_connections() assert self.conn2_handler.receiver_requested_ct args = self.conn2_handler.receiver_requested_args[-1] receiver = self.conn2.accept_receiver(args.link_handle) receiver.open() try: self.process_connections() assert False, "RuntimeError expected!" except RuntimeError: pass sender = self.conn1.create_sender("src2", "tgt2") sender.open() self.process_connections() args = self.conn2_handler.receiver_requested_args[-1] rc = ReceiverBadCallback() receiver = self.conn2.accept_receiver(args.link_handle, event_handler=rc) receiver.open() try: self.process_connections() assert False, "RuntimeError expected!" except RuntimeError: pass sender = self.conn1.create_sender("src3", "tgt3") sender.open() self.process_connections() args = self.conn2_handler.receiver_requested_args[-1] rc = ReceiverBadCallback2() receiver = self.conn2.accept_receiver(args.link_handle, event_handler=rc) receiver.open() try: self.process_connections() assert False, "RuntimeError expected!" except RuntimeError: pass