#
# 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.
#

import time

from proton import Message, symbol
from proton.handlers import MessagingHandler
from proton.reactor import Container

from system_test import TestCase, Qdrouterd, main_module, TIMEOUT, TestTimeout
from system_test import unittest, skip_test_in_ci
from system_test import Logger


class AddrTimer:
    def __init__(self, parent):
        self.parent = parent

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


class RouterTest(TestCase):

    inter_router_port = None

    @classmethod
    def setUpClass(cls):
        """Start a router"""
        super(RouterTest, cls).setUpClass()

        def router(name, mode, connection, extra=None):
            config = [
                ('router', {'mode': mode, 'id': name}),
                ('listener', {'port': cls.tester.get_port(), 'stripAnnotations': 'no'}),
                ('listener', {'port': cls.tester.get_port(), 'role': 'route-container', 'name': 'WP'}),
                ('address',  {'prefix': 'dest', 'enableFallback': 'yes'}),
                ('autoLink', {'connection': 'WP', 'address': 'dest.al', 'direction': 'out', 'fallback': 'yes'}),
                ('autoLink', {'connection': 'WP', 'address': 'dest.al', 'direction': 'in',  'fallback': 'yes'}),
                connection
            ]

            if extra:
                config.append(extra)
            config = Qdrouterd.Config(config)
            cls.routers.append(cls.tester.qdrouterd(name, config, wait=True))

        cls.routers = []

        inter_router_port = cls.tester.get_port()
        edge_port_A       = cls.tester.get_port()
        edge_port_B       = cls.tester.get_port()

        router('INT.A', 'interior', ('listener', {'role': 'inter-router', 'port': inter_router_port}),
               ('listener', {'role': 'edge', 'port': edge_port_A}))
        router('INT.B', 'interior', ('connector', {'name': 'connectorToA', 'role': 'inter-router', 'port': inter_router_port}),
               ('listener', {'role': 'edge', 'port': edge_port_B}))
        router('EA1',   'edge',     ('connector', {'name': 'edge', 'role': 'edge', 'port': edge_port_A}))
        router('EA2',   'edge',     ('connector', {'name': 'edge', 'role': 'edge', 'port': edge_port_A}))
        router('EB1',   'edge',     ('connector', {'name': 'edge', 'role': 'edge', 'port': edge_port_B}))
        router('EB2',   'edge',     ('connector', {'name': 'edge', 'role': 'edge', 'port': edge_port_B}))

        cls.routers[0].wait_router_connected('INT.B')
        cls.routers[1].wait_router_connected('INT.A')

        cls.ROUTER_INTA = cls.routers[0].addresses[0]
        cls.ROUTER_INTB = cls.routers[1].addresses[0]
        cls.ROUTER_EA1 = cls.routers[2].addresses[0]
        cls.ROUTER_EA2 = cls.routers[3].addresses[0]
        cls.ROUTER_EB1 = cls.routers[4].addresses[0]
        cls.ROUTER_EB2 = cls.routers[5].addresses[0]

        cls.ROUTER_INTA_WP = cls.routers[0].addresses[1]
        cls.ROUTER_INTB_WP = cls.routers[1].addresses[1]
        cls.ROUTER_EA1_WP = cls.routers[2].addresses[1]
        cls.ROUTER_EA2_WP = cls.routers[3].addresses[1]
        cls.ROUTER_EB1_WP = cls.routers[4].addresses[1]
        cls.ROUTER_EB2_WP = cls.routers[5].addresses[1]

    def test_01_sender_first_primary_same_interior(self):
        test = SenderFirstTest(self.ROUTER_INTA,
                               self.ROUTER_INTA,
                               'dest.01', False)
        test.run()
        self.assertIsNone(test.error)

    def test_02_sender_first_fallback_same_interior(self):
        test = SenderFirstTest(self.ROUTER_INTA,
                               self.ROUTER_INTA,
                               'dest.02', True)
        test.run()
        self.assertIsNone(test.error)

    def test_03_sender_first_primary_same_edge(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_EA1,
                               'dest.03', False)
        test.run()
        self.assertIsNone(test.error)

    def test_04_sender_first_fallback_same_edge(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_EA1,
                               'dest.04', True)
        test.run()
        self.assertIsNone(test.error)

    def test_05_sender_first_primary_interior_interior(self):
        test = SenderFirstTest(self.ROUTER_INTA,
                               self.ROUTER_INTB,
                               'dest.05', False)
        test.run()
        self.assertIsNone(test.error)

    def test_06_sender_first_fallback_interior_interior(self):
        test = SenderFirstTest(self.ROUTER_INTA,
                               self.ROUTER_INTB,
                               'dest.06', True)
        test.run()
        self.assertIsNone(test.error)

    def test_07_sender_first_primary_edge_interior(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_INTB,
                               'dest.07', False)
        test.run()
        self.assertIsNone(test.error)

    def test_08_sender_first_fallback_edge_interior(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_INTB,
                               'dest.08', True)
        test.run()
        self.assertIsNone(test.error)

    def test_09_sender_first_primary_interior_edge(self):
        test = SenderFirstTest(self.ROUTER_INTB,
                               self.ROUTER_EA1,
                               'dest.09', False)
        test.run()
        self.assertIsNone(test.error)

    def test_10_sender_first_fallback_interior_edge(self):
        test = SenderFirstTest(self.ROUTER_INTB,
                               self.ROUTER_EA1,
                               'dest.10', True)
        test.run()
        self.assertIsNone(test.error)

    def test_11_sender_first_primary_edge_edge(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_EB1,
                               'dest.11', False)
        test.run()
        self.assertIsNone(test.error)

    def test_12_sender_first_fallback_edge_edge(self):
        test = SenderFirstTest(self.ROUTER_EA1,
                               self.ROUTER_EB1,
                               'dest.12', True)
        test.run()
        self.assertIsNone(test.error)

    def test_13_receiver_first_primary_same_interior(self):
        test = ReceiverFirstTest(self.ROUTER_INTA,
                                 self.ROUTER_INTA,
                                 'dest.13', False)
        test.run()
        self.assertIsNone(test.error)

    def test_14_receiver_first_fallback_same_interior(self):
        test = ReceiverFirstTest(self.ROUTER_INTA,
                                 self.ROUTER_INTA,
                                 'dest.14', True)
        test.run()
        self.assertIsNone(test.error)

    def test_15_receiver_first_primary_same_edge(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_EA1,
                                 'dest.15', False)
        test.run()
        self.assertIsNone(test.error)

    def test_16_receiver_first_fallback_same_edge(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_EA1,
                                 'dest.16', True)
        test.run()
        self.assertIsNone(test.error)

    def test_17_receiver_first_primary_interior_interior(self):
        test = ReceiverFirstTest(self.ROUTER_INTA,
                                 self.ROUTER_INTB,
                                 'dest.17', False)
        test.run()
        self.assertIsNone(test.error)

    def test_18_receiver_first_fallback_interior_interior(self):
        test = ReceiverFirstTest(self.ROUTER_INTA,
                                 self.ROUTER_INTB,
                                 'dest.18', True)
        test.run()
        self.assertIsNone(test.error)

    def test_19_receiver_first_primary_edge_interior(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_INTB,
                                 'dest.19', False)
        test.run()
        self.assertIsNone(test.error)

    def test_20_receiver_first_fallback_edge_interior(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_INTB,
                                 'dest.20', True)
        test.run()
        self.assertIsNone(test.error)

    def test_21_receiver_first_primary_interior_edge(self):
        test = ReceiverFirstTest(self.ROUTER_INTB,
                                 self.ROUTER_EA1,
                                 'dest.21', False)
        test.run()
        self.assertIsNone(test.error)

    def test_22_receiver_first_fallback_interior_edge(self):
        test = ReceiverFirstTest(self.ROUTER_INTB,
                                 self.ROUTER_EA1,
                                 'dest.22', True)
        test.run()
        self.assertIsNone(test.error)

    def test_23_receiver_first_primary_edge_edge(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_EB1,
                                 'dest.23', False)
        test.run()
        self.assertIsNone(test.error)

    def test_24_receiver_first_fallback_edge_edge(self):
        test = ReceiverFirstTest(self.ROUTER_EA1,
                                 self.ROUTER_EB1,
                                 'dest.24', True)
        test.run()
        self.assertIsNone(test.error)

    skip_reason = 'Test skipped until switchover use case resolved'

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_25_switchover_same_edge(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EA1, "EA1"],
                              'dest.25')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_26_switchover_same_interior(self):
        test = SwitchoverTest([self.ROUTER_INTA, "INTA"],
                              [self.ROUTER_INTA, "INTA"],
                              [self.ROUTER_INTA, "INTA"],
                              'dest.26')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_27_switchover_local_edge_alt_remote_interior(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_INTA, "INTA"],
                              [self.ROUTER_EA1, "EA1"],
                              'dest.27')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_28_switchover_local_edge_alt_remote_edge(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EB1, "EB1"],
                              [self.ROUTER_EA1, "EA1"],
                              'dest.28')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_29_switchover_local_edge_pri_remote_interior(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_INTA, "INTA"],
                              'dest.29')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_30_switchover_local_interior_pri_remote_edge(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EB1, "EB1"],
                              'dest.30')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_31_switchover_local_interior_alt_remote_interior(self):
        test = SwitchoverTest([self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_INTA, "INTA"],
                              [self.ROUTER_INTB, "INTB"],
                              'dest.31')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_32_switchover_local_interior_alt_remote_edge(self):
        test = SwitchoverTest([self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_EA2, "EA2"],
                              [self.ROUTER_INTB, "INTB"],
                              'dest.32')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_33_switchover_local_interior_pri_remote_interior(self):
        test = SwitchoverTest([self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_INTA, "INTA"],
                              'dest.33')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_34_switchover_local_interior_pri_remote_edge(self):
        test = SwitchoverTest([self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_EB1, "EB1"],
                              'dest.34')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_35_switchover_mix_1(self):
        test = SwitchoverTest([self.ROUTER_INTA, "INTA"],
                              [self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_EA1, "EA1"],
                              'dest.35')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_36_switchover_mix_2(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_INTA, "INTA"],
                              'dest.36')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_37_switchover_mix_3(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_INTB, "INTB"],
                              [self.ROUTER_EB1, "EB1"],
                              'dest.37')
        test.run()
        self.assertIsNone(test.error)

    @unittest.skipIf(skip_test_in_ci('QPID_SYSTEM_TEST_SKIP_FALLBACK_SWITCHOVER_TEST'), skip_reason)
    def test_38_switchover_mix_4(self):
        test = SwitchoverTest([self.ROUTER_EA1, "EA1"],
                              [self.ROUTER_EA2, "EA2"],
                              [self.ROUTER_EB1, "EB1"],
                              'dest.38')
        test.run()
        self.assertIsNone(test.error)

    def test_39_auto_link_sender_first_fallback_same_interior(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_INTA,
                                       self.ROUTER_INTA_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_40_auto_link_sender_first_fallback_same_edge(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_EA1,
                                       self.ROUTER_EA1_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_41_auto_link_sender_first_fallback_interior_interior(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_INTA,
                                       self.ROUTER_INTB_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_42_auto_link_sender_first_fallback_edge_interior(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_EA1,
                                       self.ROUTER_INTA_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_43_auto_link_sender_first_fallback_interior_edge(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_INTB,
                                       self.ROUTER_EA1_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_44_auto_link_sender_first_fallback_edge_edge(self):
        test = SenderFirstAutoLinkTest(self.ROUTER_EA1,
                                       self.ROUTER_EB1_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_45_auto_link_receiver_first_fallback_same_interior(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_INTA,
                                         self.ROUTER_INTA_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_46_auto_link_receiver_first_fallback_same_edge(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_EA1,
                                         self.ROUTER_EA1_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_47_auto_link_receiver_first_fallback_interior_interior(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_INTA,
                                         self.ROUTER_INTB_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_48_auto_link_receiver_first_fallback_edge_interior(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_EA1,
                                         self.ROUTER_INTB_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_49_auto_link_receiver_first_fallback_interior_edge(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_INTB,
                                         self.ROUTER_EA1_WP)
        test.run()
        self.assertIsNone(test.error)

    def test_50_auto_link_receiver_first_fallback_edge_edge(self):
        test = ReceiverFirstAutoLinkTest(self.ROUTER_EA1,
                                         self.ROUTER_EB1_WP)
        test.run()
        self.assertIsNone(test.error)


class SenderFirstTest(MessagingHandler):
    def __init__(self, sender_host, receiver_host, addr, rx_fallback):
        super(SenderFirstTest, self).__init__()
        self.sender_host   = sender_host
        self.receiver_host = receiver_host
        self.addr          = addr
        self.rx_fallback   = rx_fallback
        self.count         = 300

        self.sender_conn   = None
        self.receiver_conn = None
        self.error         = None
        self.n_tx          = 0
        self.n_rx          = 0
        self.n_rel         = 0
        self.tx_seq        = 0

    def timeout(self):
        self.error = "Timeout Expired - n_tx=%d, n_rx=%d, n_rel=%d" % (self.n_tx, self.n_rx, self.n_rel)
        self.sender_conn.close()
        self.receiver_conn.close()

    def fail(self, error):
        self.error = error
        self.sender_conn.close()
        self.receiver_conn.close()
        self.timer.cancel()

    def on_start(self, event):
        self.timer         = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn   = event.container.connect(self.sender_host)
        self.receiver_conn = event.container.connect(self.receiver_host)
        self.sender        = event.container.create_sender(self.sender_conn, self.addr, name=(self.addr + "_sender"))

    def on_link_opened(self, event):
        if event.sender == self.sender:
            rname = self.addr + "_receiver_fallback_" + ("true" if self.rx_fallback else "false")
            self.receiver = event.container.create_receiver(self.receiver_conn, self.addr, name=rname)
            if self.rx_fallback:
                self.receiver.source.capabilities.put_symbol("qd.fallback")

    def on_sendable(self, event):
        if event.sender == self.sender:
            while self.sender.credit > 0 and self.n_tx < self.count:
                self.sender.send(Message("Msg %s %d %d" % (self.addr, self.tx_seq, self.n_tx)))
                self.n_tx += 1
                self.tx_seq += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_rx += 1
            if self.n_rx == self.count:
                self.fail(None)

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

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


class ReceiverFirstTest(MessagingHandler):
    def __init__(self, sender_host, receiver_host, addr, rx_fallback):
        super(ReceiverFirstTest, self).__init__()
        self.sender_host   = sender_host
        self.receiver_host = receiver_host
        self.addr          = addr
        self.rx_fallback   = rx_fallback
        self.count         = 300

        self.sender_conn   = None
        self.receiver_conn = None
        self.error         = None
        self.n_tx          = 0
        self.n_rx          = 0
        self.n_rel         = 0
        self.tx_seq        = 0

    def timeout(self):
        self.error = "Timeout Expired - n_tx=%d, n_rx=%d, n_rel=%d" % (self.n_tx, self.n_rx, self.n_rel)
        self.sender_conn.close()
        self.receiver_conn.close()

    def fail(self, error):
        self.error = error
        self.sender_conn.close()
        self.receiver_conn.close()
        self.timer.cancel()

    def on_start(self, event):
        self.timer         = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn   = event.container.connect(self.sender_host)
        self.receiver_conn = event.container.connect(self.receiver_host)
        rname = self.addr + "_receiver_fallback_" + ("true" if self.rx_fallback else "false")
        self.receiver      = event.container.create_receiver(self.receiver_conn, self.addr, name=rname)
        if self.rx_fallback:
            self.receiver.source.capabilities.put_symbol("qd.fallback")

    def on_link_opened(self, event):
        if event.receiver == self.receiver:
            self.sender = event.container.create_sender(self.sender_conn, self.addr, name=(self.addr + "_sender"))

    def on_sendable(self, event):
        if event.sender == self.sender:
            while self.sender.credit > 0 and self.n_tx < self.count:
                self.sender.send(Message("Msg %s %d %d" % (self.addr, self.tx_seq, self.n_tx)))
                self.n_tx += 1
                self.tx_seq += 1

    def on_message(self, event):
        if event.receiver == self.receiver:
            self.n_rx += 1
            if self.n_rx == self.count:
                self.fail(None)

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

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


class SwitchoverTest(MessagingHandler):
    def __init__(self, sender_host, primary_host, fallback_host, addr):
        super(SwitchoverTest, self).__init__()
        self.sender_host    = sender_host[0]
        self.primary_host   = primary_host[0]
        self.fallback_host  = fallback_host[0]
        self.sender_name    = sender_host[1]
        self.primary_name   = primary_host[1]
        self.fallback_name  = fallback_host[1]
        self.addr           = addr
        self.count          = 300

        # DISPATCH-2213 back off on logging.
        self.log_sends      = 100  # every 100th send
        self.log_recvs      = 100  # every 100th receive
        self.log_released   = 100  # every 100th sender released

        self.sender_conn    = None
        self.primary_conn   = None
        self.fallback_conn  = None
        self.primary_open   = False
        self.fallback_open  = False
        self.error          = None
        self.n_tx           = 0
        self.n_rx           = 0
        self.n_rel          = 0
        self.phase          = 0
        self.tx_seq         = 0
        self.local_rel      = 0

        self.log_prefix     = "FALLBACK_TEST %s" % self.addr
        self.logger = Logger("SwitchoverTest_%s" % addr, print_to_console=False)
        # Prepend a convenience SERVER line for scraper tool.
        # Then the logs from this test can be merged with the router logs in scraper.
        self.logger.log("SERVER (info) Container Name: %s" % self.addr)
        self.logger.log("%s SwitchoverTest sender:%s primary:%s fallback:%s" %
                        (self.log_prefix, self.sender_name, self.primary_name, self.fallback_name))

    def timeout(self):
        self.error = "Timeout Expired - n_tx=%d, n_rx=%d, n_rel=%d, phase=%d, local_rel=%d" % \
                     (self.n_tx, self.n_rx, self.n_rel, self.phase, self.local_rel)
        self.sender_conn.close()
        self.primary_conn.close()
        self.fallback_conn.close()

    def fail(self, error):
        self.error = error
        self.sender_conn.close()
        self.primary_conn.close()
        self.fallback_conn.close()
        self.timer.cancel()

    def on_start(self, event):
        self.timer              = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.logger.log("%s Opening sender connection to %s" % (self.log_prefix, self.sender_name))
        self.sender_conn        = event.container.connect(self.sender_host)
        self.logger.log("%s Opening primary receiver connection to %s" % (self.log_prefix, self.primary_name))
        self.primary_conn       = event.container.connect(self.primary_host)
        self.logger.log("%s Opening fallback receiver connection to %s" % (self.log_prefix, self.fallback_name))
        self.fallback_conn      = event.container.connect(self.fallback_host)
        self.logger.log("%s Opening primary receiver to %s" % (self.log_prefix, self.primary_name))
        self.primary_receiver   = event.container.create_receiver(self.primary_conn, self.addr, name=(self.addr + "_primary_receiver"))
        self.logger.log("%s Opening fallback receiver to %s" % (self.log_prefix, self.fallback_name))
        self.fallback_receiver  = event.container.create_receiver(self.fallback_conn, self.addr, name=(self.addr + "fallback_receiver"))
        self.fallback_receiver.source.capabilities.put_object(symbol("qd.fallback"))

    def on_link_opened(self, event):
        receiver_event = False
        if event.receiver == self.primary_receiver:
            self.logger.log("%s Primary receiver opened" % self.log_prefix)
            self.primary_open = True
            receiver_event = True
        if event.receiver == self.fallback_receiver:
            self.logger.log("%s Fallback receiver opened" % self.log_prefix)
            self.fallback_open = True
            receiver_event = True
        if receiver_event and self.primary_open and self.fallback_open:
            self.logger.log("%s Opening sender to %s" % (self.log_prefix, self.sender_name))
            self.sender = event.container.create_sender(self.sender_conn, self.addr, name=(self.addr + "_sender"))

    def on_link_closed(self, event):
        if event.receiver == self.primary_receiver:
            self.logger.log("%s Primary receiver closed. Start phase 1 send" % self.log_prefix)
            self.n_rx = 0
            self.n_tx = 0
            self.send()

    def send(self):
        e_credit = self.sender.credit
        e_n_tx = self.n_tx
        e_tx_seq = self.tx_seq
        last_message = Message("None")
        while self.sender.credit > 0 and self.n_tx < self.count and not self.sender.drain_mode:
            last_message = Message("Msg %s %d %d" % (self.addr, self.tx_seq, self.n_tx))
            self.sender.send(last_message)
            self.n_tx += 1
            self.tx_seq += 1
        if self.sender.drain_mode:
            n_drained = self.sender.drained()
            self.logger.log("%s sender.drained() drained %d credits" % (self.log_prefix, n_drained))
        if self.n_tx > e_n_tx and self.n_tx % self.log_sends == 0:  # if sent then log every Nth message
            self.logger.log("%s send() exit: last sent '%s' phase=%d, credit=%3d->%3d, n_tx=%4d->%4d, tx_seq=%4d->%4d, n_rel=%4d" %
                            (self.log_prefix, last_message.body, self.phase, e_credit, self.sender.credit,
                             e_n_tx, self.n_tx, e_tx_seq, self.tx_seq, self.n_rel))

    def on_sendable(self, event):
        if event.sender == self.sender:
            self.send()
        else:
            self.fail("%s on_sendable event not from the only sender")

    def on_message(self, event):
        if event.receiver == self.primary_receiver:
            if self.phase == 0:
                self.n_rx += 1
                if self.n_rx % self.log_recvs == 0:
                    self.logger.log("%s Received phase 0 message '%s', n_rx=%d" %
                                    (self.log_prefix, event.message.body, self.n_rx))
                if self.n_rx == self.count:
                    self.logger.log("%s Triggering fallback by closing primary receiver on %s. Test phase 0->1." %
                                    (self.log_prefix, self.primary_name))
                    self.phase = 1
                    self.primary_receiver.close()
            else:
                # Phase 1 messages are unexpected on primary receiver
                self.logger.log("%s Phase %d message received on primary: '%s'" % (self.log_prefix, self.phase, event.message.body))
                self.fail("Receive phase1 message on primary receiver")
        elif event.receiver == self.fallback_receiver:
            if self.phase == 0:
                # Phase 0 message over fallback receiver. This may happen because
                # primary receiver is on a distant router and the fallback receiver is local.
                # Release the message to keep trying until the primary receiver kicks in.
                self.release(event.delivery)
                self.n_rel += 1
                self.n_tx -= 1
                self.local_rel += 1
                if self.local_rel % self.log_recvs == 0:
                    self.logger.log("%s Released phase 0 over fallback: msg:'%s', n_rx=%d, n_tx=%d, n_rel=%d, local_rel=%d" %
                                    (self.log_prefix, event.message.body, self.n_rx, self.n_tx, self.n_rel, self.local_rel))
                    time.sleep(0.02)
            else:
                self.n_rx += 1
                if self.n_rx % self.log_recvs == 0:
                    self.logger.log("%s Received phase 1 over fallback: msg:'%s', n_rx=%d" %
                                    (self.log_prefix, event.message.body, self.n_rx))
                if self.n_rx == self.count:
                    self.logger.log("%s Success" % self.log_prefix)
                    self.fail(None)
        else:
            self.fail("%s message received on unidentified receiver" % self.addr)

    def on_released(self, event):
        # event type pn_delivery for sender
        self.n_rel += 1
        self.n_tx  -= 1
        if self.n_rel % self.log_released == 0:
            self.logger.log("%s on_released: sender delivery was released. Adjusted counts: n_rel=%d, n_tx=%d" %
                            (self.log_prefix, self.n_rel, self.n_tx))
        if event.sender is None:
            self.fail("on_released event not related to sender")

    def run(self):
        Container(self).run()
        if self.error is not None:
            self.logger.dump()


class SenderFirstAutoLinkTest(MessagingHandler):
    def __init__(self, sender_host, receiver_host):
        super(SenderFirstAutoLinkTest, self).__init__()
        self.sender_host   = sender_host
        self.receiver_host = receiver_host
        self.addr          = "dest.al"
        self.count         = 300

        self.sender_conn   = None
        self.receiver_conn = None
        self.error         = None
        self.n_tx          = 0
        self.n_rx          = 0
        self.n_rel         = 0
        self.tx_seq        = 0

    def timeout(self):
        self.error = "Timeout Expired - n_tx=%d, n_rx=%d, n_rel=%d" % (self.n_tx, self.n_rx, self.n_rel)
        self.sender_conn.close()
        self.receiver_conn.close()

    def fail(self, error):
        self.error = error
        self.sender_conn.close()
        self.receiver_conn.close()
        self.timer.cancel()

    def on_start(self, event):
        self.timer       = event.reactor.schedule(TIMEOUT, TestTimeout(self))
        self.sender_conn = event.container.connect(self.sender_host)
        self.sender      = event.container.create_sender(self.sender_conn, self.addr, name=(self.addr + "_sender"))

    def on_link_opening(self, event):
        if event.sender:
            self.alt_sender = event.sender
            event.sender.source.address = self.addr
            event.sender.open()

        elif event.receiver:
            self.alt_receiver = event.receiver
            event.receiver.target.address = self.addr
            event.receiver.open()

    def on_link_opened(self, event):
        if event.sender == self.sender:
            self.receiver_conn = event.container.connect(self.receiver_host)

    def send(self):
        while self.sender.credit > 0 and self.n_tx < self.count:
            self.sender.send(Message("Msg %s %d %d" % (self.addr, self.tx_seq, self.n_tx)))
            self.n_tx += 1
            self.tx_seq += 1

    def on_sendable(self, event):
        if event.sender == self.sender:
            self.send()

    def on_message(self, event):
        self.n_rx += 1
        if self.n_rx == self.count:
            self.fail(None)

    def on_released(self, event):
        self.n_rel += 1
        self.n_tx -= 1
        self.send()

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


class ReceiverFirstAutoLinkTest(MessagingHandler):
    def __init__(self, sender_host, receiver_host):
        super(ReceiverFirstAutoLinkTest, self).__init__()
        self.sender_host   = sender_host
        self.receiver_host = receiver_host
        self.addr          = "dest.al"
        self.count         = 300

        self.sender_conn   = None
        self.receiver_conn = None
        self.alt_receiver  = None
        self.error         = None
        self.n_tx          = 0
        self.n_rx          = 0
        self.n_rel         = 0
        self.tx_seq        = 0

    def timeout(self):
        self.error = "Timeout Expired - n_tx=%d, n_rx=%d, n_rel=%d" % (self.n_tx, self.n_rx, self.n_rel)
        self.sender_conn.close()
        self.receiver_conn.close()

    def fail(self, error):
        self.error = error
        self.sender_conn.close()
        self.receiver_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_host)

    def on_link_opening(self, event):
        if event.sender:
            self.alt_sender = event.sender
            event.sender.source.address = self.addr
            event.sender.open()

        elif event.receiver:
            self.alt_receiver = event.receiver
            event.receiver.target.address = self.addr
            event.receiver.open()

    def on_link_opened(self, event):
        if event.receiver == self.alt_receiver and not self.sender_conn:
            self.sender_conn = event.container.connect(self.sender_host)
            self.sender      = event.container.create_sender(self.sender_conn, self.addr)

    def send(self):
        while self.sender.credit > 0 and self.n_tx < self.count:
            self.sender.send(Message("Msg %s %d %d" % (self.addr, self.tx_seq, self.n_tx)))
            self.n_tx += 1
            self.tx_seq += 1

    def on_sendable(self, event):
        if event.sender == self.sender:
            self.send()

    def on_message(self, event):
        self.n_rx += 1
        if self.n_rx == self.count:
            self.fail(None)

    def on_released(self, event):
        self.n_rel += 1
        self.n_tx -= 1
        self.send()

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


if __name__ == '__main__':
    unittest.main(main_module())