# 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 mock import re import six import sys from testtools import matchers from ceilometerclient.tests import utils from ceilometerclient.v2 import alarms from ceilometerclient.v2 import samples from ceilometerclient.v2 import shell as ceilometer_shell class ShellAlarmStateCommandsTest(utils.BaseTestCase): ALARM_ID = 'foobar' def setUp(self): super(ShellAlarmStateCommandsTest, self).setUp() self.cc = mock.Mock() self.cc.alarms = mock.Mock() self.args = mock.Mock() self.args.alarm_id = self.ALARM_ID def test_alarm_state_get(self): ceilometer_shell.do_alarm_state_get(self.cc, self.args) self.cc.alarms.get_state.assert_called_once_with(self.ALARM_ID) self.assertFalse(self.cc.alarms.set_state.called) def test_alarm_state_set(self): self.args.state = 'ok' ceilometer_shell.do_alarm_state_set(self.cc, self.args) self.cc.alarms.set_state.assert_called_once_with(self.ALARM_ID, 'ok') self.assertFalse(self.cc.alarms.get_state.called) class ShellAlarmHistoryCommandTest(utils.BaseTestCase): ALARM_ID = '768ff714-8cfb-4db9-9753-d484cb33a1cc' 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": "768ff714-8cfb-4db9-9753-d484cb33a1cc", ' '"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': '768ff714-8cfb-4db9-9753-d484cb33a1cc', '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': '768ff714-8cfb-4db9-9753-d484cb33a1cc', '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': '768ff714-8cfb-4db9-9753-d484cb33a1cc', 'project_id': '57d04f24d0824b78b1ea9bcecedbda8f', 'type': 'creation'}] TIMESTAMP_RE = (' +\| (\d{4})-(\d{2})-(\d{2})T' '(\d{2})\:(\d{2})\:(\d{2})\.(\d{6}) \| +') def setUp(self): super(ShellAlarmHistoryCommandTest, self).setUp() self.cc = mock.Mock() self.cc.alarms = mock.Mock() self.args = mock.Mock() self.args.alarm_id = self.ALARM_ID def _do_test_alarm_history(self, raw_query=None, parsed_query=None): self.args.query = raw_query orig = sys.stdout sys.stdout = six.StringIO() history = [alarms.AlarmChange(mock.Mock(), change) for change in self.ALARM_HISTORY] self.cc.alarms.get_history.return_value = history try: ceilometer_shell.do_alarm_history(self.cc, self.args) self.cc.alarms.get_history.assert_called_once_with( q=parsed_query, alarm_id=self.ALARM_ID ) out = sys.stdout.getvalue() required = [ '.*creation%sname: scombo.*' % self.TIMESTAMP_RE, '.*rule change%sdescription: combination of one.*' % self.TIMESTAMP_RE, '.*state transition%sstate: alarm.*' % self.TIMESTAMP_RE, ] for r in required: self.assertThat(out, matchers.MatchesRegex(r, re.DOTALL)) finally: sys.stdout.close() sys.stdout = orig def test_alarm_all_history(self): self._do_test_alarm_history() def test_alarm_constrained_history(self): parsed_query = [dict(field='timestamp', value='2013-10-03T08:59:28', op='gt')] self._do_test_alarm_history(raw_query='timestamp>2013-10-03T08:59:28', parsed_query=parsed_query) class ShellAlarmCommandTest(utils.BaseTestCase): ALARM_ID = '768ff714-8cfb-4db9-9753-d484cb33a1cc' ALARM = {"alarm_actions": ["log://"], "ok_actions": [], "description": "instance running hot", "timestamp": "2013-11-20T10:38:42.206952", "enabled": True, "state_timestamp": "2013-11-19T17:20:44", "threshold_rule": {"meter_name": "cpu_util", "evaluation_periods": 3, "period": 600, "statistic": "avg", "threshold": 99.0, "query": [{"field": "resource_id", "value": "INSTANCE_ID", "op": "eq"}], "comparison_operator": "gt"}, "alarm_id": ALARM_ID, "state": "insufficient data", "insufficient_data_actions": [], "repeat_actions": True, "user_id": "528d9b68fa774689834b5c04b4564f8a", "project_id": "ed9d4e2be2a748bc80108053cf4598f5", "type": "threshold", "name": "cpu_high"} def setUp(self): super(ShellAlarmCommandTest, self).setUp() self.cc = mock.Mock() self.cc.alarms = mock.Mock() self.args = mock.Mock() self.args.alarm_id = self.ALARM_ID def _do_test_alarm_update_repeat_actions(self, method, repeat_actions): self.args.threshold = 42.0 if repeat_actions is not None: self.args.repeat_actions = repeat_actions orig = sys.stdout sys.stdout = six.StringIO() alarm = [alarms.Alarm(mock.Mock(), self.ALARM)] self.cc.alarms.get.return_value = alarm self.cc.alarms.update.return_value = alarm[0] try: method(self.cc, self.args) args, kwargs = self.cc.alarms.update.call_args self.assertEqual(self.ALARM_ID, args[0]) self.assertEqual(42.0, kwargs.get('threshold')) if repeat_actions is not None: self.assertEqual(repeat_actions, kwargs.get('repeat_actions')) else: self.assertFalse('repeat_actions' in kwargs) finally: sys.stdout.close() sys.stdout = orig def test_alarm_update_repeat_actions_untouched(self): method = ceilometer_shell.do_alarm_update self._do_test_alarm_update_repeat_actions(method, None) def test_alarm_update_repeat_actions_set(self): method = ceilometer_shell.do_alarm_update self._do_test_alarm_update_repeat_actions(method, True) def test_alarm_update_repeat_actions_clear(self): method = ceilometer_shell.do_alarm_update self._do_test_alarm_update_repeat_actions(method, False) def test_alarm_combination_update_repeat_actions_untouched(self): method = ceilometer_shell.do_alarm_combination_update self._do_test_alarm_update_repeat_actions(method, None) def test_alarm_combination_update_repeat_actions_set(self): method = ceilometer_shell.do_alarm_combination_update self._do_test_alarm_update_repeat_actions(method, True) def test_alarm_combination_update_repeat_actions_clear(self): method = ceilometer_shell.do_alarm_combination_update self._do_test_alarm_update_repeat_actions(method, False) def test_alarm_threshold_update_repeat_actions_untouched(self): method = ceilometer_shell.do_alarm_threshold_update self._do_test_alarm_update_repeat_actions(method, None) def test_alarm_threshold_update_repeat_actions_set(self): method = ceilometer_shell.do_alarm_threshold_update self._do_test_alarm_update_repeat_actions(method, True) def test_alarm_threshold_update_repeat_actions_clear(self): method = ceilometer_shell.do_alarm_threshold_update self._do_test_alarm_update_repeat_actions(method, False) class ShellSampleListCommandTest(utils.BaseTestCase): METER = 'cpu_util' SAMPLES = [{"counter_name": "cpu_util", "resource_id": "5dcf5537-3161-4e25-9235-407e1385bd35", "timestamp": "2013-10-15T05:50:30", "counter_unit": "%", "counter_volume": 0.261666666667, "counter_type": "gauge"}, {"counter_name": "cpu_util", "resource_id": "87d197e9-9cf6-4c25-bc66-1b1f4cedb52f", "timestamp": "2013-10-15T05:50:29", "counter_unit": "%", "counter_volume": 0.261666666667, "counter_type": "gauge"}, {"counter_name": "cpu_util", "resource_id": "5dcf5537-3161-4e25-9235-407e1385bd35", "timestamp": "2013-10-15T05:40:30", "counter_unit": "%", "counter_volume": 0.251247920133, "counter_type": "gauge"}, {"counter_name": "cpu_util", "resource_id": "87d197e9-9cf6-4c25-bc66-1b1f4cedb52f", "timestamp": "2013-10-15T05:40:29", "counter_unit": "%", "counter_volume": 0.26, "counter_type": "gauge"}] def setUp(self): super(ShellSampleListCommandTest, self).setUp() self.cc = mock.Mock() self.cc.alarms = mock.Mock() self.args = mock.Mock() self.args.meter = self.METER self.args.query = None self.args.limit = None def test_sample_list(self): sample_list = [samples.Sample(mock.Mock(), sample) for sample in self.SAMPLES] self.cc.samples.list.return_value = sample_list org_stdout = sys.stdout try: sys.stdout = output = six.StringIO() ceilometer_shell.do_sample_list(self.cc, self.args) self.cc.samples.list.assert_called_once_with( meter_name=self.METER, q=None, limit=None) finally: sys.stdout = org_stdout self.assertEqual(output.getvalue(), '''\ +--------------------------------------+----------+-------+----------------\ +------+---------------------+ | Resource ID | Name | Type | Volume \ | Unit | Timestamp | +--------------------------------------+----------+-------+----------------\ +------+---------------------+ | 5dcf5537-3161-4e25-9235-407e1385bd35 | cpu_util | gauge | 0.261666666667 \ | % | 2013-10-15T05:50:30 | | 87d197e9-9cf6-4c25-bc66-1b1f4cedb52f | cpu_util | gauge | 0.261666666667 \ | % | 2013-10-15T05:50:29 | | 5dcf5537-3161-4e25-9235-407e1385bd35 | cpu_util | gauge | 0.251247920133 \ | % | 2013-10-15T05:40:30 | | 87d197e9-9cf6-4c25-bc66-1b1f4cedb52f | cpu_util | gauge | 0.26 \ | % | 2013-10-15T05:40:29 | +--------------------------------------+----------+-------+----------------\ +------+---------------------+ ''')