#
# 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"
        t = self.snd.transport
        wire_bytes = t.peek(1024)
        assert wanted == 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"
        t = self.snd.transport
        wire_bytes = t.peek(2048)
        assert wanted == 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"
        wire_bytes = t.peek(2048)
        assert wanted == 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