#
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.
#

from __future__ import unicode_literals
from __future__ import division
from __future__ import absolute_import
from __future__ import print_function

import time

from system_test import TestCase, Qdrouterd, TIMEOUT
from system_test import AsyncTestReceiver
from system_test import unittest
from test_broker import FakeBroker

from proton import Disposition
from proton import Message
from proton.utils import BlockingConnection
from proton.utils import SyncRequestResponse
from proton.utils import SendException
from proton.utils import LinkDetached


class LinkRouteLookupTest(TestCase):
    """
    Tests link route address lookup
    """
    # hardcoded values from the router's C code
    QD_TERMINUS_ADDRESS_LOOKUP = '_$qd.addr_lookup'
    PROTOCOL_VERSION = 1
    OPCODE_LINK_ROUTE_LOOKUP = 1
    QCM_ADDR_LOOKUP_OK = 0
    QCM_ADDR_LOOKUP_NOT_FOUND = 3

    def _check_response(self, message):
        is_instance_dict = isinstance(message.properties, dict)
        self.assertTrue(is_instance_dict)
        self.assertEqual(self.PROTOCOL_VERSION, message.properties.get('version'))
        self.assertTrue(message.properties.get('opcode') is not None)
        is_instance_list = isinstance(message.body, list)
        self.assertTrue(is_instance_list)
        self.assertEqual(2, len(message.body))
        return (message.properties.get('status'),
                message.body[0],  # is link_route?
                message.body[1])  # has destinations?

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

        def router(name, mode, extra=None):
            config = [
                ('router', {'mode': mode, 'id': name}),
                ('listener', {'role': 'normal', 'host': '0.0.0.0', 'port': cls.tester.get_port(), 'saslMechanisms': 'ANONYMOUS'})
            ]

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

        cls.routers = []

        inter_router_port = cls.tester.get_port()
        edge_port_A       = cls.tester.get_port()
        broker_port_A     = cls.tester.get_port()
        broker_port_B     = cls.tester.get_port()

        router('INT.A', 'interior',
               [
                   ('listener', {'role': 'edge',
                                 'port': cls.tester.get_port()}),
                   ('listener', {'role': 'inter-router',
                                 'port': inter_router_port}),
                   ('connector', {'role': 'route-container',
                                  'port': cls.tester.get_port()}),

                   ('linkRoute', {'pattern': 'org.apache.A.#',
                                  'containerId': 'FakeBrokerA',
                                  'direction': 'in'}),
                   ('linkRoute', {'pattern': 'org.apache.A.#',
                                  'containerId': 'FakeBrokerA',
                                  'direction': 'out'})
               ])
        cls.INT_A = cls.routers[-1]
        cls.INT_A.listener = cls.INT_A.addresses[0]
        cls.INT_A.edge_listener = cls.INT_A.addresses[1]
        cls.INT_A.inter_router_listener = cls.INT_A.addresses[2]
        cls.INT_A.broker_connector = cls.INT_A.connector_addresses[0]

        router('INT.B', 'interior',
               [
                   ('listener', {'role': 'edge',
                                 'port': cls.tester.get_port()}),
                   ('connector', {'role': 'inter-router',
                                  'name': 'connectorToA',
                                  'port': inter_router_port}),
                   ('connector', {'role': 'route-container',
                                  'port': cls.tester.get_port()}),

                   ('linkRoute', {'pattern': 'org.apache.B.#',
                                  'containerId': 'FakeBrokerB',
                                  'direction': 'in'}),
                   ('linkRoute', {'pattern': 'org.apache.B.#',
                                  'containerId': 'FakeBrokerB',
                                  'direction': 'out'})
                ])
        cls.INT_B = cls.routers[-1]
        cls.INT_B.edge_listener = cls.INT_B.addresses[1]
        cls.INT_B.broker_connector = cls.INT_B.connector_addresses[1]

        cls.INT_A.wait_router_connected('INT.B')
        cls.INT_B.wait_router_connected('INT.A')

    def _lookup_request(self, lr_address, direction):
        """
        Construct a link route lookup request message
        """
        return Message(body=[lr_address,
                             direction],
                       properties={"version": self.PROTOCOL_VERSION,
                                   "opcode": self.OPCODE_LINK_ROUTE_LOOKUP})

    def test_link_route_lookup_ok(self):
        """
        verify a link route address can be looked up successfully for both
        locally attached and remotely attached containers
        """

        # fire up a fake broker attached to the router local to the edge router
        # wait until both in and out addresses are ready
        fb = FakeBroker(self.INT_A.broker_connector, container_id='FakeBrokerA')
        self.INT_A.wait_address("org.apache.A", containers=1, count=2)

        # create a fake edge and lookup the target address
        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        srr = SyncRequestResponse(bc, self.QD_TERMINUS_ADDRESS_LOOKUP)

        for direction in [True, False]:
            # True = direction inbound (receiver) False = direction outbound (sender)
            rsp = self._check_response(srr.call(self._lookup_request("org.apache.A.foo", direction)))
            self.assertEqual(self.QCM_ADDR_LOOKUP_OK, rsp[0])
            self.assertTrue(rsp[1])
            self.assertTrue(rsp[2])

        # shutdown fake router
        fb.join()

        # now fire up a fake broker attached to the remote router
        fb = FakeBroker(self.INT_B.broker_connector, container_id='FakeBrokerB')
        self.INT_A.wait_address("org.apache.B", remotes=1, count=2)

        for direction in [True, False]:
            rsp = self._check_response(srr.call(self._lookup_request("org.apache.B.foo", direction)))
            self.assertEqual(self.QCM_ADDR_LOOKUP_OK, rsp[0])
            self.assertTrue(rsp[1])
            self.assertTrue(rsp[2])

        fb.join()
        bc.close()

    def test_link_route_lookup_not_found(self):
        """
        verify correct handling of lookup misses
        """
        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        srr = SyncRequestResponse(bc, self.QD_TERMINUS_ADDRESS_LOOKUP)

        rsp = self._check_response(srr.call(self._lookup_request("not.found.address", True)))
        self.assertEqual(self.QCM_ADDR_LOOKUP_NOT_FOUND, rsp[0])

    def test_link_route_lookup_not_link_route(self):
        """
        verify correct handling of matches to mobile addresses
        """
        addr = "not.a.linkroute"
        client = AsyncTestReceiver(self.INT_A.listener, addr)
        self.INT_A.wait_address(addr)

        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        srr = SyncRequestResponse(bc, self.QD_TERMINUS_ADDRESS_LOOKUP)

        rsp = self._check_response(srr.call(self._lookup_request(addr, True)))
        # self.assertEqual(self.QCM_ADDR_LOOKUP_OK, rsp[0])
        self.assertEqual(False, rsp[1])
        bc.close()
        client.stop()

    def test_link_route_lookup_no_dest(self):
        """
        verify correct handling of matches to mobile addresses
        """
        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        srr = SyncRequestResponse(bc, self.QD_TERMINUS_ADDRESS_LOOKUP)
        rsp = self._check_response(srr.call(self._lookup_request("org.apache.A.nope", True)))
        self.assertEqual(self.QCM_ADDR_LOOKUP_OK, rsp[0])
        self.assertEqual(True, rsp[1])
        self.assertEqual(False, rsp[2])
        bc.close()

    def _invalid_request_test(self, msg, disposition=Disposition.REJECTED):
        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        srr = SyncRequestResponse(bc, self.QD_TERMINUS_ADDRESS_LOOKUP)
        # @TODO(kgiusti) - self.assertRaises does not work here:
        try:
            srr.call(msg)
            self.assertTrue(False)
        except SendException as exc:
            self.assertEqual(disposition, exc.state)
        bc.close()

    def test_link_route_invalid_request(self):
        """
        Test various invalid message content
        """

        # empty message
        self._invalid_request_test(Message())

        # missing properties
        msg = self._lookup_request("ignore", False)
        msg.properties = None
        self._invalid_request_test(msg)

        # missing version
        msg = self._lookup_request("ignore", False)
        del msg.properties['version']
        self._invalid_request_test(msg)

        # invalid version
        msg = self._lookup_request("ignore", False)
        msg.properties['version'] = "blech"
        self._invalid_request_test(msg)

        # unsupported version
        msg = self._lookup_request("ignore", False)
        msg.properties['version'] = 97387187
        self._invalid_request_test(msg)

        # missing opcode
        msg = self._lookup_request("ignore", False)
        del msg.properties['opcode']
        self._invalid_request_test(msg)

        # bad opcode
        msg = self._lookup_request("ignore", False)
        msg.properties['opcode'] = "snarf"
        self._invalid_request_test(msg)

        # bad body
        msg = self._lookup_request("ignore", False)
        msg.body = [71]
        self._invalid_request_test(msg)

    def test_lookup_bad_connection(self):
        """
        Verify that clients connected via non-edge connections fail
        """
        bc = BlockingConnection(self.INT_A.listener, timeout=TIMEOUT)
        self.assertRaises(LinkDetached, SyncRequestResponse, bc, self.QD_TERMINUS_ADDRESS_LOOKUP)
        bc.close()

        bc = BlockingConnection(self.INT_A.inter_router_listener, timeout=TIMEOUT)
        self.assertRaises(LinkDetached, SyncRequestResponse, bc, self.QD_TERMINUS_ADDRESS_LOOKUP)
        bc.close()

        # consuming from the lookup address is forbidden:
        bc = BlockingConnection(self.INT_A.edge_listener, timeout=TIMEOUT)
        self.assertRaises(LinkDetached, bc.create_receiver, self.QD_TERMINUS_ADDRESS_LOOKUP)
        bc.close()