4ead7cb2dc
We already expose functions to handle the instantiation of classes such as RPCServer and RPCTransport but the same was never done for RPCClient so the API is inconsistent in its enforcement. This adds a get_rpc_client function that should be used instead of instatiating the RPCClient class directly to be more consistent. This also allows to handle more logic inside the function in the future such as if implementations for an async client is implemented, as investigation in [1] has shown. [1] https://review.opendev.org/c/openstack/oslo.messaging/+/858936 Change-Id: Ia4d1f0497b9e2728bde02f4ff05fdc175ddffe66
968 lines
35 KiB
Python
968 lines
35 KiB
Python
|
|
# 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 unittest import mock
|
|
|
|
|
|
import eventlet
|
|
import fixtures
|
|
from oslo_config import cfg
|
|
from oslo_utils import eventletutils
|
|
import testscenarios
|
|
|
|
import oslo_messaging
|
|
from oslo_messaging import rpc
|
|
from oslo_messaging.rpc import dispatcher
|
|
from oslo_messaging.rpc import server as rpc_server_module
|
|
from oslo_messaging import server as server_module
|
|
from oslo_messaging.tests import utils as test_utils
|
|
|
|
load_tests = testscenarios.load_tests_apply_scenarios
|
|
|
|
|
|
class ServerSetupMixin(object):
|
|
|
|
class Server(object):
|
|
def __init__(self, transport, topic, server, endpoint, serializer,
|
|
exchange):
|
|
self.controller = ServerSetupMixin.ServerController()
|
|
target = oslo_messaging.Target(topic=topic, server=server,
|
|
exchange=exchange)
|
|
self.server = oslo_messaging.get_rpc_server(transport,
|
|
target,
|
|
[endpoint,
|
|
self.controller],
|
|
serializer=serializer)
|
|
|
|
def wait(self):
|
|
# Wait for the executor to process the stop message, indicating all
|
|
# test messages have been processed
|
|
self.controller.stopped.wait()
|
|
|
|
# Check start() does nothing with a running server
|
|
self.server.start()
|
|
self.server.stop()
|
|
self.server.wait()
|
|
|
|
def start(self):
|
|
self.server.start()
|
|
|
|
class ServerController(object):
|
|
def __init__(self):
|
|
self.stopped = eventletutils.Event()
|
|
|
|
def stop(self, ctxt):
|
|
self.stopped.set()
|
|
|
|
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,
|
|
exchange=None):
|
|
server = self.Server(transport,
|
|
topic=topic or 'testtopic',
|
|
server=server or 'testserver',
|
|
endpoint=endpoint,
|
|
serializer=self.serializer,
|
|
exchange=exchange)
|
|
|
|
server.start()
|
|
return server
|
|
|
|
def _stop_server(self, client, server, topic=None, exchange=None):
|
|
client.cast({}, 'stop')
|
|
server.wait()
|
|
|
|
def _setup_client(self, transport, topic='testtopic', exchange=None):
|
|
target = oslo_messaging.Target(topic=topic, exchange=exchange)
|
|
return oslo_messaging.get_rpc_client(transport, target=target,
|
|
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())
|
|
# FakeExchangeManager uses a class-level exchanges mapping; "reset" it
|
|
# before tests assert amount of items stored
|
|
self.useFixture(fixtures.MonkeyPatch(
|
|
'oslo_messaging._drivers.impl_fake.FakeExchangeManager._exchanges',
|
|
new_value={}))
|
|
|
|
def test_constructor(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
target = oslo_messaging.Target(topic='foo', server='bar')
|
|
endpoints = [object()]
|
|
serializer = object()
|
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
|
|
|
server = oslo_messaging.get_rpc_server(transport,
|
|
target,
|
|
endpoints,
|
|
serializer=serializer,
|
|
access_policy=access_policy,
|
|
executor='threading')
|
|
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('threading', server.executor_type)
|
|
|
|
def test_constructor_with_eventlet_executor(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
target = oslo_messaging.Target(topic='foo', server='bar')
|
|
endpoints = [object()]
|
|
serializer = object()
|
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
|
|
|
server = oslo_messaging.get_rpc_server(transport,
|
|
target,
|
|
endpoints,
|
|
serializer=serializer,
|
|
access_policy=access_policy,
|
|
executor='eventlet')
|
|
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('eventlet', server.executor_type)
|
|
|
|
def test_constructor_with_unrecognized_executor(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
target = oslo_messaging.Target(topic='foo', server='bar')
|
|
endpoints = [object()]
|
|
serializer = object()
|
|
access_policy = dispatcher.DefaultRPCAccessPolicy
|
|
|
|
self.assertRaises(
|
|
server_module.ExecutorLoadFailure,
|
|
oslo_messaging.get_rpc_server,
|
|
transport=transport,
|
|
target=target,
|
|
endpoints=endpoints,
|
|
serializer=serializer,
|
|
access_policy=access_policy,
|
|
executor='boom')
|
|
|
|
def test_server_wait_method(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
target = oslo_messaging.Target(topic='foo', server='bar')
|
|
endpoints = [object()]
|
|
serializer = object()
|
|
|
|
class MagicMockIgnoreArgs(mock.MagicMock):
|
|
"""MagicMock ignores arguments.
|
|
|
|
A MagicMock which can never misinterpret the arguments passed to
|
|
it during construction.
|
|
"""
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super(MagicMockIgnoreArgs, self).__init__()
|
|
|
|
server = oslo_messaging.get_rpc_server(transport, target, endpoints,
|
|
serializer=serializer)
|
|
# Mocking executor
|
|
server._executor_cls = MagicMockIgnoreArgs
|
|
server._create_listener = MagicMockIgnoreArgs()
|
|
server.dispatcher = MagicMockIgnoreArgs()
|
|
# Here assigning executor's listener object to listener variable
|
|
# before calling wait method, because in wait method we are
|
|
# setting executor to None.
|
|
server.start()
|
|
listener = server.listener
|
|
server.stop()
|
|
# call server wait method
|
|
server.wait()
|
|
self.assertEqual(1, listener.cleanup.call_count)
|
|
|
|
def test_no_target_server(self):
|
|
transport = oslo_messaging.get_rpc_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_rpc_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)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
def _test_no_client_topic(self, call=True):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
client = self._setup_client(transport, topic=None)
|
|
|
|
method = client.call if call else client.cast
|
|
|
|
try:
|
|
method({}, 'ping', arg='foo')
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex, oslo_messaging.InvalidTarget, ex)
|
|
self.assertIsNotNone(ex.target)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
def test_no_client_topic_call(self):
|
|
self._test_no_client_topic(call=True)
|
|
|
|
def test_no_client_topic_cast(self):
|
|
self._test_no_client_topic(call=False)
|
|
|
|
def test_client_call_timeout(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
finished = False
|
|
wait = threading.Condition()
|
|
|
|
class TestEndpoint(object):
|
|
def ping(self, ctxt, arg):
|
|
with wait:
|
|
if not finished:
|
|
wait.wait()
|
|
|
|
server_thread = self._setup_server(transport, TestEndpoint())
|
|
client = self._setup_client(transport)
|
|
|
|
try:
|
|
client.prepare(timeout=0).call({}, 'ping', arg='foo')
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex, oslo_messaging.MessagingTimeout, ex)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
with wait:
|
|
finished = True
|
|
wait.notify()
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
def test_unknown_executor(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
try:
|
|
oslo_messaging.get_rpc_server(transport, None, [], executor='foo')
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex, oslo_messaging.ExecutorLoadFailure)
|
|
self.assertEqual('foo', ex.executor)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
def test_cast(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
class TestEndpoint(object):
|
|
def __init__(self):
|
|
self.pings = []
|
|
|
|
def ping(self, ctxt, arg):
|
|
self.pings.append(arg)
|
|
|
|
endpoint = TestEndpoint()
|
|
server_thread = self._setup_server(transport, endpoint)
|
|
client = self._setup_client(transport)
|
|
|
|
client.cast({}, 'ping', arg='foo')
|
|
client.cast({}, 'ping', arg='bar')
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
self.assertEqual(['dsfoo', 'dsbar'], endpoint.pings)
|
|
|
|
def test_call(self):
|
|
# NOTE(milan): using a separate transport instance for each the client
|
|
# and the server to be able to check independent transport instances
|
|
# can communicate over same exchange&topic
|
|
transport_srv = oslo_messaging.get_rpc_transport(self.conf,
|
|
url='fake:')
|
|
transport_cli = oslo_messaging.get_rpc_transport(self.conf,
|
|
url='fake:')
|
|
|
|
class TestEndpoint(object):
|
|
def ping(self, ctxt, arg):
|
|
return arg
|
|
|
|
server_thread = self._setup_server(transport_srv, TestEndpoint())
|
|
client = self._setup_client(transport_cli)
|
|
|
|
self.assertIsNone(client.call({}, 'ping', arg=None))
|
|
self.assertEqual(0, client.call({}, 'ping', arg=0))
|
|
self.assertFalse(client.call({}, 'ping', arg=False))
|
|
self.assertEqual([], client.call({}, 'ping', arg=[]))
|
|
self.assertEqual({}, client.call({}, 'ping', arg={}))
|
|
self.assertEqual('dsdsfoo', client.call({}, 'ping', arg='foo'))
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
def test_direct_call(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
class TestEndpoint(object):
|
|
def ping(self, ctxt, arg):
|
|
return arg
|
|
|
|
server_thread = self._setup_server(transport, TestEndpoint())
|
|
client = self._setup_client(transport)
|
|
|
|
direct = client.prepare(server='testserver')
|
|
self.assertIsNone(direct.call({}, 'ping', arg=None))
|
|
self.assertEqual(0, client.call({}, 'ping', arg=0))
|
|
self.assertFalse(client.call({}, 'ping', arg=False))
|
|
self.assertEqual([], client.call({}, 'ping', arg=[]))
|
|
self.assertEqual({}, client.call({}, 'ping', arg={}))
|
|
self.assertEqual('dsdsfoo', direct.call({}, 'ping', arg='foo'))
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
def test_context(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
class TestEndpoint(object):
|
|
def ctxt_check(self, ctxt, key):
|
|
return ctxt[key]
|
|
|
|
server_thread = self._setup_server(transport, TestEndpoint())
|
|
client = self._setup_client(transport)
|
|
|
|
self.assertEqual('dsdsb',
|
|
client.call({'dsa': 'b'},
|
|
'ctxt_check',
|
|
key='a'))
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
def test_failure(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
class TestEndpoint(object):
|
|
def ping(self, ctxt, arg):
|
|
raise ValueError(arg)
|
|
|
|
debugs = []
|
|
errors = []
|
|
|
|
def stub_debug(msg, *a, **kw):
|
|
if (a and len(a) == 1 and isinstance(a[0], dict) and a[0]):
|
|
a = a[0]
|
|
debugs.append(str(msg) % a)
|
|
|
|
def stub_error(msg, *a, **kw):
|
|
if (a and len(a) == 1 and isinstance(a[0], dict) and a[0]):
|
|
a = a[0]
|
|
errors.append(str(msg) % a)
|
|
|
|
self.useFixture(fixtures.MockPatchObject(
|
|
rpc_server_module.LOG, 'debug', stub_debug))
|
|
self.useFixture(fixtures.MockPatchObject(
|
|
rpc_server_module.LOG, 'error', stub_error))
|
|
|
|
server_thread = self._setup_server(transport, TestEndpoint())
|
|
client = self._setup_client(transport)
|
|
|
|
try:
|
|
client.call({}, 'ping', arg='foo')
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex, ValueError)
|
|
self.assertEqual('dsfoo', str(ex))
|
|
self.assertTrue(len(debugs) == 0)
|
|
self.assertGreater(len(errors), 0)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
def test_expected_failure(self):
|
|
transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
|
|
|
|
debugs = []
|
|
errors = []
|
|
|
|
def stub_debug(msg, *a, **kw):
|
|
if (a and len(a) == 1 and isinstance(a[0], dict) and a[0]):
|
|
a = a[0]
|
|
debugs.append(str(msg) % a)
|
|
|
|
def stub_error(msg, *a, **kw):
|
|
if (a and len(a) == 1 and isinstance(a[0], dict) and a[0]):
|
|
a = a[0]
|
|
errors.append(str(msg) % a)
|
|
|
|
self.useFixture(fixtures.MockPatchObject(
|
|
rpc_server_module.LOG, 'debug', stub_debug))
|
|
self.useFixture(fixtures.MockPatchObject(
|
|
rpc_server_module.LOG, 'error', stub_error))
|
|
|
|
class TestEndpoint(object):
|
|
@oslo_messaging.expected_exceptions(ValueError)
|
|
def ping(self, ctxt, arg):
|
|
raise ValueError(arg)
|
|
|
|
server_thread = self._setup_server(transport, TestEndpoint())
|
|
client = self._setup_client(transport)
|
|
|
|
try:
|
|
client.call({}, 'ping', arg='foo')
|
|
except Exception as ex:
|
|
self.assertIsInstance(ex, ValueError)
|
|
self.assertEqual('dsfoo', str(ex))
|
|
self.assertGreater(len(debugs), 0)
|
|
self.assertTrue(len(errors) == 0)
|
|
else:
|
|
self.assertTrue(False)
|
|
|
|
self._stop_server(client, server_thread)
|
|
|
|
@mock.patch('oslo_messaging.rpc.server.LOG')
|
|
def test_warning_when_notifier_transport(self, log):
|
|
transport = oslo_messaging.get_notification_transport(self.conf)
|
|
target = oslo_messaging.Target(topic='foo', server='bar')
|
|
endpoints = [object()]
|
|
serializer = object()
|
|
|
|
oslo_messaging.get_rpc_server(transport, target,
|
|
endpoints, serializer=serializer)
|
|
log.warning.assert_called_once_with(
|
|
"Using notification transport for RPC. Please use "
|
|
"get_rpc_transport to obtain an RPC transport "
|
|
"instance.")
|
|
|
|
|
|
class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):
|
|
|
|
_exchanges = [
|
|
('same_exchange', dict(exchange1=None, exchange2=None)),
|
|
('diff_exchange', dict(exchange1='x1', exchange2='x2')),
|
|
]
|
|
|
|
_topics = [
|
|
('same_topic', dict(topic1='t', topic2='t')),
|
|
('diff_topic', dict(topic1='t1', topic2='t2')),
|
|
]
|
|
|
|
_server = [
|
|
('same_server', dict(server1=None, server2=None)),
|
|
('diff_server', dict(server1='s1', server2='s2')),
|
|
]
|
|
|
|
_fanout = [
|
|
('not_fanout', dict(fanout1=None, fanout2=None)),
|
|
('fanout', dict(fanout1=True, fanout2=True)),
|
|
]
|
|
|
|
_method = [
|
|
('call', dict(call1=True, call2=True)),
|
|
('cast', dict(call1=False, call2=False)),
|
|
]
|
|
|
|
_endpoints = [
|
|
('one_endpoint',
|
|
dict(multi_endpoints=False,
|
|
expect1=['ds1', 'ds2'],
|
|
expect2=['ds1', 'ds2'])),
|
|
('two_endpoints',
|
|
dict(multi_endpoints=True,
|
|
expect1=['ds1'],
|
|
expect2=['ds2'])),
|
|
]
|
|
|
|
@classmethod
|
|
def generate_scenarios(cls):
|
|
cls.scenarios = testscenarios.multiply_scenarios(cls._exchanges,
|
|
cls._topics,
|
|
cls._server,
|
|
cls._fanout,
|
|
cls._method,
|
|
cls._endpoints)
|
|
|
|
# fanout call not supported
|
|
def filter_fanout_call(scenario):
|
|
params = scenario[1]
|
|
fanout = params['fanout1'] or params['fanout2']
|
|
call = params['call1'] or params['call2']
|
|
return not (call and fanout)
|
|
|
|
# listening multiple times on same topic/server pair not supported
|
|
def filter_same_topic_and_server(scenario):
|
|
params = scenario[1]
|
|
single_topic = params['topic1'] == params['topic2']
|
|
single_server = params['server1'] == params['server2']
|
|
return not (single_topic and single_server)
|
|
|
|
# fanout to multiple servers on same topic and exchange each endpoint
|
|
# will receive both messages
|
|
def fanout_to_servers(scenario):
|
|
params = scenario[1]
|
|
fanout = params['fanout1'] or params['fanout2']
|
|
single_exchange = params['exchange1'] == params['exchange2']
|
|
single_topic = params['topic1'] == params['topic2']
|
|
multi_servers = params['server1'] != params['server2']
|
|
if fanout and single_exchange and single_topic and multi_servers:
|
|
params['expect1'] = params['expect1'][:] + params['expect1']
|
|
params['expect2'] = params['expect2'][:] + params['expect2']
|
|
return scenario
|
|
|
|
# multiple endpoints on same topic and exchange
|
|
# either endpoint can get either message
|
|
def single_topic_multi_endpoints(scenario):
|
|
params = scenario[1]
|
|
single_exchange = params['exchange1'] == params['exchange2']
|
|
single_topic = params['topic1'] == params['topic2']
|
|
if single_topic and single_exchange and params['multi_endpoints']:
|
|
params['expect_either'] = (params['expect1'] +
|
|
params['expect2'])
|
|
params['expect1'] = params['expect2'] = []
|
|
else:
|
|
params['expect_either'] = []
|
|
return scenario
|
|
|
|
for f in [filter_fanout_call, filter_same_topic_and_server]:
|
|
cls.scenarios = [i for i in cls.scenarios if f(i)]
|
|
for m in [fanout_to_servers, single_topic_multi_endpoints]:
|
|
cls.scenarios = [m(i) for i in cls.scenarios]
|
|
|
|
def __init__(self, *args):
|
|
super(TestMultipleServers, self).__init__(*args)
|
|
ServerSetupMixin.__init__(self)
|
|
|
|
def setUp(self):
|
|
super(TestMultipleServers, self).setUp(conf=cfg.ConfigOpts())
|
|
self.useFixture(fixtures.MonkeyPatch(
|
|
'oslo_messaging._drivers.impl_fake.FakeExchangeManager._exchanges',
|
|
new_value={}))
|
|
|
|
def test_multiple_servers(self):
|
|
transport1 = oslo_messaging.get_rpc_transport(self.conf,
|
|
url='fake:')
|
|
if self.exchange1 != self.exchange2:
|
|
transport2 = oslo_messaging.get_rpc_transport(self.conf,
|
|
url='fake:')
|
|
else:
|
|
transport2 = transport1
|
|
|
|
class TestEndpoint(object):
|
|
def __init__(self):
|
|
self.pings = []
|
|
|
|
def ping(self, ctxt, arg):
|
|
self.pings.append(arg)
|
|
|
|
def alive(self, ctxt):
|
|
return 'alive'
|
|
|
|
if self.multi_endpoints:
|
|
endpoint1, endpoint2 = TestEndpoint(), TestEndpoint()
|
|
else:
|
|
endpoint1 = endpoint2 = TestEndpoint()
|
|
|
|
server1 = self._setup_server(transport1, endpoint1,
|
|
topic=self.topic1,
|
|
exchange=self.exchange1,
|
|
server=self.server1)
|
|
server2 = self._setup_server(transport2, endpoint2,
|
|
topic=self.topic2,
|
|
exchange=self.exchange2,
|
|
server=self.server2)
|
|
|
|
client1 = self._setup_client(transport1, topic=self.topic1,
|
|
exchange=self.exchange1)
|
|
client2 = self._setup_client(transport2, topic=self.topic2,
|
|
exchange=self.exchange2)
|
|
|
|
client1 = client1.prepare(server=self.server1)
|
|
client2 = client2.prepare(server=self.server2)
|
|
|
|
if self.fanout1:
|
|
client1.call({}, 'alive')
|
|
client1 = client1.prepare(fanout=True)
|
|
if self.fanout2:
|
|
client2.call({}, 'alive')
|
|
client2 = client2.prepare(fanout=True)
|
|
|
|
(client1.call if self.call1 else client1.cast)({}, 'ping', arg='1')
|
|
(client2.call if self.call2 else client2.cast)({}, 'ping', arg='2')
|
|
|
|
self._stop_server(client1.prepare(fanout=None),
|
|
server1, topic=self.topic1, exchange=self.exchange1)
|
|
self._stop_server(client2.prepare(fanout=None),
|
|
server2, topic=self.topic2, exchange=self.exchange2)
|
|
|
|
def check(pings, expect):
|
|
self.assertEqual(len(expect), len(pings))
|
|
for a in expect:
|
|
self.assertIn(a, pings)
|
|
|
|
if self.expect_either:
|
|
check(endpoint1.pings + endpoint2.pings, self.expect_either)
|
|
else:
|
|
check(endpoint1.pings, self.expect1)
|
|
check(endpoint2.pings, self.expect2)
|
|
|
|
|
|
TestMultipleServers.generate_scenarios()
|
|
|
|
|
|
class TestServerLocking(test_utils.BaseTestCase):
|
|
def setUp(self):
|
|
super(TestServerLocking, self).setUp(conf=cfg.ConfigOpts())
|
|
|
|
def _logmethod(name):
|
|
def method(self, *args, **kwargs):
|
|
with self._lock:
|
|
self._calls.append(name)
|
|
return method
|
|
|
|
executors = []
|
|
|
|
class FakeExecutor(object):
|
|
def __init__(self, *args, **kwargs):
|
|
self._lock = threading.Lock()
|
|
self._calls = []
|
|
executors.append(self)
|
|
|
|
submit = _logmethod('submit')
|
|
shutdown = _logmethod('shutdown')
|
|
|
|
self.executors = executors
|
|
|
|
class MessageHandlingServerImpl(oslo_messaging.MessageHandlingServer):
|
|
def _create_listener(self):
|
|
return mock.Mock()
|
|
|
|
def _process_incoming(self, incoming):
|
|
pass
|
|
|
|
self.server = MessageHandlingServerImpl(mock.Mock(), mock.Mock())
|
|
self.server._executor_cls = FakeExecutor
|
|
|
|
def test_start_stop_wait(self):
|
|
# Test a simple execution of start, stop, wait in order
|
|
|
|
eventlet.spawn(self.server.start)
|
|
self.server.stop()
|
|
self.server.wait()
|
|
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
self.assertTrue(self.server.listener.cleanup.called)
|
|
|
|
def test_reversed_order(self):
|
|
# Test that if we call wait, stop, start, these will be correctly
|
|
# reordered
|
|
|
|
eventlet.spawn(self.server.wait)
|
|
# This is non-deterministic, but there's not a great deal we can do
|
|
# about that
|
|
eventlet.sleep(0)
|
|
|
|
eventlet.spawn(self.server.stop)
|
|
eventlet.sleep(0)
|
|
|
|
eventlet.spawn(self.server.start)
|
|
|
|
self.server.wait()
|
|
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
|
|
def test_wait_for_running_task(self):
|
|
# Test that if 2 threads call a method simultaneously, both will wait,
|
|
# but only 1 will call the underlying executor method.
|
|
|
|
start_event = eventletutils.Event()
|
|
finish_event = eventletutils.Event()
|
|
|
|
running_event = eventletutils.Event()
|
|
done_event = eventletutils.Event()
|
|
|
|
_runner = [None]
|
|
|
|
class SteppingFakeExecutor(self.server._executor_cls):
|
|
def __init__(self, *args, **kwargs):
|
|
# Tell the test which thread won the race
|
|
_runner[0] = eventlet.getcurrent()
|
|
running_event.set()
|
|
|
|
start_event.wait()
|
|
super(SteppingFakeExecutor, self).__init__(*args, **kwargs)
|
|
done_event.set()
|
|
|
|
finish_event.wait()
|
|
|
|
self.server._executor_cls = SteppingFakeExecutor
|
|
|
|
start1 = eventlet.spawn(self.server.start)
|
|
start2 = eventlet.spawn(self.server.start)
|
|
|
|
# Wait until one of the threads starts running
|
|
running_event.wait()
|
|
runner = _runner[0]
|
|
waiter = start2 if runner == start1 else start2
|
|
|
|
waiter_finished = eventletutils.Event()
|
|
waiter.link(lambda _: waiter_finished.set())
|
|
|
|
# At this point, runner is running start(), and waiter() is waiting for
|
|
# it to complete. runner has not yet logged anything.
|
|
self.assertEqual(0, len(self.executors))
|
|
self.assertFalse(waiter_finished.is_set())
|
|
|
|
# Let the runner log the call
|
|
start_event.set()
|
|
done_event.wait()
|
|
|
|
# We haven't signalled completion yet, so submit shouldn't have run
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual([], self.executors[0]._calls)
|
|
self.assertFalse(waiter_finished.is_set())
|
|
|
|
# Let the runner complete
|
|
finish_event.set()
|
|
waiter.wait()
|
|
runner.wait()
|
|
|
|
# Check that both threads have finished, start was only called once,
|
|
# and execute ran
|
|
self.assertTrue(waiter_finished.is_set())
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual([], self.executors[0]._calls)
|
|
|
|
def test_start_stop_wait_stop_wait(self):
|
|
# Test that we behave correctly when calling stop/wait more than once.
|
|
# Subsequent calls should be noops.
|
|
|
|
self.server.start()
|
|
self.server.stop()
|
|
self.server.wait()
|
|
self.server.stop()
|
|
self.server.wait()
|
|
|
|
self.assertEqual(len(self.executors), 1)
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
self.assertTrue(self.server.listener.cleanup.called)
|
|
|
|
def test_state_wrapping(self):
|
|
# Test that we behave correctly if a thread waits, and the server state
|
|
# has wrapped when it it next scheduled
|
|
|
|
# Ensure that if 2 threads wait for the completion of 'start', the
|
|
# first will wait until complete_event is signalled, but the second
|
|
# will continue
|
|
complete_event = eventletutils.Event()
|
|
complete_waiting_callback = eventletutils.Event()
|
|
|
|
start_state = self.server._states['start']
|
|
old_wait_for_completion = start_state.wait_for_completion
|
|
waited = [False]
|
|
|
|
def new_wait_for_completion(*args, **kwargs):
|
|
if not waited[0]:
|
|
waited[0] = True
|
|
complete_waiting_callback.set()
|
|
complete_event.wait()
|
|
old_wait_for_completion(*args, **kwargs)
|
|
|
|
start_state.wait_for_completion = new_wait_for_completion
|
|
|
|
# thread1 will wait for start to complete until we signal it
|
|
thread1 = eventlet.spawn(self.server.stop)
|
|
thread1_finished = eventletutils.Event()
|
|
thread1.link(lambda _: thread1_finished.set())
|
|
|
|
self.server.start()
|
|
complete_waiting_callback.wait()
|
|
|
|
# The server should have started, but stop should not have been called
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual([], self.executors[0]._calls)
|
|
self.assertFalse(thread1_finished.is_set())
|
|
|
|
self.server.stop()
|
|
self.server.wait()
|
|
|
|
# We should have gone through all the states, and thread1 should still
|
|
# be waiting
|
|
self.assertEqual(1, len(self.executors))
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
self.assertFalse(thread1_finished.is_set())
|
|
|
|
# Start again
|
|
self.server.start()
|
|
|
|
# We should now record 4 executors (2 for each server)
|
|
self.assertEqual(2, len(self.executors))
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
self.assertEqual([], self.executors[1]._calls)
|
|
self.assertFalse(thread1_finished.is_set())
|
|
|
|
# Allow thread1 to complete
|
|
complete_event.set()
|
|
thread1_finished.wait()
|
|
|
|
# thread1 should now have finished, and stop should not have been
|
|
# called again on either the first or second executor
|
|
self.assertEqual(2, len(self.executors))
|
|
self.assertEqual(['shutdown'], self.executors[0]._calls)
|
|
self.assertEqual([], self.executors[1]._calls)
|
|
self.assertTrue(thread1_finished.is_set())
|
|
|
|
@mock.patch.object(server_module, 'DEFAULT_LOG_AFTER', 1)
|
|
@mock.patch.object(server_module, 'LOG')
|
|
def test_logging(self, mock_log):
|
|
# Test that we generate a log message if we wait longer than
|
|
# DEFAULT_LOG_AFTER
|
|
|
|
log_event = eventletutils.Event()
|
|
mock_log.warning.side_effect = lambda _, __: log_event.set()
|
|
|
|
# Call stop without calling start. We should log a wait after 1 second
|
|
thread = eventlet.spawn(self.server.stop)
|
|
log_event.wait()
|
|
|
|
# Redundant given that we already waited, but it's nice to assert
|
|
self.assertTrue(mock_log.warning.called)
|
|
thread.kill()
|
|
|
|
@mock.patch.object(server_module, 'LOG')
|
|
def test_logging_explicit_wait(self, mock_log):
|
|
# Test that we generate a log message if we wait longer than
|
|
# the number of seconds passed to log_after
|
|
|
|
log_event = eventletutils.Event()
|
|
mock_log.warning.side_effect = lambda _, __: log_event.set()
|
|
|
|
# Call stop without calling start. We should log a wait after 1 second
|
|
thread = eventlet.spawn(self.server.stop, log_after=1)
|
|
log_event.wait()
|
|
|
|
# Redundant given that we already waited, but it's nice to assert
|
|
self.assertTrue(mock_log.warning.called)
|
|
thread.kill()
|
|
|
|
@mock.patch.object(server_module, 'LOG')
|
|
def test_logging_with_timeout(self, mock_log):
|
|
# Test that we log a message after log_after seconds if we've also
|
|
# specified an absolute timeout
|
|
|
|
log_event = eventletutils.Event()
|
|
mock_log.warning.side_effect = lambda _, __: log_event.set()
|
|
|
|
# Call stop without calling start. We should log a wait after 1 second
|
|
thread = eventlet.spawn(self.server.stop, log_after=1, timeout=2)
|
|
log_event.wait()
|
|
|
|
# Redundant given that we already waited, but it's nice to assert
|
|
self.assertTrue(mock_log.warning.called)
|
|
thread.kill()
|
|
|
|
def test_timeout_wait(self):
|
|
# Test that we will eventually timeout when passing the timeout option
|
|
# if a preceding condition is not satisfied.
|
|
|
|
self.assertRaises(server_module.TaskTimeout,
|
|
self.server.stop, timeout=1)
|
|
|
|
def test_timeout_running(self):
|
|
# Test that we will eventually timeout if we're waiting for another
|
|
# thread to complete this task
|
|
|
|
# Start the server, which will also instantiate an executor
|
|
self.server.start()
|
|
self.server.stop()
|
|
shutdown_called = eventletutils.Event()
|
|
|
|
# Patch the executor's stop method to be very slow
|
|
def slow_shutdown(wait):
|
|
shutdown_called.set()
|
|
eventlet.sleep(10)
|
|
self.executors[0].shutdown = slow_shutdown
|
|
|
|
# Call wait in a new thread
|
|
thread = eventlet.spawn(self.server.wait)
|
|
|
|
# Wait until the thread is in the slow stop method
|
|
shutdown_called.wait()
|
|
|
|
# Call wait again in the main thread with a timeout
|
|
self.assertRaises(server_module.TaskTimeout,
|
|
self.server.wait, timeout=1)
|
|
thread.kill()
|
|
|
|
@mock.patch.object(server_module, 'LOG')
|
|
def test_log_after_zero(self, mock_log):
|
|
# Test that we do not log a message after DEFAULT_LOG_AFTER if the
|
|
# caller gave log_after=1
|
|
|
|
# Call stop without calling start.
|
|
self.assertRaises(server_module.TaskTimeout,
|
|
self.server.stop, log_after=0, timeout=2)
|
|
|
|
# We timed out. Ensure we didn't log anything.
|
|
self.assertFalse(mock_log.warning.called)
|
|
|
|
|
|
class TestRPCExposeDecorator(test_utils.BaseTestCase):
|
|
|
|
def foo(self):
|
|
pass
|
|
|
|
@rpc.expose
|
|
def bar(self):
|
|
"""bar docstring"""
|
|
pass
|
|
|
|
def test_undecorated(self):
|
|
self.assertRaises(AttributeError, lambda: self.foo.exposed)
|
|
|
|
def test_decorated(self):
|
|
self.assertEqual(True, self.bar.exposed)
|
|
self.assertEqual("""bar docstring""", self.bar.__doc__)
|
|
self.assertEqual('bar', self.bar.__name__)
|