
# Copyright 2013 Red Hat, Inc.
#
#    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.

import threading

from oslo_config import cfg
import testscenarios

import oslo_messaging
from oslo_messaging.tests import utils as test_utils
from six.moves import mock

load_tests = testscenarios.load_tests_apply_scenarios


class ServerSetupMixin(object):

    class Server(object):
        def __init__(self, transport, topic, server, endpoint, serializer):
            target = oslo_messaging.Target(topic=topic, server=server)
            self._server = oslo_messaging.get_rpc_server(transport,
                                                         target,
                                                         [endpoint, self],
                                                         serializer=serializer)

        def stop(self, ctxt):
            # Check start() does nothing with a running server
            self._server.start()
            self._server.stop()
            self._server.wait()

        def start(self):
            self._server.start()

    class TestSerializer(object):

        def serialize_entity(self, ctxt, entity):
            return ('s' + entity) if entity else entity

        def deserialize_entity(self, ctxt, entity):
            return ('d' + entity) if entity else entity

        def serialize_context(self, ctxt):
            return dict([(k, 's' + v) for k, v in ctxt.items()])

        def deserialize_context(self, ctxt):
            return dict([(k, 'd' + v) for k, v in ctxt.items()])

    def __init__(self):
        self.serializer = self.TestSerializer()

    def _setup_server(self, transport, endpoint, topic=None, server=None):
        server = self.Server(transport,
                             topic=topic or 'testtopic',
                             server=server or 'testserver',
                             endpoint=endpoint,
                             serializer=self.serializer)

        thread = threading.Thread(target=server.start)
        thread.daemon = True
        thread.start()

        return thread

    def _stop_server(self, client, server_thread, topic=None):
        if topic is not None:
            client = client.prepare(topic=topic)
        client.cast({}, 'stop')
        server_thread.join(timeout=30)

    def _setup_client(self, transport, topic='testtopic'):
        return oslo_messaging.RPCClient(transport,
                                        oslo_messaging.Target(topic=topic),
                                        serializer=self.serializer)


class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):

    def __init__(self, *args):
        super(TestRPCServer, self).__init__(*args)
        ServerSetupMixin.__init__(self)

    def setUp(self):
        super(TestRPCServer, self).setUp(conf=cfg.ConfigOpts())

    def test_constructor(self):
        transport = oslo_messaging.get_transport(self.conf, url='fake:')
        target = oslo_messaging.Target(topic='foo', server='bar')
        endpoints = [object()]
        serializer = object()

        server = oslo_messaging.get_rpc_server(transport, target, endpoints,
                                               serializer=serializer)

        self.assertIs(server.conf, self.conf)
        self.assertIs(server.transport, transport)
        self.assertIsInstance(server.dispatcher, oslo_messaging.RPCDispatcher)
        self.assertIs(server.dispatcher.endpoints, endpoints)
        self.assertIs(server.dispatcher.serializer, serializer)
        self.assertEqual('blocking', server.executor)

    def test_server_wait_method(self):
        transport = oslo_messaging.get_transport(self.conf, url='fake:')
        target = oslo_messaging.Target(topic='foo', server='bar')
        endpoints = [object()]
        serializer = object()

        server = oslo_messaging.get_rpc_server(transport, target, endpoints,
                                               serializer=serializer)
        # Mocking executor
        server._executor = mock.Mock()
        # Here assigning executor's listener object to listener variable
        # before calling wait method, beacuse in wait method we are
        # setting executor to None.
        listener = server._executor.listener
        # call server wait method
        server.wait()
        self.assertIsNone(server._executor)
        self.assertEqual(1, listener.cleanup.call_count)

    def test_no_target_server(self):
        transport = oslo_messaging.get_transport(self.conf, url='fake:')

        server = oslo_messaging.get_rpc_server(
            transport,
            oslo_messaging.Target(topic='testtopic'),
            [])
        try:
            server.start()
        except Exception as ex:
            self.assertIsInstance(ex, oslo_messaging.InvalidTarget, ex)
            self.assertEqual('testtopic', ex.target.topic)
        else:
            self.assertTrue(False)

    def test_no_server_topic(self):
        transport = oslo_messaging.get_transport(self.conf, url='fake:')
        target = oslo_messaging.Target(server='testserver')
        server = oslo_messaging.get_rpc_server(transport, target, [])
        try:
            server.start()
        except Exception as ex:
            self.assertIsInstance(ex, oslo_messaging.InvalidTarget, ex)
            self.assertEqual('testserver', ex.target.server)