# Licensed 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 unittest import mock

from oslo_config import cfg
import oslo_messaging as messaging
from oslo_messaging import conffixture as messaging_conffixture
from oslo_messaging import exceptions as oslomsg_exc
import testtools

from neutron_lib import fixture
from neutron_lib import rpc
from neutron_lib.tests import _base as base


CONF = cfg.CONF


class TestRPC(base.BaseTestCase):
    def setUp(self):
        super().setUp()
        self.useFixture(fixture.RPCFixture())

    @mock.patch.object(rpc, 'RequestContextSerializer')
    @mock.patch.object(messaging, 'get_rpc_transport')
    @mock.patch.object(messaging, 'get_notification_transport')
    def test_init(self, mock_noti_trans, mock_trans, mock_ser):
        transport = mock.Mock()
        noti_transport = mock.Mock()
        conf = mock.Mock()

        mock_trans.return_value = transport
        mock_noti_trans.return_value = noti_transport

        rpc.init(conf, rpc_ext_mods=['foo'])

        expected_mods = list(set(['foo'] + rpc._DFT_EXMODS))
        mock_trans.assert_called_once_with(
            conf, allowed_remote_exmods=expected_mods)
        mock_noti_trans.assert_called_once_with(
            conf, allowed_remote_exmods=expected_mods)
        self.assertIsNotNone(rpc.TRANSPORT)
        self.assertIsNotNone(rpc.NOTIFICATION_TRANSPORT)

    def test_cleanup_transport_null(self):
        rpc.NOTIFICATION_TRANSPORT = mock.Mock()
        rpc.TRANSPORT = None
        self.assertRaises(AssertionError, rpc.cleanup)
        rpc.TRANSPORT = mock.Mock()

    def test_cleanup_notification_transport_null(self):
        rpc.TRANSPORT = mock.Mock()
        rpc.NOTIFICATION_TRANSPORT = None
        self.assertRaises(AssertionError, rpc.cleanup)
        rpc.NOTIFICATION_TRANSPORT = mock.Mock()

    def test_cleanup(self):
        rpc.NOTIFICATION_TRANSPORT = mock.Mock()
        rpc.TRANSPORT = mock.Mock()
        trans_cleanup = mock.Mock()
        not_trans_cleanup = mock.Mock()
        rpc.TRANSPORT.cleanup = trans_cleanup
        rpc.NOTIFICATION_TRANSPORT.cleanup = not_trans_cleanup

        rpc.cleanup()

        trans_cleanup.assert_called_once_with()
        not_trans_cleanup.assert_called_once_with()
        self.assertIsNone(rpc.TRANSPORT)
        self.assertIsNone(rpc.NOTIFICATION_TRANSPORT)

        rpc.TRANSPORT = mock.Mock()
        rpc.NOTIFICATION_TRANSPORT = mock.Mock()

    @mock.patch.object(rpc, 'RequestContextSerializer')
    @mock.patch.object(messaging, 'get_rpc_client')
    def test_get_client(self, mock_get, mock_ser):
        rpc.TRANSPORT = mock.Mock()
        tgt = mock.Mock()
        ser = mock.Mock()
        mock_get.return_value = 'client'
        mock_ser.return_value = ser

        client = rpc.get_client(tgt, version_cap='1.0', serializer='foo')

        mock_ser.assert_called_once_with('foo')
        mock_get.assert_called_once_with(rpc.TRANSPORT,
                                         tgt, version_cap='1.0',
                                         serializer=ser,
                                         client_cls=rpc.BackingOffClient)
        self.assertEqual('client', client)

    @mock.patch.object(rpc, 'RequestContextSerializer')
    @mock.patch.object(messaging, 'get_rpc_server')
    def test_get_server(self, mock_get, mock_ser):
        ser = mock.Mock()
        tgt = mock.Mock()
        ends = mock.Mock()
        mock_ser.return_value = ser
        mock_get.return_value = 'server'

        server = rpc.get_server(tgt, ends, serializer='foo')

        mock_ser.assert_called_once_with('foo')
        mock_get.assert_called_once_with(rpc.TRANSPORT, tgt, ends,
                                         serializer=ser)
        self.assertEqual('server', server)

    def test_get_notifier(self):
        mock_notifier = mock.Mock(return_value=None)
        messaging.Notifier.__init__ = mock_notifier
        rpc.get_notifier('service', publisher_id='foo')
        mock_notifier.assert_called_once_with(
            mock.ANY, serializer=mock.ANY, publisher_id='foo')

    def test_get_notifier_null_publisher(self):
        mock_notifier = mock.Mock(return_value=None)
        messaging.Notifier.__init__ = mock_notifier
        rpc.get_notifier('service', host='bar')
        mock_notifier.assert_called_once_with(
            mock.ANY, serializer=mock.ANY, publisher_id='service.bar')


class TestRequestContextSerializer(base.BaseTestCase):
    def setUp(self):
        super().setUp()
        self.mock_base = mock.Mock()
        self.ser = rpc.RequestContextSerializer(self.mock_base)
        self.ser_null = rpc.RequestContextSerializer(None)

    def test_serialize_entity(self):
        self.mock_base.serialize_entity.return_value = 'foo'

        ser_ent = self.ser.serialize_entity('context', 'entity')

        self.mock_base.serialize_entity.assert_called_once_with('context',
                                                                'entity')
        self.assertEqual('foo', ser_ent)

    def test_deserialize_entity(self):
        self.mock_base.deserialize_entity.return_value = 'foo'

        deser_ent = self.ser.deserialize_entity('context', 'entity')

        self.mock_base.deserialize_entity.assert_called_once_with('context',
                                                                  'entity')
        self.assertEqual('foo', deser_ent)

    def test_deserialize_entity_null_base(self):
        deser_ent = self.ser_null.deserialize_entity('context', 'entity')

        self.assertEqual('entity', deser_ent)

    def test_serialize_context(self):
        context = mock.Mock()

        self.ser.serialize_context(context)

        context.to_dict.assert_called_once_with()

    def test_deserialize_context(self):
        context_dict = {'foo': 'bar',
                        'user_id': 1,
                        'tenant_id': 1,
                        'is_admin': True}

        c = self.ser.deserialize_context(context_dict)

        self.assertEqual(1, c.user_id)
        self.assertEqual(1, c.project_id)

    def test_deserialize_context_no_user_id(self):
        context_dict = {'foo': 'bar',
                        'user': 1,
                        'tenant_id': 1,
                        'is_admin': True}

        c = self.ser.deserialize_context(context_dict)

        self.assertEqual(1, c.user_id)
        self.assertEqual(1, c.project_id)

    def test_deserialize_context_no_tenant_id(self):
        context_dict = {'foo': 'bar',
                        'user_id': 1,
                        'project_id': 1,
                        'is_admin': True}

        c = self.ser.deserialize_context(context_dict)

        self.assertEqual(1, c.user_id)
        self.assertEqual(1, c.project_id)

    def test_deserialize_context_no_ids(self):
        context_dict = {'foo': 'bar', 'is_admin': True}

        c = self.ser.deserialize_context(context_dict)

        self.assertIsNone(c.user_id)
        self.assertIsNone(c.project_id)


class ServiceTestCase(base.BaseTestCase):
    # the class cannot be based on BaseTestCase since it mocks rpc.Connection

    def setUp(self):
        super().setUp()
        self.host = 'foo'
        self.topic = 'neutron-agent'

        self.target_mock = mock.patch('oslo_messaging.Target')
        self.target_mock.start()

        self.messaging_conf = messaging_conffixture.ConfFixture(CONF)
        self.messaging_conf.transport_url = 'fake://'
        self.messaging_conf.response_timeout = 0
        self.useFixture(self.messaging_conf)

        self.addCleanup(rpc.cleanup)
        rpc.init(CONF)

    @mock.patch.object(cfg, 'CONF')
    def test_operations(self, mock_conf):
        mock_conf.host = self.host
        with mock.patch('oslo_messaging.get_rpc_server') as get_rpc_server:
            rpc_server = get_rpc_server.return_value

            service = rpc.Service(self.host, self.topic)
            service.start()
            rpc_server.start.assert_called_once_with()

            service.stop()
            rpc_server.stop.assert_called_once_with()
            rpc_server.wait.assert_called_once_with()


class TimeoutTestCase(base.BaseTestCase):
    def setUp(self):
        super().setUp()

        self.messaging_conf = messaging_conffixture.ConfFixture(CONF)
        self.messaging_conf.transport_url = 'fake://'
        self.messaging_conf.response_timeout = 0
        self.useFixture(self.messaging_conf)

        self.addCleanup(rpc.cleanup)
        rpc.init(CONF)
        rpc.TRANSPORT = mock.MagicMock()
        rpc.TRANSPORT._send.side_effect = messaging.MessagingTimeout
        rpc.TRANSPORT.conf.oslo_messaging_metrics.metrics_enabled = False
        target = messaging.Target(version='1.0', topic='testing')
        self.client = rpc.get_client(target)
        self.call_context = mock.Mock()
        self.sleep = mock.patch('time.sleep').start()
        rpc.TRANSPORT.conf.rpc_response_timeout = 10
        rpc.TRANSPORT.conf.rpc_response_max_timeout = 300

    def test_timeout_unaffected_when_explicitly_set(self):
        rpc.TRANSPORT.conf.rpc_response_timeout = 5
        ctx = self.client.prepare(topic='sandwiches', timeout=77)
        with testtools.ExpectedException(messaging.MessagingTimeout):
            ctx.call(self.call_context, 'create_pb_and_j')
        # ensure that the timeout was not increased and the back-off sleep
        # wasn't called
        self.assertEqual(
            5,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['create_pb_and_j'])
        self.assertFalse(self.sleep.called)

    def test_timeout_store_defaults(self):
        # any method should default to the configured timeout
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_2'])
        # a change to an existing should not affect new or existing ones
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_2'] = 7000
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_3'])

    def test_method_timeout_sleep(self):
        rpc.TRANSPORT.conf.rpc_response_timeout = 2
        for i in range(100):
            with testtools.ExpectedException(messaging.MessagingTimeout):
                self.client.call(self.call_context, 'method_1')
            # sleep value should always be between 0 and configured timeout
            self.assertGreaterEqual(self.sleep.call_args_list[0][0][0], 0)
            self.assertLessEqual(self.sleep.call_args_list[0][0][0], 2)
            self.sleep.reset_mock()

    def test_method_timeout_increases_on_timeout_exception(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 1
        for i in range(5):
            with testtools.ExpectedException(messaging.MessagingTimeout):
                self.client.call(self.call_context, 'method_1')

        # we only care to check the timeouts sent to the transport
        timeouts = [call[1]['timeout']
                    for call in rpc.TRANSPORT._send.call_args_list]
        self.assertEqual([1, 2, 4, 8, 16], timeouts)

    def test_method_timeout_config_ceiling(self):
        rpc.TRANSPORT.conf.rpc_response_timeout = 10
        # 5 doublings should max out at the 10xdefault ceiling
        for i in range(5):
            with testtools.ExpectedException(messaging.MessagingTimeout):
                self.client.call(self.call_context, 'method_1')
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_max_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])
        with testtools.ExpectedException(messaging.MessagingTimeout):
            self.client.call(self.call_context, 'method_1')
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_max_timeout,
            rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])

    def test_timeout_unchanged_on_other_exception(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 1
        rpc.TRANSPORT._send.side_effect = ValueError
        with testtools.ExpectedException(ValueError):
            self.client.call(self.call_context, 'method_1')
        rpc.TRANSPORT._send.side_effect = messaging.MessagingTimeout
        with testtools.ExpectedException(messaging.MessagingTimeout):
            self.client.call(self.call_context, 'method_1')
        timeouts = [call[1]['timeout']
                    for call in rpc.TRANSPORT._send.call_args_list]
        self.assertEqual([1, 1], timeouts)

    def test_timeouts_for_methods_tracked_independently(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 1
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_2'] = 1
        for method in ('method_1', 'method_1', 'method_2',
                       'method_1', 'method_2'):
            with testtools.ExpectedException(messaging.MessagingTimeout):
                self.client.call(self.call_context, method)
        timeouts = [call[1]['timeout']
                    for call in rpc.TRANSPORT._send.call_args_list]
        self.assertEqual([1, 2, 1, 4, 2], timeouts)

    def test_timeouts_for_namespaces_tracked_independently(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['ns1.method'] = 1
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['ns2.method'] = 1
        for ns in ('ns1', 'ns2'):
            self.client.target.namespace = ns
            for i in range(4):
                with testtools.ExpectedException(messaging.MessagingTimeout):
                    self.client.call(self.call_context, 'method')
        timeouts = [call[1]['timeout']
                    for call in rpc.TRANSPORT._send.call_args_list]
        self.assertEqual([1, 2, 4, 8, 1, 2, 4, 8], timeouts)

    def test_method_timeout_increases_with_prepare(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 1
        ctx = self.client.prepare(version='1.4')
        with testtools.ExpectedException(messaging.MessagingTimeout):
            ctx.call(self.call_context, 'method_1')
        with testtools.ExpectedException(messaging.MessagingTimeout):
            ctx.call(self.call_context, 'method_1')

        # we only care to check the timeouts sent to the transport
        timeouts = [call[1]['timeout']
                    for call in rpc.TRANSPORT._send.call_args_list]
        self.assertEqual([1, 2], timeouts)

    def test_set_max_timeout_caps_all_methods(self):
        rpc.TRANSPORT.conf.rpc_response_timeout = 300
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 100
        rpc.BackingOffClient.set_max_timeout(50)
        # both explicitly tracked
        self.assertEqual(
            50, rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])
        # as well as new methods
        self.assertEqual(
            50, rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_2'])

    def test_set_max_timeout_retains_lower_timeouts(self):
        rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'] = 10
        rpc.BackingOffClient.set_max_timeout(50)
        self.assertEqual(
            10, rpc._BackingOffContextWrapper._METHOD_TIMEOUTS['method_1'])

    def test_set_max_timeout_overrides_default_timeout(self):
        rpc.TRANSPORT.conf.rpc_response_timeout = 10
        self.assertEqual(
            rpc.TRANSPORT.conf.rpc_response_max_timeout,
            rpc._BackingOffContextWrapper.get_max_timeout())
        rpc._BackingOffContextWrapper.set_max_timeout(10)
        self.assertEqual(10, rpc._BackingOffContextWrapper.get_max_timeout())


class CastExceptionTestCase(base.BaseTestCase):
    def setUp(self):
        super().setUp()

        self.messaging_conf = messaging_conffixture.ConfFixture(CONF)
        self.messaging_conf.transport_url = 'fake://'
        self.messaging_conf.response_timeout = 0
        self.useFixture(self.messaging_conf)

        self.addCleanup(rpc.cleanup)
        rpc.init(CONF)
        rpc.TRANSPORT = mock.MagicMock()
        rpc.TRANSPORT._send.side_effect = oslomsg_exc.MessageDeliveryFailure
        rpc.TRANSPORT.conf.oslo_messaging_metrics.metrics_enabled = False
        target = messaging.Target(version='1.0', topic='testing')
        self.client = rpc.get_client(target)
        self.cast_context = mock.Mock()

    def test_cast_catches_exception(self):
        self.client.cast(self.cast_context, 'method_1')


class TestConnection(base.BaseTestCase):
    def setUp(self):
        super().setUp()
        self.conn = rpc.Connection()

    @mock.patch.object(messaging, 'Target')
    @mock.patch.object(cfg, 'CONF')
    @mock.patch.object(rpc, 'get_server')
    def test_create_consumer(self, mock_get, mock_cfg, mock_tgt):
        mock_cfg.host = 'foo'
        server = mock.Mock()
        target = mock.Mock()
        mock_get.return_value = server
        mock_tgt.return_value = target

        self.conn.create_consumer('topic', 'endpoints', fanout=True)

        mock_tgt.assert_called_once_with(topic='topic', server='foo',
                                         fanout=True)
        mock_get.assert_called_once_with(target, 'endpoints')
        self.assertEqual([server], self.conn.servers)

    def test_consume_in_threads(self):
        self.conn.servers = [mock.Mock(), mock.Mock()]

        servs = self.conn.consume_in_threads()

        for serv in self.conn.servers:
            serv.start.assert_called_once_with()
        self.assertEqual(servs, self.conn.servers)

    def test_close(self):
        self.conn.servers = [mock.Mock(), mock.Mock()]

        self.conn.close()

        for serv in self.conn.servers:
            serv.stop.assert_called_once_with()
            serv.wait.assert_called_once_with()