File: test_alarming_api_admin.py

package info (click to toggle)
telemetry-tempest-plugin 2.5.0-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 444 kB
  • sloc: python: 1,842; makefile: 20; sh: 11
file content (115 lines) | stat: -rw-r--r-- 4,928 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
#    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.
from tempest import config
from tempest.lib.common.utils import data_utils
from tempest.lib import decorators
from tempest.lib import exceptions as lib_exc

from telemetry_tempest_plugin.aodh.api import base

CONF = config.CONF


class TelemetryAlarmingAPIAdminTest(base.BaseAlarmingAdminTest):

    @classmethod
    def resource_setup(cls):
        super(TelemetryAlarmingAPIAdminTest, cls).resource_setup()

        if CONF.alarming_plugin.create_alarms:
            cls.rule = {
                "event_type": "compute.instance.*",
                "query": [
                    {"field": "traits.name",
                     "type": "string",
                     "op": "eq",
                     "value": "test"}]
            }
            for i in range(2):
                cls.create_alarm(event_rule=cls.rule)

    @decorators.idempotent_id('1c918e06-210b-41eb-bd45-14676dd77cd7')
    def test_alarm_list(self):
        # List alarms
        alarm_list = self.admin_client.list_alarms(query=['all_projects',
                                                          'eq',
                                                          'true'])

        # Verify created alarm in the list
        fetched_ids = [a['alarm_id'] for a in alarm_list]
        missing_alarms = [a for a in self.alarm_ids if a not in fetched_ids]
        self.assertEqual(0, len(missing_alarms),
                         "Failed to find the following created alarm(s)"
                         " in a fetched list: %s" %
                         ', '.join(str(a) for a in missing_alarms))

    @decorators.idempotent_id('f9966992-405d-475c-aa41-47213cecdf94')
    def test_alarm_create_set_log_test_actions(self):
        # Create an alarm
        alarm_name = data_utils.rand_name('telemetry_alarm')
        alarm_actions = ['test://', 'log://']
        body = self.admin_client.create_alarm(
            name=alarm_name, type='event',
            event_rule=self.rule,
            alarm_actions=alarm_actions)
        self.assertEqual(alarm_name, body['name'])
        alarm_id = body['alarm_id']
        self.assertLessEqual(self.rule.items(), body['event_rule'].items())
        self.assertEqual(['test:', 'log:'], body['alarm_actions'])

        # Verify not seen by non-admin
        self.assertRaises(lib_exc.NotFound,
                          self.alarming_client.show_alarm, alarm_id)

        # Delete alarm and verify if deleted
        self.admin_client.delete_alarm(alarm_id)
        self.assertRaises(lib_exc.NotFound,
                          self.admin_client.show_alarm, alarm_id)

    @decorators.idempotent_id('5d9f41d4-febc-47a8-a9cb-29ef77229e7b')
    def test_alarm_update_and_delete(self):
        # Create an alarm as normal user
        alarm_name = data_utils.rand_name('telemetry_alarm')
        body = self.alarming_client.create_alarm(
            name=alarm_name, type='event',
            event_rule=self.rule)
        self.assertEqual(alarm_name, body['name'])
        alarm_id = body['alarm_id']
        self.assertLessEqual(self.rule.items(), body['event_rule'].items())

        # Verify initial state as non-admin
        body = self.alarming_client.show_alarm(alarm_id)
        self.assertEqual("insufficient data", body['state'])
        self.assertEqual("Not evaluated yet", body['state_reason'])

        # Verify initial state as admin
        body = self.admin_client.show_alarm(alarm_id)
        self.assertEqual("insufficient data", body['state'])
        self.assertEqual("Not evaluated yet", body['state_reason'])

        # Update state and verify as admin
        self.admin_client.alarm_set_state(alarm_id, state='ok')
        body = self.admin_client.show_alarm(alarm_id)
        self.assertEqual('ok', body['state'])
        self.assertEqual('Manually set via API', body['state_reason'])

        # Verify state as non-admin
        body = self.alarming_client.show_alarm(alarm_id)
        self.assertEqual('ok', body['state'])
        self.assertEqual('Manually set via API', body['state_reason'])

        # Delete alarm as admin and verify if deleted
        self.admin_client.delete_alarm(alarm_id)
        self.assertRaises(lib_exc.NotFound,
                          self.alarming_client.show_alarm, alarm_id)
        self.assertRaises(lib_exc.NotFound,
                          self.admin_client.show_alarm, alarm_id)