← Back to branch summary

~ubuntu-branches/ubuntu/trusty/ceilometer/trusty-proposed

« back to all changes in this revision

Viewing changes to ceilometer/tests/test_notification.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, James Page, Chuck Short
  • Date: 2014-01-23 15:08:11 UTC
  • mfrom: (1.1.11)
  • Revision ID: package-import@ubuntu.com-20140123150811-1zaismsuyh1hcl8y
Tags: 2014.1~b2-0ubuntu1
https://launchpad.net/bugs/1261101
https://launchpad.net/bugs/1250002
https://launchpad.net/bugs/1224223
[ James Page ]
* d/control: Add python-jsonpath-rw to BD's.
* d/p/fix-setup-requirements.patch: Bump WebOb to support < 1.4.
 (LP: #1261101)

[ Chuck Short ]
* New upstream version.
* debian/control, debian/ceilometer-common.install: Split out
  ceilometer-alarm-evaluator and ceilometer-alarm-notifier into their
  own packages. (LP: #1250002)
* debian/ceilometer-agent-central.logrotate,
  debian/ceilometer-agent-compute.logrotate,
  debian/ceilometer-api.logrotate,
  debian/ceilometer-collector.logrotate: Add logrotate files, 
  thanks to Ahmed Rahal. (LP: #1224223)
* Fix typos in upstart files.

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/tests/test_notification.py
17
17
# under the License.
18
18
"""Tests for Ceilometer notify daemon."""
19
19
 
20
 
import datetime
21
20
import mock
22
21
 
23
22
from stevedore import extension
24
23
from stevedore.tests import manager as test_manager
25
24
 
26
 
from ceilometer.compute import notifications
 
25
from ceilometer.compute.notifications import instance
27
26
from ceilometer import notification
28
27
from ceilometer.openstack.common.fixture import config
29
 
from ceilometer.openstack.common import timeutils
30
28
from ceilometer.storage import models
31
29
from ceilometer.tests import base as tests_base
32
30
 
89
87
        self.CONF.set_override("connection", "log://", group='database')
90
88
 
91
89
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
90
    @mock.patch('ceilometer.event.converter.setup_events', mock.MagicMock())
92
91
    def test_process_notification(self):
93
92
        # If we try to create a real RPC connection, init_host() never
94
93
        # returns. Mock it out so we can establish the service
101
100
            [extension.Extension('test',
102
101
                                 None,
103
102
                                 None,
104
 
                                 notifications.Instance(),
 
103
                                 instance.Instance(),
105
104
                                 ),
106
105
             ])
107
106
        self.srv.process_notification(TEST_NOTICE)
124
123
            self.srv.process_notification({})
125
124
            self.assertTrue(fake_msg_to_event.called)
126
125
 
127
 
    def test_message_to_event_missing_keys(self):
128
 
        now = timeutils.utcnow()
129
 
        timeutils.set_time_override(now)
130
 
        message = {'event_type': "foo",
131
 
                   'message_id': "abc",
132
 
                   'publisher_id': "1"}
133
 
 
134
 
        mock_dispatcher = mock.MagicMock()
135
 
        self.srv.dispatcher_manager = test_manager.TestExtensionManager(
136
 
            [extension.Extension('test',
137
 
                                 None,
138
 
                                 None,
139
 
                                 mock_dispatcher
140
 
                                 ),
141
 
             ])
142
 
 
143
 
        self.srv._message_to_event(message)
144
 
        events = mock_dispatcher.record_events.call_args[0]
145
 
        self.assertEqual(1, len(events))
146
 
        event = events[0]
147
 
        self.assertEqual("foo", event.event_type)
148
 
        self.assertEqual(now, event.generated)
149
 
        self.assertEqual(1, len(event.traits))
150
 
 
151
126
    def test_message_to_event_duplicate(self):
152
127
        self.CONF.set_override("store_events", True, group="notification")
153
128
        mock_dispatcher = mock.MagicMock()
 
129
        self.srv.event_converter = mock.MagicMock()
 
130
        self.srv.event_converter.to_event.return_value = mock.MagicMock(
 
131
            event_type='test.test')
154
132
        self.srv.dispatcher_manager = test_manager.TestExtensionManager(
155
133
            [extension.Extension('test',
156
134
                                 None,
166
144
    def test_message_to_event_bad_event(self):
167
145
        self.CONF.set_override("store_events", True, group="notification")
168
146
        mock_dispatcher = mock.MagicMock()
 
147
        self.srv.event_converter = mock.MagicMock()
 
148
        self.srv.event_converter.to_event.return_value = mock.MagicMock(
 
149
            event_type='test.test')
169
150
        self.srv.dispatcher_manager = test_manager.TestExtensionManager(
170
151
            [extension.Extension('test',
171
152
                                 None,
178
159
        message = {'event_type': "foo", 'message_id': "abc"}
179
160
        self.assertRaises(notification.UnableToSaveEventException,
180
161
                          self.srv._message_to_event, message)
181
 
 
182
 
    def test_extract_when(self):
183
 
        now = timeutils.utcnow()
184
 
        modified = now + datetime.timedelta(minutes=1)
185
 
        timeutils.set_time_override(now)
186
 
 
187
 
        body = {"timestamp": str(modified)}
188
 
        when = notification.NotificationService._extract_when(body)
189
 
        self.assertTimestampEqual(modified, when)
190
 
 
191
 
        body = {"_context_timestamp": str(modified)}
192
 
        when = notification.NotificationService._extract_when(body)
193
 
        self.assertTimestampEqual(modified, when)
194
 
 
195
 
        then = now + datetime.timedelta(hours=1)
196
 
        body = {"timestamp": str(modified), "_context_timestamp": str(then)}
197
 
        when = notification.NotificationService._extract_when(body)
198
 
        self.assertTimestampEqual(modified, when)
199
 
 
200
 
        when = notification.NotificationService._extract_when({})
201
 
        self.assertTimestampEqual(now, when)