← Back to branch summary

~ubuntu-branches/ubuntu/vivid/ceilometer/vivid

« back to all changes in this revision

Viewing changes to ceilometer/tests/network/statistics/test_statistics.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2014-03-06 14:44:28 UTC
  • mto: (28.1.1 utopic-proposed) (1.2.1)
  • mto: This revision was merged to the branch mainline in revision 19.
  • Revision ID: package-import@ubuntu.com-20140306144428-rvphsh4igwyulzf0
Tags: upstream-2014.1~b3
Import upstream version 2014.1~b3

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/tests/network/statistics/test_statistics.py
 
1
#
 
2
# Copyright 2014 NEC Corporation.  All rights reserved.
 
3
#
 
4
# Licensed under the Apache License, Version 2.0 (the "License"); you may
 
5
# not use this file except in compliance with the License. You may obtain
 
6
# a copy of the License at
 
7
#
 
8
#      http://www.apache.org/licenses/LICENSE-2.0
 
9
#
 
10
# Unless required by applicable law or agreed to in writing, software
 
11
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 
12
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 
13
# License for the specific language governing permissions and limitations
 
14
# under the License.
 
15
 
 
16
from stevedore import extension
 
17
from stevedore.tests import manager as test_manager
 
18
 
 
19
from ceilometer.network import statistics
 
20
from ceilometer.network.statistics import driver
 
21
from ceilometer.openstack.common import test
 
22
from ceilometer.openstack.common import timeutils
 
23
from ceilometer import sample
 
24
 
 
25
 
 
26
class TestBase(test.BaseTestCase):
 
27
 
 
28
    def test_subclass_ok(self):
 
29
 
 
30
        class OkSubclass(statistics._Base):
 
31
 
 
32
            meter_name = 'foo'
 
33
            meter_type = sample.TYPE_GAUGE
 
34
            meter_unit = 'B'
 
35
 
 
36
        OkSubclass()
 
37
 
 
38
    def test_subclass_ng(self):
 
39
 
 
40
        class NgSubclass1(statistics._Base):
 
41
            '''meter_name is lost.'''
 
42
 
 
43
            meter_type = sample.TYPE_GAUGE
 
44
            meter_unit = 'B'
 
45
 
 
46
        class NgSubclass2(statistics._Base):
 
47
            '''meter_type is lost.'''
 
48
 
 
49
            meter_name = 'foo'
 
50
            meter_unit = 'B'
 
51
 
 
52
        class NgSubclass3(statistics._Base):
 
53
            '''meter_unit is lost.'''
 
54
 
 
55
            meter_name = 'foo'
 
56
            meter_type = sample.TYPE_GAUGE
 
57
 
 
58
        self.assertRaises(TypeError, NgSubclass1)
 
59
        self.assertRaises(TypeError, NgSubclass2)
 
60
        self.assertRaises(TypeError, NgSubclass3)
 
61
 
 
62
 
 
63
class TestBaseGetSamples(test.BaseTestCase):
 
64
 
 
65
    def setUp(self):
 
66
        super(TestBaseGetSamples, self).setUp()
 
67
 
 
68
        class FakePollster(statistics._Base):
 
69
            meter_name = 'foo'
 
70
            meter_type = sample.TYPE_CUMULATIVE
 
71
            meter_unit = 'bar'
 
72
 
 
73
        self.pollster = FakePollster()
 
74
 
 
75
    def _setup_ext_mgr(self, *drivers):
 
76
        extensions = [
 
77
            extension.Extension(str(d), None, None, d())
 
78
            for d in drivers]
 
79
        ext_mgr = test_manager.TestExtensionManager(
 
80
            extensions)
 
81
        self.pollster.extension_manager = ext_mgr
 
82
 
 
83
    def _make_fake_driver(self, *return_values):
 
84
        class FakeDriver(driver.Driver):
 
85
 
 
86
            def __init__(self):
 
87
                self.index = 0
 
88
 
 
89
            def get_sample_data(self, meter_name, resource, cache):
 
90
                if self.index >= len(return_values):
 
91
                    return None
 
92
                retval = return_values[self.index]
 
93
                self.index += 1
 
94
                return retval
 
95
        return FakeDriver
 
96
 
 
97
    def _make_timestamps(self, count):
 
98
        return [timeutils.isotime() for i in range(count)]
 
99
 
 
100
    def _get_samples(self, *resources):
 
101
 
 
102
        return [v for v in self.pollster.get_samples(self, {}, resources)]
 
103
 
 
104
    def _assert_sample(self, s, volume, resource_id, resource_metadata,
 
105
                       timestamp):
 
106
            self.assertEqual(s.name, 'foo')
 
107
            self.assertEqual(s.type, sample.TYPE_CUMULATIVE)
 
108
            self.assertEqual(s.unit, 'bar')
 
109
            self.assertEqual(s.volume, volume)
 
110
            self.assertIsNone(s.user_id)
 
111
            self.assertIsNone(s.project_id)
 
112
            self.assertEqual(s.resource_id, resource_id)
 
113
            self.assertEqual(s.timestamp, timestamp)
 
114
            self.assertEqual(s.resource_metadata, resource_metadata)
 
115
 
 
116
    def test_get_samples_one_driver_one_resource(self):
 
117
        times = self._make_timestamps(2)
 
118
        fake_driver = self._make_fake_driver((1, 'a', {'spam': 'egg'},
 
119
                                              times[0]),
 
120
                                             (2, 'b', None, times[1]))
 
121
 
 
122
        self._setup_ext_mgr(fake_driver)
 
123
 
 
124
        samples = self._get_samples('http://foo')
 
125
 
 
126
        self.assertEqual(len(samples), 1)
 
127
        self._assert_sample(samples[0], 1, 'a', {'spam': 'egg'}, times[0])
 
128
 
 
129
    def test_get_samples_one_driver_two_resource(self):
 
130
        times = self._make_timestamps(3)
 
131
        fake_driver = self._make_fake_driver((1, 'a', {'spam': 'egg'},
 
132
                                              times[0]),
 
133
                                             (2, 'b', None, times[1]),
 
134
                                             (3, 'c', None, times[2]))
 
135
 
 
136
        self._setup_ext_mgr(fake_driver)
 
137
 
 
138
        samples = self._get_samples('http://foo', 'http://bar')
 
139
 
 
140
        self.assertEqual(len(samples), 2)
 
141
        self._assert_sample(samples[0], 1, 'a', {'spam': 'egg'}, times[2])
 
142
        self._assert_sample(samples[1], 2, 'b', None, times[1])
 
143
 
 
144
    def test_get_samples_two_driver_one_resource(self):
 
145
        times = self._make_timestamps(4)
 
146
        fake_driver1 = self._make_fake_driver((1, 'a', {'spam': 'egg'},
 
147
                                               times[0]),
 
148
                                              (2, 'b', None), times[1])
 
149
 
 
150
        fake_driver2 = self._make_fake_driver((11, 'A', None, times[2]),
 
151
                                              (12, 'B', None, times[3]))
 
152
 
 
153
        self._setup_ext_mgr(fake_driver1, fake_driver2)
 
154
 
 
155
        samples = self._get_samples('http://foo')
 
156
 
 
157
        self.assertEqual(len(samples), 2)
 
158
        self._assert_sample(samples[0], 1, 'a', {'spam': 'egg'}, times[0])
 
159
        self._assert_sample(samples[1], 11, 'A', None, times[2])
 
160
 
 
161
    def test_get_samples_multi_samples(self):
 
162
        times = self._make_timestamps(2)
 
163
        fake_driver = self._make_fake_driver([(1, 'a', {'spam': 'egg'},
 
164
                                               times[0]),
 
165
                                              (2, 'b', None, times[1])])
 
166
 
 
167
        self._setup_ext_mgr(fake_driver)
 
168
 
 
169
        samples = self._get_samples('http://foo')
 
170
 
 
171
        self.assertEqual(len(samples), 2)
 
172
        self._assert_sample(samples[0], 1, 'a', {'spam': 'egg'}, times[0])
 
173
        self._assert_sample(samples[1], 2, 'b', None, times[1])
 
174
 
 
175
    def test_get_samples_return_none(self):
 
176
        fake_driver = self._make_fake_driver(None)
 
177
 
 
178
        self._setup_ext_mgr(fake_driver)
 
179
 
 
180
        samples = self._get_samples('http://foo')
 
181
 
 
182
        self.assertEqual(len(samples), 0)