← Back to branch summary

~ubuntu-branches/ubuntu/saucy/ceilometer/saucy

« back to all changes in this revision

Viewing changes to tests/api/v2/test_alarm.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, Yolanda Robla, Chuck Short, James Page
  • Date: 2013-07-19 10:04:36 UTC
  • mfrom: (1.1.5)
  • Revision ID: package-import@ubuntu.com-20130719100436-8ppgb9laoyr60zyr
Tags: 2013.2~b2-0ubuntu1
[ Yolanda Robla ]
* debian/patches/default-dbconnection-sqlite.patch: updated db section

[ Chuck Short ]
* New upstream version.
* debian/patches/default-dbconnection-sqlite.patch: Refreshed.
* debian/control: Bump requirements for stevedore to 0.9.
* debian/control: Add python-simplejson
* debian/control: Drop python-keystoneclient hardcoded version.
* debian/control: Add python-testscenarios as a build depends.
* debian/control: Add python-cinderclient as a build depends.
* debian/control: Add python-ceilometerclient as a build depends.
* debian/control: Add python-alembic as build depends.
* debian/control: Add python-oslo.sphinx as build-depends.
* debian/control: Update runtime depends.
* debian/control: Add python-happybase.
* debian/ceilometer-common.install: Add ceilometer-alarm-singleton,
  ceilometer-alarm-notifier, and ceilometer-expirer.
* debian/patches/skip-database-tests.patch: Skip database tests
  if the database is not installed.

[ James Page ]
* d/control: Update VCS fields for new branch locations.

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/api/v2/test_alarm.py
1
 
# -*- encoding: utf-8 -*-
2
 
#
3
 
# Copyright © 2013 eNovance <licensing@enovance.com>
4
 
#
5
 
# Author: Mehdi Abaakouk <mehdi.abaakouk@enovance.com>
6
 
#         Angus Salkeld <asalkeld@redhat.com>
7
 
#
8
 
# Licensed under the Apache License, Version 2.0 (the 'License'); you may
9
 
# not use this file except in compliance with the License. You may obtain
10
 
# a copy of the License at
11
 
#
12
 
#      http://www.apache.org/licenses/LICENSE-2.0
13
 
#
14
 
# Unless required by applicable law or agreed to in writing, software
15
 
# distributed under the License is distributed on an 'AS IS' BASIS, WITHOUT
16
 
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
17
 
# License for the specific language governing permissions and limitations
18
 
# under the License.
19
 
'''Tests alarm operation
20
 
'''
21
 
 
22
 
import logging
23
 
import uuid
24
 
 
25
 
from .base import FunctionalTest
26
 
 
27
 
from ceilometer.storage.models import Alarm
28
 
 
29
 
LOG = logging.getLogger(__name__)
30
 
 
31
 
 
32
 
class TestListEmptyAlarms(FunctionalTest):
33
 
 
34
 
    def test_empty(self):
35
 
        data = self.get_json('/alarms')
36
 
        self.assertEquals([], data)
37
 
 
38
 
 
39
 
class TestAlarms(FunctionalTest):
40
 
 
41
 
    def setUp(self):
42
 
        super(TestAlarms, self).setUp()
43
 
 
44
 
        self.auth_headers = {'X-User-Id': str(uuid.uuid1()),
45
 
                             'X-Project-Id': str(uuid.uuid1())}
46
 
        for alarm in [Alarm(name='name1',
47
 
                            counter_name='meter.test',
48
 
                            comparison_operator='gt', threshold=2.0,
49
 
                            statistic='avg',
50
 
                            user_id=self.auth_headers['X-User-Id'],
51
 
                            project_id=self.auth_headers['X-Project-Id']),
52
 
                      Alarm(name='name2',
53
 
                            counter_name='meter.mine',
54
 
                            comparison_operator='gt', threshold=2.0,
55
 
                            statistic='avg',
56
 
                            user_id=self.auth_headers['X-User-Id'],
57
 
                            project_id=self.auth_headers['X-Project-Id']),
58
 
                      Alarm(name='name3',
59
 
                            counter_name='meter.test',
60
 
                            comparison_operator='gt', threshold=2.0,
61
 
                            statistic='avg',
62
 
                            user_id=self.auth_headers['X-User-Id'],
63
 
                            project_id=self.auth_headers['X-Project-Id'])]:
64
 
            self.conn.update_alarm(alarm)
65
 
 
66
 
    def test_list_alarms(self):
67
 
        data = self.get_json('/alarms')
68
 
        self.assertEquals(3, len(data))
69
 
        self.assertEquals(set(r['name'] for r in data),
70
 
                          set(['name1',
71
 
                               'name2',
72
 
                               'name3']))
73
 
        self.assertEquals(set(r['counter_name'] for r in data),
74
 
                          set(['meter.test',
75
 
                               'meter.mine']))
76
 
 
77
 
    def test_get_alarm(self):
78
 
        alarms = self.get_json('/alarms',
79
 
                               q=[{'field': 'name',
80
 
                                   'value': 'name1',
81
 
                                   }])
82
 
        for a in alarms:
83
 
            print '%s: %s' % (a['name'], a['alarm_id'])
84
 
        self.assertEquals(alarms[0]['name'], 'name1')
85
 
        self.assertEquals(alarms[0]['counter_name'], 'meter.test')
86
 
 
87
 
        one = self.get_json('/alarms/%s' % alarms[0]['alarm_id'])
88
 
        self.assertEquals(one['name'], 'name1')
89
 
        self.assertEquals(one['counter_name'], 'meter.test')
90
 
        self.assertEquals(one['alarm_id'], alarms[0]['alarm_id'])
91
 
 
92
 
    def test_post_invalid_alarm(self):
93
 
        json = {
94
 
            'name': 'added_alarm',
95
 
            'counter_name': 'ameter',
96
 
            'comparison_operator': 'gt',
97
 
            'threshold': 2.0,
98
 
            'statistic': 'magic',
99
 
        }
100
 
        self.post_json('/alarms', params=json, expect_errors=True, status=400,
101
 
                       headers=self.auth_headers)
102
 
        alarms = list(self.conn.get_alarms())
103
 
        self.assertEquals(3, len(alarms))
104
 
 
105
 
    def test_post_alarm(self):
106
 
        json = {
107
 
            'name': 'added_alarm',
108
 
            'counter_name': 'ameter',
109
 
            'comparison_operator': 'gt',
110
 
            'threshold': 2.0,
111
 
            'statistic': 'avg',
112
 
        }
113
 
        self.post_json('/alarms', params=json, status=200,
114
 
                       headers=self.auth_headers)
115
 
        alarms = list(self.conn.get_alarms())
116
 
        self.assertEquals(4, len(alarms))
117
 
 
118
 
    def test_put_alarm(self):
119
 
        json = {
120
 
            'name': 'renamed_alarm',
121
 
        }
122
 
        data = self.get_json('/alarms',
123
 
                             q=[{'field': 'name',
124
 
                                 'value': 'name1',
125
 
                                 }])
126
 
        self.assertEquals(1, len(data))
127
 
        alarm_id = data[0]['alarm_id']
128
 
 
129
 
        self.put_json('/alarms/%s' % alarm_id,
130
 
                      params=json,
131
 
                      headers=self.auth_headers)
132
 
        alarm = list(self.conn.get_alarms(alarm_id=alarm_id))[0]
133
 
        self.assertEquals(alarm.name, json['name'])
134
 
 
135
 
    def test_put_alarm_wrong_field(self):
136
 
        '''
137
 
        Note: wsme will ignore unknown fields so will
138
 
        just not appear in the Alarm.
139
 
        '''
140
 
        json = {
141
 
            'name': 'renamed_alarm',
142
 
            'this_can_not_be_correct': 'ha',
143
 
        }
144
 
        data = self.get_json('/alarms',
145
 
                             q=[{'field': 'name',
146
 
                                 'value': 'name1',
147
 
                                 }],
148
 
                             headers=self.auth_headers)
149
 
        self.assertEquals(1, len(data))
150
 
 
151
 
        resp = self.put_json('/alarms/%s' % data[0]['alarm_id'],
152
 
                             params=json,
153
 
                             expect_errors=True,
154
 
                             headers=self.auth_headers)
155
 
        self.assertEquals(resp.status_code, 200)
156
 
 
157
 
    def test_delete_alarm(self):
158
 
        data = self.get_json('/alarms')
159
 
        self.assertEquals(3, len(data))
160
 
 
161
 
        self.delete('/alarms/%s' % data[0]['alarm_id'],
162
 
                    status=200)
163
 
        alarms = list(self.conn.get_alarms())
164
 
        self.assertEquals(2, len(alarms))