← Back to branch summary

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

« back to all changes in this revision

Viewing changes to tests/energy/test_kwapi.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/energy/test_kwapi.py
52
52
        self.keystone = None
53
53
 
54
54
 
55
 
class TestKwapiPollster(base.TestCase):
56
 
 
57
 
    @staticmethod
58
 
    def fake_kwapi_iter_probes(self, ksclient):
59
 
        probes = PROBE_DICT['probes']
60
 
        for key, value in probes.iteritems():
61
 
            probe_dict = value
62
 
            probe_dict['id'] = key
63
 
            yield probe_dict
64
 
 
65
 
    @staticmethod
66
 
    def fake_kwapi_get_kwapi_client(self, ksclient):
 
55
class TestKwapi(base.TestCase):
 
56
 
 
57
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
58
    def setUp(self):
 
59
        super(TestKwapi, self).setUp()
 
60
        self.context = context.get_admin_context()
 
61
        self.manager = TestManager()
 
62
 
 
63
    @staticmethod
 
64
    def fake_get_kwapi_client(self, ksclient):
67
65
        raise exceptions.EndpointNotFound("fake keystone exception")
68
66
 
69
 
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
70
 
    def setUp(self):
71
 
        super(TestKwapiPollster, self).setUp()
72
 
        self.context = context.get_admin_context()
73
 
        self.manager = TestManager()
74
 
 
75
 
    def test_kwapi_endpoint_not_exist(self):
 
67
    def test_endpoint_not_exist(self):
76
68
        self.stubs.Set(kwapi._Base, 'get_kwapi_client',
77
 
                       self.fake_kwapi_get_kwapi_client)
 
69
                       self.fake_get_kwapi_client)
78
70
 
79
 
        counters = list(kwapi.KwapiPollster().get_counters(self.manager))
 
71
        counters = list(kwapi.EnergyPollster().get_counters(self.manager, {}))
80
72
        self.assertEqual(len(counters), 0)
81
73
 
82
 
    def test_kwapi_counter(self):
83
 
        self.stubs.Set(kwapi._Base, 'iter_probes', self.fake_kwapi_iter_probes)
84
 
 
85
 
        counters = list(kwapi.KwapiPollster().get_counters(self.manager))
86
 
        self.assertEqual(len(counters), 6)
87
 
        energy_counters = [counter for counter in counters
88
 
                           if counter.name == "energy"]
89
 
        power_counters = [counter for counter in counters
90
 
                          if counter.name == "power"]
91
 
        for probe in PROBE_DICT['probes'].values():
92
 
            self.assert_(
93
 
                any(map(lambda counter: counter.timestamp ==
94
 
                    datetime.datetime.fromtimestamp(
95
 
                        probe['timestamp']).isoformat(),
96
 
                        counters)))
97
 
            self.assert_(
98
 
                any(map(lambda counter: counter.volume == probe['kwh'],
99
 
                        energy_counters)))
100
 
            self.assert_(
101
 
                any(map(lambda counter: counter.volume == probe['w'],
102
 
                        power_counters)))
103
 
 
104
 
    def test_kwapi_counter_list(self):
105
 
        self.stubs.Set(kwapi._Base, 'iter_probes', self.fake_kwapi_iter_probes)
106
 
 
107
 
        counters = list(kwapi.KwapiPollster().get_counters(self.manager))
108
 
        self.assertEqual(set([c.name for c in counters]),
109
 
                         set(kwapi.KwapiPollster().get_counter_names()))
 
74
 
 
75
class TestEnergyPollster(base.TestCase):
 
76
 
 
77
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
78
    def setUp(self):
 
79
        super(TestEnergyPollster, self).setUp()
 
80
        self.context = context.get_admin_context()
 
81
        self.manager = TestManager()
 
82
        self.stubs.Set(kwapi._Base, '_iter_probes',
 
83
                       self.fake_iter_probes)
 
84
 
 
85
    @staticmethod
 
86
    def fake_iter_probes(self, ksclient, cache):
 
87
        probes = PROBE_DICT['probes']
 
88
        for key, value in probes.iteritems():
 
89
            probe_dict = value
 
90
            probe_dict['id'] = key
 
91
            yield probe_dict
 
92
 
 
93
    def test_counter(self):
 
94
        cache = {}
 
95
        counters = list(kwapi.EnergyPollster().get_counters(
 
96
            self.manager,
 
97
            cache,
 
98
        ))
 
99
        self.assertEqual(len(counters), 3)
 
100
        counters_by_name = dict((c.resource_id, c) for c in counters)
 
101
        for name, probe in PROBE_DICT['probes'].items():
 
102
            counter = counters_by_name[name]
 
103
            expected = datetime.datetime.fromtimestamp(
 
104
                probe['timestamp']
 
105
            ).isoformat()
 
106
            self.assertEqual(counter.timestamp, expected)
 
107
            self.assertEqual(counter.volume, probe['kwh'])
 
108
            # self.assert_(
 
109
            #     any(map(lambda counter: counter.volume == probe['w'],
 
110
            #             power_counters)))
 
111
 
 
112
 
 
113
class TestEnergyPollsterCache(base.TestCase):
 
114
 
 
115
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
116
    def setUp(self):
 
117
        super(TestEnergyPollsterCache, self).setUp()
 
118
        self.context = context.get_admin_context()
 
119
        self.manager = TestManager()
 
120
 
 
121
    def test_get_counters_cached(self):
 
122
        probe = {'id': 'A'}
 
123
        probe.update(PROBE_DICT['probes']['A'])
 
124
        cache = {
 
125
            kwapi.EnergyPollster.CACHE_KEY_PROBE: [probe],
 
126
        }
 
127
        self.manager.keystone = mock.Mock()
 
128
        pollster = kwapi.EnergyPollster()
 
129
        with mock.patch.object(pollster, '_get_probes') as do_not_call:
 
130
            do_not_call.side_effect = AssertionError('should not be called')
 
131
            counters = list(pollster.get_counters(self.manager, cache))
 
132
        self.assertEqual(len(counters), 1)
 
133
 
 
134
 
 
135
class TestPowerPollster(base.TestCase):
 
136
 
 
137
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
138
    def setUp(self):
 
139
        super(TestPowerPollster, self).setUp()
 
140
        self.context = context.get_admin_context()
 
141
        self.manager = TestManager()
 
142
        self.stubs.Set(kwapi._Base, '_iter_probes',
 
143
                       self.fake_iter_probes)
 
144
 
 
145
    @staticmethod
 
146
    def fake_iter_probes(self, ksclient, cache):
 
147
        probes = PROBE_DICT['probes']
 
148
        for key, value in probes.iteritems():
 
149
            probe_dict = value
 
150
            probe_dict['id'] = key
 
151
            yield probe_dict
 
152
 
 
153
    def test_counter(self):
 
154
        cache = {}
 
155
        counters = list(kwapi.PowerPollster().get_counters(
 
156
            self.manager,
 
157
            cache,
 
158
        ))
 
159
        self.assertEqual(len(counters), 3)
 
160
        counters_by_name = dict((c.resource_id, c) for c in counters)
 
161
        for name, probe in PROBE_DICT['probes'].items():
 
162
            counter = counters_by_name[name]
 
163
            expected = datetime.datetime.fromtimestamp(
 
164
                probe['timestamp']
 
165
            ).isoformat()
 
166
            self.assertEqual(counter.timestamp, expected)
 
167
            self.assertEqual(counter.volume, probe['w'])
 
168
 
 
169
 
 
170
class TestPowerPollsterCache(base.TestCase):
 
171
 
 
172
    @mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
 
173
    def setUp(self):
 
174
        super(TestPowerPollsterCache, self).setUp()
 
175
        self.context = context.get_admin_context()
 
176
        self.manager = TestManager()
 
177
 
 
178
    def test_get_counters_cached(self):
 
179
        probe = {'id': 'A'}
 
180
        probe.update(PROBE_DICT['probes']['A'])
 
181
        cache = {
 
182
            kwapi.PowerPollster.CACHE_KEY_PROBE: [probe],
 
183
        }
 
184
        self.manager.keystone = mock.Mock()
 
185
        pollster = kwapi.PowerPollster()
 
186
        with mock.patch.object(pollster, '_get_probes') as do_not_call:
 
187
            do_not_call.side_effect = AssertionError('should not be called')
 
188
            counters = list(pollster.get_counters(self.manager, cache))
 
189
        self.assertEqual(len(counters), 1)