OpenStack library for messaging
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 
 
 

412 lines
16 KiB

# Copyright 2013 eNovance
#
# 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
import time
import mock
import testscenarios
from oslo.config import cfg
import oslo_messaging
from oslo_messaging.notify import dispatcher
from oslo_messaging.tests import utils as test_utils
load_tests = testscenarios.load_tests_apply_scenarios
class RestartableServerThread(object):
def __init__(self, server):
self.server = server
self.thread = None
def start(self):
if self.thread is None:
self.thread = threading.Thread(target=self.server.start)
self.thread.daemon = True
self.thread.start()
def stop(self):
if self.thread is not None:
# Check start() does nothing with a running listener
self.server.start()
self.server.stop()
self.server.wait()
self.thread.join(timeout=15)
ret = self.thread.isAlive()
self.thread = None
return ret
return True
class ListenerSetupMixin(object):
class ThreadTracker(object):
def __init__(self):
self._received_msgs = 0
self.threads = []
self.lock = threading.Lock()
def info(self, ctxt, publisher_id, event_type, payload, metadata):
# NOTE(sileht): this run into an other thread
with self.lock:
self._received_msgs += 1
def wait_for_messages(self, expect_messages):
while self._received_msgs < expect_messages:
time.sleep(0.01)
def stop(self):
for thread in self.threads:
thread.stop()
self.threads = []
def start(self, thread):
self.threads.append(thread)
thread.start()
def setUp(self):
self.trackers = {}
self.addCleanup(self._stop_trackers)
def _stop_trackers(self):
for pool in self.trackers:
self.trackers[pool].stop()
self.trackers = {}
def _setup_listener(self, transport, endpoints,
targets=None, pool=None):
if pool is None:
tracker_name = '__default__'
else:
tracker_name = pool
if targets is None:
targets = [oslo_messaging.Target(topic='testtopic')]
tracker = self.trackers.setdefault(
tracker_name, self.ThreadTracker())
listener = oslo_messaging.get_notification_listener(
transport, targets=targets, endpoints=[tracker] + endpoints,
allow_requeue=True, pool=pool)
thread = RestartableServerThread(listener)
tracker.start(thread)
return thread
def wait_for_messages(self, expect_messages, tracker_name='__default__'):
self.trackers[tracker_name].wait_for_messages(expect_messages)
def _setup_notifier(self, transport, topic='testtopic',
publisher_id='testpublisher'):
return oslo_messaging.Notifier(transport, topic=topic,
driver='messaging',
publisher_id=publisher_id)
class TestNotifyListener(test_utils.BaseTestCase, ListenerSetupMixin):
def __init__(self, *args):
super(TestNotifyListener, self).__init__(*args)
ListenerSetupMixin.__init__(self)
def setUp(self):
super(TestNotifyListener, self).setUp(conf=cfg.ConfigOpts())
ListenerSetupMixin.setUp(self)
def test_constructor(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
target = oslo_messaging.Target(topic='foo')
endpoints = [object()]
listener = oslo_messaging.get_notification_listener(
transport, [target], endpoints)
self.assertIs(listener.conf, self.conf)
self.assertIs(listener.transport, transport)
self.assertIsInstance(listener.dispatcher,
dispatcher.NotificationDispatcher)
self.assertIs(listener.dispatcher.endpoints, endpoints)
self.assertEqual('blocking', listener.executor)
def test_no_target_topic(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
listener = oslo_messaging.get_notification_listener(
transport,
[oslo_messaging.Target()],
[mock.Mock()])
try:
listener.start()
except Exception as ex:
self.assertIsInstance(ex, oslo_messaging.InvalidTarget, ex)
else:
self.assertTrue(False)
def test_unknown_executor(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
try:
oslo_messaging.get_notification_listener(transport, [], [],
executor='foo')
except Exception as ex:
self.assertIsInstance(ex, oslo_messaging.ExecutorLoadFailure)
self.assertEqual('foo', ex.executor)
else:
self.assertTrue(False)
def test_one_topic(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint = mock.Mock()
endpoint.info.return_value = None
listener_thread = self._setup_listener(transport, [endpoint])
notifier = self._setup_notifier(transport)
notifier.info({}, 'an_event.start', 'test message')
self.wait_for_messages(1)
self.assertFalse(listener_thread.stop())
endpoint.info.assert_called_once_with(
{}, 'testpublisher', 'an_event.start', 'test message',
{'message_id': mock.ANY, 'timestamp': mock.ANY})
def test_two_topics(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint = mock.Mock()
endpoint.info.return_value = None
targets = [oslo_messaging.Target(topic="topic1"),
oslo_messaging.Target(topic="topic2")]
listener_thread = self._setup_listener(transport, [endpoint],
targets=targets)
notifier = self._setup_notifier(transport, topic='topic1')
notifier.info({'ctxt': '1'}, 'an_event.start1', 'test')
notifier = self._setup_notifier(transport, topic='topic2')
notifier.info({'ctxt': '2'}, 'an_event.start2', 'test')
self.wait_for_messages(2)
self.assertFalse(listener_thread.stop())
endpoint.info.assert_has_calls([
mock.call({'ctxt': '1'}, 'testpublisher',
'an_event.start1', 'test',
{'timestamp': mock.ANY, 'message_id': mock.ANY}),
mock.call({'ctxt': '2'}, 'testpublisher',
'an_event.start2', 'test',
{'timestamp': mock.ANY, 'message_id': mock.ANY})],
any_order=True)
def test_two_exchanges(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint = mock.Mock()
endpoint.info.return_value = None
targets = [oslo_messaging.Target(topic="topic",
exchange="exchange1"),
oslo_messaging.Target(topic="topic",
exchange="exchange2")]
listener_thread = self._setup_listener(transport, [endpoint],
targets=targets)
notifier = self._setup_notifier(transport, topic="topic")
def mock_notifier_exchange(name):
def side_effect(target, ctxt, message, version, retry):
target.exchange = name
return transport._driver.send_notification(target, ctxt,
message, version,
retry=retry)
transport._send_notification = mock.MagicMock(
side_effect=side_effect)
notifier.info({'ctxt': '0'},
'an_event.start', 'test message default exchange')
mock_notifier_exchange('exchange1')
notifier.info({'ctxt': '1'},
'an_event.start', 'test message exchange1')
mock_notifier_exchange('exchange2')
notifier.info({'ctxt': '2'},
'an_event.start', 'test message exchange2')
self.wait_for_messages(2)
self.assertFalse(listener_thread.stop())
endpoint.info.assert_has_calls([
mock.call({'ctxt': '1'}, 'testpublisher', 'an_event.start',
'test message exchange1',
{'timestamp': mock.ANY, 'message_id': mock.ANY}),
mock.call({'ctxt': '2'}, 'testpublisher', 'an_event.start',
'test message exchange2',
{'timestamp': mock.ANY, 'message_id': mock.ANY})],
any_order=True)
def test_two_endpoints(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint1 = mock.Mock()
endpoint1.info.return_value = None
endpoint2 = mock.Mock()
endpoint2.info.return_value = oslo_messaging.NotificationResult.HANDLED
listener_thread = self._setup_listener(transport,
[endpoint1, endpoint2])
notifier = self._setup_notifier(transport)
notifier.info({}, 'an_event.start', 'test')
self.wait_for_messages(1)
self.assertFalse(listener_thread.stop())
endpoint1.info.assert_called_once_with(
{}, 'testpublisher', 'an_event.start', 'test', {
'timestamp': mock.ANY,
'message_id': mock.ANY})
endpoint2.info.assert_called_once_with(
{}, 'testpublisher', 'an_event.start', 'test', {
'timestamp': mock.ANY,
'message_id': mock.ANY})
def test_requeue(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint = mock.Mock()
endpoint.info = mock.Mock()
def side_effect_requeue(*args, **kwargs):
if endpoint.info.call_count == 1:
return oslo_messaging.NotificationResult.REQUEUE
return oslo_messaging.NotificationResult.HANDLED
endpoint.info.side_effect = side_effect_requeue
listener_thread = self._setup_listener(transport, [endpoint])
notifier = self._setup_notifier(transport)
notifier.info({}, 'an_event.start', 'test')
self.wait_for_messages(2)
self.assertFalse(listener_thread.stop())
endpoint.info.assert_has_calls([
mock.call({}, 'testpublisher', 'an_event.start', 'test',
{'timestamp': mock.ANY, 'message_id': mock.ANY}),
mock.call({}, 'testpublisher', 'an_event.start', 'test',
{'timestamp': mock.ANY, 'message_id': mock.ANY})])
def test_two_pools(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint1 = mock.Mock()
endpoint1.info.return_value = None
endpoint2 = mock.Mock()
endpoint2.info.return_value = None
targets = [oslo_messaging.Target(topic="topic")]
listener1_thread = self._setup_listener(transport, [endpoint1],
targets=targets, pool="pool1")
listener2_thread = self._setup_listener(transport, [endpoint2],
targets=targets, pool="pool2")
notifier = self._setup_notifier(transport, topic="topic")
notifier.info({'ctxt': '0'}, 'an_event.start', 'test message0')
notifier.info({'ctxt': '1'}, 'an_event.start', 'test message1')
self.wait_for_messages(2, "pool1")
self.wait_for_messages(2, "pool2")
self.assertFalse(listener2_thread.stop())
self.assertFalse(listener1_thread.stop())
def mocked_endpoint_call(i):
return mock.call({'ctxt': '%d' % i}, 'testpublisher',
'an_event.start', 'test message%d' % i,
{'timestamp': mock.ANY, 'message_id': mock.ANY})
endpoint1.info.assert_has_calls([mocked_endpoint_call(0),
mocked_endpoint_call(1)])
endpoint2.info.assert_has_calls([mocked_endpoint_call(0),
mocked_endpoint_call(1)])
def test_two_pools_three_listener(self):
transport = oslo_messaging.get_transport(self.conf, url='fake:')
endpoint1 = mock.Mock()
endpoint1.info.return_value = None
endpoint2 = mock.Mock()
endpoint2.info.return_value = None
endpoint3 = mock.Mock()
endpoint3.info.return_value = None
targets = [oslo_messaging.Target(topic="topic")]
listener1_thread = self._setup_listener(transport, [endpoint1],
targets=targets, pool="pool1")
listener2_thread = self._setup_listener(transport, [endpoint2],
targets=targets, pool="pool2")
listener3_thread = self._setup_listener(transport, [endpoint3],
targets=targets, pool="pool2")
def mocked_endpoint_call(i):
return mock.call({'ctxt': '%d' % i}, 'testpublisher',
'an_event.start', 'test message%d' % i,
{'timestamp': mock.ANY, 'message_id': mock.ANY})
notifier = self._setup_notifier(transport, topic="topic")
mocked_endpoint1_calls = []
for i in range(0, 25):
notifier.info({'ctxt': '%d' % i}, 'an_event.start',
'test message%d' % i)
mocked_endpoint1_calls.append(mocked_endpoint_call(i))
self.wait_for_messages(25, 'pool2')
listener2_thread.stop()
for i in range(0, 25):
notifier.info({'ctxt': '%d' % i}, 'an_event.start',
'test message%d' % i)
mocked_endpoint1_calls.append(mocked_endpoint_call(i))
self.wait_for_messages(50, 'pool2')
listener2_thread.start()
listener3_thread.stop()
for i in range(0, 25):
notifier.info({'ctxt': '%d' % i}, 'an_event.start',
'test message%d' % i)
mocked_endpoint1_calls.append(mocked_endpoint_call(i))
self.wait_for_messages(75, 'pool2')
listener3_thread.start()
for i in range(0, 25):
notifier.info({'ctxt': '%d' % i}, 'an_event.start',
'test message%d' % i)
mocked_endpoint1_calls.append(mocked_endpoint_call(i))
self.wait_for_messages(100, 'pool1')
self.wait_for_messages(100, 'pool2')
self.assertFalse(listener3_thread.stop())
self.assertFalse(listener2_thread.stop())
self.assertFalse(listener1_thread.stop())
self.assertEqual(100, endpoint1.info.call_count)
endpoint1.info.assert_has_calls(mocked_endpoint1_calls)
self.assertLessEqual(25, endpoint2.info.call_count)
self.assertLessEqual(25, endpoint3.info.call_count)
self.assertEqual(100, endpoint2.info.call_count +
endpoint3.info.call_count)
for call in mocked_endpoint1_calls:
self.assertIn(call, endpoint2.info.mock_calls +
endpoint3.info.mock_calls)