# # 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 __future__ import absolute_import import os import gc from time import time, sleep from proton import * from proton.reactor import Container from . import common from .common import pump, Skipped # older versions of gc do not provide the garbage list if not hasattr(gc, "garbage"): gc.garbage = [] # future test areas # + different permutations of setup # - creating deliveries and calling input/output before opening the session/link # + shrinking output_size down to something small? should the engine buffer? # + resuming # - locally and remotely created deliveries with the same tag OUTPUT_SIZE = 10 * 1024 class Test(common.Test): def __init__(self, *args): common.Test.__init__(self, *args) self._wires = [] def connection(self): c1 = Connection() c2 = Connection() t1 = Transport() t1.bind(c1) t2 = Transport() t2.bind(c2) self._wires.append((c1, t1, c2, t2)) mask1 = 0 mask2 = 0 for cat in ("TRACE_FRM", "TRACE_RAW"): trc = os.environ.get("PN_%s" % cat) if trc and trc.lower() in ("1", "2", "yes", "true"): mask1 = mask1 | getattr(Transport, cat) if trc == "2": mask2 = mask2 | getattr(Transport, cat) t1.trace(mask1) t2.trace(mask2) return c1, c2 def link(self, name, max_frame=None, idle_timeout=None): c1, c2 = self.connection() if max_frame: c1.transport.max_frame_size = max_frame[0] c2.transport.max_frame_size = max_frame[1] if idle_timeout: # idle_timeout in seconds expressed as float c1.transport.idle_timeout = idle_timeout[0] c2.transport.idle_timeout = idle_timeout[1] c1.open() c2.open() ssn1 = c1.session() ssn1.open() self.pump() ssn2 = c2.session_head(Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE) ssn2.open() self.pump() snd = ssn1.sender(name) rcv = ssn2.receiver(name) return snd, rcv def cleanup(self): self._wires = [] def pump(self, buffer_size=OUTPUT_SIZE): for c1, t1, c2, t2 in self._wires: pump(t1, t2, buffer_size) class ConnectionTest(Test): def setUp(self): gc.enable() self.c1, self.c2 = self.connection() def cleanup(self): # release resources created by this class super(ConnectionTest, self).cleanup() self.c1 = None self.c2 = None def tearDown(self): self.cleanup() gc.collect() assert not gc.garbage def test_open_close(self): assert self.c1.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert self.c2.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.c1.open() self.pump() assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT assert self.c2.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE self.c2.open() self.pump() assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.c1.close() self.pump() assert self.c1.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED self.c2.close() self.pump() assert self.c1.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert self.c2.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_simultaneous_open_close(self): assert self.c1.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert self.c2.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.c1.open() self.c2.open() self.pump() assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.c1.close() self.c2.close() self.pump() assert self.c1.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert self.c2.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_capabilities_array(self): self.c1.offered_capabilities = Array(UNDESCRIBED, Data.SYMBOL, symbol("O_one"), symbol("O_two"), symbol("O_three")) self.c1.desired_capabilities = Array(UNDESCRIBED, Data.SYMBOL, symbol("D_one"), symbol("D_two"), symbol("D_three")) self.c1.open() assert self.c2.remote_offered_capabilities is None assert self.c2.remote_desired_capabilities is None self.pump() assert self.c2.remote_offered_capabilities == self.c1.offered_capabilities, \ (self.c2.remote_offered_capabilities, self.c1.offered_capabilities) assert self.c2.remote_desired_capabilities == self.c1.desired_capabilities, \ (self.c2.remote_desired_capabilities, self.c1.desired_capabilities) def test_capabilities_symbol_list(self): self.c1.offered_capabilities = SymbolList(['O_one', 'O_two', symbol('O_three')]) self.c1.desired_capabilities = SymbolList([symbol('D_one'), 'D_two', 'D_three']) self.c1.open() assert self.c2.remote_offered_capabilities is None assert self.c2.remote_desired_capabilities is None self.pump() assert self.c2.remote_offered_capabilities == self.c1.offered_capabilities, \ (self.c2.remote_offered_capabilities, self.c1.offered_capabilities) assert self.c2.remote_desired_capabilities == self.c1.desired_capabilities, \ (self.c2.remote_desired_capabilities, self.c1.desired_capabilities) def test_condition(self): self.c1.open() self.c2.open() self.pump() assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE cond = Condition("blah:bleh", "this is a description", {symbol("foo"): "bar"}) self.c1.condition = cond self.c1.close() self.pump() assert self.c1.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED rcond = self.c2.remote_condition assert rcond == cond, (rcond, cond) def test_properties(self, p1=PropertyDict(key=symbol("value")), p2=None): self.c1.properties = p1 self.c2.properties = p2 self.c1.open() self.c2.open() self.pump() assert self.c2.remote_properties == p1, (self.c2.remote_properties, p1) assert self.c1.remote_properties == p2, (self.c1.remote_properties, p2) # The proton implementation limits channel_max to 32767. # If I set the application's limit lower than that, I should # get my wish. If I set it higher -- not. def test_channel_max_low(self, value=1234): self.c1.transport.channel_max = value self.c1.open() self.pump() assert self.c1.transport.channel_max == value, (self.c1.transport.channel_max, value) def test_channel_max_high(self, value=65535): self.c1.transport.channel_max = value self.c1.open() self.pump() assert self.c1.transport.channel_max == 32767, (self.c1.transport.channel_max, value) def test_channel_max_raise_and_lower(self): upper_limit = 32767 # It's OK to lower the max below upper_limit. self.c1.transport.channel_max = 12345 assert self.c1.transport.channel_max == 12345 # But it won't let us raise the limit above PN_IMPL_CHANNEL_MAX. self.c1.transport.channel_max = 65535 assert self.c1.transport.channel_max == upper_limit # send the OPEN frame self.c1.open() self.pump() # Now it's too late to make any change, because # we have already sent the OPEN frame. try: self.c1.transport.channel_max = 666 assert False, "expected session exception" except: pass assert self.c1.transport.channel_max == upper_limit # TODO: Currently failing test - PROTON-1759 - skip def test_channel_max_limits_sessions(self): raise Skipped('Test fails - PROTON-1759') # This is an index -- so max number of channels should be 1. self.c1.transport.channel_max = 0 self.c1.open() self.c2.open() ssn_0 = self.c2.session() assert ssn_0 is not None ssn_0.open() self.pump() try: ssn_1 = self.c2.session() ssn_1.open() self.pump() assert False, "expected session exception" except SessionException: pass def test_cleanup(self): self.c1.open() self.c2.open() self.pump() assert self.c1.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE t1 = self.c1.transport t2 = self.c2.transport c2 = self.c2 self.c1.close() # release all references to C1, except that held by the transport self.cleanup() gc.collect() # transport should flush last state from C1: pump(t1, t2) assert c2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED def test_user_config(self): self.c1.user = "vindaloo" self.c1.password = "secret" self.c1.open() self.pump() self.c2.user = "leela" self.c2.password = "trustno1" self.c2.open() self.pump() assert self.c1.user == "vindaloo", self.c1.user assert self.c1.password is None, self.c1.password assert self.c2.user == "leela", self.c2.user assert self.c2.password is None, self.c2.password class SessionTest(Test): def setUp(self): gc.enable() self.c1, self.c2 = self.connection() self.ssn = self.c1.session() self.c1.open() self.c2.open() def cleanup(self): # release resources created by this class super(SessionTest, self).cleanup() self.c1 = None self.c2 = None self.ssn = None def tearDown(self): self.cleanup() gc.collect() assert not gc.garbage def test_open_close(self): assert self.ssn.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.ssn.open() assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT self.pump() assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT ssn = self.c2.session_head(Endpoint.REMOTE_ACTIVE | Endpoint.LOCAL_UNINIT) assert ssn is not None assert ssn.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT ssn.open() assert ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT self.pump() assert ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE ssn.close() assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.pump() assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED self.ssn.close() assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED self.pump() assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert self.ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_simultaneous_close(self): self.ssn.open() self.pump() ssn = self.c2.session_head(Endpoint.REMOTE_ACTIVE | Endpoint.LOCAL_UNINIT) assert ssn is not None ssn.open() self.pump() assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.ssn.close() ssn.close() assert self.ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE self.pump() assert self.ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_closing_connection(self): self.ssn.open() self.pump() self.c1.close() self.pump() self.ssn.close() self.pump() def test_condition(self): self.ssn.open() self.pump() ssn = self.c2.session_head(Endpoint.REMOTE_ACTIVE | Endpoint.LOCAL_UNINIT) assert ssn is not None ssn.open() self.pump() assert self.ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE cond = Condition("blah:bleh", "this is a description", {symbol("foo"): "bar"}) self.ssn.condition = cond self.ssn.close() self.pump() assert self.ssn.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED rcond = ssn.remote_condition assert rcond == cond, (rcond, cond) def test_cleanup(self): snd, rcv = self.link("test-link") snd.open() rcv.open() self.pump() snd_ssn = snd.session rcv_ssn = rcv.session assert rcv_ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.ssn = None snd_ssn.close() snd_ssn.free() del snd_ssn gc.collect() self.pump() assert rcv_ssn.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED def test_reopen_on_same_session_without_free(self): """ confirm that a link is correctly opened when attaching to a previously closed link *that has not been freed yet* on the same session """ self.ssn.open() self.pump() ssn2 = self.c2.session_head(Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE) ssn2.open() self.pump() snd = self.ssn.sender("test-link") rcv = ssn2.receiver("test-link") assert snd.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT snd.open() rcv.open() self.pump() assert snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE snd.close() rcv.close() self.pump() assert snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert rcv.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED snd = self.ssn.sender("test-link") rcv = ssn2.receiver("test-link") assert snd.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT snd.open() rcv.open() self.pump() assert snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE def test_set_get_outgoing_window(self): assert self.ssn.outgoing_window == 2147483647 self.ssn.outgoing_window = 1024 assert self.ssn.outgoing_window == 1024 class LinkTest(Test): def setUp(self): gc.enable() self.snd, self.rcv = self.link("test-link") def cleanup(self): # release resources created by this class super(LinkTest, self).cleanup() self.snd = None self.rcv = None def tearDown(self): self.cleanup() gc.collect() assert not gc.garbage, gc.garbage def test_open_close(self): assert self.snd.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.snd.open() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.pump() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE self.rcv.open() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.pump() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.snd.close() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.pump() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED self.rcv.close() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED self.pump() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert self.rcv.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_simultaneous_open_close(self): assert self.snd.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_UNINIT self.snd.open() self.rcv.open() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_UNINIT self.pump() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.snd.close() self.rcv.close() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE self.pump() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED assert self.rcv.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_CLOSED def test_multiple(self): rcv = self.snd.session.receiver("second-rcv") assert rcv.name == "second-rcv" self.snd.open() rcv.open() self.pump() c2 = self.rcv.session.connection l = c2.link_head(Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE) while l: l.open() l = l.next(Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE) self.pump() assert self.snd assert rcv self.snd.close() rcv.close() ssn = rcv.session conn = ssn.connection ssn.close() conn.close() self.pump() def test_closing_session(self): self.snd.open() self.rcv.open() ssn1 = self.snd.session self.pump() ssn1.close() self.pump() self.snd.close() self.pump() def test_closing_connection(self): self.snd.open() self.rcv.open() ssn1 = self.snd.session c1 = ssn1.connection self.pump() c1.close() self.pump() self.snd.close() self.pump() def assertEqualTermini(self, t1, t2): assert t1.type == t2.type, (t1.type, t2.type) assert t1.address == t2.address, (t1.address, t2.address) assert t1.durability == t2.durability, (t1.durability, t2.durability) assert t1.expiry_policy == t2.expiry_policy, (t1.expiry_policy, t2.expiry_policy) assert t1.timeout == t2.timeout, (t1.timeout, t2.timeout) assert t1.dynamic == t2.dynamic, (t1.dynamic, t2.dynamic) for attr in ["properties", "capabilities", "outcomes", "filter"]: d1 = getattr(t1, attr) d2 = getattr(t2, attr) assert d1.format() == d2.format(), (attr, d1.format(), d2.format()) def _test_source_target(self, config_source, config_target): if config_source is None: self.snd.source.type = Terminus.UNSPECIFIED else: config_source(self.snd.source) if config_target is None: self.snd.target.type = Terminus.UNSPECIFIED else: config_target(self.snd.target) self.snd.open() self.pump() self.assertEqualTermini(self.rcv.remote_source, self.snd.source) self.assertEqualTermini(self.rcv.remote_target, self.snd.target) self.rcv.target.copy(self.rcv.remote_target) self.rcv.source.copy(self.rcv.remote_source) self.rcv.open() self.pump() self.assertEqualTermini(self.snd.remote_target, self.snd.target) self.assertEqualTermini(self.snd.remote_source, self.snd.source) def test_source_target(self): self._test_source_target(TerminusConfig(address="source"), TerminusConfig(address="target")) def test_source(self): self._test_source_target(TerminusConfig(address="source"), None) def test_target(self): self._test_source_target(None, TerminusConfig(address="target")) def test_coordinator(self): caps = Array(UNDESCRIBED, Data.SYMBOL, symbol("amqp:local-transactions")) self._test_source_target(None, TerminusConfig(type=Terminus.COORDINATOR, capabilities=caps)) def test_source_target_full(self): self._test_source_target(TerminusConfig(address="source", timeout=3, dist_mode=Terminus.DIST_MODE_MOVE, filter=[("int", 1), ("symbol", "two"), ("string", "three")], capabilities=["one", "two", "three"]), TerminusConfig(address="source", timeout=7, capabilities=None)) def test_distribution_mode(self): self._test_source_target(TerminusConfig(address="source", dist_mode=Terminus.DIST_MODE_COPY), TerminusConfig(address="target")) assert self.rcv.remote_source.distribution_mode == Terminus.DIST_MODE_COPY assert self.rcv.remote_target.distribution_mode == Terminus.DIST_MODE_UNSPECIFIED def test_dynamic_link(self): self._test_source_target(TerminusConfig(address=None, dynamic=True), None) assert self.rcv.remote_source.dynamic assert self.rcv.remote_source.address is None def test_condition(self): self.snd.open() self.rcv.open() self.pump() assert self.snd.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE cond = Condition("blah:bleh", "this is a description", {symbol("foo"): "bar"}) self.snd.condition = cond self.snd.close() self.pump() assert self.snd.state == Endpoint.LOCAL_CLOSED | Endpoint.REMOTE_ACTIVE assert self.rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED rcond = self.rcv.remote_condition assert rcond == cond, (rcond, cond) def test_settle_mode(self): self.snd.snd_settle_mode = Link.SND_UNSETTLED assert self.snd.snd_settle_mode == Link.SND_UNSETTLED self.rcv.rcv_settle_mode = Link.RCV_SECOND assert self.rcv.rcv_settle_mode == Link.RCV_SECOND assert self.snd.remote_rcv_settle_mode != Link.RCV_SECOND assert self.rcv.remote_snd_settle_mode != Link.SND_UNSETTLED self.snd.open() self.rcv.open() self.pump() assert self.snd.remote_rcv_settle_mode == Link.RCV_SECOND assert self.rcv.remote_snd_settle_mode == Link.SND_UNSETTLED def test_max_message_size(self): assert self.snd.max_message_size == 0 assert self.rcv.remote_max_message_size == 0 self.snd.max_message_size = 13579 self.snd.open() self.rcv.open() self.pump() assert self.rcv.remote_max_message_size == 13579 def test_properties(self): sender_props = {symbol('key1'): 'value1', symbol('key2'): 'value2'} self.snd.properties = sender_props self.snd.open() self.rcv.open() self.pump() assert self.rcv.remote_properties == sender_props, (self.rcv.remote_properties, sender_props) assert self.snd.remote_properties is None, (self.snd.remote_properties, None) def test_cleanup(self): snd, rcv = self.link("test-link") snd.open() rcv.open() self.pump() assert rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE snd.close() snd.free() del snd gc.collect() self.pump() assert rcv.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED class TerminusConfig: def __init__(self, type=None, address=None, timeout=None, durability=None, filter=None, capabilities=None, dynamic=False, dist_mode=None): self.address = address self.timeout = timeout self.durability = durability self.filter = filter self.capabilities = capabilities self.dynamic = dynamic self.dist_mode = dist_mode self.type = type def __call__(self, terminus): if self.type is not None: terminus.type = self.type if self.address is not None: terminus.address = self.address if self.timeout is not None: terminus.timeout = self.timeout if self.durability is not None: terminus.durability = self.durability if self.capabilities is not None: terminus.capabilities.put_array(False, Data.SYMBOL) terminus.capabilities.enter() for c in self.capabilities: terminus.capabilities.put_symbol(c) if self.filter is not None: terminus.filter.put_map() terminus.filter.enter() for (t, v) in self.filter: setter = getattr(terminus.filter, "put_%s" % t) setter(v) if self.dynamic: terminus.dynamic = True if self.dist_mode is not None: terminus.distribution_mode = self.dist_mode class TransferTest(Test): def setUp(self): gc.enable() self.snd, self.rcv = self.link("test-link") self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() def cleanup(self): # release resources created by this class super(TransferTest, self).cleanup() self.c1 = None self.c2 = None self.snd = None self.rcv = None def tearDown(self): self.cleanup() gc.collect() assert not gc.garbage def test_work_queue(self): assert self.c1.work_head is None self.snd.delivery("tag") assert self.c1.work_head is None self.rcv.flow(1) self.pump() d = self.c1.work_head assert d is not None tag = d.tag assert tag == "tag", tag assert d.writable n = self.snd.send(b"this is a test") assert self.snd.advance() assert self.c1.work_head is None self.pump() d = self.c2.work_head assert d.tag == "tag" assert d.readable def test_multiframe(self): self.rcv.flow(1) self.snd.delivery("tag") msg = b"this is a test" n = self.snd.send(msg) assert n == len(msg) self.pump() d = self.rcv.current assert d assert d.tag == "tag", repr(d.tag) assert d.readable binary = self.rcv.recv(1024) assert binary == msg, (binary, msg) binary = self.rcv.recv(1024) assert binary == b"" msg = b"this is more" n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() binary = self.rcv.recv(1024) assert binary == msg, (binary, msg) binary = self.rcv.recv(1024) assert binary is None def test_multiframe_abort(self): self.rcv.flow(1) sd = self.snd.delivery("tag") msg = b"this is a test" n = self.snd.send(msg) assert n == len(msg) self.pump() binary = self.rcv.recv(1024) assert binary == msg, (binary, msg) msg = b"this is more. Error if not discarded." n = self.snd.send(msg) assert n == len(msg) sd.abort() assert sd.aborted # Confirm abort discards the sender's buffered content, i.e. no data in generated transfer frame. # We want: # @transfer(20) [handle=0, delivery-id=0, delivery-tag=b"tag", message-format=0, settled=true, aborted=true] # wanted = b"\x00\x00\x00%\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x15\x00\x00\x00\nR\x00R\x00\xa0\x03tagR\x00A@@@@A" # wanted = b"\x00\x00\x00\x26\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x16\x00\x00\x00\x0bR\x00R\x00\xa0\x03tagR\x00A@@@@A@" # wanted = b'\x00\x00\x00\x20\x02\x00\x00\x00\x00S\x14\xc0\x13\x0bR\x00R\x00\xa0\x03tagR\x00A@@@@A@' # wanted = b'\x00\x00\x00"\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x12\x00\x00\x00\nCC\xa0\x03tagCA@@@@A' # wanted = b'\x00\x00\x00\x1d\x02\x00\x00\x00\x00S\x14\xc0\x10\x0bCC\xa0\x03tagCA@@@@A@' wanted = b'\x00\x00\x00\x1c\x02\x00\x00\x00\x00S\x14\xc0\x0f\x0aCC\xa0\x03tagCA@@@@A' t = self.snd.transport wire_bytes = t.peek(1024) assert wanted == wire_bytes, wire_bytes self.pump() assert self.rcv.current.aborted def test_multiframe_last_empty(self): self.rcv.flow(1) sd = self.snd.delivery("tag") msg_p1 = b"this is a test" n = self.snd.send(msg_p1) assert n == len(msg_p1) self.pump() assert len(msg_p1) == self.rcv.current.pending assert self.rcv.current.partial msg_p2 = b"this is more." n = self.snd.send(msg_p2) assert n == len(msg_p2) self.pump() msg = msg_p1 + msg_p2 assert len(msg) == self.rcv.current.pending assert self.rcv.current.partial # Advance. Should send empty xfer frame with more flag false. assert self.snd.advance() self.pump() assert len(msg) == self.rcv.current.pending assert not self.rcv.current.partial binary = self.rcv.recv(self.rcv.current.pending) assert binary == msg def test_disposition(self): self.rcv.flow(1) self.pump() sd = self.snd.delivery("tag") msg = b"this is a test" n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() rd = self.rcv.current assert rd is not None assert rd.tag == sd.tag rmsg = self.rcv.recv(1024) assert rmsg == msg rd.update(Delivery.ACCEPTED) self.pump() rdisp = sd.remote_state ldisp = rd.local_state assert rdisp == ldisp == Delivery.ACCEPTED, (rdisp, ldisp) assert sd.updated sd.update(Delivery.ACCEPTED) self.pump() assert sd.local_state == rd.remote_state == Delivery.ACCEPTED sd.settle() def test_delivery_id_ordering(self): self.rcv.flow(1024) self.pump(buffer_size=64 * 1024) # fill up delivery buffer on sender for m in range(1024): sd = self.snd.delivery("tag%s" % m) msg = ("message %s" % m).encode('ascii') n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump(buffer_size=64 * 1024) # receive a session-windows worth of messages and accept them for m in range(1024): rd = self.rcv.current assert rd is not None, m assert rd.tag == ("tag%s" % m), (rd.tag, m) msg = self.rcv.recv(1024) assert msg == ("message %s" % m).encode('ascii'), (msg, m) rd.update(Delivery.ACCEPTED) rd.settle() self.pump(buffer_size=64 * 1024) # add some new deliveries for m in range(1024, 1450): sd = self.snd.delivery("tag%s" % m) msg = ("message %s" % m).encode('ascii') n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() # handle all disposition changes to sent messages d = self.c1.work_head while d: next_d = d.work_next if d.updated: d.update(Delivery.ACCEPTED) d.settle() d = next_d # submit some more deliveries for m in range(1450, 1500): sd = self.snd.delivery("tag%s" % m) msg = ("message %s" % m).encode('ascii') n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump(buffer_size=64 * 1024) self.rcv.flow(1024) self.pump(buffer_size=64 * 1024) # verify remaining messages can be received and accepted for m in range(1024, 1500): rd = self.rcv.current assert rd is not None, m assert rd.tag == ("tag%s" % m), (rd.tag, m) msg = self.rcv.recv(1024) assert msg == ("message %s" % m).encode('ascii'), (msg, m) rd.update(Delivery.ACCEPTED) rd.settle() def test_cleanup(self): self.rcv.flow(10) self.pump() for x in range(10): self.snd.delivery("tag%d" % x) msg = b"this is a test" n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.snd.close() self.snd.free() self.snd = None gc.collect() self.pump() for x in range(10): rd = self.rcv.current assert rd is not None assert rd.tag == "tag%d" % x rmsg = self.rcv.recv(1024) assert self.rcv.advance() assert rmsg == msg # close of snd should've settled: assert rd.settled rd.settle() class MaxFrameTransferTest(Test): def setUp(self): pass def cleanup(self): # release resources created by this class super(MaxFrameTransferTest, self).cleanup() self.c1 = None self.c2 = None self.snd = None self.rcv = None def tearDown(self): self.cleanup() def message(self, size): parts = [] for i in range(size): parts.append(str(i)) return "/".join(parts)[:size].encode("utf-8") def testMinFrame(self): """ Configure receiver to support minimum max-frame as defined by AMQP-1.0. Verify transfer of messages larger than 512. """ self.snd, self.rcv = self.link("test-link", max_frame=[0, 512]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() assert self.rcv.session.connection.transport.max_frame_size == 512 assert self.snd.session.connection.transport.remote_max_frame_size == 512 self.rcv.flow(1) self.snd.delivery("tag") msg = self.message(513) n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() binary = self.rcv.recv(513) assert binary == msg binary = self.rcv.recv(1024) assert binary is None def testOddFrame(self): """ Test an odd sized max limit with data that will require multiple frames to be transfered. """ self.snd, self.rcv = self.link("test-link", max_frame=[0, 521]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() assert self.rcv.session.connection.transport.max_frame_size == 521 assert self.snd.session.connection.transport.remote_max_frame_size == 521 self.rcv.flow(2) self.snd.delivery("tag") msg = ("X" * 1699).encode('utf-8') n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() binary = self.rcv.recv(1699) assert binary == msg binary = self.rcv.recv(1024) assert binary is None self.rcv.advance() self.snd.delivery("gat") msg = self.message(1426) n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() binary = self.rcv.recv(1426) assert binary == msg self.pump() binary = self.rcv.recv(1024) assert binary is None def testSendQueuedMultiFrameMessages(self, sendSingleFrameMsg=False): """ Test that multiple queued messages on the same link with multi-frame content are sent correctly. Use an odd max frame size, send enough data to use many. """ self.snd, self.rcv = self.link("test-link", max_frame=[0, 517]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() assert self.rcv.session.connection.transport.max_frame_size == 517 assert self.snd.session.connection.transport.remote_max_frame_size == 517 self.rcv.flow(5) self.pump() # Send a delivery with 5 frames, all bytes as X1234 self.snd.delivery("tag") msg = ("X1234" * 425).encode('utf-8') assert 2125 == len(msg) n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() # Send a delivery with 5 frames, all bytes as Y5678 self.snd.delivery("tag2") msg2 = ("Y5678" * 425).encode('utf-8') assert 2125 == len(msg2) n = self.snd.send(msg2) assert n == len(msg2) assert self.snd.advance() self.pump() if sendSingleFrameMsg: # Send a delivery with 1 frame self.snd.delivery("tag3") msg3 = ("Z").encode('utf-8') assert 1 == len(msg3) n = self.snd.send(msg3) assert n == len(msg3) assert self.snd.advance() self.pump() binary = self.rcv.recv(5000) self.assertEqual(binary, msg) self.rcv.advance() binary2 = self.rcv.recv(5000) self.assertEqual(binary2, msg2) self.rcv.advance() if sendSingleFrameMsg: binary3 = self.rcv.recv(5000) self.assertEqual(binary3, msg3) self.rcv.advance() self.pump() def testSendQueuedMultiFrameMessagesThenSingleFrameMessage(self): self.testSendQueuedMultiFrameMessages(sendSingleFrameMsg=True) def testBigMessage(self): """ Test transferring a big message. """ self.snd, self.rcv = self.link("test-link") self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() self.rcv.flow(2) self.snd.delivery("tag") msg = self.message(1024 * 256) n = self.snd.send(msg) assert n == len(msg) assert self.snd.advance() self.pump() binary = self.rcv.recv(1024 * 256) assert binary == msg binary = self.rcv.recv(1024) assert binary is None def testMaxFrameAbort(self): self.snd, self.rcv = self.link("test-link", max_frame=[0, 512]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.rcv.flow(1) sndt = self.snd.transport dblfrmbytes = ("0123456789" * 52).encode('utf-8') assert 520 == len(dblfrmbytes) self.pump() # Part 1: abort a delivery that would generate two frames - before they are generated. # Expect that no output is generated and credit is unaffected. assert self.snd.credit == 1 sd = self.snd.delivery("tag_0") n = self.snd.send(dblfrmbytes) assert n == len(dblfrmbytes) sd.abort() generated = sndt.peek(2048) assert generated == b"" assert self.snd.credit == 1 # Part 2: abort a streaming delivery whose last content would generate two frames. # First send some un-aborted data across the wire. sd = self.snd.delivery("tag_1") n = self.snd.send(dblfrmbytes) assert n == len(dblfrmbytes) self.pump() binary = self.rcv.recv(2048) assert binary == dblfrmbytes, (binary, dblfrmbytes) # Now send more data spanning two frames and immediately abort. # Unlike part 1, an abort frame is required to sync with peer. n = self.snd.send(dblfrmbytes) assert n == len(dblfrmbytes) sd.abort() assert sd.aborted # Expect a single abort transfer frame with no content. One credit is consumed. # @transfer(20) [handle=0, delivery-id=0, delivery-tag=b"tag_1", message-format=0, settled=true, aborted=true] # wanted = b"\x00\x00\x00\x27\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x17\x00\x00\x00\nR\x00R\x00\xa0\x05tag_1R\x00A@@@@A" # wanted = b"\x00\x00\x00\x28\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x18\x00\x00\x00\x0bR\x00R\x00\xa0\x05tag_1R\x00A@@@@A@" # wanted = b'\x00\x00\x00\x22\x02\x00\x00\x00\x00S\x14\xc0\x15\x0bR\x00R\x00\xa0\x05tag_1R\x00A@@@@A@' # wanted = b'\x00\x00\x00\x24\x02\x00\x00\x00\x00S\x14\xd0\x00\x00\x00\x14\x00\x00\x00\nCC\xa0\x05tag_1CA@@@@A' # wanted = b'\x00\x00\x00\x1f\x02\x00\x00\x00\x00S\x14\xc0\x12\x0bCC\xa0\x05tag_1CA@@@@A@' wanted = b'\x00\x00\x00\x1e\x02\x00\x00\x00\x00S\x14\xc0\x11\x0aCC\xa0\x05tag_1CA@@@@A' t = self.snd.transport wire_bytes = t.peek(2048) assert wanted == wire_bytes, wire_bytes assert self.snd.credit == 0 self.pump() assert self.rcv.current.aborted # Confirm no lingering transfers by closing the link. self.snd.close() # Expect just the detach frame. # @detach(22) [handle=0, closed=true] # wanted = b"\x00\x00\x00\x17\x02\x00\x00\x00\x00S\x16\xd0\x00\x00\x00\x07\x00\x00\x00\x02R\x00A" # wanted = b"\x00\x00\x00\x18\x02\x00\x00\x00\x00S\x16\xd0\x00\x00\x00\x08\x00\x00\x00\x03R\x00A@" # wanted = b'\x00\x00\x00\x12\x02\x00\x00\x00\x00S\x16\xc0\x05\x03R\x00A@' # wanted = b"\x00\x00\x00\x16\x02\x00\x00\x00\x00S\x16\xd0\x00\x00\x00\x06\x00\x00\x00\x02CA" # wanted = b'\x00\x00\x00\x11\x02\x00\x00\x00\x00S\x16\xc0\x04\x03CA@' wanted = b'\x00\x00\x00\x10\x02\x00\x00\x00\x00S\x16\xc0\x03\x02CA' wire_bytes = t.peek(2048) assert wanted == wire_bytes, wire_bytes class IdleTimeoutTest(Test): def setUp(self): pass def cleanup(self): # release resources created by this class super(IdleTimeoutTest, self).cleanup() self.snd = None self.rcv = None self.c1 = None self.c2 = None def tearDown(self): self.cleanup() def message(self, size): parts = [] for i in range(size): parts.append(str(i)) return "/".join(parts)[:size] def testGetSet(self): """ Verify the configuration and negotiation of the idle timeout. """ self.snd, self.rcv = self.link("test-link", idle_timeout=[1.0, 2.0]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() # proton advertises 1/2 the configured timeout to the peer: assert self.rcv.session.connection.transport.idle_timeout == 2.0 assert self.rcv.session.connection.transport.remote_idle_timeout == 0.5 assert self.snd.session.connection.transport.idle_timeout == 1.0 assert self.snd.session.connection.transport.remote_idle_timeout == 1.0 def testTimeout(self): """ Verify the AMQP Connection idle timeout. """ # snd will timeout the Connection if no frame is received within 1000 ticks self.snd, self.rcv = self.link("test-link", idle_timeout=[1.0, 0]) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() t_snd = self.snd.session.connection.transport t_rcv = self.rcv.session.connection.transport assert t_rcv.idle_timeout == 0.0 # proton advertises 1/2 the timeout (see spec) assert t_rcv.remote_idle_timeout == 0.5 assert t_snd.idle_timeout == 1.0 assert t_snd.remote_idle_timeout == 0.0 sndr_frames_in = t_snd.frames_input rcvr_frames_out = t_rcv.frames_output # at t+1msec, nothing should happen: clock = 0.001 assert t_snd.tick(clock) == 1.001, "deadline for remote timeout" assert t_rcv.tick(clock) == 0.251, "deadline to send keepalive" self.pump() assert sndr_frames_in == t_snd.frames_input, "unexpected received frame" # at one tick from expected idle frame send, nothing should happen: clock = 0.250 assert t_snd.tick(clock) == 1.001, "deadline for remote timeout" assert t_rcv.tick(clock) == 0.251, "deadline to send keepalive" self.pump() assert sndr_frames_in == t_snd.frames_input, "unexpected received frame" # this should cause rcvr to expire and send a keepalive clock = 0.251 assert t_snd.tick(clock) == 1.001, "deadline for remote timeout" assert t_rcv.tick(clock) == 0.501, "deadline to send keepalive" self.pump() sndr_frames_in += 1 rcvr_frames_out += 1 assert sndr_frames_in == t_snd.frames_input, "unexpected received frame" assert rcvr_frames_out == t_rcv.frames_output, "unexpected frame" # since a keepalive was received, sndr will rebase its clock against this tick: # and the receiver should not change its deadline clock = 0.498 assert t_snd.tick(clock) == 1.498, "deadline for remote timeout" assert t_rcv.tick(clock) == 0.501, "deadline to send keepalive" self.pump() assert sndr_frames_in == t_snd.frames_input, "unexpected received frame" # now expire sndr clock = 1.499 t_snd.tick(clock) self.pump() assert self.c2.state & Endpoint.REMOTE_CLOSED assert self.c2.remote_condition.name == "amqp:resource-limit-exceeded" class CreditTest(Test): def setUp(self): self.snd, self.rcv = self.link("test-link", max_frame=(16 * 1024, 16 * 1024)) self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() def cleanup(self): # release resources created by this class super(CreditTest, self).cleanup() self.c1 = None self.snd = None self.c2 = None self.rcv2 = None self.snd2 = None def tearDown(self): self.cleanup() def testCreditSender(self, count=1024): credit = self.snd.credit assert credit == 0, credit self.rcv.flow(10) self.pump() credit = self.snd.credit assert credit == 10, credit self.rcv.flow(count) self.pump() credit = self.snd.credit assert credit == 10 + count, credit def testCreditReceiver(self): self.rcv.flow(10) self.pump() assert self.rcv.credit == 10, self.rcv.credit d = self.snd.delivery("tag") assert d assert self.snd.advance() self.pump() assert self.rcv.credit == 10, self.rcv.credit assert self.rcv.queued == 1, self.rcv.queued c = self.rcv.current assert c.tag == "tag", c.tag assert self.rcv.advance() assert self.rcv.credit == 9, self.rcv.credit assert self.rcv.queued == 0, self.rcv.queued def _testBufferingOnClose(self, a, b): for i in range(10): d = self.snd.delivery("tag-%s" % i) assert d d.settle() self.pump() assert self.snd.queued == 10 endpoints = {"connection": (self.c1, self.c2), "session": (self.snd.session, self.rcv.session), "link": (self.snd, self.rcv)} local_a, remote_a = endpoints[a] local_b, remote_b = endpoints[b] remote_b.close() self.pump() assert local_b.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED local_a.close() self.pump() assert remote_a.state & Endpoint.REMOTE_CLOSED assert self.snd.queued == 10 def testBufferingOnCloseLinkLink(self): self._testBufferingOnClose("link", "link") def testBufferingOnCloseLinkSession(self): self._testBufferingOnClose("link", "session") def testBufferingOnCloseLinkConnection(self): self._testBufferingOnClose("link", "connection") def testBufferingOnCloseSessionLink(self): self._testBufferingOnClose("session", "link") def testBufferingOnCloseSessionSession(self): self._testBufferingOnClose("session", "session") def testBufferingOnCloseSessionConnection(self): self._testBufferingOnClose("session", "connection") def testBufferingOnCloseConnectionLink(self): self._testBufferingOnClose("connection", "link") def testBufferingOnCloseConnectionSession(self): self._testBufferingOnClose("connection", "session") def testBufferingOnCloseConnectionConnection(self): self._testBufferingOnClose("connection", "connection") def testFullDrain(self): assert self.rcv.credit == 0 assert self.snd.credit == 0 self.rcv.drain(10) assert self.rcv.draining() assert self.rcv.credit == 10 assert self.snd.credit == 0 self.pump() assert self.rcv.credit == 10 assert self.snd.credit == 10 assert self.rcv.draining() self.snd.drained() assert self.rcv.credit == 10 assert self.snd.credit == 0 assert self.rcv.draining() self.pump() assert self.rcv.credit == 0 assert self.snd.credit == 0 assert not self.rcv.draining() drained = self.rcv.drained() assert drained == 10, drained def testPartialDrain(self): self.rcv.drain(2) assert self.rcv.draining() self.pump() d = self.snd.delivery("tag") assert d assert self.snd.advance() self.snd.drained() assert self.rcv.draining() self.pump() assert not self.rcv.draining() c = self.rcv.current assert self.rcv.queued == 1, self.rcv.queued assert c.tag == d.tag, c.tag assert self.rcv.advance() assert not self.rcv.current assert self.rcv.credit == 0, self.rcv.credit assert not self.rcv.draining() drained = self.rcv.drained() assert drained == 1, drained def testDrainFlow(self): assert self.rcv.credit == 0 assert self.snd.credit == 0 self.rcv.drain(10) assert self.rcv.credit == 10 assert self.snd.credit == 0 self.pump() assert self.rcv.credit == 10 assert self.snd.credit == 10 self.snd.drained() assert self.rcv.credit == 10 assert self.snd.credit == 0 self.pump() assert self.rcv.credit == 0 assert self.snd.credit == 0 self.rcv.flow(10) assert self.rcv.credit == 10 assert self.snd.credit == 0 self.pump() assert self.rcv.credit == 10 assert self.snd.credit == 10 self.snd.drained() assert self.rcv.credit == 10 assert self.snd.credit == 10 self.pump() assert self.rcv.credit == 10 assert self.snd.credit == 10 drained = self.rcv.drained() assert drained == 10, drained def testNegative(self): assert self.snd.credit == 0 d = self.snd.delivery("tag") assert d assert self.snd.advance() self.pump() assert self.rcv.credit == 0 assert self.rcv.queued == 0 self.rcv.flow(1) assert self.rcv.credit == 1 assert self.rcv.queued == 0 self.pump() assert self.rcv.credit == 1 assert self.rcv.queued == 1, self.rcv.queued c = self.rcv.current assert c assert c.tag == "tag" assert self.rcv.advance() assert self.rcv.credit == 0 assert self.rcv.queued == 0 def testDrainZero(self): assert self.snd.credit == 0 assert self.rcv.credit == 0 assert self.rcv.queued == 0 drained = self.rcv.drained() assert drained == 0 self.rcv.flow(10) self.pump() assert self.snd.credit == 10 assert self.rcv.credit == 10 assert self.rcv.queued == 0 self.snd.drained() self.pump() assert self.snd.credit == 10 assert self.rcv.credit == 10 assert self.rcv.queued == 0 drained = self.rcv.drained() assert drained == 0 self.rcv.drain(0) assert self.snd.credit == 10 assert self.rcv.credit == 10 assert self.rcv.queued == 0 self.pump() assert self.snd.credit == 10 assert self.rcv.credit == 10 assert self.rcv.queued == 0 self.snd.drained() assert self.snd.credit == 0 assert self.rcv.credit == 10 assert self.rcv.queued == 0 drained = self.rcv.drained() assert drained == 0 self.pump() assert self.snd.credit == 0 assert self.rcv.credit == 0 assert self.rcv.queued == 0 drained = self.rcv.drained() assert drained == 10 def testDrainOrder(self): """ Verify drain/drained works regardless of ordering. See PROTON-401 """ assert self.snd.credit == 0 assert self.rcv.credit == 0 assert self.rcv.queued == 0 # self.rcv.session.connection.transport.trace(Transport.TRACE_FRM) # self.snd.session.connection.transport.trace(Transport.TRACE_FRM) # verify that a sender that has reached the drain state will respond # promptly to a drain issued by the peer. self.rcv.flow(10) self.pump() assert self.snd.credit == 10, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit sd = self.snd.delivery("tagA") assert sd n = self.snd.send(b"A") assert n == 1 self.pump() self.snd.advance() # done sending, so signal that we are drained: self.snd.drained() self.pump() assert self.snd.credit == 9, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit self.rcv.drain(0) self.pump() assert self.snd.credit == 9, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit data = self.rcv.recv(10) assert data == b"A", data self.rcv.advance() self.pump() assert self.snd.credit == 9, self.snd.credit assert self.rcv.credit == 9, self.rcv.credit self.snd.drained() self.pump() assert self.snd.credit == 0, self.snd.credit assert self.rcv.credit == 0, self.rcv.credit # verify that a drain requested by the peer is not "acknowledged" until # after the sender has completed sending its pending messages self.rcv.flow(10) self.pump() assert self.snd.credit == 10, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit sd = self.snd.delivery("tagB") assert sd n = self.snd.send(b"B") assert n == 1 self.snd.advance() self.pump() assert self.snd.credit == 9, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit self.rcv.drain(0) self.pump() assert self.snd.credit == 9, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit sd = self.snd.delivery("tagC") assert sd n = self.snd.send(b"C") assert n == 1 self.snd.advance() self.pump() assert self.snd.credit == 8, self.snd.credit assert self.rcv.credit == 10, self.rcv.credit # now that the sender has finished sending everything, it can signal # drained self.snd.drained() self.pump() assert self.snd.credit == 0, self.snd.credit assert self.rcv.credit == 2, self.rcv.credit data = self.rcv.recv(10) assert data == b"B", data self.rcv.advance() data = self.rcv.recv(10) assert data == b"C", data self.rcv.advance() self.pump() assert self.snd.credit == 0, self.snd.credit assert self.rcv.credit == 0, self.rcv.credit def testPushback(self, count=10): assert self.snd.credit == 0 assert self.rcv.credit == 0 self.rcv.flow(count) self.pump() for i in range(count): d = self.snd.delivery("tag%s" % i) assert d self.snd.advance() assert self.snd.queued == count assert self.rcv.queued == 0 self.pump() assert self.snd.queued == 0 assert self.rcv.queued == count d = self.snd.delivery("extra") self.snd.advance() assert self.snd.queued == 1 assert self.rcv.queued == count self.pump() assert self.snd.queued == 1 assert self.rcv.queued == count def testHeadOfLineBlocking(self): self.snd2 = self.snd.session.sender("link-2") self.rcv2 = self.rcv.session.receiver("link-2") self.snd2.open() self.rcv2.open() self.pump() assert self.snd2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE assert self.rcv2.state == Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE self.rcv.flow(5) self.rcv2.flow(10) self.pump() assert self.snd.credit == 5 assert self.snd2.credit == 10 for i in range(10): tag = "test %d" % i self.snd.delivery(tag) self.snd.send(tag.encode("ascii")) assert self.snd.advance() self.snd2.delivery(tag) self.snd2.send(tag.encode("ascii")) assert self.snd2.advance() self.pump() for i in range(5): b = self.rcv.recv(512) assert self.rcv.advance() b = self.rcv2.recv(512) assert self.rcv2.advance() for i in range(5): b = self.rcv2.recv(512) assert self.rcv2.advance() class SessionCreditTest(Test): def tearDown(self): self.cleanup() def testBuffering(self, count=32, size=1024, capacity=16 * 1024, max_frame=1024): snd, rcv = self.link("test-link", max_frame=(max_frame, max_frame)) rcv.session.incoming_capacity = capacity snd.open() rcv.open() rcv.flow(count) self.pump() assert count > 0 total_bytes = count * size assert snd.session.outgoing_bytes == 0, snd.session.outgoing_bytes assert rcv.session.incoming_bytes == 0, rcv.session.incoming_bytes assert snd.queued == 0, snd.queued assert rcv.queued == 0, rcv.queued data = bytes(bytearray(size)) idx = 0 while snd.credit: d = snd.delivery("tag%s" % idx) assert d n = snd.send(data) assert n == size, (n, size) assert snd.advance() self.pump() idx += 1 assert idx == count, (idx, count) assert snd.session.outgoing_bytes < total_bytes, (snd.session.outgoing_bytes, total_bytes) assert rcv.session.incoming_bytes < capacity, (rcv.session.incoming_bytes, capacity) assert snd.session.outgoing_bytes + rcv.session.incoming_bytes == total_bytes, \ (snd.session.outgoing_bytes, rcv.session.incoming_bytes, total_bytes) if snd.session.outgoing_bytes > 0: available = rcv.session.incoming_capacity - rcv.session.incoming_bytes assert available < max_frame, (available, max_frame) for i in range(count): d = rcv.current assert d, i pending = d.pending before = rcv.session.incoming_bytes assert rcv.advance() after = rcv.session.incoming_bytes assert before - after == pending, (before, after, pending) snd_before = snd.session.incoming_bytes self.pump() snd_after = snd.session.incoming_bytes assert rcv.session.incoming_bytes < capacity if snd_before > 0: assert capacity - after <= max_frame assert snd_before > snd_after if snd_after > 0: available = rcv.session.incoming_capacity - rcv.session.incoming_bytes assert available < max_frame, available def testBufferingSize16(self): self.testBuffering(size=16) def testBufferingSize256(self): self.testBuffering(size=256) def testBufferingSize512(self): self.testBuffering(size=512) def testBufferingSize2048(self): self.testBuffering(size=2048) def testBufferingSize1025(self): self.testBuffering(size=1025) def testBufferingSize1023(self): self.testBuffering(size=1023) def testBufferingSize989(self): self.testBuffering(size=989) def testBufferingSize1059(self): self.testBuffering(size=1059) def testCreditWithBuffering(self): snd, rcv = self.link("test-link", max_frame=(1024, 1024)) rcv.session.incoming_capacity = 64 * 1024 snd.open() rcv.open() rcv.flow(128) self.pump() assert snd.credit == 128, snd.credit assert rcv.queued == 0, rcv.queued idx = 0 while snd.credit: d = snd.delivery("tag%s" % idx) snd.send(("x" * 1024).encode('ascii')) assert d assert snd.advance() self.pump() idx += 1 assert idx == 128, idx assert rcv.queued < 128, rcv.queued rcv.flow(1) self.pump() assert snd.credit == 1, snd.credit class SettlementTest(Test): def setUp(self): self.snd, self.rcv = self.link("test-link") self.c1 = self.snd.session.connection self.c2 = self.rcv.session.connection self.snd.open() self.rcv.open() self.pump() def cleanup(self): # release resources created by this class super(SettlementTest, self).cleanup() self.c1 = None self.snd = None self.c2 = None self.rcv2 = None self.snd2 = None def tearDown(self): self.cleanup() def testSettleCurrent(self): self.rcv.flow(10) self.pump() assert self.snd.credit == 10, self.snd.credit d = self.snd.delivery("tag") e = self.snd.delivery("tag2") assert d assert e c = self.snd.current assert c.tag == "tag", c.tag c.settle() c = self.snd.current assert c.tag == "tag2", c.tag c.settle() c = self.snd.current assert not c self.pump() c = self.rcv.current assert c assert c.tag == "tag", c.tag assert c.settled c.settle() c = self.rcv.current assert c assert c.tag == "tag2", c.tag assert c.settled c.settle() c = self.rcv.current assert not c def testUnsettled(self): self.rcv.flow(10) self.pump() assert self.snd.unsettled == 0, self.snd.unsettled assert self.rcv.unsettled == 0, self.rcv.unsettled d = self.snd.delivery("tag") assert d assert self.snd.unsettled == 1, self.snd.unsettled assert self.rcv.unsettled == 0, self.rcv.unsettled assert self.snd.advance() self.pump() assert self.snd.unsettled == 1, self.snd.unsettled assert self.rcv.unsettled == 1, self.rcv.unsettled c = self.rcv.current assert c c.settle() assert self.snd.unsettled == 1, self.snd.unsettled assert self.rcv.unsettled == 0, self.rcv.unsettled def testMultipleUnsettled(self, count=1024, size=1024): self.rcv.flow(count) self.pump() assert self.snd.unsettled == 0, self.snd.unsettled assert self.rcv.unsettled == 0, self.rcv.unsettled unsettled = [] for i in range(count): sd = self.snd.delivery("tag%s" % i) assert sd n = self.snd.send(("x" * size).encode('ascii')) assert n == size, n assert self.snd.advance() self.pump() rd = self.rcv.current assert rd, "did not receive delivery %s" % i n = rd.pending b = self.rcv.recv(n) assert len(b) == n, (b, n) rd.update(Delivery.ACCEPTED) assert self.rcv.advance() self.pump() unsettled.append(rd) assert self.rcv.unsettled == count for rd in unsettled: rd.settle() def testMultipleUnsettled2K1K(self): self.testMultipleUnsettled(2048, 1024) def testMultipleUnsettled4K1K(self): self.testMultipleUnsettled(4096, 1024) def testMultipleUnsettled1K2K(self): self.testMultipleUnsettled(1024, 2048) def testMultipleUnsettled2K2K(self): self.testMultipleUnsettled(2048, 2048) def testMultipleUnsettled4K2K(self): self.testMultipleUnsettled(4096, 2048) class PipelineTest(Test): def setUp(self): self.c1, self.c2 = self.connection() def cleanup(self): # release resources created by this class super(PipelineTest, self).cleanup() self.c1 = None self.c2 = None def tearDown(self): self.cleanup() def test(self): ssn = self.c1.session() snd = ssn.sender("sender") self.c1.open() ssn.open() snd.open() for i in range(10): t = "delivery-%s" % i d = snd.delivery(t) snd.send(t.encode('ascii')) d.settle() snd.close() ssn.close() self.c1.close() self.pump() state = self.c2.state assert state == (Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE), "%x" % state ssn2 = self.c2.session_head(Endpoint.LOCAL_UNINIT) assert ssn2 state == ssn2.state assert state == (Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE), "%x" % state rcv = self.c2.link_head(Endpoint.LOCAL_UNINIT) assert rcv state = rcv.state assert state == (Endpoint.LOCAL_UNINIT | Endpoint.REMOTE_ACTIVE), "%x" % state self.c2.open() ssn2.open() rcv.open() rcv.flow(10) assert rcv.queued == 0, rcv.queued self.pump() assert rcv.queued == 10, rcv.queued state = rcv.state assert state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED), "%x" % state state = ssn2.state assert state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED), "%x" % state state = self.c2.state assert state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_CLOSED), "%x" % state for i in range(rcv.queued): d = rcv.current assert d assert d.tag == "delivery-%s" % i d.settle() assert rcv.queued == 0, rcv.queued class ServerTest(Test): def testKeepalive(self): """ Verify that idle frames are sent to keep a Connection alive """ idle_timeout = self.delay server = common.TestServer() server.start() class Program: def on_reactor_init(self, event): self.conn = event.reactor.connection() self.conn.hostname = "%s:%s" % (server.host, server.port) self.conn.open() self.old_count = None event.reactor.schedule(3 * idle_timeout, self) def on_connection_bound(self, event): event.transport.idle_timeout = idle_timeout def on_connection_remote_open(self, event): self.old_count = event.transport.frames_input def on_timer_task(self, event): assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection terminated" assert self.conn.transport.frames_input > self.old_count, "No idle frames received" self.conn.close() Container(Program()).run() server.stop() def testIdleTimeout(self): """ Verify that a Connection is terminated properly when Idle frames do not arrive in a timely manner. """ idle_timeout = self.delay server = common.TestServer(idle_timeout=idle_timeout) server.start() class Program: def on_reactor_init(self, event): self.conn = event.reactor.connection() self.conn.hostname = "%s:%s" % (server.host, server.port) self.conn.open() self.remote_condition = None self.old_count = None # verify the connection stays up even if we don't explicitly send stuff # wait up to 3x the idle timeout event.reactor.schedule(3 * idle_timeout, self) def on_connection_bound(self, event): self.transport = event.transport def on_connection_remote_open(self, event): self.old_count = event.transport.frames_output def on_connection_remote_close(self, event): assert self.conn.remote_condition assert self.conn.remote_condition.name == "amqp:resource-limit-exceeded" self.remote_condition = self.conn.remote_condition def on_timer_task(self, event): assert self.conn.state == (Endpoint.LOCAL_ACTIVE | Endpoint.REMOTE_ACTIVE), "Connection terminated" assert self.conn.transport.frames_output > self.old_count, "No idle frames sent" # now wait to explicitly cause the other side to expire: suspend_time = 3 * idle_timeout if os.name == "nt": # On windows, the full delay gets too close to the graceful/hard close tipping point suspend_time = 2.5 * idle_timeout sleep(suspend_time) p = Program() Container(p).run() assert p.remote_condition assert p.remote_condition.name == "amqp:resource-limit-exceeded" server.stop() class NoValue: def __init__(self): pass def apply(self, dlv): pass def check(self, dlv): assert dlv.data is None assert dlv.section_number == 0 assert dlv.section_offset == 0 assert dlv.condition is None assert dlv.failed == False assert dlv.undeliverable == False assert dlv.annotations is None class RejectValue: def __init__(self, condition): self.condition = condition def apply(self, dlv): dlv.condition = self.condition def check(self, dlv): assert dlv.data is None, dlv.data assert dlv.section_number == 0 assert dlv.section_offset == 0 assert dlv.condition == self.condition, (dlv.condition, self.condition) assert dlv.failed == False assert dlv.undeliverable == False assert dlv.annotations is None class ReceivedValue: def __init__(self, section_number, section_offset): self.section_number = section_number self.section_offset = section_offset def apply(self, dlv): dlv.section_number = self.section_number dlv.section_offset = self.section_offset def check(self, dlv): assert dlv.data is None, dlv.data assert dlv.section_number == self.section_number, (dlv.section_number, self.section_number) assert dlv.section_offset == self.section_offset assert dlv.condition is None assert dlv.failed == False assert dlv.undeliverable == False assert dlv.annotations is None class ModifiedValue: def __init__(self, failed, undeliverable, annotations): self.failed = failed self.undeliverable = undeliverable self.annotations = annotations def apply(self, dlv): dlv.failed = self.failed dlv.undeliverable = self.undeliverable dlv.annotations = self.annotations def check(self, dlv): assert dlv.data is None, dlv.data assert dlv.section_number == 0 assert dlv.section_offset == 0 assert dlv.condition is None assert dlv.failed == self.failed assert dlv.undeliverable == self.undeliverable assert dlv.annotations == self.annotations, (dlv.annotations, self.annotations) class CustomValue: def __init__(self, data): self.data = data def apply(self, dlv): dlv.data = self.data def check(self, dlv): assert dlv.data == self.data, (dlv.data, self.data) assert dlv.section_number == 0 assert dlv.section_offset == 0 assert dlv.condition is None assert dlv.failed == False assert dlv.undeliverable == False assert dlv.annotations is None class DeliveryTest(Test): def tearDown(self): self.cleanup() def testDisposition(self, count=1, tag="tag%i", type=Delivery.ACCEPTED, value=NoValue()): snd, rcv = self.link("test-link") snd.open() rcv.open() snd_deliveries = [] for i in range(count): d = snd.delivery(tag % i) snd_deliveries.append(d) snd.advance() rcv.flow(count) self.pump() rcv_deliveries = [] for i in range(count): d = rcv.current assert d.tag == (tag % i) rcv_deliveries.append(d) rcv.advance() for d in rcv_deliveries: value.apply(d.local) d.update(type) self.pump() for d in snd_deliveries: assert d.remote_state == type assert d.remote.type == type value.check(d.remote) value.apply(d.local) d.update(type) self.pump() for d in rcv_deliveries: assert d.remote_state == type assert d.remote.type == type value.check(d.remote) for d in snd_deliveries: d.settle() self.pump() for d in rcv_deliveries: assert d.settled, d.settled d.settle() def testReceived(self): self.testDisposition(type=Disposition.RECEIVED, value=ReceivedValue(1, 2)) def testRejected(self): self.testDisposition(type=Disposition.REJECTED, value=RejectValue(Condition(symbol("foo")))) def testReleased(self): self.testDisposition(type=Disposition.RELEASED) def testModified(self): self.testDisposition(type=Disposition.MODIFIED, value=ModifiedValue(failed=True, undeliverable=True, annotations={"key": "value"})) def testCustom(self): self.testDisposition(type=0x12345, value=CustomValue([1, 2, 3])) class CollectorTest(Test): def setUp(self): self.collector = Collector() def drain(self): result = [] while True: e = self.collector.peek() if e: result.append(e) self.collector.pop() else: break return result def expect(self, *types): return self.expect_oneof(types) def expect_oneof(self, *sequences): events = self.drain() types = tuple([e.type for e in events]) for alternative in sequences: if types == alternative: if len(events) == 1: return events[0] elif len(events) > 1: return events else: return assert False, "actual events %s did not match any of the expected sequences: %s" % (events, sequences) def expect_until(self, *types): events = self.drain() etypes = tuple([e.type for e in events[-len(types):]]) assert etypes == types, "actual events %s did not end in expect sequence: %s" % (events, types) class EventTest(CollectorTest): def tearDown(self): self.cleanup() def testEndpointEvents(self): c1, c2 = self.connection() c1.collect(self.collector) self.expect(Event.CONNECTION_INIT) self.pump() self.expect() c2.open() self.pump() self.expect(Event.CONNECTION_REMOTE_OPEN) self.pump() self.expect() ssn = c2.session() snd = ssn.sender("sender") ssn.open() snd.open() self.expect() self.pump() self.expect(Event.SESSION_INIT, Event.SESSION_REMOTE_OPEN, Event.LINK_INIT, Event.LINK_REMOTE_OPEN) c1.open() ssn2 = c1.session() ssn2.open() rcv = ssn2.receiver("receiver") rcv.open() self.pump() self.expect(Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, Event.SESSION_INIT, Event.SESSION_LOCAL_OPEN, Event.TRANSPORT, Event.LINK_INIT, Event.LINK_LOCAL_OPEN, Event.TRANSPORT) rcv.close() self.expect(Event.LINK_LOCAL_CLOSE, Event.TRANSPORT) self.pump() rcv.free() del rcv self.expect(Event.LINK_FINAL) ssn2.free() del ssn2 self.pump() c1.free() c1.transport.unbind() self.expect_oneof((Event.SESSION_FINAL, Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL), (Event.CONNECTION_UNBOUND, Event.SESSION_FINAL, Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_FINAL)) def testConnectionINIT_FINAL(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) c.free() self.expect(Event.CONNECTION_FINAL) def testSessionINIT_FINAL(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) s = c.session() self.expect(Event.SESSION_INIT) s.free() self.expect(Event.SESSION_FINAL) c.free() self.expect(Event.CONNECTION_FINAL) def testLinkINIT_FINAL(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) s = c.session() self.expect(Event.SESSION_INIT) r = s.receiver("asdf") self.expect(Event.LINK_INIT) r.free() self.expect(Event.LINK_FINAL) c.free() self.expect(Event.SESSION_FINAL, Event.CONNECTION_FINAL) def testFlowEvents(self): snd, rcv = self.link("test-link") snd.session.connection.collect(self.collector) rcv.open() rcv.flow(10) self.pump() self.expect(Event.CONNECTION_INIT, Event.SESSION_INIT, Event.LINK_INIT, Event.LINK_REMOTE_OPEN, Event.LINK_FLOW) rcv.flow(10) self.pump() self.expect(Event.LINK_FLOW) return snd, rcv def testDeliveryEvents(self): snd, rcv = self.link("test-link") rcv.session.connection.collect(self.collector) rcv.open() rcv.flow(10) self.pump() self.expect(Event.CONNECTION_INIT, Event.SESSION_INIT, Event.LINK_INIT, Event.LINK_LOCAL_OPEN, Event.TRANSPORT) snd.delivery("delivery") snd.send(b"Hello World!") snd.advance() self.pump() self.expect() snd.open() self.pump() self.expect(Event.LINK_REMOTE_OPEN, Event.DELIVERY) rcv.session.connection.transport.unbind() rcv.session.connection.free() self.expect(Event.CONNECTION_UNBOUND, Event.TRANSPORT, Event.LINK_FINAL, Event.SESSION_FINAL, Event.CONNECTION_FINAL) def testDeliveryEventsDisp(self): snd, rcv = self.testFlowEvents() snd.open() dlv = snd.delivery("delivery") snd.send(b"Hello World!") assert snd.advance() self.expect(Event.LINK_LOCAL_OPEN, Event.TRANSPORT) self.pump() self.expect(Event.LINK_FLOW) rdlv = rcv.current assert rdlv is not None assert rdlv.tag == "delivery" rdlv.update(Delivery.ACCEPTED) self.pump() event = self.expect(Event.DELIVERY) assert event.context == dlv, (dlv, event.context) def testConnectionBOUND_UNBOUND(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) t = Transport() t.bind(c) self.expect(Event.CONNECTION_BOUND) t.unbind() self.expect(Event.CONNECTION_UNBOUND, Event.TRANSPORT) def testTransportERROR_CLOSE(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) t = Transport() t.bind(c) self.expect(Event.CONNECTION_BOUND) assert t.condition is None t.push(b"asdf") self.expect(Event.TRANSPORT_ERROR, Event.TRANSPORT_TAIL_CLOSED) assert t.condition is not None assert t.condition.name == "amqp:connection:framing-error" assert "AMQP protocol header" in t.condition.description p = t.pending() assert p > 0 t.pop(p) self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) def testTransportCLOSED(self): c = Connection() c.collect(self.collector) self.expect(Event.CONNECTION_INIT) t = Transport() t.bind(c) c.open() self.expect(Event.CONNECTION_BOUND, Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT) c2 = Connection() t2 = Transport() t2.bind(c2) c2.open() c2.close() pump(t, t2) self.expect(Event.CONNECTION_REMOTE_OPEN, Event.CONNECTION_REMOTE_CLOSE, Event.TRANSPORT_TAIL_CLOSED) c.close() pump(t, t2) self.expect(Event.CONNECTION_LOCAL_CLOSE, Event.TRANSPORT, Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) def testLinkDetach(self): c1 = Connection() c1.collect(self.collector) t1 = Transport() t1.bind(c1) c1.open() s1 = c1.session() s1.open() l1 = s1.sender("asdf") l1.open() l1.detach() self.expect_until(Event.LINK_LOCAL_DETACH, Event.TRANSPORT) c2 = Connection() c2.collect(self.collector) t2 = Transport() t2.bind(c2) pump(t1, t2) self.expect_until(Event.LINK_REMOTE_DETACH) class PeerTest(CollectorTest): def setUp(self): CollectorTest.setUp(self) self.connection = Connection() self.connection.collect(self.collector) self.transport = Transport() self.transport.bind(self.connection) self.peer = Connection() self.peer_transport = Transport() self.peer_transport.bind(self.peer) self.peer_transport.trace(Transport.TRACE_OFF) def pump(self): pump(self.transport, self.peer_transport) class TeardownLeakTest(PeerTest): def doLeak(self, local, remote): self.connection.open() self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT) ssn = self.connection.session() ssn.open() self.expect(Event.SESSION_INIT, Event.SESSION_LOCAL_OPEN, Event.TRANSPORT) snd = ssn.sender("sender") snd.open() self.expect(Event.LINK_INIT, Event.LINK_LOCAL_OPEN, Event.TRANSPORT) self.pump() self.peer.open() self.peer.session_head(0).open() self.peer.link_head(0).open() self.pump() self.expect_oneof((Event.CONNECTION_REMOTE_OPEN, Event.SESSION_REMOTE_OPEN, Event.LINK_REMOTE_OPEN, Event.LINK_FLOW), (Event.CONNECTION_REMOTE_OPEN, Event.SESSION_REMOTE_OPEN, Event.LINK_REMOTE_OPEN)) if local: snd.close() # ha!! self.expect(Event.LINK_LOCAL_CLOSE, Event.TRANSPORT) ssn.close() self.expect(Event.SESSION_LOCAL_CLOSE, Event.TRANSPORT) self.connection.close() self.expect(Event.CONNECTION_LOCAL_CLOSE, Event.TRANSPORT) if remote: self.peer.link_head(0).close() # ha!! self.peer.session_head(0).close() self.peer.close() self.pump() if remote: self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.LINK_REMOTE_CLOSE, Event.SESSION_REMOTE_CLOSE, Event.CONNECTION_REMOTE_CLOSE, Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_CLOSED) else: self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.SESSION_REMOTE_CLOSE, Event.CONNECTION_REMOTE_CLOSE, Event.TRANSPORT_TAIL_CLOSED, Event.TRANSPORT_CLOSED) self.connection.free() self.expect(Event.LINK_FINAL, Event.SESSION_FINAL) self.transport.unbind() self.expect(Event.CONNECTION_UNBOUND, Event.CONNECTION_FINAL) def testLocalRemoteLeak(self): self.doLeak(True, True) def testLocalLeak(self): self.doLeak(True, False) def testRemoteLeak(self): self.doLeak(False, True) def testLeak(self): self.doLeak(False, False) class IdleTimeoutEventTest(PeerTest): def half_pump(self): p = self.transport.pending() if p > 0: self.transport.pop(p) def testTimeoutWithZombieServer(self, expectOpenCloseFrames=True): self.transport.idle_timeout = self.delay self.connection.open() self.half_pump() t = time() self.transport.tick(t) self.transport.tick(t + self.delay * 4) self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND, Event.CONNECTION_LOCAL_OPEN, Event.TRANSPORT, Event.TRANSPORT_ERROR, Event.TRANSPORT_TAIL_CLOSED) assert self.transport.capacity() < 0 if expectOpenCloseFrames: assert self.transport.pending() > 0 self.half_pump() self.expect(Event.TRANSPORT_HEAD_CLOSED, Event.TRANSPORT_CLOSED) assert self.transport.pending() < 0 def testTimeoutWithZombieServerAndSASL(self): sasl = self.transport.sasl() self.testTimeoutWithZombieServer(expectOpenCloseFrames=False) class DeliverySegFaultTest(Test): def testDeliveryAfterUnbind(self): conn = Connection() t = Transport() ssn = conn.session() snd = ssn.sender("sender") dlv = snd.delivery("tag") dlv.settle() del dlv t.bind(conn) t.unbind() dlv = snd.delivery("tag") class SaslEventTest(CollectorTest): def testAnonymousNoInitialResponse(self): conn = Connection() conn.collect(self.collector) transport = Transport(Transport.SERVER) transport.bind(conn) self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND) transport.push(b'AMQP\x03\x01\x00\x00\x00\x00\x00 \x02\x01\x00\x00\x00SA' b'\xd0\x00\x00\x00\x10\x00\x00\x00\x02\xa3\tANONYMOUS@' b'AMQP\x00\x01\x00\x00') self.expect(Event.TRANSPORT) for i in range(1024): p = transport.pending() self.drain() p = transport.pending() self.expect() def testPipelinedServerReadFirst(self): conn = Connection() conn.collect(self.collector) transport = Transport(Transport.CLIENT) s = transport.sasl() s.allowed_mechs("ANONYMOUS PLAIN") transport.bind(conn) self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND) transport.push( # SASL b'AMQP\x03\x01\x00\x00' # @sasl-mechanisms(64) [sasl-server-mechanisms=@PN_SYMBOL[:ANONYMOUS]] b'\x00\x00\x00\x1c\x02\x01\x00\x00\x00S@\xc0\x0f\x01\xe0\x0c\x01\xa3\tANONYMOUS' # @sasl-outcome(68) [code=0] b'\x00\x00\x00\x10\x02\x01\x00\x00\x00SD\xc0\x03\x01P\x00' # AMQP b'AMQP\x00\x01\x00\x00' ) self.expect(Event.TRANSPORT) p = transport.pending() bytes = transport.peek(p) transport.pop(p) server = Transport(Transport.SERVER) server.push(bytes) assert s.outcome == SASL.OK assert server.sasl().outcome == SASL.OK def testPipelinedServerWriteFirst(self): conn = Connection() conn.collect(self.collector) transport = Transport(Transport.CLIENT) s = transport.sasl() s.allowed_mechs("ANONYMOUS") transport.bind(conn) p = transport.pending() bytes = transport.peek(p) transport.pop(p) self.expect(Event.CONNECTION_INIT, Event.CONNECTION_BOUND) transport.push( # SASL b'AMQP\x03\x01\x00\x00' # @sasl-mechanisms(64) [sasl-server-mechanisms=@PN_SYMBOL[:ANONYMOUS]] b'\x00\x00\x00\x1c\x02\x01\x00\x00\x00S@\xc0\x0f\x01\xe0\x0c\x01\xa3\tANONYMOUS' # @sasl-outcome(68) [code=0] b'\x00\x00\x00\x10\x02\x01\x00\x00\x00SD\xc0\x03\x01P\x00' # AMQP b'AMQP\x00\x01\x00\x00' ) self.expect(Event.TRANSPORT) p = transport.pending() bytes = transport.peek(p) transport.pop(p) assert s.outcome == SASL.OK # XXX: the bytes above appear to be correct, but we don't get any # sort of event indicating that the transport is authenticated