#
# 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 time import sleep

from proton import Message, Delivery
from system_test import TestCase, Qdrouterd, TIMEOUT, get_link_info, \
    get_inter_router_links, has_mobile_dest_in_address_table, PollTimeout, TestTimeout
from proton.handlers import MessagingHandler
from proton.reactor import Container
from qpid_dispatch.management.client import Node

LARGE_PAYLOAD = ("X" * 1024) * 30


_LINK_STATISTIC_KEYS = set(['unsettledCount',
                            'undeliveredCount',
                            'releasedCount',
                            'presettledCount',
                            'acceptedCount',
                            'droppedPresettledCount',
                            'rejectedCount',
                            'deliveryCount',
                            'modifiedCount'])


def get_body(n_sent, large_message=False):
    if large_message:
        body = {'number': n_sent, 'msg': LARGE_PAYLOAD}
    else:
        body = {'number': n_sent}


def _link_stats_are_zero(statistics, keys):
    """
    Verify that all statistics whose keys are present are zero
    """
    for key in keys:
        if statistics.get(key) != 0:
            return False
    return True


class OneRouterModifiedTest(TestCase):
    @classmethod
    def setUpClass(cls):
        """Start three routers"""
        super(OneRouterModifiedTest, cls).setUpClass()

        listen_port = cls.tester.get_port()
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'A'}),
            ('listener', {'port': listen_port, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'})])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def router_modified_counts(self, large_message=False):
        address = self.router.addresses[0]

        local_node = Node.connect(address, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_modified_index = outs.attribute_names.index('modifiedDeliveries')
        results = outs.results[0]
        num_modified_deliveries_pre_test = results[deliveries_modified_index]

        num_messages = 10
        test = ModifiedDeliveriesTest(address, num_messages, large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        results = outs.results[0]

        self.assertEqual(results[deliveries_modified_index] - num_modified_deliveries_pre_test, num_messages)

        # check link statistics
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'modifiedCount'])))
        self.assertEqual(test.sender_stats['deliveryCount'], num_messages)
        self.assertEqual(test.sender_stats['modifiedCount'], num_messages)

        # receiver just drops the link, so these are not counted as modified
        # but unsettled instead
        self.assertEqual(test.receiver_stats['deliveryCount'], num_messages)
        self.assertEqual(test.receiver_stats['unsettledCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'unsettledCount'])))

    def test_one_router_modified_counts(self):
        self.router_modified_counts()

    def test_one_router_large_message_modified_counts(self):
        self.router_modified_counts(True)


class OneRouterRejectedTest(TestCase):
    @classmethod
    def setUpClass(cls):
        """Start three routers"""
        super(OneRouterRejectedTest, cls).setUpClass()

        listen_port = cls.tester.get_port()
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'A'}),
            ('listener', {'port': listen_port, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'})])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def one_router_rejected_counts(self, large_message=False):
        address = self.router.addresses[0]

        local_node = Node.connect(address, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_rejected_index = outs.attribute_names.index('rejectedDeliveries')
        results = outs.results[0]
        deliveries_rejected_pre_test = results[deliveries_rejected_index]

        num_messages = 10
        test = RejectedDeliveriesTest(address, num_messages, large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        results = outs.results[0]

        self.assertEqual(results[deliveries_rejected_index] - deliveries_rejected_pre_test, num_messages)

        # check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], num_messages)
        self.assertEqual(test.sender_stats['rejectedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'rejectedCount'])))

        self.assertEqual(test.receiver_stats['deliveryCount'], num_messages)
        self.assertEqual(test.receiver_stats['rejectedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'rejectedCount'])))

    def test_one_router_rejected_counts(self):
        self.one_router_rejected_counts()

    def test_one_router_large_message_rejected_counts(self):
        self.one_router_rejected_counts(True)


class OneRouterReleasedDroppedPresettledTest(TestCase):
    @classmethod
    def setUpClass(cls):
        """Start three routers"""
        super(OneRouterReleasedDroppedPresettledTest, cls).setUpClass()

        listen_port = cls.tester.get_port()
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'A'}),
            ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
            ('listener', {'port': listen_port, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'})])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def one_router_released_dropped_count(self, large_message=False):
        address = self.router.addresses[0]

        local_node = Node.connect(address, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        deliveries_dropped_presettled_index = outs.attribute_names.index('droppedPresettledDeliveries')
        deliveries_released_index = outs.attribute_names.index('releasedDeliveries')
        deliveries_presettled_index = outs.attribute_names.index('presettledDeliveries')
        results = outs.results[0]

        deliveries_dropped_presettled_pre_test = results[deliveries_dropped_presettled_index]
        deliveries_released_pre_test = results[deliveries_released_index]
        deliveries_presettled_pre_test = results[deliveries_presettled_index]
        num_messages = 20
        test = ReleasedDroppedPresettledCountTest(address, num_messages, large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        results = outs.results[0]

        self.assertEqual(results[deliveries_dropped_presettled_index] - deliveries_dropped_presettled_pre_test, 10)
        self.assertEqual(results[deliveries_released_index] - deliveries_released_pre_test, 10)
        self.assertEqual(results[deliveries_presettled_index] - deliveries_presettled_pre_test, 10)

        # check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], test.n_sent)
        self.assertEqual(test.sender_stats['releasedCount'], 10)
        self.assertEqual(test.sender_stats['presettledCount'], 10)
        self.assertEqual(test.sender_stats['droppedPresettledCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'releasedCount',
                                                                         'presettledCount',
                                                                         'droppedPresettledCount'])))

    def test_one_router_released_dropped_counts(self):
        self.one_router_released_dropped_count()

    def test_one_router_large_message_released_dropped_counts(self):
        self.one_router_released_dropped_count(True)


class TwoRouterReleasedDroppedPresettledTest(TestCase):
    @classmethod
    def setUpClass(cls):
        super(TwoRouterReleasedDroppedPresettledTest, cls).setUpClass()

        listen_port_1 = cls.tester.get_port()
        listen_port_2 = cls.tester.get_port()
        listen_port_inter_router = cls.tester.get_port()

        config_1 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'A'}),
            ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
            ('listener', {'port': listen_port_1, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('listener', {'role': 'inter-router', 'port': listen_port_inter_router, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
        ])

        config_2 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'B'}),
            ('listener', {'port': listen_port_2, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': listen_port_inter_router}),
        ])

        cls.routers = []
        cls.routers.append(cls.tester.qdrouterd("A", config_1, wait=True))
        cls.routers.append(cls.tester.qdrouterd("B", config_2, wait=True))
        cls.routers[1].wait_router_connected('A')

    def two_router_released_dropped_counts(self, large_message=False):
        address = self.routers[0].addresses[0]

        # Send presettled and settled messages to router 1.
        # Make sure the hello messages (which are presettled dont show up in the counts

        local_node = Node.connect(address, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_dropped_presettled_index = outs.attribute_names.index('droppedPresettledDeliveries')
        deliveries_released_index = outs.attribute_names.index('releasedDeliveries')
        deliveries_presettled_index = outs.attribute_names.index('presettledDeliveries')
        results = outs.results[0]
        deliveries_dropped_presettled_pre_test = results[deliveries_dropped_presettled_index]
        deliveries_released_pre_test = results[deliveries_released_index]
        deliveries_presettled_pre_test = results[deliveries_presettled_index]
        num_messages = 20
        test = ReleasedDroppedPresettledCountTest(address, num_messages, large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        results = outs.results[0]

        self.assertEqual(results[deliveries_dropped_presettled_index] - deliveries_dropped_presettled_pre_test, 10)
        self.assertEqual(results[deliveries_released_index] - deliveries_released_pre_test, 10)
        self.assertEqual(results[deliveries_presettled_index] - deliveries_presettled_pre_test, 10)

        # check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], test.n_sent)
        self.assertEqual(test.sender_stats['releasedCount'], 10)
        self.assertEqual(test.sender_stats['presettledCount'], 10)
        self.assertEqual(test.sender_stats['droppedPresettledCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'releasedCount',
                                                                         'presettledCount',
                                                                         'droppedPresettledCount'])))

    def test_two_router_released_dropped_counts(self):
        self.two_router_released_dropped_counts()

    def test_two_router_large_message_released_dropped_counts(self):
        self.two_router_released_dropped_counts(True)


class AddressCheckerTimeout (object):
    def __init__(self, parent):
        self.parent = parent

    def on_timer_task(self, event):
        self.parent.address_check_timeout()


class CounterCheckerTimeout (object):
    def __init__(self, parent):
        self.parent = parent

    def on_timer_task(self, event):
        self.parent.count_check_timeout()


class LargePresettledLinkCounterTest(MessagingHandler):
    def __init__(self, sender_addr, receiver_addr):
        super(LargePresettledLinkCounterTest, self).__init__()
        self.timer = None
        self.sender_conn = None
        self.receiver_conn = None
        self.receiver = None
        self.error = None
        self.n_sent = 0
        self.n_received = 0
        self.num_messages = 25
        self.sender_addr = sender_addr
        self.receiver_addr = receiver_addr
        self.dest = "LargePresettledLinkCounterTest"
        self.links = None
        self.success = False
        self.address_check_timer = None
        self.container = None
        self.num_attempts = 0
        self.reactor = None
        self.done = False

    def check_if_done(self):
        if self.done:
            # Step 5: All messages have been received by receiver.
            # Check the presettled count on the inter-router link of
            # Router B (where the receiver is attached).
            self.links = get_inter_router_links(self.receiver_addr)
            for link in self.links:
                # The self.num_messages + 1 is because before this test started the presettledCount was 1
                if link.get("linkDir") == "in" and link.get("presettledCount") == self.num_messages + 1:
                    self.success = True
                    break
            self.sender_conn.close()
            self.receiver_conn.close()
            self.timer.cancel()

    def address_check_timeout(self):
        if has_mobile_dest_in_address_table(self.sender_addr, self.dest):
            # Step 3: The address has propagated to Router A. Now attach a sender
            # to router A.
            self.sender_conn = self.container.connect(self.sender_addr)
            self.sender = self.container.create_sender(self.sender_conn,
                                                       self.dest,
                                                       name='SenderA')
        else:
            if self.num_attempts < 2:
                self.address_check_timer = self.reactor.schedule(2,
                                                                 AddressCheckerTimeout(self))
                self.num_attempts += 1

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d, self.self.n_received=%d  " % (self.n_sent, self.n_received)
        self.sender_conn.close()
        self.receiver_conn.close()

    def on_start(self, event):
        self.container = event.container
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        # Step 1: Create a receiver with name ReceiverA to address LargePresettledLinkCounterTest
        # This receiver is attached to router B. Later a sender will be
        # created which will be connected to Router A. The sender will send
        # on the same address that the receiver is receiving on.
        self.receiver_conn = event.container.connect(self.receiver_addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        self.dest,
                                                        name='ReceiverA')

    def on_link_opened(self, event):
        self.reactor = event.reactor
        if event.receiver:
            # Step 2: The receiver link has been opened.
            # Give 2 seconds for the address to propagate to the other router (Router A)
            self.address_check_timer = event.reactor.schedule(2, AddressCheckerTimeout(self))
            self.num_attempts += 1

    def on_sendable(self, event):
        # Step 4: Send self.num_messages multi-frame large pre-settled messages.
        # These messages will travel over inter-router link to Router B.
        if self.n_sent < self.num_messages:
            msg = Message(body=LARGE_PAYLOAD)
            dlv = self.sender.send(msg)
            # We are sending a pre-settled large multi frame message.
            dlv.settle()
            self.n_sent += 1

    def on_message(self, event):
        if self.receiver == event.receiver:
            self.n_received += 1
            if self.n_received == self.num_messages:
                self.done = True
            self.check_if_done()

    def run(self):
        Container(self).run()


class LargePresettledReleasedLinkCounterTest(MessagingHandler):
    def __init__(self, sender_addr, receiver_addr):
        super(LargePresettledReleasedLinkCounterTest, self).__init__(prefetch=0)
        self.sender_addr = sender_addr
        self.receiver_addr = receiver_addr
        self.dest = "LargePresettledReleasedLinkCounterTest"
        self.receiver_dropoff_count = 50
        self.num_messages = 200
        self.num_attempts = 0
        self.n_sent = 0
        self.done = False
        self.n_received = 0
        self.count_check_timer = None
        self.success = False
        self.links = None
        self.receiver_conn_closed = False

    def check_if_done(self):
        # Step 6:
        # Check the counts on the inter-router link of
        # Router B (where the receiver is attached). There
        # should be no released or modified messages.
        self.links = get_inter_router_links(self.receiver_addr)
        for link in self.links:
            # We don't know how many deliveries got from one side of the
            # inter-router link to the other but there should at least be as
            # many as was sent to the receiver
            if link.get("linkDir") == "in" \
                    and link.get("presettledCount") > self.receiver_dropoff_count \
                    and link.get("deliveryCount") > self.receiver_dropoff_count \
                    and link.get("releasedCount") == 0\
                    and link.get("modifiedCount") == 0:
                self.success = True
                break
        self.sender_conn.close()
        self.timer.cancel()

    def count_check_timeout(self):
        self.check_if_done()

    def address_check_timeout(self):
        if has_mobile_dest_in_address_table(self.sender_addr, self.dest):
            # Step 3: The address has propagated to Router A. Now attach a sender
            # to router A.
            self.sender_conn = self.container.connect(self.sender_addr)
            self.sender = self.container.create_sender(self.sender_conn,
                                                       self.dest,
                                                       name='SenderA')
        else:
            if self.num_attempts < 2:
                self.address_check_timer = self.reactor.schedule(2, AddressCheckerTimeout(self))
                self.num_attempts += 1

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d, self.self.n_received=%d  " % (self.n_sent, self.n_received)
        self.sender_conn.close()
        if not self.receiver_conn_closed:
            self.receiver_conn.close()

    def on_start(self, event):
        self.container = event.container
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        # Step 1: Create a receiver with name ReceiverA to address LargePresettledReleasedLinkCounterTest
        # This receiver is attached to router B. Later a sender will be
        # created which will be connected to Router A. The sender will send
        # on the same address that the receiver is receiving on.
        self.receiver_conn = event.container.connect(self.receiver_addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        self.dest,
                                                        name='ReceiverA')
        self.receiver.flow(self.receiver_dropoff_count)

    def on_link_opened(self, event):
        self.reactor = event.reactor
        if event.receiver:
            # Step 2: The receiver link has been opened.
            # Give 2 seconds for the address to propagate to the other router (Router A)
            self.address_check_timer = event.reactor.schedule(2, AddressCheckerTimeout(self))
            self.num_attempts += 1

    def on_sendable(self, event):
        # Step 4: Send self.num_messages multi-frame large pre-settled messages.
        # These messages will travel over inter-router link to Router B.
        while self.n_sent < self.num_messages:
            msg = Message(body=LARGE_PAYLOAD)
            dlv = self.sender.send(msg)
            # We are sending a pre-settled large multi frame message.
            dlv.settle()
            self.n_sent += 1

    def on_message(self, event):
        if self.receiver == event.receiver and not self.done:
            self.n_received += 1
            # Step 5: The receiver receives only 50 messages out of the 200
            # messages and drops out.
            if self.n_received == self.receiver_dropoff_count:
                self.done = True
                self.receiver_conn.close()
                self.receiver_conn_closed = True
                self.count_check_timer = event.reactor.schedule(3, CounterCheckerTimeout(self))

    def run(self):
        Container(self).run()


class TwoRouterLargeMessagePresettledCountTest(TestCase):
    @classmethod
    def setUpClass(cls):
        super(TwoRouterLargeMessagePresettledCountTest, cls).setUpClass()

        listen_port_1 = cls.tester.get_port()
        listen_port_2 = cls.tester.get_port()
        listen_port_inter_router = cls.tester.get_port()

        config_1 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'A'}),
            ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
            ('listener', {'port': listen_port_1, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('listener', {'role': 'inter-router', 'port': listen_port_inter_router, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
        ])

        config_2 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'B'}),
            ('listener', {'port': listen_port_2, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': listen_port_inter_router}),
        ])

        cls.routers = []
        cls.routers.append(cls.tester.qdrouterd("A", config_1, wait=True))
        cls.routers.append(cls.tester.qdrouterd("B", config_2, wait=True))
        cls.routers[1].wait_router_connected('A')

    def test_verify_inter_router_presettled_count_DISPATCH_1540(self):
        sender_address = self.routers[0].addresses[0]
        receiver_address = self.routers[1].addresses[0]
        # Sends presettled large messages across routers and checks
        # the pre-settled count on the inter-router link of the downstream
        # router (i.e. that to which receiver is attached)
        # This test will fail if DISPATCH-1540 is not fixed since the
        # pre-settled count will show zero
        test = LargePresettledLinkCounterTest(sender_address, receiver_address)
        test.run()
        self.assertTrue(test.success)


class TwoRouterLargeMessagePresettledReleasedCountTest(TestCase):
    @classmethod
    def setUpClass(cls):
        super(TwoRouterLargeMessagePresettledReleasedCountTest, cls).setUpClass()

        listen_port_1 = cls.tester.get_port()
        listen_port_2 = cls.tester.get_port()
        listen_port_inter_router = cls.tester.get_port()

        config_1 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'A'}),
            ('address', {'prefix': 'multicast', 'distribution': 'multicast'}),
            ('listener', {'port': listen_port_1, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('listener', {'role': 'inter-router', 'port': listen_port_inter_router, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
        ])

        config_2 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'B'}),
            ('listener', {'port': listen_port_2, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': listen_port_inter_router}),
        ])

        cls.routers = []
        cls.routers.append(cls.tester.qdrouterd("A", config_1, wait=True))
        cls.routers.append(cls.tester.qdrouterd("B", config_2, wait=True))
        cls.routers[1].wait_router_connected('A')

    def test_verify_inter_router_presettled_released_count_DISPATCH_1541(self):
        # This test sends presettled large messages across routers. A sender is on
        # router A and a receiver on B. The sender sends 200 messages, the receiver
        # receives 50 messages and goes away by closing its connection. There should be no released or
        # modified messages on the incoming inter-router link on Router B
        # This test will fail without the patch to DISPATCH-1541
        sender_address = self.routers[0].addresses[0]
        receiver_address = self.routers[1].addresses[0]
        test = LargePresettledReleasedLinkCounterTest(sender_address, receiver_address)
        test.run()
        self.assertTrue(test.success)


class LinkRouteIngressEgressTransitTest(TestCase):
    @classmethod
    def setUpClass(cls):
        """Start three routers"""
        super(LinkRouteIngressEgressTransitTest, cls).setUpClass()

        def router(name, connection):

            config = [
                ('router', {'mode': 'interior', 'id': 'QDR.%s' % name}),
            ] + connection

            config = Qdrouterd.Config(config)
            cls.routers.append(cls.tester.qdrouterd(name, config, wait=False))

        cls.routers = []
        a_listener_port = cls.tester.get_port()
        b_listener_port = cls.tester.get_port()
        c_listener_port = cls.tester.get_port()
        test_tag_listener_port = cls.tester.get_port()

        router('A',
               [
                   ('listener', {'role': 'normal', 'host': '0.0.0.0', 'port': a_listener_port, 'saslMechanisms': 'ANONYMOUS'}),
               ])
        router('B',
               [
                   ('listener', {'role': 'normal', 'host': '0.0.0.0', 'port': b_listener_port, 'saslMechanisms': 'ANONYMOUS'}),
                   ('listener', {'name': 'test-tag', 'role': 'route-container', 'host': '0.0.0.0', 'port': test_tag_listener_port, 'saslMechanisms': 'ANONYMOUS'}),

                   ('connector', {'name': 'broker', 'role': 'route-container', 'host': '0.0.0.0', 'port': a_listener_port, 'saslMechanisms': 'ANONYMOUS'}),
                   ('connector', {'name': 'routerC', 'role': 'inter-router', 'host': '0.0.0.0', 'port': c_listener_port}),


                   ('linkRoute', {'prefix': 'pulp.task', 'connection': 'broker', 'direction': 'in'}),
                   ('linkRoute', {'prefix': 'pulp.task', 'connection': 'broker', 'direction': 'out'}),

               ]
               )
        router('C',
               [
                   ('listener', {'host': '0.0.0.0', 'role': 'normal', 'port': cls.tester.get_port(), 'saslMechanisms': 'ANONYMOUS'}),
                   ('listener', {'host': '0.0.0.0', 'role': 'inter-router', 'port': c_listener_port, 'saslMechanisms': 'ANONYMOUS'}),

                   ('linkRoute', {'prefix': 'pulp.task', 'direction': 'in'}),
                   ('linkRoute', {'prefix': 'pulp.task', 'direction': 'out'}),
               ]
               )

        # Wait for the routers to locate each other, and for route propagation
        # to settle
        cls.routers[2].wait_router_connected('QDR.B')
        cls.routers[1].wait_router_connected('QDR.C')
        cls.routers[2].wait_address("pulp.task", remotes=1, delay=3, count=2)

        # This is not a classic router network in the sense that QDR.A and D are acting as brokers. We allow a little
        # bit more time for the routers to stabilize.
        sleep(2)

    def link_route_ingress_egress_transit_counts(self, large_message=False):
        address1 = self.routers[2].addresses[0]
        address2 = self.routers[2].addresses[0]

        local_node = Node.connect(address1, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        deliveries_ingress_index = outs.attribute_names.index('deliveriesIngress')
        deliveries_egress_index = outs.attribute_names.index('deliveriesEgress')
        deliveries_transit_index = outs.attribute_names.index('deliveriesTransit')

        results = outs.results[0]

        pre_ingress_count = results[deliveries_ingress_index]
        pre_egress_count = results[deliveries_egress_index]
        pre_transit_count = results[deliveries_transit_index]

        num_messages = 10
        # Send and receive on the same router, router C
        test = IngressEgressTransitLinkRouteTest(address1, address2, num_messages, large_message=large_message)
        test.run()
        local_node = Node.connect(address1, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        results = outs.results[0]

        post_ingress_count = results[deliveries_ingress_index]
        post_egress_count = results[deliveries_egress_index]
        post_transit_count = results[deliveries_transit_index]

        # 10 messages entered the router, and 10 messages were echoed by router A and 3 mgmt requests
        self.assertEqual(post_ingress_count - pre_ingress_count, 23)

        # 10 messages + 3 mgmt request
        self.assertEqual(post_egress_count - pre_egress_count, 13)

        # 10 messages went out this router
        self.assertEqual(post_transit_count - pre_transit_count, 10)

        # Check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], 10)
        self.assertEqual(test.sender_stats['acceptedCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

        self.assertEqual(test.receiver_stats['deliveryCount'], 10)
        self.assertEqual(test.receiver_stats['acceptedCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

    def test_link_route_ingress_egress_transit_counts(self):
        self.link_route_ingress_egress_transit_counts()

    def test_link_route_large_message_ingress_egress_transit_counts(self):
        self.link_route_ingress_egress_transit_counts(True)


class TwoRouterIngressEgressTest(TestCase):
    @classmethod
    def setUpClass(cls):
        super(TwoRouterIngressEgressTest, cls).setUpClass()

        listen_port_1 = cls.tester.get_port()
        listen_port_2 = cls.tester.get_port()
        listen_port_inter_router = cls.tester.get_port()

        config_1 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'A'}),
            ('listener', {'port': listen_port_1, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('listener', {'role': 'inter-router', 'port': listen_port_inter_router, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
        ])

        config_2 = Qdrouterd.Config([
            ('router', {'mode': 'interior', 'id': 'B'}),
            ('listener', {'port': listen_port_2, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'}),
            ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': listen_port_inter_router}),
        ])

        cls.routers = []
        cls.routers.append(cls.tester.qdrouterd("A", config_1, wait=True))
        cls.routers.append(cls.tester.qdrouterd("B", config_2, wait=True))
        cls.routers[1].wait_router_connected('A')

    def two_router_ingress_egress_counts(self, large_message=False):
        in_router_addr = self.routers[0].addresses[0]
        out_router_addr = self.routers[1].addresses[0]

        # Gather the values for deliveries_ingress and deliveries_egress before running the test.

        local_node = Node.connect(in_router_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_ingress_index = outs.attribute_names.index('deliveriesIngress')
        results = outs.results[0]

        pre_deliveries_ingresss = results[deliveries_ingress_index]

        local_node = Node.connect(out_router_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_egress_index = outs.attribute_names.index('deliveriesEgress')
        deliveries_accepted_index = outs.attribute_names.index('acceptedDeliveries')
        results = outs.results[0]

        pre_deliveries_egress = results[deliveries_egress_index]
        pre_deliveries_accepted = results[deliveries_accepted_index]

        # Now run the test.  At the end of the test each router will be queried
        # for the per-link stats
        num_messages = 10
        test = IngressEgressTwoRouterTest(in_router_addr, out_router_addr, num_messages, large_message=large_message)
        test.run()

        # Gather the values for deliveries_ingress and deliveries_egress after running the test.
        local_node = Node.connect(in_router_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        results = outs.results[0]

        post_deliveries_ingresss = results[deliveries_ingress_index]

        local_node = Node.connect(out_router_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        results = outs.results[0]

        post_deliveries_egress = results[deliveries_egress_index]
        post_deliveries_accepted = results[deliveries_accepted_index]

        accepted_deliveries_diff = post_deliveries_accepted - pre_deliveries_accepted

        # 12 = 10 msgs + 2 mgmt requests
        self.assertEqual(post_deliveries_ingresss - pre_deliveries_ingresss, 12)
        self.assertEqual(post_deliveries_egress - pre_deliveries_egress, 12)

        # check the link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], num_messages)
        self.assertEqual(test.sender_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

        self.assertEqual(test.receiver_stats['deliveryCount'], num_messages)
        self.assertEqual(test.receiver_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

        # The management requests are counted in the acceptedDeliveries, so it is difficult to measure the
        # exact number of accepted deliveries at this point in time. But it must at least be 10 since
        # we know for sure from the test that the 10 dispositions related to the 10 sent messages
        # were definitely received
        self.assertTrue(accepted_deliveries_diff >= num_messages)

    def test_two_router_ingress_egress_counts(self):
        self.two_router_ingress_egress_counts()

    def test_two_router_large_message_ingress_egress_counts(self):
        self.two_router_ingress_egress_counts(True)


class OneRouterIngressEgressTest(TestCase):
    @classmethod
    def setUpClass(cls):
        """Start a router and a messenger"""
        super(OneRouterIngressEgressTest, cls).setUpClass()

        listen_port = cls.tester.get_port()
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'A'}),
            ('listener', {'port': listen_port, 'authenticatePeer': False, 'saslMechanisms': 'ANONYMOUS'})])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def one_router_ingress_egress_counts(self, large_message=False):
        address = self.router.addresses[0]

        local_node = Node.connect(address, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        deliveries_ingress_index = outs.attribute_names.index('deliveriesIngress')
        deliveries_egress_index = outs.attribute_names.index('deliveriesEgress')
        results = outs.results[0]
        deliveries_ingress_pre_test = results[deliveries_ingress_index]
        deliveries_egress_pre_test = results[deliveries_egress_index]

        num_messages = 10
        test = IngressEgressOneRouterTest(address, num_messages, large_message=large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        results = outs.results[0]

        # 13 = ten msgs + 3 mgmt requests
        self.assertEqual(results[deliveries_ingress_index] - deliveries_ingress_pre_test, 13)
        # 12 = ten msgs + 2 mgmt requests
        self.assertEqual(results[deliveries_egress_index] - deliveries_egress_pre_test, 13)

        # check the link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], num_messages)
        self.assertEqual(test.sender_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

        self.assertEqual(test.receiver_stats['deliveryCount'], num_messages)
        self.assertEqual(test.receiver_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

    def test_one_router_ingress_egress_counts(self):
        self.one_router_ingress_egress_counts()

    def test_one_router_large_message_ingress_egress_counts(self):
        self.one_router_ingress_egress_counts(True)


class RouteContainerEgressCount(TestCase):
    @classmethod
    def setUpClass(cls):
        super(RouteContainerEgressCount, cls).setUpClass()

        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'QDR'}),

            #
            # Create a general-purpose listener for sending and receiving deliveries
            #
            ('listener', {'port': cls.tester.get_port()}),

            # Create a route-container listener and give it a name myListener.
            # Later on we will create an autoLink which has a connection property of myListener.
            #
            ('listener', {'role': 'route-container', 'name': 'myListener', 'port': cls.tester.get_port()}),
            #
            # Note here that the connection is set to a previously declared 'myListener'
            #
            ('autoLink', {'address': 'myListener.1', 'connection': 'myListener', 'direction': 'in'}),
            ('autoLink', {'address': 'myListener.1', 'connection': 'myListener', 'direction': 'out'}),
        ])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def route_container_egress(self , large_message=False):
        regular_addr = self.router.addresses[0]
        route_container_addr = self.router.addresses[1]
        num_messages = 10
        local_node = Node.connect(regular_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_egress_route_container_index = outs.attribute_names.index('deliveriesEgressRouteContainer')
        results = outs.results[0]

        deliveries_egress_pre_test = results[deliveries_egress_route_container_index]

        test = RouteContainerEgressTest(route_container_addr, regular_addr, num_messages, large_message=large_message)
        test.run()

        outs = local_node.query(type='org.apache.qpid.dispatch.router')
        deliveries_egress_route_container_index = outs.attribute_names.index('deliveriesEgressRouteContainer')

        results = outs.results[0]
        # 11 = 10 msgs + 1 mgmt msg
        self.assertEqual(results[deliveries_egress_route_container_index] - deliveries_egress_pre_test, 11)

        # check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], num_messages)
        self.assertEqual(test.sender_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

        self.assertEqual(test.receiver_stats['deliveryCount'], num_messages)
        self.assertEqual(test.receiver_stats['acceptedCount'], num_messages)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))

    def test_route_container_egress_count(self):
        self.route_container_egress()

    def test_route_container_large_message_egress_count(self):
        self.route_container_egress(True)


class OneRouterLinkCountersTest(TestCase):
    """
    A set of tests that validate link-level counters
    """
    CREDIT = 20  # default issued by test receiver client
    COUNT  = 40  # default total msgs the sender client generates

    @classmethod
    def setUpClass(cls):
        # create one router
        super(OneRouterLinkCountersTest, cls).setUpClass()

        listen_port = cls.tester.get_port()
        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'LinkCounters'}),
            ('listener', {'port': listen_port,
                          'authenticatePeer': False,
                          'saslMechanisms': 'ANONYMOUS',
                          'linkCapacity': cls.CREDIT})])

        cls.router = cls.tester.qdrouterd(name="LinkCounters", config=config, wait=True)

    class LinkCountersTest(MessagingHandler):
        """
        Create 1 sender and 1 receiver to router_addr.  Send count messages.
        The test ends when the receivers deliveryCount reaches rx_limit.
        Explicitly set the receiver credit and whether to sender sends
        presettled or unsettled messages.
        """

        def __init__(self, router_addr, count=None, rx_limit=None,
                     credit=None, presettled=False, outcome=None,
                     large_message=False):
            super(OneRouterLinkCountersTest.LinkCountersTest,
                  self).__init__(auto_accept=False,
                                 auto_settle=False,
                                 prefetch=0)
            self.router_addr = router_addr
            self.presettled = presettled
            self.outcome = outcome
            self.count = OneRouterLinkCountersTest.COUNT \
                if count is None else count
            self.credit = OneRouterLinkCountersTest.COUNT \
                if credit is None else credit
            self.rx_limit = OneRouterLinkCountersTest.COUNT \
                if rx_limit is None else rx_limit

            self.sent = 0
            self.timer = 0
            self.poll_timer = None
            self.conn = None
            self.sender_stats = None
            self.receiver_stats = None
            self.large_message = large_message

        def timeout(self):
            self._cleanup()

        def _cleanup(self):
            if self.conn:
                self.conn.close()
                self.conn = None
            if self.poll_timer:
                self.poll_timer.cancel()
                self.poll_timer = None
            if self.timer:
                self.timer.cancel()
                self.timer = None

        def poll_timeout(self):
            """
            Periodically check the deliveryCount on the receiver.  Once it
            reaches rx_limit the test is complete: gather link statistics
            before closing the clients
            """
            li = get_link_info("Rx_Test01", self.router_addr)
            if li and li['deliveryCount'] == self.rx_limit:
                self.receiver_stats = li
                self.sender_stats = get_link_info("Tx_Test01", self.router_addr)
                self._cleanup()
            else:
                self.poll_timer = self.reactor.schedule(0.5, PollTimeout(self))

        def on_start(self, event):
            self.reactor = event.reactor
            self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
            self.poll_timer = event.reactor.schedule(0.5, PollTimeout(self))
            self.conn = event.container.connect(self.router_addr)
            self.receiver = event.container.create_receiver(self.conn,
                                                            source="Test01",
                                                            name='Rx_Test01')
            self.receiver.flow(self.credit)
            self.sender = event.container.create_sender(self.conn,
                                                        target="Test01",
                                                        name="Tx_Test01")

        def on_sendable(self, event):
            if self.sent < self.count:
                if self.large_message:
                    dlv = self.sender.send(Message(body=LARGE_PAYLOAD))
                else:
                    dlv = self.sender.send(Message(body="Test01"))
                if self.presettled:
                    dlv.settle()
                self.sent += 1

        def on_message(self, event):
            if self.outcome:
                event.delivery.update(self.outcome)
                event.delivery.settle()
                # otherwise just drop it

        def run(self):
            Container(self).run()

    def verify_released(self, large_message=False):
        """
        Verify the link released count by releasing all received messages
        """
        test = self.LinkCountersTest(self.router.addresses[0],
                                     outcome=Delivery.RELEASED,
                                     large_message=large_message)
        test.run()
        self.assertEqual(test.receiver_stats['deliveryCount'], self.COUNT)
        self.assertEqual(test.receiver_stats['releasedCount'], self.COUNT)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'releasedCount'])))

        self.assertEqual(test.sender_stats['deliveryCount'], self.COUNT)
        self.assertEqual(test.sender_stats['releasedCount'], self.COUNT)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'releasedCount'])))

    def verify_unsettled_count(self, large_message=False):
        """
        Verify the link unsettled count by granting less credit than required
        by the sender
        """
        test = self.LinkCountersTest(self.router.addresses[0],
                                     presettled=False,
                                     count=self.COUNT,
                                     rx_limit=self.CREDIT,
                                     credit=self.CREDIT,
                                     large_message=large_message)
        test.run()

        # expect the receiver to get rx_limit worth of unsettled deliveries
        self.assertEqual(test.receiver_stats['deliveryCount'], self.CREDIT)
        self.assertEqual(test.receiver_stats['unsettledCount'], self.CREDIT)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'unsettledCount'])))

        # expect sender only to be able to send as much as credit
        self.assertEqual(test.sender_stats['deliveryCount'], self.CREDIT)
        self.assertEqual(test.sender_stats['unsettledCount'], self.CREDIT)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'unsettledCount'])))

    def verify_presettled_count(self, large_message=False):
        """
        Verify the presettled dropped count link counter by exhausting credit
        before sending is complete
        """
        limit = self.CREDIT // 2  # 1/2 the capacity given the sender
        test = self.LinkCountersTest(self.router.addresses[0],
                                     presettled=True,
                                     count=self.COUNT,
                                     rx_limit=limit,
                                     credit=limit,
                                     large_message=large_message)
        test.run()

        # since these are presettled the sender should have credit
        # replenished by the router after each message.
        self.assertEqual(test.sender_stats['deliveryCount'], self.COUNT)
        self.assertEqual(test.sender_stats['presettledCount'], self.COUNT)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'presettledCount'])))

        # since credit is fixed at limit, exactly that number of msgs can be received
        self.assertEqual(test.receiver_stats['deliveryCount'], limit)

        # verify that some messages were dropped and some are stuck on the
        # undelivered list
        self.assertTrue(test.receiver_stats['undeliveredCount'] > 0)
        self.assertTrue(test.receiver_stats['droppedPresettledCount'] > 0)

        # expect that whatever was not dropped was delivered
        self.assertEqual(test.receiver_stats['deliveryCount'],
                         (test.receiver_stats['presettledCount']
                          - test.receiver_stats['droppedPresettledCount']))

        # expect the sum of dropped+delivered+undelivered accounts for all
        # messages sent
        self.assertEqual(self.COUNT,
                         (test.receiver_stats['deliveryCount']
                          + test.receiver_stats['undeliveredCount']
                          + test.receiver_stats['droppedPresettledCount']))

        # all other counters must be zero
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'undeliveredCount',
                                                    'droppedPresettledCount',
                                                    'presettledCount'])))

    def verify_one_credit_accepted(self, large_message=False):
        """
        Verify counters on a credit-blocked link
        """
        test = self.LinkCountersTest(self.router.addresses[0],
                                     outcome=Delivery.ACCEPTED,
                                     rx_limit=1,
                                     credit=1)
        test.run()
        # expect only 1 delivery, an link credit worth of queued up messages
        self.assertEqual(test.receiver_stats['deliveryCount'], 1)
        self.assertEqual(test.receiver_stats['acceptedCount'], 1)
        self.assertEqual(test.receiver_stats['undeliveredCount'], self.CREDIT)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'undeliveredCount',
                                                    'acceptedCount'])))

        # expect that one message will be delivered, then link capacity
        # messages will be enqueued internally
        self.assertEqual(test.sender_stats['unsettledCount'], self.CREDIT)
        self.assertEqual(test.sender_stats['deliveryCount'], self.CREDIT + 1)
        self.assertEqual(test.sender_stats['acceptedCount'], 1)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS
                                             - set(['deliveryCount',
                                                    'unsettledCount',
                                                    'acceptedCount'])))

    def test_01_presettled(self):
        self.verify_presettled_count()

    def test_02_large_mesage_presettled(self):
        self.verify_presettled_count(True)

    def test_03_unsettled(self):
        self.verify_presettled_count()

    def test_04_large_message_unsettled(self):
        self.verify_presettled_count(True)

    def test_05_released(self):
        self.verify_released()

    def test_06_large_message_released(self):
        self.verify_released(True)

    def test_07_one_credit_accepted(self):
        self.verify_one_credit_accepted()

    def test_08_large_message_one_credit_accepted(self):
        self.verify_one_credit_accepted(True)


class RouteContainerIngressCount(TestCase):
    @classmethod
    def setUpClass(cls):
        super(RouteContainerIngressCount, cls).setUpClass()

        config = Qdrouterd.Config([
            ('router', {'mode': 'standalone', 'id': 'QDR'}),

            #
            # Create a general-purpose listener for sending and receiving deliveries
            #
            ('listener', {'port': cls.tester.get_port()}),

            # Create a route-container listener and give it a name myListener.
            # Later on we will create an autoLink which has a connection property of myListener.
            #
            ('listener', {'role': 'route-container', 'name': 'myListener', 'port': cls.tester.get_port()}),
            #
            # Note here that the connection is set to a previously declared 'myListener'
            #
            ('autoLink', {'address': 'myListener.1', 'connection': 'myListener', 'direction': 'in'}),
            ('autoLink', {'address': 'myListener.1', 'connection': 'myListener', 'direction': 'out'}),
        ])

        cls.router = cls.tester.qdrouterd(name="A", config=config, wait=True)

    def test_route_container_ingress(self):
        regular_addr = self.router.addresses[0]
        route_container_addr = self.router.addresses[1]
        test = RouteContainerIngressTest(route_container_addr, regular_addr)
        test.run()

        local_node = Node.connect(regular_addr, timeout=TIMEOUT)
        outs = local_node.query(type='org.apache.qpid.dispatch.router')

        deliveries_ingress_route_container_index = outs.attribute_names.index('deliveriesIngressRouteContainer')

        results = outs.results[0]
        # 22 = 20 msgs + 2 mgmt msgs
        self.assertEqual(results[deliveries_ingress_route_container_index], 22)

        # check link statistics
        self.assertEqual(test.sender_stats['deliveryCount'], 10)
        self.assertEqual(test.sender_stats['acceptedCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.sender_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))
        self.assertEqual(test.sender1_stats['deliveryCount'], 10)
        self.assertEqual(test.sender1_stats['acceptedCount'], 10)
        self.assertTrue(_link_stats_are_zero(test.sender1_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))
        self.assertEqual(test.receiver_stats['deliveryCount'], 20)
        self.assertEqual(test.receiver_stats['acceptedCount'], 20)
        self.assertTrue(_link_stats_are_zero(test.receiver_stats,
                                             _LINK_STATISTIC_KEYS - set(['deliveryCount',
                                                                         'acceptedCount'])))


class IngressEgressTwoRouterTest(MessagingHandler):
    def __init__(self, sender_address, receiver_address, num_messages, large_message=False):
        super(IngressEgressTwoRouterTest, self).__init__()
        self.sender = None
        self.receiver = None
        self.conn_sender = None
        self.conn_recv = None
        self.timer = None
        self.dest = 'examples'
        self.sender_address = sender_address
        self.receiver_address = receiver_address
        self.n_sent = 0
        self.n_received = 0
        self.num_messages = num_messages
        self.start = False
        self.n_accept = 0
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False
        self.large_message = large_message

    def timeout(self):
        self.conn_sender.close()
        self.conn_recv.close()

    def check_if_done(self):
        if not self.done and self.num_messages == self.n_received and self.n_accept == self.num_messages:
            self.done = True
            self.sender_stats = get_link_info('Tx_IngressEgressTwoRouterTest',
                                              self.sender_address)
            self.receiver_stats = get_link_info('Rx_IngressEgressTwoRouterTest',
                                                self.receiver_address)
            self.conn_sender.close()
            self.conn_recv.close()
            self.timer.cancel()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn_recv = event.container.connect(self.receiver_address)
        self.receiver = event.container.create_receiver(self.conn_recv,
                                                        source=self.dest,
                                                        name='Rx_IngressEgressTwoRouterTest')

    def on_sendable(self, event):
        if not self.start:
            return

        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            self.sender.send(msg)
            self.n_sent += 1

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.start = True
            self.conn_sender = event.container.connect(self.sender_address)
            self.sender = event.container.create_sender(self.conn_sender,
                                                        target=self.dest,
                                                        name='Tx_IngressEgressTwoRouterTest')

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_received += 1

    def on_accepted(self, event):
        if event.sender:
            self.n_accept += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class IngressEgressOneRouterTest(MessagingHandler):
    def __init__(self, address, num_messages, large_message=False):
        super(IngressEgressOneRouterTest, self).__init__()
        self.sender = None
        self.receiver = None
        self.conn = None
        self.timer = None
        self.dest = 'examples'
        self.address = address
        self.n_sent = 0
        self.n_received = 0
        self.n_accepted = 0
        self.num_messages = num_messages
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False
        self.large_message = large_message

    def timeout(self):
        self.conn.close()

    def check_if_done(self):
        if not self.done and (self.n_sent == self.n_received
                              and self.n_sent == self.n_accepted):
            self.done = True
            self.sender_stats = get_link_info('Tx_IngressEgressOneRouterTest', self.address)
            self.receiver_stats = get_link_info('Rx_IngressEgressOneRouterTest', self.address)
            self.conn.close()
            self.timer.cancel()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.conn = event.container.connect(self.address)
        self.sender = event.container.create_sender(self.conn,
                                                    target=self.dest,
                                                    name='Tx_IngressEgressOneRouterTest')
        self.receiver = event.container.create_receiver(self.conn,
                                                        source=self.dest,
                                                        name='Rx_IngressEgressOneRouterTest')

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_received += 1

    def on_accepted(self, event):
        self.n_accepted += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class RouteContainerEgressTest(MessagingHandler):
    def __init__(self, route_container_addr, sender_addr, num_messages, large_message=False):
        super(RouteContainerEgressTest, self).__init__()
        self.sender_addr = sender_addr
        self.route_container_addr = route_container_addr
        self.timer = None
        self.error = None
        self.receiver = None
        self.receiver_conn = None
        self.dest = "myListener.1"
        self.sender_conn = None
        self.sender = None
        self.start = False
        self.n_sent = 0
        self.n_received = 0
        self.n_accepted = 0
        self.num_messages = num_messages
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False

    def check_if_done(self):
        if not self.done and (self.n_sent == self.n_received
                              and self.n_sent == self.n_accepted):
            self.done = True
            self.sender_stats = get_link_info('Tx_RouteContainerEgressTest', self.sender_addr)
            self.receiver_stats = get_link_info('Rx_RouteContainerEgressTest', self.route_container_addr)
            self.receiver_conn.close()
            self.sender_conn.close()
            self.timer.cancel()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.receiver_conn = event.container.connect(self.route_container_addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        source=self.dest,
                                                        name='Rx_RouteContainerEgressTest')

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d self.n_received=%d" % (self.n_sent, self.self.n_received)
        self.sender_conn.close()
        self.receiver_conn.close()

    def on_sendable(self, event):
        if not self.start:
            return

        if self.n_sent < self.num_messages:
            msg = Message(body={'number': self.n_sent})
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_received += 1

    def on_accepted(self, event):
        self.n_accepted += 1
        self.check_if_done()

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.start = True
            self.sender_conn = event.container.connect(self.sender_addr)
            self.sender = event.container.create_sender(self.sender_conn,
                                                        target=self.dest,
                                                        name='Tx_RouteContainerEgressTest')

    def run(self):
        Container(self).run()


class RouteContainerIngressTest(MessagingHandler):
    def __init__(self, route_container_addr, receiver_addr):
        super(RouteContainerIngressTest, self).__init__()
        self.receiver_addr = receiver_addr
        self.route_container_addr = route_container_addr
        self.timer = None
        self.error = None
        self.receiver = None
        self.receiver_conn = None
        self.dest = "myListener.1"
        self.sender_conn = None
        self.sender = None
        self.sender1 = None
        self.start = False
        self.n_sent = 0
        self.n_received = 0
        self.n_accepted = 0
        self.num_messages = 20
        self.sender_stats = None
        self.sender1_stats = None
        self.receiver_stats = None
        self.done = False

    def check_if_done(self):
        if not self.done and (self.n_sent == self.n_received
                              and self.n_sent == self.n_accepted):
            self.done = True
            self.sender_stats = get_link_info('A', self.route_container_addr)
            self.sender1_stats = get_link_info('B', self.route_container_addr)
            self.receiver_stats = get_link_info('Rx_RouteContainerIngressTest', self.receiver_addr)
            self.receiver_conn.close()
            self.sender_conn.close()
            self.timer.cancel()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.receiver_conn = event.container.connect(self.receiver_addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        source=self.dest,
                                                        name='Rx_RouteContainerIngressTest')
        self.sender_conn = event.container.connect(self.route_container_addr)

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d self.n_received=%d" % (self.n_sent, self.self.n_received)
        self.sender_conn.close()
        self.receiver_conn.close()

    def on_sendable(self, event):
        if not self.start:
            return

        if self.n_sent < self.num_messages:
            msg = Message(body={'number': self.n_sent})
            self.sender.send(msg)
            self.n_sent += 1

            self.sender1.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_received += 1

    def on_accepted(self, event):
        self.n_accepted += 1
        self.check_if_done()

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.start = True
            # Create 2 senders. Each sender will send 10 messages each
            self.sender = event.container.create_sender(self.sender_conn, self.dest, name="A")
            self.sender1 = event.container.create_sender(self.sender_conn, self.dest, name="B")

    def run(self):
        Container(self).run()


class IngressEgressTransitLinkRouteTest(MessagingHandler):
    def __init__(self, sender_addr, receiver_addr, num_messages, large_message=False):
        super(IngressEgressTransitLinkRouteTest, self).__init__()
        self.timer = None
        self.receiver_conn = None
        self.receiver = None
        self.sender = None
        self.sender_conn = None
        self.dest = "pulp.task"
        self.start = False
        self.n_sent = 0
        self.num_messages = num_messages
        self.n_received = 0
        self.n_accepted = 0
        self.sender_addr = sender_addr
        self.receiver_addr = receiver_addr
        self.error = None
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False
        self.large_message = large_message

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d self.n_received=%d" % (self.n_sent, self.self.n_received)
        self.sender_conn.close()
        self.receiver_conn.close()

    def check_if_done(self):
        if not self.done and (self.n_sent == self.n_received
                              and self.n_sent == self.n_accepted):
            self.done = True
            self.sender_stats = get_link_info('Tx_IngressEgressTransitLinkRouteTest',
                                              self.sender_addr)
            self.receiver_stats = get_link_info('Rx_IngressEgressTransitLinkRouteTest',
                                                self.receiver_addr)
            self.receiver_conn.close()
            self.sender_conn.close()
            self.timer.cancel()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.receiver_conn = event.container.connect(self.receiver_addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        source=self.dest,
                                                        name='Rx_IngressEgressTransitLinkRouteTest')

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.sender_conn = event.container.connect(self.sender_addr)
            self.sender = event.container.create_sender(self.sender_conn,
                                                        target=self.dest,
                                                        name='Tx_IngressEgressTransitLinkRouteTest')
            self.start = True

    def on_sendable(self, event):
        if not self.start:
            return

        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            self.sender.send(msg)
            self.n_sent += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_received += 1

    def on_accepted(self, event):
        self.n_accepted += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class ReleasedDroppedPresettledCountTest(MessagingHandler):
    def __init__(self, sender_addr, num_messages, large_message=False):
        super(ReleasedDroppedPresettledCountTest, self).__init__()
        self.timer = None
        self.sender_conn = None
        self.sender = None
        self.error = None
        self.n_sent = 0
        self.num_messages = num_messages
        self.sender_addr = sender_addr
        self.sender_stats = None

        # We are sending to a multicast address
        self.dest = "multicast"
        self.n_released = 0
        self.expect_released = 10
        self.done = False
        self.large_message = large_message

    def check_if_done(self):
        if not self.done and self.expect_released == self.n_released:
            self.done = True
            self.sender_stats = get_link_info('ReleasedDroppedPresettledCountTest',
                                              self.sender_addr)
            self.sender_conn.close()
            self.timer.cancel()

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d, self.self.n_released=%d  " % (self.n_sent, self.n_released)
        self.sender_conn.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn = event.container.connect(self.sender_addr)

        # Note that this is an anonymous link which will be granted credit w/o
        # blocking for consumers.  Therefore all messages sent to this address
        # will be dropped
        self.sender = event.container.create_sender(self.sender_conn,
                                                    name='ReleasedDroppedPresettledCountTest')

    def on_sendable(self, event):
        # We are sending a total of 20 deliveries. 10 unsettled and 10 pre-settled to a multicast address
        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            msg.address = self.dest
            dlv = self.sender.send(msg)
            if self.n_sent < 10:
                dlv.settle()
            self.n_sent += 1

    def on_released(self, event):
        self.n_released += 1
        self.check_if_done()

    def run(self):
        Container(self).run()


class RejectedDeliveriesTest(MessagingHandler):
    def __init__(self, addr, num_messages, large_message=False):
        super(RejectedDeliveriesTest, self).__init__(auto_accept=False)
        self.addr = addr
        self.dest = "someaddress"
        self.error = None
        self.n_sent = 0
        self.num_messages = num_messages
        self.n_rejected = 0
        self.sender_conn = None
        self.receiver_conn = None
        self.timer = None
        self.sender = None
        self.receiver = None
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False
        self.large_message = large_message

    def check_if_done(self):
        if not self.done and self.n_rejected == self.num_messages:
            self.done = True
            self.sender_stats = get_link_info('Tx_RejectedDeliveriesTest', self.addr)
            self.receiver_stats = get_link_info('Rx_RejectedDeliveriesTest', self.addr)
            self.sender_conn.close()
            self.receiver_conn.close()
            self.timer.cancel()

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d, self.self.n_rejected=%d  " % (self.n_sent, self.n_rejected)
        self.sender_conn.close()
        self.receiver_conn.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn = event.container.connect(self.addr)
        self.sender = event.container.create_sender(self.sender_conn,
                                                    target=self.dest,
                                                    name='Tx_RejectedDeliveriesTest')
        self.receiver_conn = event.container.connect(self.addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        source=self.dest,
                                                        name='Rx_RejectedDeliveriesTest')

    def on_rejected(self, event):
        self.n_rejected += 1
        self.check_if_done()

    def on_message(self, event):
        # We will reject every delivery we receive.
        self.reject(event.delivery)

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            self.sender.send(msg)
            self.n_sent += 1

    def run(self):
        Container(self).run()


class ModifiedDeliveriesTest(MessagingHandler):
    def __init__(self, addr, num_messages, large_message=False):
        super(ModifiedDeliveriesTest, self).__init__(auto_accept=False)
        self.addr = addr
        self.dest = "someaddress"
        self.error = None
        self.n_sent = 0
        self.num_messages = num_messages
        self.n_modified = 0
        self.sender_conn = None
        self.receiver_conn = None
        self.timer = None
        self.sender = None
        self.receiver = None
        self.n_received = 0
        self.sender_stats = None
        self.receiver_stats = None
        self.done = False
        self.large_message = large_message

    def check_if_done(self):
        if not self.done and self.n_modified == self.num_messages:
            self.done = True
            self.sender_stats = get_link_info('Tx_ModifiedDeliveriesTest', self.addr)
            self.sender_conn.close()
            self.receiver_conn.close()
            self.timer.cancel()

    def timeout(self):
        self.error = "Timeout Expired: self.n_sent=%d, self.self.n_modified=%d  " % (self.n_sent, self.n_modified)
        self.sender_conn.close()
        self.receiver_conn.close()

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn = event.container.connect(self.addr)
        self.sender = event.container.create_sender(self.sender_conn,
                                                    target=self.dest,
                                                    name='Tx_ModifiedDeliveriesTest')
        self.receiver_conn = event.container.connect(self.addr)
        self.receiver = event.container.create_receiver(self.receiver_conn,
                                                        source=self.dest,
                                                        name='Rx_ModifiedDeliveriesTest')

    def on_released(self, event):
        if event.delivery.remote_state == Delivery.MODIFIED:
            self.n_modified += 1
        self.check_if_done()

    def on_message(self, event):
        # The messages have arrived at the receiver but we will not settle the message and instead just closed the
        # connection. Since the router did not receive the acknowledgements, it will send back MODIFIED dispositions
        # to the sender.
        self.n_received += 1
        # After 10 messages are received, simply close the receiver connection without acknowledging the messages
        if self.n_received == self.num_messages:
            self.receiver_stats = get_link_info('Rx_ModifiedDeliveriesTest', self.addr)
            self.receiver_conn.close()

    def on_sendable(self, event):
        if self.n_sent < self.num_messages:
            msg = Message(body=get_body(self.n_sent, self.large_message))
            self.sender.send(msg)
            self.n_sent += 1

    def run(self):
        Container(self).run()