openstack/python-aodhclient
Code Issues Proposed changes
python-aodhclient/aodhclient/tests/functional/test_alarm.py
liusheng 3377106557 Return the error message of API request failure 
For now, the error message from Aodh API can not be return when a request
failure occur(e.g. BadRequest, Conflict). That is because the fualtstring
from the response isn't extracted correctly.

Change-Id: I5e0048f969d2e852042c4aa2c7e6e0bff254d15b
2016-02-17 15:26:20 +08:00

599 lines
25 KiB
Python
Raw Blame History

# 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 uuid
import six
from tempest_lib import exceptions
from aodhclient.tests.functional import base
class AodhClientTest(base.ClientTestBase):
def test_help(self):
self.aodh("help", params="alarm create")
self.aodh("help", params="alarm delete")
self.aodh("help", params="alarm list")
self.aodh("help", params="alarm search")
self.aodh("help", params="alarm show")
self.aodh("help", params="alarm update")
def test_event_scenario(self):
PROJECT_ID = str(uuid.uuid4())
# CREATE
result = self.aodh(u'alarm',
params=(u"create --type event --name ev_alarm1 "
"--project-id %s" % PROJECT_ID))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('ev_alarm1', alarm['name'])
self.assertEqual('*', alarm['event_type'])
# CREATE FAIL
result = self.aodh(u'alarm',
params=(u"create --type event --name ev_alarm1 "
"--project-id %s" % PROJECT_ID),
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(
result.split('\n'),
"Alarm with name='ev_alarm1' exists (HTTP 409)")
# UPDATE IGNORE INVALID
result = self.aodh(
'alarm', params=("update %s --severity critical --threshold 10"
% ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('critical', alarm_updated['severity'])
# UPDATE IGNORE INVALID
result = self.aodh(
'alarm', params=("update %s --event-type dummy" % ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('dummy', alarm_updated['event_type'])
# GET
result = self.aodh(
'alarm', params="show %s" % ALARM_ID)
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual(PROJECT_ID, alarm_show["project_id"])
self.assertEqual('ev_alarm1', alarm_show['name'])
self.assertEqual('dummy', alarm_show['event_type'])
# LIST
result = self.aodh('alarm', params="list --type event")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('ev_alarm1', alarm_list['name'])
# SEARCH ALL
result = self.aodh('alarm', params=("search --type event"))
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('ev_alarm1', alarm_list['name'])
# SEARCH SOME
result = self.aodh('alarm',
params=("search --type event --query "
"'{\"=\": {\"project_id\": \"%s\"}}'"
% PROJECT_ID))
alarm_list = self.parser.listing(result)[0]
self.assertEqual(ALARM_ID, alarm_list["alarm_id"])
self.assertEqual('ev_alarm1', alarm_list['name'])
# DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result)
# GET FAIL
result = self.aodh('alarm', params="show %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
expected = "Alarm %s not found (HTTP 404)" % ALARM_ID
self.assertFirstLineStartsWith(result.split('\n'), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.split('\n'), expected)
# LIST DOES NOT HAVE ALARM
result = self.aodh('alarm', params="list --type event")
self.assertNotIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
def test_threshold_scenario(self):
PROJECT_ID = str(uuid.uuid4())
# CREATE
result = self.aodh(u'alarm',
params=(u"create --type threshold --name alarm1 "
" -m meter_name --threshold 5 "
"--project-id %s" % PROJECT_ID))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('alarm1', alarm['name'])
self.assertEqual('meter_name', alarm['meter_name'])
self.assertEqual('5.0', alarm['threshold'])
# CREATE FAIL
result = self.aodh(u'alarm',
params=(u"create --type threshold --name alarm1 "
"-m meter_name --threshold 5 "
"--project-id %s" % PROJECT_ID),
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(
result.split('\n'), "Alarm with name='alarm1' exists (HTTP 409)")
# CREATE FAIL MISSING PARAM
self.assertRaises(exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"create --type threshold --name alarm1 "
"--project-id %s" % PROJECT_ID))
# UPDATE
result = self.aodh(
'alarm', params=("update %s --severity critical --threshold 10"
% ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('critical', alarm_updated['severity'])
self.assertEqual('10.0', alarm_updated["threshold"])
# GET
result = self.aodh(
'alarm', params="show %s" % ALARM_ID)
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual(PROJECT_ID, alarm_show["project_id"])
self.assertEqual('alarm1', alarm_show['name'])
self.assertEqual('meter_name', alarm_show['meter_name'])
self.assertEqual('10.0', alarm_show['threshold'])
# LIST
result = self.aodh('alarm', params="list --type threshold")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH ALL
result = self.aodh('alarm', params=("search --type threshold"))
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH SOME
result = self.aodh('alarm',
params=("search --type threshold --query "
"'{\"=\": {\"project_id\": \"%s\"}}'"
% PROJECT_ID))
alarm_list = self.parser.listing(result)[0]
self.assertEqual(ALARM_ID, alarm_list["alarm_id"])
self.assertEqual('alarm1', alarm_list['name'])
# DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result)
# GET FAIL
result = self.aodh('alarm', params="show %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
expected = "Alarm %s not found (HTTP 404)" % ALARM_ID
self.assertFirstLineStartsWith(result.split('\n'), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.split('\n'), expected)
# LIST DOES NOT HAVE ALARM
result = self.aodh('alarm', params="list --type threshold")
self.assertNotIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
class AodhClientGnocchiRulesTest(base.ClientTestBase):
def test_gnocchi_resources_threshold_scenario(self):
PROJECT_ID = str(uuid.uuid4())
# NOTE(gordc): taken from setup-tests.sh
RESOURCE_ID = '6868DA77-FA82-4E67-ABA9-270C5AE8CBCA'
# CREATE
result = self.aodh(u'alarm',
params=(u"create "
"--type gnocchi_resources_threshold "
"--name alarm1 --metric cpu_util "
"--threshold 80 "
"--resource-id %s --resource-type instance "
"--aggregation-method last "
"--project-id %s"
% (RESOURCE_ID, PROJECT_ID)))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('alarm1', alarm['name'])
self.assertEqual('cpu_util', alarm['metric'])
self.assertEqual('80.0', alarm['threshold'])
self.assertEqual('last', alarm['aggregation_method'])
self.assertEqual('6868DA77-FA82-4E67-ABA9-270C5AE8CBCA',
alarm['resource_id'])
self.assertEqual('instance', alarm['resource_type'])
# CREATE FAIL
result = self.aodh(u'alarm',
params=(u"create "
"--type gnocchi_resources_threshold "
"--name alarm1 --metric cpu_util "
"--threshold 80 "
"--resource-id %s --resource-type instance "
"--aggregation-method last "
"--project-id %s"
% (RESOURCE_ID, PROJECT_ID)),
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(
result.split('\n'), "Alarm with name='alarm1' exists (HTTP 409)")
# CREATE FAIL MISSING PARAM
self.assertRaises(exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"create "
"--type gnocchi_resources_threshold "
"--name alarm1 --metric cpu_util "
"--resource-id %s --resource-type instance "
"--aggregation-method last "
"--project-id %s"
% (RESOURCE_ID, PROJECT_ID)))
# UPDATE
result = self.aodh(
'alarm', params=("update %s --severity critical --threshold 90"
% ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('critical', alarm_updated['severity'])
self.assertEqual('90.0', alarm_updated["threshold"])
# GET
result = self.aodh(
'alarm', params="show %s" % ALARM_ID)
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual(PROJECT_ID, alarm_show["project_id"])
self.assertEqual('alarm1', alarm_show['name'])
self.assertEqual('cpu_util', alarm_show['metric'])
self.assertEqual('90.0', alarm_show['threshold'])
self.assertEqual('critical', alarm_show['severity'])
self.assertEqual('last', alarm_show['aggregation_method'])
self.assertEqual('instance', alarm_show['resource_type'])
# LIST
result = self.aodh(
'alarm', params="list --type gnocchi_resources_threshold")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH ALL
result = self.aodh(
'alarm', params=("search --type gnocchi_resources_threshold"))
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH SOME
result = self.aodh('alarm',
params=("search --type gnocchi_resources_threshold "
"--query "
"'{\"=\": {\"project_id\": \"%s\"}}'"
% PROJECT_ID))
alarm_list = self.parser.listing(result)[0]
self.assertEqual(ALARM_ID, alarm_list["alarm_id"])
self.assertEqual('alarm1', alarm_list['name'])
# DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result)
# GET FAIL
result = self.aodh('alarm', params="show %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
expected = "Alarm %s not found (HTTP 404)" % ALARM_ID
self.assertFirstLineStartsWith(result.split('\n'), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.split('\n'), expected)
# LIST DOES NOT HAVE ALARM
result = self.aodh('alarm',
params="list --type gnocchi_resources_threshold")
self.assertNotIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
def test_gnocchi_aggr_by_resources_scenario(self):
PROJECT_ID = str(uuid.uuid4())
# CREATE
result = self.aodh(
u'alarm',
params=(u"create "
"--type "
"gnocchi_aggregation_by_resources_threshold "
"--name alarm1 --metric cpu --threshold 80 "
"--query "
"'{\"=\": {\"server_group\": \"my_group\"}}' "
"--resource-type instance "
"--aggregation-method last "
"--project-id %s" % PROJECT_ID))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('alarm1', alarm['name'])
self.assertEqual('cpu', alarm['metric'])
self.assertEqual('80.0', alarm['threshold'])
self.assertEqual('last', alarm['aggregation_method'])
self.assertEqual('instance', alarm['resource_type'])
self.assertEqual('{"=": {"server_group": "my_group"}}',
alarm['query'])
# CREATE FAIL
result = self.aodh(
u'alarm',
params=(u"create "
"--type "
"gnocchi_aggregation_by_resources_threshold "
"--name alarm1 --metric cpu --threshold 80 "
"--query "
"'{\"=\": {\"server_group\": \"my_group\"}}' "
"--resource-type instance "
"--aggregation-method last "
"--project-id %s" % PROJECT_ID),
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(
result.split('\n'), "Alarm with name='alarm1' exists (HTTP 409)")
# CREATE FAIL MISSING PARAM
self.assertRaises(
exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"create "
"--type "
"gnocchi_aggregation_by_resources_threshold "
"--name alarm1 --metric cpu "
"--query "
"'{\"=\": {\"server_group\": \"my_group\"}}' "
"--resource-type instance "
"--aggregation-method last "
"--project-id %s" % PROJECT_ID))
# UPDATE
result = self.aodh(
'alarm', params=("update %s --severity critical --threshold 90"
% ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('critical', alarm_updated['severity'])
self.assertEqual('90.0', alarm_updated["threshold"])
# GET
result = self.aodh(
'alarm', params="show %s" % ALARM_ID)
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual(PROJECT_ID, alarm_show["project_id"])
self.assertEqual('alarm1', alarm_show['name'])
self.assertEqual('cpu', alarm_show['metric'])
self.assertEqual('90.0', alarm_show['threshold'])
self.assertEqual('critical', alarm_show['severity'])
self.assertEqual('last', alarm_show['aggregation_method'])
self.assertEqual('instance', alarm_show['resource_type'])
# LIST
result = self.aodh(
'alarm', params="list --type "
"gnocchi_aggregation_by_resources_threshold")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH ALL
result = self.aodh(
'alarm', params=("search --type "
"gnocchi_aggregation_by_resources_threshold"))
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH SOME
result = self.aodh(
'alarm', params=("search --type "
"gnocchi_aggregation_by_resources_threshold "
"--query '{\"=\": {\"project_id\": \"%s\"}}'"
% PROJECT_ID))
alarm_list = self.parser.listing(result)[0]
self.assertEqual(ALARM_ID, alarm_list["alarm_id"])
self.assertEqual('alarm1', alarm_list['name'])
# DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result)
# GET FAIL
result = self.aodh('alarm', params="show %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
expected = "Alarm %s not found (HTTP 404)" % ALARM_ID
self.assertFirstLineStartsWith(result.split('\n'), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.split('\n'), expected)
# LIST DOES NOT HAVE ALARM
result = self.aodh('alarm', params="list --type "
"gnocchi_aggregation_by_resources_threshold")
self.assertNotIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
def test_gnocchi_aggr_by_metrics_scenario(self):
PROJECT_ID = str(uuid.uuid4())
METRIC1 = 'cpu'
METRIC2 = 'cpu_util'
# CREATE
result = self.aodh(
u'alarm',
params=(u"create "
"--type gnocchi_aggregation_by_metrics_threshold "
"--name alarm1 "
"--metrics %s "
"--metrics %s "
"--threshold 80 "
"--aggregation-method last "
"--project-id %s"
% (METRIC1, METRIC2, PROJECT_ID)))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('alarm1', alarm['name'])
metrics = "[u'cpu', u'cpu_util']" if six.PY2 else "['cpu', 'cpu_util']"
self.assertEqual(metrics, alarm['metrics'])
self.assertEqual('80.0', alarm['threshold'])
self.assertEqual('last', alarm['aggregation_method'])
# CREATE FAIL
result = self.aodh(
u'alarm',
params=(u"create "
"--type gnocchi_aggregation_by_metrics_threshold "
"--name alarm1 "
"--metrics %s "
"--metrics %s "
"--threshold 80 "
"--aggregation-method last "
"--project-id %s"
% (METRIC1, METRIC2, PROJECT_ID)),
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(
result.split('\n'), "Alarm with name='alarm1' exists (HTTP 409)")
# CREATE FAIL MISSING PARAM
self.assertRaises(
exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"create "
"--type gnocchi_aggregation_by_metrics_threshold "
"--name alarm1 "
"--metrics %s "
"--metrics %s "
"--aggregation-method last "
"--project-id %s"
% (METRIC1, METRIC2, PROJECT_ID)))
# UPDATE
result = self.aodh(
'alarm', params=("update %s --severity critical --threshold 90"
% ALARM_ID))
alarm_updated = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_updated["alarm_id"])
self.assertEqual('critical', alarm_updated['severity'])
self.assertEqual('90.0', alarm_updated["threshold"])
# GET
result = self.aodh(
'alarm', params="show %s" % ALARM_ID)
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual(PROJECT_ID, alarm_show["project_id"])
self.assertEqual('alarm1', alarm_show['name'])
self.assertEqual(metrics, alarm_show['metrics'])
self.assertEqual('90.0', alarm_show['threshold'])
self.assertEqual('critical', alarm_show['severity'])
self.assertEqual('last', alarm_show['aggregation_method'])
# LIST
result = self.aodh(
'alarm', params="list --type "
"gnocchi_aggregation_by_metrics_threshold")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH ALL
result = self.aodh(
'alarm', params=("search --type "
"gnocchi_aggregation_by_metrics_threshold"))
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
for alarm_list in self.parser.listing(result):
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm1', alarm_list['name'])
# SEARCH SOME
result = self.aodh(
'alarm',
params=("search --type "
"gnocchi_aggregation_by_metrics_threshold "
"--query '{\"=\": {\"project_id\": \"%s\"}}'"
% PROJECT_ID))
alarm_list = self.parser.listing(result)[0]
self.assertEqual(ALARM_ID, alarm_list["alarm_id"])
self.assertEqual('alarm1', alarm_list['name'])
# DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result)
# GET FAIL
result = self.aodh('alarm', params="show %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
expected = "Alarm %s not found (HTTP 404)" % ALARM_ID
self.assertFirstLineStartsWith(result.split('\n'), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.split('\n'), expected)
# LIST DOES NOT HAVE ALARM
result = self.aodh(
'alarm', params="list --type "
"gnocchi_aggregation_by_metrics_threshold")
self.assertNotIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
View Git Blame Copy Permalink