openstack/python-aodhclient
Code Issues Proposed changes

Merge "Add functional tests for prometheus type alarms"

Browse Source
This commit is contained in:
Zuul 2024-03-07 11:55:05 +00:00 committed by Gerrit Code Review
commit dc6cf1392f
3 changed files with 194 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
.zuul.yaml
View File

@ -9,15 +9,18 @@
# We neeed ceilometer's devstack plugin to install gnocchi # We neeed ceilometer's devstack plugin to install gnocchi
- openstack/ceilometer - openstack/ceilometer
- gnocchixyz/gnocchi - gnocchixyz/gnocchi
- infrawatch/sg-core
timeout: 4200 timeout: 4200
vars: vars:
devstack_localrc: devstack_localrc:
USE_PYTHON3: True USE_PYTHON3: True
GLOBAL_VENV: False GLOBAL_VENV: False
CEILOMETER_BACKENDS: "gnocchi" CEILOMETER_BACKENDS: "gnocchi,sg-core"
PROMETHEUS_SERVICE_SCRAPE_TARGETS: prometheus,sg-core
devstack_plugins: devstack_plugins:
aodh: https://opendev.org/openstack/aodh aodh: https://opendev.org/openstack/aodh
ceilometer: https://opendev.org/openstack/ceilometer ceilometer: https://opendev.org/openstack/ceilometer
sg-core: https://github.com/infrawatch/sg-core
- project: - project:
templates: templates:

189
aodhclient/tests/functional/test_alarm.py
View File

@ -890,3 +890,192 @@ class AodhClientGnocchiRulesTest(base.ClientTestBase):
# DELETE # DELETE
result = self.aodh('alarm', params="delete %s" % ALARM_ID) result = self.aodh('alarm', params="delete %s" % ALARM_ID)
self.assertEqual("", result) self.assertEqual("", result)
class AodhClientPrometheusRulesTest(base.ClientTestBase):
def test_prometheus_type_scenario(self):
QUERY = "ceilometer_image_size"
req = requests.get(
os.environ.get("PROMETHEUS_ENDPOINT") +
"/api/v1/status/runtimeinfo",
)
self.assertEqual(200, req.status_code)
# CREATE
result = self.aodh(u'alarm',
params=(u"create "
"--type prometheus "
"--name alarm_p1 "
"--threshold 80 "
"--query %s "
% QUERY))
alarm = self.details_multiple(result)[0]
ALARM_ID = alarm['alarm_id']
self.assertEqual('alarm_p1', alarm['name'])
self.assertEqual(QUERY, alarm['query'])
self.assertEqual('80.0', alarm['threshold'])
# CREATE WITH --TIME-CONSTRAINT
result = self.aodh(
u'alarm',
params=(u"create --type prometheus "
"--name alarm_ptc --threshold 80 "
"--time-constraint "
"name=cons1;start='0 11 * * *';duration=300 "
"--time-constraint "
"name=cons2;start='0 23 * * *';duration=600 "
"--query %s "
% QUERY))
alarm = self.details_multiple(result)[0]
alarm_ptc_id = alarm["alarm_id"]
self.assertEqual('alarm_ptc', alarm['name'])
self.assertEqual('80.0', alarm['threshold'])
self.assertIsNotNone(alarm['time_constraints'])
# CREATE FAIL MISSING PARAM
self.assertRaises(exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"create "
"--type prometheus "
"--name alarm1 "
"--query %s "
% QUERY))
# 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('alarm_p1', alarm_show['name'])
self.assertEqual('90.0', alarm_show['threshold'])
self.assertEqual('critical', alarm_show['severity'])
self.assertEqual(QUERY, alarm_show['query'])
# GET BY NAME
result = self.aodh(
'alarm', params="show --name alarm_p1")
alarm_show = self.details_multiple(result)[0]
self.assertEqual(ALARM_ID, alarm_show["alarm_id"])
self.assertEqual('alarm_p1', alarm_show['name'])
self.assertEqual('90.0', alarm_show['threshold'])
self.assertEqual('critical', alarm_show['severity'])
self.assertEqual(QUERY, alarm_show['query'])
# GET BY NAME AND ID ERROR
self.assertRaises(exceptions.CommandFailed,
self.aodh, u'alarm',
params=(u"show %s --name alarm_p1" %
ALARM_ID))
# LIST
result = self.aodh('alarm', params="list --filter all_projects=true")
self.assertIn(ALARM_ID,
[r['alarm_id'] for r in self.parser.listing(result)])
output_colums = ['alarm_id', 'type', 'name', 'state', 'severity',
'enabled']
for alarm_list in self.parser.listing(result):
self.assertEqual(sorted(output_colums), sorted(alarm_list.keys()))
if alarm_list["alarm_id"] == ALARM_ID:
self.assertEqual('alarm_p1', alarm_list['name'])
# LIST WITH PAGINATION
# list with limit
result = self.aodh('alarm',
params="list --filter all_projects=true --limit 1")
alarm_list = self.parser.listing(result)
self.assertEqual(1, len(alarm_list))
# list with sort with key=name dir=asc
result = self.aodh(
'alarm',
params="list --filter all_projects=true --sort name:asc")
names = [r['name'] for r in self.parser.listing(result)]
sorted_name = sorted(names)
self.assertEqual(sorted_name, names)
# list with sort with key=name dir=asc and key=alarm_id dir=asc
result = self.aodh(
u'alarm',
params=(u"create --type prometheus "
"--name alarm_p2 --threshold 80 "
"--query %s"
% QUERY))
created_alarm_id = self.details_multiple(result)[0]['alarm_id']
result = self.aodh(
'alarm',
params="list --filter all_projects=true --sort name:asc "
"--sort alarm_id:asc")
alarm_list = self.parser.listing(result)
ids_with_same_name = []
names = []
for alarm in alarm_list:
names.append(['alarm_name'])
if alarm['name'] == 'alarm_p2':
ids_with_same_name.append(alarm['alarm_id'])
sorted_ids = sorted(ids_with_same_name)
sorted_names = sorted(names)
self.assertEqual(sorted_names, names)
self.assertEqual(sorted_ids, ids_with_same_name)
# List with sort with key=name dir=desc and with the marker
# verify that we handle non full and empty pages correctly.
# Verify that marker works as expected
# We have 3 alarms. List the first 2.
result = self.aodh(
'alarm',
params="list --filter all_projects=true --filter "
"type=prometheus --sort name:asc --limit 2")
alarm_list = self.parser.listing(result)
self.assertNotIn('alarm_ptc',
[r['name'] for r in alarm_list])
self.assertEqual(2, len(alarm_list))
# List the next page. Only the 3rd alarm should be returned
result = self.aodh(
'alarm',
params="list --filter all_projects=true --filter "
"type=prometheus --sort name:asc "
"--marker %s --limit 2" % alarm_list[1]['alarm_id'])
alarm_list = self.parser.listing(result)
self.assertIn('alarm_ptc',
[r['name'] for r in alarm_list])
self.assertEqual(1, len(alarm_list))
# List the next page. Empty page should be returned
result = self.aodh(
'alarm',
params="list --filter all_projects=true --filter "
"type=prometheus --sort name:asc "
"--marker %s --limit 2" % alarm_list[0]['alarm_id'])
alarm_list = self.parser.listing(result)
self.assertEqual(0, len(alarm_list))
# Delete the last created alarm
self.aodh('alarm', params="delete %s" % created_alarm_id)
# Delete alarm_ptc
self.aodh('alarm', params="delete %s" % alarm_ptc_id)
# 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.splitlines(), expected)
# DELETE FAIL
result = self.aodh('alarm', params="delete %s" % ALARM_ID,
fail_ok=True, merge_stderr=True)
self.assertFirstLineStartsWith(result.splitlines(), expected)

1
tox.ini
View File

@ -29,6 +29,7 @@ setenv =
AODH_CLIENT_EXEC_DIR={envdir}/bin AODH_CLIENT_EXEC_DIR={envdir}/bin
AODH_ENDPOINT=http://localhost:8042 AODH_ENDPOINT=http://localhost:8042
GNOCCHI_ENDPOINT=http://localhost/metric/ GNOCCHI_ENDPOINT=http://localhost/metric/
PROMETHEUS_ENDPOINT=http://localhost:9090
allowlist_externals = allowlist_externals =
bash bash
deps = .[test] deps = .[test]