path: root/ceilometerclient/tests/unit/v2/test_alarms.py

#
# 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 copy

import six
from six.moves import xrange  # noqa
import testtools

from ceilometerclient import exc
from ceilometerclient.openstack.common.apiclient import client
from ceilometerclient.openstack.common.apiclient import fake_client
from ceilometerclient.v2 import alarms

AN_ALARM = {u'alarm_actions': [u'http://site:8000/alarm'],
            u'ok_actions': [u'http://site:8000/ok'],
            u'description': u'An alarm',
            u'type': u'threshold',
            u'severity': 'low',
            u'threshold_rule': {
                u'meter_name': u'storage.objects',
                u'query': [{u'field': u'key_name',
                            u'op': u'eq',
                            u'value': u'key_value'}],
                u'evaluation_periods': 2,
                u'period': 240.0,
                u'statistic': u'avg',
                u'threshold': 200.0,
                u'comparison_operator': 'gt'},
            u'time_constraints': [
                {
                    u'name': u'cons1',
                    u'description': u'desc1',
                    u'start': u'0 11 * * *',
                    u'duration': 300,
                    u'timezone': u''},
                {
                    u'name': u'cons2',
                    u'description': u'desc2',
                    u'start': u'0 23 * * *',
                    u'duration': 600,
                    u'timezone': ''}],
            u'timestamp': u'2013-05-09T13:41:23.085000',
            u'enabled': True,
            u'alarm_id': u'alarm-id',
            u'state': u'ok',
            u'insufficient_data_actions': [u'http://site:8000/nodata'],
            u'user_id': u'user-id',
            u'project_id': u'project-id',
            u'state_timestamp': u'2013-05-09T13:41:23.085000',
            u'repeat_actions': False,
            u'name': 'SwiftObjectAlarm'}
CREATE_ALARM = copy.deepcopy(AN_ALARM)
del CREATE_ALARM['timestamp']
del CREATE_ALARM['state_timestamp']
del CREATE_ALARM['alarm_id']
CREATE_ALARM_WITHOUT_TC = copy.deepcopy(CREATE_ALARM)
del CREATE_ALARM_WITHOUT_TC['time_constraints']
DELTA_ALARM = {u'alarm_actions': ['url1', 'url2']}
DELTA_ALARM_RULE = {u'comparison_operator': u'lt',
                    u'threshold': 42.1,
                    u'meter_name': u'foobar',
                    u'query': [{u'field': u'key_name',
                                u'op': u'eq',
                                u'value': u'key_value'}]}
DELTA_ALARM_TC = [{u'name': u'cons1',
                  u'duration': 500}]
DELTA_ALARM['time_constraints'] = DELTA_ALARM_TC
DELTA_ALARM['user_id'] = u'new-user-id'
UPDATED_ALARM = copy.deepcopy(AN_ALARM)
UPDATED_ALARM.update(DELTA_ALARM)
UPDATED_ALARM['threshold_rule'].update(DELTA_ALARM_RULE)
DELTA_ALARM['remove_time_constraints'] = 'cons2'
UPDATED_ALARM['time_constraints'] = [{u'name': u'cons1',
                                      u'description': u'desc1',
                                      u'start': u'0 11 * * *',
                                      u'duration': 500,
                                      u'timezone': u''}]
DELTA_ALARM['threshold_rule'] = DELTA_ALARM_RULE
UPDATE_ALARM = copy.deepcopy(UPDATED_ALARM)
UPDATE_ALARM['remove_time_constraints'] = 'cons2'
UPDATE_ALARM['user_id'] = u'new-user-id'
del UPDATE_ALARM['project_id']
del UPDATE_ALARM['name']
del UPDATE_ALARM['alarm_id']
del UPDATE_ALARM['timestamp']
del UPDATE_ALARM['state_timestamp']

AN_LEGACY_ALARM = {u'alarm_actions': [u'http://site:8000/alarm'],
                   u'ok_actions': [u'http://site:8000/ok'],
                   u'description': u'An alarm',
                   u'matching_metadata': {u'key_name': u'key_value'},
                   u'evaluation_periods': 2,
                   u'timestamp': u'2013-05-09T13:41:23.085000',
                   u'enabled': True,
                   u'meter_name': u'storage.objects',
                   u'period': 240.0,
                   u'alarm_id': u'alarm-id',
                   u'state': u'ok',
                   u'severity': u'low',
                   u'insufficient_data_actions': [u'http://site:8000/nodata'],
                   u'statistic': u'avg',
                   u'threshold': 200.0,
                   u'user_id': u'user-id',
                   u'project_id': u'project-id',
                   u'state_timestamp': u'2013-05-09T13:41:23.085000',
                   u'comparison_operator': 'gt',
                   u'repeat_actions': False,
                   u'name': 'SwiftObjectAlarm'}
CREATE_LEGACY_ALARM = copy.deepcopy(AN_LEGACY_ALARM)
del CREATE_LEGACY_ALARM['timestamp']
del CREATE_LEGACY_ALARM['state_timestamp']
del CREATE_LEGACY_ALARM['alarm_id']
DELTA_LEGACY_ALARM = {u'alarm_actions': ['url1', 'url2'],
                      u'comparison_operator': u'lt',
                      u'meter_name': u'foobar',
                      u'threshold': 42.1}
DELTA_LEGACY_ALARM['time_constraints'] = [{u'name': u'cons1',
                                           u'duration': 500}]
DELTA_LEGACY_ALARM['user_id'] = u'new-user-id'
DELTA_LEGACY_ALARM['remove_time_constraints'] = 'cons2'
UPDATED_LEGACY_ALARM = copy.deepcopy(AN_LEGACY_ALARM)
UPDATED_LEGACY_ALARM.update(DELTA_LEGACY_ALARM)
UPDATE_LEGACY_ALARM = copy.deepcopy(UPDATED_LEGACY_ALARM)
UPDATE_LEGACY_ALARM['user_id'] = u'new-user-id'
del UPDATE_LEGACY_ALARM['project_id']
del UPDATE_LEGACY_ALARM['name']
del UPDATE_LEGACY_ALARM['alarm_id']
del UPDATE_LEGACY_ALARM['timestamp']
del UPDATE_LEGACY_ALARM['state_timestamp']

FULL_DETAIL = ('{"alarm_actions": [], '
               '"user_id": "8185aa72421a4fd396d4122cba50e1b5", '
               '"name": "scombo", '
               '"timestamp": "2013-10-03T08:58:33.647912", '
               '"enabled": true, '
               '"state_timestamp": "2013-10-03T08:58:33.647912", '
               '"rule": {"operator": "or", "alarm_ids": '
               '["062cc907-3a9f-4867-ab3b-fa83212b39f7"]}, '
               '"alarm_id": "alarm-id, '
               '"state": "insufficient data", '
               '"insufficient_data_actions": [], '
               '"repeat_actions": false, '
               '"ok_actions": [], '
               '"project_id": "57d04f24d0824b78b1ea9bcecedbda8f", '
               '"type": "combination", '
               '"description": "Combined state of alarms '
               '062cc907-3a9f-4867-ab3b-fa83212b39f7"}')
ALARM_HISTORY = [{'on_behalf_of': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'user_id': '8185aa72421a4fd396d4122cba50e1b5',
                  'event_id': 'c74a8611-6553-4764-a860-c15a6aabb5d0',
                  'timestamp': '2013-10-03T08:59:28.326000',
                  'detail': '{"state": "alarm"}',
                  'alarm_id': 'alarm-id',
                  'project_id': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'type': 'state transition'},
                 {'on_behalf_of': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'user_id': '8185aa72421a4fd396d4122cba50e1b5',
                  'event_id': 'c74a8611-6553-4764-a860-c15a6aabb5d0',
                  'timestamp': '2013-10-03T08:59:28.326000',
                  'detail': '{"description": "combination of one"}',
                  'alarm_id': 'alarm-id',
                  'project_id': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'type': 'rule change'},
                 {'on_behalf_of': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'user_id': '8185aa72421a4fd396d4122cba50e1b5',
                  'event_id': '4fd7df9e-190d-4471-8884-dc5a33d5d4bb',
                  'timestamp': '2013-10-03T08:58:33.647000',
                  'detail': FULL_DETAIL,
                  'alarm_id': 'alarm-id',
                  'project_id': '57d04f24d0824b78b1ea9bcecedbda8f',
                  'type': 'creation'}]

fixtures = {
    '/v2/alarms':
    {
        'GET': (
            {},
            [AN_ALARM],
        ),
        'POST': (
            {},
            CREATE_ALARM,
        ),
    },
    '/v2/alarms/alarm-id':
    {
        'GET': (
            {},
            AN_ALARM,
        ),
        'PUT': (
            {},
            UPDATED_ALARM,
        ),
        'DELETE': (
            {},
            None,
        ),
    },
    '/v2/alarms/unk-alarm-id':
    {
        'GET': (
            {},
            None,
        ),
        'PUT': (
            {},
            None,
        ),
    },
    '/v2/alarms/alarm-id/state':
    {
        'PUT': (
            {},
            {'alarm': 'alarm'}
        ),
        'GET': (
            {},
            {'alarm': 'alarm'}
        ),

    },
    '/v2/alarms?q.field=project_id&q.field=name&q.op=&q.op='
    '&q.type=&q.type=&q.value=project-id&q.value=SwiftObjectAlarm':
    {
        'GET': (
            {},
            [AN_ALARM],
        ),
    },
    '/v2/alarms/victim-id':
    {
        'DELETE': (
            {},
            None,
        ),
    },
    '/v2/alarms/alarm-id/history':
    {
        'GET': (
            {},
            ALARM_HISTORY,
        ),
    },
    '/v2/alarms/alarm-id/history?q.field=timestamp&q.op=&q.type=&q.value=NOW':
    {
        'GET': (
            {},
            ALARM_HISTORY,
        ),
    },
}


class AlarmManagerTest(testtools.TestCase):

    def setUp(self):
        super(AlarmManagerTest, self).setUp()
        self.http_client = fake_client.FakeHTTPClient(fixtures=fixtures)
        self.api = client.BaseClient(self.http_client)
        self.mgr = alarms.AlarmManager(self.api)

    def test_list_all(self):
        alarms = list(self.mgr.list())
        expect = [
            'GET', '/v2/alarms'
        ]
        self.http_client.assert_called(*expect)
        self.assertEqual(1, len(alarms))
        self.assertEqual('alarm-id', alarms[0].alarm_id)

    def test_list_with_query(self):
        alarms = list(self.mgr.list(q=[{"field": "project_id",
                                        "value": "project-id"},
                                       {"field": "name",
                                        "value": "SwiftObjectAlarm"}]))
        expect = [
            'GET',
            '/v2/alarms?q.field=project_id&q.field=name&q.op=&q.op='
            '&q.type=&q.type=&q.value=project-id&q.value=SwiftObjectAlarm',
        ]
        self.http_client.assert_called(*expect)
        self.assertEqual(1, len(alarms))
        self.assertEqual('alarm-id', alarms[0].alarm_id)

    def test_get(self):
        alarm = self.mgr.get(alarm_id='alarm-id')
        expect = [
            'GET', '/v2/alarms/alarm-id'
        ]
        self.http_client.assert_called(*expect)
        self.assertIsNotNone(alarm)
        self.assertEqual('alarm-id', alarm.alarm_id)
        self.assertEqual(alarm.rule, alarm.threshold_rule)

    def test_create(self):
        alarm = self.mgr.create(**CREATE_ALARM)
        expect = [
            'POST', '/v2/alarms'
        ]
        self.http_client.assert_called(*expect, body=CREATE_ALARM)
        self.assertIsNotNone(alarm)

    def test_update(self):
        alarm = self.mgr.update(alarm_id='alarm-id', **UPDATE_ALARM)
        expect_get = [
            'GET', '/v2/alarms/alarm-id'
        ]
        expect_put = [
            'PUT', '/v2/alarms/alarm-id', UPDATED_ALARM
        ]
        self.http_client.assert_called(*expect_get, pos=0)
        self.http_client.assert_called(*expect_put, pos=1)
        self.assertIsNotNone(alarm)
        self.assertEqual('alarm-id', alarm.alarm_id)
        for (key, value) in six.iteritems(UPDATED_ALARM):
            self.assertEqual(getattr(alarm, key), value)

    def test_update_delta(self):
        alarm = self.mgr.update(alarm_id='alarm-id', **DELTA_ALARM)
        expect_get = [
            'GET', '/v2/alarms/alarm-id'
        ]
        expect_put = [
            'PUT', '/v2/alarms/alarm-id', UPDATED_ALARM
        ]
        self.http_client.assert_called(*expect_get, pos=0)
        self.http_client.assert_called(*expect_put, pos=1)
        self.assertIsNotNone(alarm)
        self.assertEqual('alarm-id', alarm.alarm_id)
        for (key, value) in six.iteritems(UPDATED_ALARM):
            self.assertEqual(getattr(alarm, key), value)

    def test_set_state(self):
        state = self.mgr.set_state(alarm_id='alarm-id', state='alarm')
        expect = [
            'PUT', '/v2/alarms/alarm-id/state'
        ]
        self.http_client.assert_called(*expect, body='alarm')
        self.assertEqual({'alarm': 'alarm'}, state)

    def test_get_state(self):
        state = self.mgr.get_state(alarm_id='alarm-id')
        expect = [
            'GET', '/v2/alarms/alarm-id/state'
        ]
        self.http_client.assert_called(*expect)
        self.assertEqual({'alarm': 'alarm'}, state)

    def test_delete(self):
        deleted = self.mgr.delete(alarm_id='victim-id')
        expect = [
            'DELETE', '/v2/alarms/victim-id'
        ]
        self.http_client.assert_called(*expect)
        self.assertIsNone(deleted)

    def test_get_from_alarm_class(self):
        alarm = self.mgr.get(alarm_id='alarm-id')
        self.assertIsNotNone(alarm)
        alarm.get()
        expect = [
            'GET', '/v2/alarms/alarm-id'
        ]
        self.http_client.assert_called(*expect, pos=0)
        self.http_client.assert_called(*expect, pos=1)
        self.assertEqual('alarm-id', alarm.alarm_id)
        self.assertEqual(alarm.threshold_rule, alarm.rule)

    def test_get_state_from_alarm_class(self):
        alarm = self.mgr.get(alarm_id='alarm-id')
        self.assertIsNotNone(alarm)
        state = alarm.get_state()
        expect_get_1 = [
            'GET', '/v2/alarms/alarm-id'
        ]
        expect_get_2 = [
            'GET', '/v2/alarms/alarm-id/state'
        ]
        self.http_client.assert_called(*expect_get_1, pos=0)
        self.http_client.assert_called(*expect_get_2, pos=1)
        self.assertEqual('alarm', state)

    def test_update_missing(self):
        alarm = None
        try:
            alarm = self.mgr.update(alarm_id='unk-alarm-id', **UPDATE_ALARM)
        except exc.CommandError:
            pass
        self.assertEqual(alarm, None)

    def test_delete_from_alarm_class(self):
        alarm = self.mgr.get(alarm_id='alarm-id')
        self.assertIsNotNone(alarm)
        deleted = alarm.delete()
        expect_get = [
            'GET', '/v2/alarms/alarm-id'
        ]
        expect_delete = [
            'DELETE', '/v2/alarms/alarm-id'
        ]
        self.http_client.assert_called(*expect_get, pos=0)
        self.http_client.assert_called(*expect_delete, pos=1)
        self.assertIsNone(deleted)

    def _do_test_get_history(self, q, url):
        history = self.mgr.get_history(q=q, alarm_id='alarm-id')
        expect = ['GET', url]
        self.http_client.assert_called(*expect)
        for i in xrange(len(history)):
            change = history[i]
            self.assertIsInstance(change, alarms.AlarmChange)
            for k, v in six.iteritems(ALARM_HISTORY[i]):
                self.assertEqual(getattr(change, k), v)

    def test_get_all_history(self):
        url = '/v2/alarms/alarm-id/history'
        self._do_test_get_history(None, url)

    def test_get_constrained_history(self):
        q = [dict(field='timestamp', value='NOW')]
        url = ('/v2/alarms/alarm-id/history?q.field=timestamp'
               '&q.op=&q.type=&q.value=NOW')
        self._do_test_get_history(q, url)


class AlarmLegacyManagerTest(testtools.TestCase):

    def setUp(self):
        super(AlarmLegacyManagerTest, self).setUp()
        self.http_client = fake_client.FakeHTTPClient(fixtures=fixtures)
        self.api = client.BaseClient(self.http_client)
        self.mgr = alarms.AlarmManager(self.api)

    def test_create(self):
        alarm = self.mgr.create(**CREATE_LEGACY_ALARM)
        expect = [
            'POST', '/v2/alarms', CREATE_ALARM_WITHOUT_TC,
        ]
        self.http_client.assert_called(*expect)
        self.assertIsNotNone(alarm)

    def test_create_counter_name(self):
        create = {}
        create.update(CREATE_LEGACY_ALARM)
        create['counter_name'] = CREATE_LEGACY_ALARM['meter_name']
        del create['meter_name']
        alarm = self.mgr.create(**create)
        expect = [
            'POST', '/v2/alarms', CREATE_ALARM_WITHOUT_TC,
        ]
        self.http_client.assert_called(*expect)
        self.assertIsNotNone(alarm)

    def test_update(self):
        alarm = self.mgr.update(alarm_id='alarm-id', **DELTA_LEGACY_ALARM)
        expect_put = [
            'PUT', '/v2/alarms/alarm-id', UPDATED_ALARM
        ]
        self.http_client.assert_called(*expect_put)
        self.assertIsNotNone(alarm)
        self.assertEqual('alarm-id', alarm.alarm_id)
        for (key, value) in six.iteritems(UPDATED_ALARM):
            self.assertEqual(getattr(alarm, key), value)

    def test_update_counter_name(self):
        updated = {}
        updated.update(UPDATE_LEGACY_ALARM)
        updated['counter_name'] = UPDATED_LEGACY_ALARM['meter_name']
        del updated['meter_name']
        alarm = self.mgr.update(alarm_id='alarm-id', **updated)
        expect_put = [
            'PUT', '/v2/alarms/alarm-id', UPDATED_ALARM
        ]
        self.http_client.assert_called(*expect_put)
        self.assertIsNotNone(alarm)
        self.assertEqual('alarm-id', alarm.alarm_id)
        for (key, value) in six.iteritems(UPDATED_ALARM):
            self.assertEqual(getattr(alarm, key), value)


class AlarmTimeConstraintTest(testtools.TestCase):

    def setUp(self):
        super(AlarmTimeConstraintTest, self).setUp()
        self.http_client = fake_client.FakeHTTPClient(fixtures=fixtures)
        self.api = client.BaseClient(self.http_client)
        self.mgr = alarms.AlarmManager(self.api)

    def test_add_new(self):
        new_constraint = dict(name='cons3',
                              start='0 0 * * *',
                              duration=500)
        kwargs = dict(time_constraints=[new_constraint])
        self.mgr.update(alarm_id='alarm-id', **kwargs)
        body = copy.deepcopy(AN_ALARM)
        body[u'time_constraints'] = \
            AN_ALARM[u'time_constraints'] + [new_constraint]
        expect = [
            'PUT', '/v2/alarms/alarm-id', body
        ]
        self.http_client.assert_called(*expect)

    def test_update_existing(self):
        updated_constraint = dict(name='cons2',
                                  duration=500)
        kwargs = dict(time_constraints=[updated_constraint])
        self.mgr.update(alarm_id='alarm-id', **kwargs)
        body = copy.deepcopy(AN_ALARM)
        body[u'time_constraints'][1] = dict(name='cons2',
                                            description='desc2',
                                            start='0 23 * * *',
                                            duration=500,
                                            timezone='')

        expect = [
            'PUT', '/v2/alarms/alarm-id', body
        ]
        self.http_client.assert_called(*expect)

    def test_update_time_constraint_no_name(self):
        updated_constraint = {
            'start': '0 23 * * *',
            'duration': 500
        }
        kwargs = dict(time_constraints=[updated_constraint])
        self.mgr.update(alarm_id='alarm-id', **kwargs)
        body = copy.deepcopy(AN_ALARM)
        body[u'time_constraints'].append({
            'start': '0 23 * * *',
            'duration': 500,
        })
        expect = [
            'PUT', '/v2/alarms/alarm-id', body
        ]
        self.http_client.assert_called(*expect)

    def test_remove(self):
        kwargs = dict(remove_time_constraints=['cons2'])
        self.mgr.update(alarm_id='alarm-id', **kwargs)
        body = copy.deepcopy(AN_ALARM)
        body[u'time_constraints'] = AN_ALARM[u'time_constraints'][:1]
        expect = [
            'PUT', '/v2/alarms/alarm-id', body
        ]
        self.http_client.assert_called(*expect)