← Back to branch summary

~ubuntu-branches/ubuntu/utopic/ceilometer/utopic

« back to all changes in this revision

Viewing changes to .pc/skip-database-tests.patch/tests/api/v2/list_events.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2013-07-28 21:58:31 UTC
  • Revision ID: package-import@ubuntu.com-20130728215831-t26vfzgt0yz99oul
Tags: 2013.2~b2-0ubuntu4
debian/patches/skip-database-tests.patch: Refreshed due to precise failures. 

Show diffs side-by-side

added added

removed removed

Lines of Context:
.pc/skip-database-tests.patch/tests/api/v2/list_events.py
 
1
# -*- encoding: utf-8 -*-
 
2
#
 
3
# Copyright © 2012 New Dream Network, LLC (DreamHost)
 
4
#
 
5
# Author: Doug Hellmann <doug.hellmann@dreamhost.com>
 
6
#
 
7
# Licensed under the Apache License, Version 2.0 (the "License"); you may
 
8
# not use this file except in compliance with the License. You may obtain
 
9
# a copy of the License at
 
10
#
 
11
#      http://www.apache.org/licenses/LICENSE-2.0
 
12
#
 
13
# Unless required by applicable law or agreed to in writing, software
 
14
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
15
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
16
# License for the specific language governing permissions and limitations
 
17
# under the License.
 
18
"""Test listing raw events.
 
19
"""
 
20
 
 
21
import datetime
 
22
import logging
 
23
import webtest.app
 
24
 
 
25
from oslo.config import cfg
 
26
 
 
27
from ceilometer.publisher import rpc
 
28
from ceilometer import counter
 
29
 
 
30
from .base import FunctionalTest
 
31
 
 
32
LOG = logging.getLogger(__name__)
 
33
 
 
34
 
 
35
class TestListEvents(FunctionalTest):
 
36
 
 
37
    def setUp(self):
 
38
        super(TestListEvents, self).setUp()
 
39
        self.counter1 = counter.Counter(
 
40
            'instance',
 
41
            'cumulative',
 
42
            '',
 
43
            1,
 
44
            'user-id',
 
45
            'project1',
 
46
            'resource-id',
 
47
            timestamp=datetime.datetime(2012, 7, 2, 10, 40),
 
48
            resource_metadata={'display_name': 'test-server',
 
49
                               'tag': 'self.counter',
 
50
                               'ignored_dict': {'key': 'value'},
 
51
                               'ignored_list': ['not-returned'],
 
52
                               }
 
53
        )
 
54
        msg = rpc.meter_message_from_counter(
 
55
            self.counter1,
 
56
            cfg.CONF.publisher_rpc.metering_secret,
 
57
            'test_source',
 
58
        )
 
59
        self.conn.record_metering_data(msg)
 
60
 
 
61
        self.counter2 = counter.Counter(
 
62
            'instance',
 
63
            'cumulative',
 
64
            '',
 
65
            1,
 
66
            'user-id2',
 
67
            'project2',
 
68
            'resource-id-alternate',
 
69
            timestamp=datetime.datetime(2012, 7, 2, 10, 41),
 
70
            resource_metadata={'display_name': 'test-server',
 
71
                               'tag': 'self.counter2',
 
72
                               }
 
73
        )
 
74
        msg2 = rpc.meter_message_from_counter(
 
75
            self.counter2,
 
76
            cfg.CONF.publisher_rpc.metering_secret,
 
77
            'source2',
 
78
        )
 
79
        self.conn.record_metering_data(msg2)
 
80
 
 
81
    def test_all(self):
 
82
        data = self.get_json('/meters/instance')
 
83
        self.assertEquals(2, len(data))
 
84
 
 
85
    def test_all_limit(self):
 
86
        data = self.get_json('/meters/instance?limit=1')
 
87
        self.assertEquals(1, len(data))
 
88
 
 
89
    def test_all_limit_negative(self):
 
90
        self.assertRaises(webtest.app.AppError,
 
91
                          self.get_json,
 
92
                          '/meters/instance?limit=-2')
 
93
 
 
94
    def test_all_limit_bigger(self):
 
95
        data = self.get_json('/meters/instance?limit=42')
 
96
        self.assertEquals(2, len(data))
 
97
 
 
98
    def test_empty_project(self):
 
99
        data = self.get_json('/meters/instance',
 
100
                             q=[{'field': 'project_id',
 
101
                                 'value': 'no-such-project',
 
102
                                 }])
 
103
        self.assertEquals([], data)
 
104
 
 
105
    def test_by_project(self):
 
106
        data = self.get_json('/meters/instance',
 
107
                             q=[{'field': 'project_id',
 
108
                                 'value': 'project1',
 
109
                                 }])
 
110
        self.assertEquals(1, len(data))
 
111
 
 
112
    def test_empty_resource(self):
 
113
        data = self.get_json('/meters/instance',
 
114
                             q=[{'field': 'resource_id',
 
115
                                 'value': 'no-such-resource',
 
116
                                 }])
 
117
        self.assertEquals([], data)
 
118
 
 
119
    def test_by_resource(self):
 
120
        data = self.get_json('/meters/instance',
 
121
                             q=[{'field': 'resource_id',
 
122
                                 'value': 'resource-id',
 
123
                                 }])
 
124
        self.assertEquals(1, len(data))
 
125
 
 
126
    def test_empty_source(self):
 
127
        data = self.get_json('/meters/instance',
 
128
                             q=[{'field': 'source',
 
129
                                 'value': 'no-such-source',
 
130
                                 }])
 
131
        self.assertEquals(0, len(data))
 
132
 
 
133
    def test_by_source(self):
 
134
        data = self.get_json('/meters/instance',
 
135
                             q=[{'field': 'source',
 
136
                                 'value': 'test_source',
 
137
                                 }])
 
138
        self.assertEquals(1, len(data))
 
139
 
 
140
    def test_empty_user(self):
 
141
        data = self.get_json('/meters/instance',
 
142
                             q=[{'field': 'user_id',
 
143
                                 'value': 'no-such-user',
 
144
                                 }])
 
145
        self.assertEquals([], data)
 
146
 
 
147
    def test_by_user(self):
 
148
        data = self.get_json('/meters/instance',
 
149
                             q=[{'field': 'user_id',
 
150
                                 'value': 'user-id',
 
151
                                 }])
 
152
        self.assertEquals(1, len(data))
 
153
 
 
154
    def test_metadata(self):
 
155
        data = self.get_json('/meters/instance',
 
156
                             q=[{'field': 'resource_id',
 
157
                                 'value': 'resource-id',
 
158
                                 }])
 
159
        sample = data[0]
 
160
        self.assert_('resource_metadata' in sample)
 
161
        self.assertEqual(
 
162
            list(sorted(sample['resource_metadata'].iteritems())),
 
163
            [('display_name', 'test-server'),
 
164
             ('tag', 'self.counter'),
 
165
             ])