#
# 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 system_test import TestCase, Qdrouterd, QdManager, TIMEOUT

from proton import Message
from proton import Endpoint
from proton.handlers import MessagingHandler
from proton.reactor import Container

# test the request/response core client messaging API
#
# These tests rely on enabling the router test hooks, which instantiates a test
# client (see modules/test_hooks/core_test_hooks) see core_test_hooks.c

CONTAINER_ID = "org.apache.qpid.dispatch.test_core_client"
TARGET_ADDR = "test_core_client_address"


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

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

        cls.router = cls.tester.qdrouterd("A", config, cl_args=["-T"])

    def test_send_receive(self):
        ts = TestService(self.router.addresses[0], credit=250)
        ts.run()
        self.assertTrue(ts.error is None)
        self.assertEqual(250, ts.in_count)
        self.assertEqual(250, ts.out_count)

    def test_credit_starve(self):
        ts = TestCreditStarve(self.router.addresses[0])
        ts.run()
        self.assertTrue(ts.error is None)
        self.assertTrue(ts.starved)
        self.assertEqual(10, ts.in_count)

    def test_unexpected_conn_close(self):
        ts = TestEarlyClose(self.router.addresses[0])
        ts.run()
        self.assertTrue(ts.error is None)
        self.assertTrue(ts.in_count >= 1)

    def test_bad_format(self):
        ts = TestNoCorrelationId(self.router.addresses[0])
        ts.run()
        self.assertTrue(ts.error is None)
        self.assertTrue(ts.rejected)

    def test_old_cid(self):
        ts = TestOldCorrelationId(self.router.addresses[0])
        ts.run()
        self.assertTrue(ts.error is None)
        self.assertTrue(ts.accepted)

    def test_call_timeout(self):
        qm = QdManager(self, self.router.addresses[0])
        ts = TestCallTimeout(self.router.addresses[0], qm)
        ts.run()
        self.assertEqual("TIMED OUT!", ts.error)


class TestService(MessagingHandler):
    """a service that the core client can communicate with"""
    __test__ = False

    class Timeout:
        def __init__(self, service):
            self.service = service

        def on_timer_task(self, event):
            self.service.timeout()

    def __init__(self, address, container_id=CONTAINER_ID, credit=1):
        super(TestService, self).__init__(prefetch=0)
        self._container = Container(self)
        self._container.container_id = CONTAINER_ID
        self._conn = None
        self.address = address
        self.timer = None
        self.error = None
        self.reply_link = None
        self.incoming_link = None
        self.credit = credit
        self.in_count = 0
        self.out_count = 0

    def fail(self, error):
        self.error = error
        if self._conn:
            self._conn.close()

    def timeout(self):
        self.fail("Timeout expired")

    def on_start(self, event):
        self.timer = event.reactor.schedule(TIMEOUT, self.Timeout(self))
        self._conn = event.container.connect(self.address)

    def on_link_opening(self, event):
        link = event.link
        if link.state & Endpoint.LOCAL_UNINIT:
            link.source.copy(link.remote_source)
            link.target.copy(link.remote_target)
        if event.sender:
            if not link.remote_source.dynamic:
                self.fail("expected dynamic source terminus")
                return
            link.source.dynamic = False
            link.source.address = "a/reply/address"
            self.reply_link = link
        else:
            link.flow(self.credit)
            self.incoming_link = link

    def create_reply(self, message):
        return Message(body=message.body,
                       correlation_id=message.correlation_id)

    # echo back to sender
    def on_message(self, event):
        self.in_count += 1
        cid = event.message.correlation_id
        self.reply_link.send(self.create_reply(event.message))

    # stop when all sent messages have settled
    def on_settled(self, event):
        self.out_count += 1
        self.credit -= 1
        if self.credit == 0:
            self._conn.close()

    def on_connection_closed(self, event):
        self._conn = None

    def run(self):
        self._container.timeout = 1.0
        self._container.start()
        while self._container.process():
            if self._conn is None and self._container.quiesced:
                break
        self._container.stop()
        self._container.process()


class TestCreditStarve(TestService):
    """wait until all credit is exhausted, then re-flow more credit"""
    __test__ = False

    def __init__(self, address):
        super(TestCreditStarve, self).__init__(address, credit=5)
        self.starved = False

    def on_settled(self, event):
        self.credit -= 1
        if self.credit == 0:
            if not self.starved:
                self.starved = True
                self.credit = 5
                self.incoming_link.drain(self.credit)
            else:
                self._conn.close()


class TestEarlyClose(TestService):
    """grant 10, but don't respond and close early"""
    __test__ = False

    def __init__(self, address):
        super(TestEarlyClose, self).__init__(address, credit=10)

    def on_message(self, event):
        self.in_count += 1
        if self.in_count == 1:
            self._conn.close()


class TestNoCorrelationId(TestService):
    __test__ = False

    def __init__(self, address):
        super(TestNoCorrelationId, self).__init__(address, credit=1)
        self.rejected = False

    def create_reply(self, message):
        return Message(body=dict())

    def on_rejected(self, event):
        self.rejected = True


class TestOldCorrelationId(TestService):
    __test__ = False

    def __init__(self, address):
        super(TestOldCorrelationId, self).__init__(address, credit=1)
        self.accepted = False

    def create_reply(self, message):
        return Message(body=dict(),
                       correlation_id="not going to match")

    def on_accepted(self, event):
        self.accepted = True


class TestCallTimeout(TestService):
    """test that the timeout is handled properly"""
    __test__ = False

    class PeriodicLogScrape:
        # periodically scan the log for the timeout error
        def __init__(self, service):
            self.service = service

        def on_timer_task(self, event):
            log = self.service.qm.get_log()
            for e in log:
                if (e[0] == 'ROUTER_CORE' and e[1] == 'error'
                        and e[2] == 'client test request done '
                                    'error=Timed out'):
                    # yes this is the line you're looking for:
                    self.service.error = "TIMED OUT!"
                    if self.service._conn:
                        self.service._conn.close()
                    return
            event.reactor.schedule(1, TestCallTimeout.PeriodicLogScrape(self.service))

    def __init__(self, address, qm):
        super(TestCallTimeout, self).__init__(address, credit=1)
        self.qm = qm

    def on_message(self, event):
        # drop it
        event.reactor.schedule(1, self.PeriodicLogScrape(self))