Source code for ceilometer.tests.unit.publisher.test_messaging_publisher

#
# Copyright 2012 New Dream Network, LLC (DreamHost)
#
# 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.
"""Tests for ceilometer/publisher/messaging.py
"""
import datetime
import uuid

import mock
from oslo_config import fixture as fixture_config
from oslo_utils import netutils
import testscenarios.testcase

from ceilometer.event.storage import models as event
from ceilometer.publisher import messaging as msg_publisher
from ceilometer import sample
from ceilometer.tests import base as tests_base


[docs]class BasePublisherTestCase(tests_base.BaseTestCase): test_event_data = [ event.Event(message_id=uuid.uuid4(), event_type='event_%d' % i, generated=datetime.datetime.utcnow(), traits=[], raw={}) for i in range(0, 5) ] test_sample_data = [ sample.Sample( name='test', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id='test_run_tasks', timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='test', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id='test_run_tasks', timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='test2', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id='test_run_tasks', timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='test2', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id='test_run_tasks', timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), sample.Sample( name='test3', type=sample.TYPE_CUMULATIVE, unit='', volume=1, user_id='test', project_id='test', resource_id='test_run_tasks', timestamp=datetime.datetime.utcnow().isoformat(), resource_metadata={'name': 'TestPublish'}, ), ]
[docs] def setUp(self): super(BasePublisherTestCase, self).setUp() self.CONF = self.useFixture(fixture_config.Config()).conf self.setup_messaging(self.CONF)
[docs]class NotifierOnlyPublisherTest(BasePublisherTestCase): @mock.patch('oslo_messaging.Notifier')
[docs] def test_publish_topic_override(self, notifier): msg_publisher.SampleNotifierPublisher( netutils.urlsplit('notifier://?topic=custom_topic')) notifier.assert_called_with(mock.ANY, topics=['custom_topic'], driver=mock.ANY, retry=mock.ANY, publisher_id=mock.ANY) msg_publisher.EventNotifierPublisher( netutils.urlsplit('notifier://?topic=custom_event_topic')) notifier.assert_called_with(mock.ANY, topics=['custom_event_topic'], driver=mock.ANY, retry=mock.ANY, publisher_id=mock.ANY)
@mock.patch('ceilometer.messaging.get_transport')
[docs] def test_publish_other_host(self, cgt): msg_publisher.SampleNotifierPublisher( netutils.urlsplit('notifier://foo:foo@127.0.0.1:1234')) cgt.assert_called_with('rabbit://foo:foo@127.0.0.1:1234') msg_publisher.EventNotifierPublisher( netutils.urlsplit('notifier://foo:foo@127.0.0.1:1234')) cgt.assert_called_with('rabbit://foo:foo@127.0.0.1:1234')
@mock.patch('ceilometer.messaging.get_transport')
[docs] def test_publish_other_host_vhost_and_query(self, cgt): msg_publisher.SampleNotifierPublisher( netutils.urlsplit('notifier://foo:foo@127.0.0.1:1234/foo' '?driver=amqp&amqp_auto_delete=true')) cgt.assert_called_with('amqp://foo:foo@127.0.0.1:1234/foo' '?amqp_auto_delete=true') msg_publisher.EventNotifierPublisher( netutils.urlsplit('notifier://foo:foo@127.0.0.1:1234/foo' '?driver=amqp&amqp_auto_delete=true')) cgt.assert_called_with('amqp://foo:foo@127.0.0.1:1234/foo' '?amqp_auto_delete=true')
[docs]class TestPublisher(testscenarios.testcase.WithScenarios, BasePublisherTestCase): scenarios = [ ('notifier', dict(protocol="notifier", publisher_cls=msg_publisher.SampleNotifierPublisher, test_data=BasePublisherTestCase.test_sample_data, pub_func='publish_samples', attr='source')), ('event_notifier', dict(protocol="notifier", publisher_cls=msg_publisher.EventNotifierPublisher, test_data=BasePublisherTestCase.test_event_data, pub_func='publish_events', attr='event_type')), ]
[docs] def setUp(self): super(TestPublisher, self).setUp() self.topic = (self.CONF.publisher_notifier.event_topic if self.pub_func == 'publish_events' else self.CONF.publisher_notifier.metering_topic)
[docs]class TestPublisherPolicy(TestPublisher): @mock.patch('ceilometer.publisher.messaging.LOG')
[docs] def test_published_with_no_policy(self, mylog): publisher = self.publisher_cls( netutils.urlsplit('%s://' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect self.assertRaises( msg_publisher.DeliveryFailure, getattr(publisher, self.pub_func), self.test_data) self.assertTrue(mylog.info.called) self.assertEqual('default', publisher.policy) self.assertEqual(0, len(publisher.local_queue)) fake_send.assert_called_once_with( self.topic, mock.ANY)
@mock.patch('ceilometer.publisher.messaging.LOG')
[docs] def test_published_with_policy_block(self, mylog): publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=default' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect self.assertRaises( msg_publisher.DeliveryFailure, getattr(publisher, self.pub_func), self.test_data) self.assertTrue(mylog.info.called) self.assertEqual(0, len(publisher.local_queue)) fake_send.assert_called_once_with( self.topic, mock.ANY)
@mock.patch('ceilometer.publisher.messaging.LOG')
[docs] def test_published_with_policy_incorrect(self, mylog): publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=notexist' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect self.assertRaises( msg_publisher.DeliveryFailure, getattr(publisher, self.pub_func), self.test_data) self.assertTrue(mylog.warning.called) self.assertEqual('default', publisher.policy) self.assertEqual(0, len(publisher.local_queue)) fake_send.assert_called_once_with( self.topic, mock.ANY)
@mock.patch('ceilometer.publisher.messaging.LOG', mock.Mock())
[docs]class TestPublisherPolicyReactions(TestPublisher):
[docs] def test_published_with_policy_drop_and_rpc_down(self): publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=drop' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(0, len(publisher.local_queue)) fake_send.assert_called_once_with( self.topic, mock.ANY)
[docs] def test_published_with_policy_queue_and_rpc_down(self): publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=queue' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(1, len(publisher.local_queue)) fake_send.assert_called_once_with( self.topic, mock.ANY)
[docs] def test_published_with_policy_queue_and_rpc_down_up(self): self.rpc_unreachable = True publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=queue' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(1, len(publisher.local_queue)) fake_send.side_effect = mock.MagicMock() getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(0, len(publisher.local_queue)) topic = self.topic expected = [mock.call(topic, mock.ANY), mock.call(topic, mock.ANY), mock.call(topic, mock.ANY)] self.assertEqual(expected, fake_send.mock_calls)
[docs] def test_published_with_policy_sized_queue_and_rpc_down(self): publisher = self.publisher_cls(netutils.urlsplit( '%s://?policy=queue&max_queue_length=3' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect for i in range(0, 5): for s in self.test_data: setattr(s, self.attr, 'test-%d' % i) getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(3, len(publisher.local_queue)) self.assertEqual( 'test-2', publisher.local_queue[0][1][0][self.attr] ) self.assertEqual( 'test-3', publisher.local_queue[1][1][0][self.attr] ) self.assertEqual( 'test-4', publisher.local_queue[2][1][0][self.attr] )
[docs] def test_published_with_policy_default_sized_queue_and_rpc_down(self): publisher = self.publisher_cls( netutils.urlsplit('%s://?policy=queue' % self.protocol)) side_effect = msg_publisher.DeliveryFailure() with mock.patch.object(publisher, '_send') as fake_send: fake_send.side_effect = side_effect for i in range(0, 2000): for s in self.test_data: setattr(s, self.attr, 'test-%d' % i) getattr(publisher, self.pub_func)(self.test_data) self.assertEqual(1024, len(publisher.local_queue)) self.assertEqual( 'test-976', publisher.local_queue[0][1][0][self.attr] ) self.assertEqual( 'test-1999', publisher.local_queue[1023][1][0][self.attr] )

Project Source