diff --git a/oslo_messaging/_drivers/impl_fake.py b/oslo_messaging/_drivers/impl_fake.py
index f25e8b7d1..6898350e7 100644
--- a/oslo_messaging/_drivers/impl_fake.py
+++ b/oslo_messaging/_drivers/impl_fake.py
@@ -142,10 +142,11 @@ class FakeExchange(object):
class FakeExchangeManager(object):
+ _exchanges_lock = threading.Lock()
+ _exchanges = {}
+
def __init__(self, default_exchange):
self._default_exchange = default_exchange
- self._exchanges_lock = threading.Lock()
- self._exchanges = {}
def get_exchange(self, name):
if name is None:
diff --git a/oslo_messaging/tests/notify/test_listener.py b/oslo_messaging/tests/notify/test_listener.py
index 6582ccd03..8467c0bf0 100644
--- a/oslo_messaging/tests/notify/test_listener.py
+++ b/oslo_messaging/tests/notify/test_listener.py
@@ -13,6 +13,7 @@
# License for the specific language governing permissions and limitations
# under the License.
+import fixtures
import threading
from oslo_config import cfg
@@ -131,6 +132,9 @@ class TestNotifyListener(test_utils.BaseTestCase, ListenerSetupMixin):
def setUp(self):
super(TestNotifyListener, self).setUp(conf=cfg.ConfigOpts())
ListenerSetupMixin.setUp(self)
+ self.useFixture(fixtures.MonkeyPatch(
+ 'oslo_messaging._drivers.impl_fake.FakeExchangeManager._exchanges',
+ new_value={}))
@mock.patch('debtcollector.deprecate')
def test_constructor(self, deprecate):
diff --git a/oslo_messaging/tests/rpc/test_server.py b/oslo_messaging/tests/rpc/test_server.py
index 22793242b..4d5c6d5b8 100644
--- a/oslo_messaging/tests/rpc/test_server.py
+++ b/oslo_messaging/tests/rpc/test_server.py
@@ -35,9 +35,11 @@ load_tests = testscenarios.load_tests_apply_scenarios
class ServerSetupMixin(object):
class Server(object):
- def __init__(self, transport, topic, server, endpoint, serializer):
+ def __init__(self, transport, topic, server, endpoint, serializer,
+ exchange):
self.controller = ServerSetupMixin.ServerController()
- target = oslo_messaging.Target(topic=topic, server=server)
+ target = oslo_messaging.Target(topic=topic, server=server,
+ exchange=exchange)
self.server = oslo_messaging.get_rpc_server(transport,
target,
[endpoint,
@@ -81,25 +83,25 @@ class ServerSetupMixin(object):
def __init__(self):
self.serializer = self.TestSerializer()
- def _setup_server(self, transport, endpoint, topic=None, server=None):
+ 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)
+ serializer=self.serializer,
+ exchange=exchange)
server.start()
return server
- def _stop_server(self, client, server, topic=None):
- if topic is not None:
- client = client.prepare(topic=topic)
+ def _stop_server(self, client, server, topic=None, exchange=None):
client.cast({}, 'stop')
server.wait()
- def _setup_client(self, transport, topic='testtopic'):
- return oslo_messaging.RPCClient(transport,
- oslo_messaging.Target(topic=topic),
+ def _setup_client(self, transport, topic='testtopic', exchange=None):
+ target = oslo_messaging.Target(topic=topic, exchange=exchange)
+ return oslo_messaging.RPCClient(transport, target=target,
serializer=self.serializer)
@@ -111,6 +113,11 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
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={}))
@mock.patch('warnings.warn')
def test_constructor(self, warn):
@@ -300,14 +307,20 @@ class TestRPCServer(test_utils.BaseTestCase, ServerSetupMixin):
self.assertEqual(['dsfoo', 'dsbar'], endpoint.pings)
def test_call(self):
- transport = oslo_messaging.get_rpc_transport(self.conf, url='fake:')
+ # 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, TestEndpoint())
- client = self._setup_client(transport)
+ 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))
@@ -498,8 +511,8 @@ class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):
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
+ # 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']
@@ -536,14 +549,16 @@ class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):
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):
- url1 = 'fake:///' + (self.exchange1 or '')
- url2 = 'fake:///' + (self.exchange2 or '')
-
- transport1 = oslo_messaging.get_rpc_transport(self.conf, url=url1)
- if url1 != url2:
- transport2 = oslo_messaging.get_rpc_transport(self.conf, url=url1)
+ 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
@@ -563,12 +578,18 @@ class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):
endpoint1 = endpoint2 = TestEndpoint()
server1 = self._setup_server(transport1, endpoint1,
- topic=self.topic1, server=self.server1)
+ topic=self.topic1,
+ exchange=self.exchange1,
+ server=self.server1)
server2 = self._setup_server(transport2, endpoint2,
- topic=self.topic2, server=self.server2)
+ topic=self.topic2,
+ exchange=self.exchange2,
+ server=self.server2)
- client1 = self._setup_client(transport1, topic=self.topic1)
- client2 = self._setup_client(transport2, topic=self.topic2)
+ 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)
@@ -584,9 +605,9 @@ class TestMultipleServers(test_utils.BaseTestCase, ServerSetupMixin):
(client2.call if self.call2 else client2.cast)({}, 'ping', arg='2')
self._stop_server(client1.prepare(fanout=None),
- server1, topic=self.topic1)
+ server1, topic=self.topic1, exchange=self.exchange1)
self._stop_server(client2.prepare(fanout=None),
- server2, topic=self.topic2)
+ server2, topic=self.topic2, exchange=self.exchange2)
def check(pings, expect):
self.assertEqual(len(expect), len(pings))