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
|
# 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.lib import decorators
from heat_tempest_plugin.tests.functional import functional_base
class StackEventsTest(functional_base.FunctionalTestsBase):
template = '''
heat_template_version: 2014-10-16
parameters:
resources:
test_resource:
type: OS::Heat::TestResource
properties:
value: 'test1'
fail: False
update_replace: False
wait_secs: 0
outputs:
resource_id:
description: 'ID of resource'
value: { get_resource: test_resource }
'''
def _verify_event_fields(self, event, event_characteristics):
self.assertIsNotNone(event_characteristics)
self.assertIsNotNone(event.event_time)
self.assertIsNotNone(event.links)
self.assertIsNotNone(event.logical_resource_id)
self.assertIsNotNone(event.resource_status)
self.assertIn(event.resource_status, event_characteristics[1])
self.assertIsNotNone(event.resource_status_reason)
self.assertIsNotNone(event.id)
@decorators.idempotent_id('620f4f7c-74f8-48a4-a8b0-d06d0337f133')
def test_event(self):
parameters = {}
test_stack_name = self._stack_rand_name()
stack_identifier = self.stack_create(
stack_name=test_stack_name,
template=self.template,
parameters=parameters
)
expected_status = ['CREATE_IN_PROGRESS', 'CREATE_COMPLETE']
event_characteristics = {
test_stack_name: ('OS::Heat::Stack', expected_status),
'test_resource': ('OS::Heat::TestResource', expected_status)}
# List stack events
# API: GET /v1/{tenant_id}/stacks/{stack_name}/{stack_id}/events
stack_events = self.client.events.list(stack_identifier)
for stack_event in stack_events:
# Key on an expected/valid resource name
self._verify_event_fields(
stack_event,
event_characteristics[stack_event.resource_name])
# Test the event filtering API based on this resource_name
# /v1/{tenant_id}/stacks/{stack_name}/{stack_id}/resources/{resource_name}/events
resource_events = self.client.events.list(
stack_identifier,
stack_event.resource_name)
# Resource events are a subset of the original stack event list
self.assertLess(len(resource_events), len(stack_events))
# Get the event details for each resource event
for resource_event in resource_events:
# A resource_event should be in the original stack event list
self.assertIn(resource_event, stack_events)
# Given a filtered list, the resource names should be identical
self.assertEqual(
resource_event.resource_name,
stack_event.resource_name)
# Verify all fields, keying off the resource_name
self._verify_event_fields(
resource_event,
event_characteristics[resource_event.resource_name])
# Exercise the event details API
# /v1/{tenant_id}/stacks/{stack_name}/{stack_id}/resources/{resource_name}/events/{event_id}
event_details = self.client.events.get(
stack_identifier,
resource_event.resource_name,
resource_event.id)
self._verify_event_fields(
event_details,
event_characteristics[event_details.resource_name])
# The names should be identical to the non-detailed event
self.assertEqual(
resource_event.resource_name,
event_details.resource_name)
# Verify the extra field in the detail results
self.assertIsNotNone(event_details.resource_type)
self.assertEqual(
event_characteristics[event_details.resource_name][0],
event_details.resource_type)
|
