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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
|
# Copyright 2016 IBM Corp.
#
# 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 unittest import mock
from oslo_policy import policy as oslo_policy
from nova import exception
from nova import policy
from nova import test
from nova.tests import fixtures as nova_fixtures
from nova.tests.functional.api import client
from nova.tests.functional import fixtures as func_fixtures
from nova.tests.functional import integrated_helpers
class InstanceActionsTestV2(integrated_helpers._IntegratedTestBase):
"""Tests Instance Actions API"""
def test_get_instance_actions(self):
server = self._create_server()
actions = self.api.get_instance_actions(server['id'])
self.assertEqual('create', actions[0]['action'])
def test_get_instance_actions_deleted(self):
server = self._create_server()
self._delete_server(server)
self.assertRaises(client.OpenStackApiNotFoundException,
self.api.get_instance_actions,
server['id'])
class InstanceActionsTestV21(InstanceActionsTestV2):
api_major_version = 'v2.1'
class InstanceActionsTestV221(InstanceActionsTestV21):
microversion = '2.21'
def setUp(self):
super(InstanceActionsTestV221, self).setUp()
self.api.microversion = self.microversion
def test_get_instance_actions_deleted(self):
server = self._create_server()
self._delete_server(server)
actions = self.api.get_instance_actions(server['id'])
self.assertEqual('delete', actions[0]['action'])
self.assertEqual('create', actions[1]['action'])
def test_get_instance_actions_shelve_deleted(self):
server = self._create_server()
self._shelve_server(server)
self._delete_server(server)
actions = self.api.get_instance_actions(server['id'])
self.assertEqual('delete', actions[0]['action'])
self.assertEqual('shelve', actions[1]['action'])
self.assertEqual('create', actions[2]['action'])
class HypervisorError(Exception):
"""This is just used to make sure the exception type is in the events."""
pass
class InstanceActionEventFaultsTestCase(
test.TestCase, integrated_helpers.InstanceHelperMixin):
"""Tests for the instance action event details reporting from the API"""
def setUp(self):
super(InstanceActionEventFaultsTestCase, self).setUp()
# Setup the standard fixtures.
self.useFixture(nova_fixtures.GlanceFixture(self))
self.useFixture(nova_fixtures.NeutronFixture(self))
self.useFixture(func_fixtures.PlacementFixture())
self.useFixture(nova_fixtures.RealPolicyFixture())
# Start the compute services.
self.start_service('conductor')
self.start_service('scheduler')
self.compute = self.start_service('compute')
api_fixture = self.useFixture(nova_fixtures.OSAPIFixture(
api_version='v2.1'))
self.api = api_fixture.api
self.admin_api = api_fixture.admin_api
def _set_policy_rules(self, overwrite=True):
rules = {'os_compute_api:os-instance-actions:show': '',
'os_compute_api:os-instance-actions:events:details':
'project_id:%(project_id)s'}
policy.set_rules(oslo_policy.Rules.from_dict(rules),
overwrite=overwrite)
def test_instance_action_event_details_non_nova_exception(self):
"""Creates a server using the non-admin user, then reboot it which
will generate a non-NovaException fault and put the instance into
ERROR status. Then checks that fault details are visible.
"""
# Create the server with the non-admin user.
server = self._build_server(
networks=[{'port': nova_fixtures.NeutronFixture.port_1['id']}])
server = self.api.post_server({'server': server})
server = self._wait_for_state_change(server, 'ACTIVE')
# Stop the server before rebooting it so that after the driver.reboot
# method raises an exception, the fake driver does not report the
# instance power state as running - that will make the compute manager
# set the instance vm_state to error.
self.api.post_server_action(server['id'], {'os-stop': None})
server = self._wait_for_state_change(server, 'SHUTOFF')
# Stub out the compute driver reboot method to raise a non-nova
# exception to simulate some error from the underlying hypervisor
# which in this case we are going to say has sensitive content.
error_msg = 'sensitive info'
with mock.patch.object(
self.compute.manager.driver, 'reboot',
side_effect=HypervisorError(error_msg)) as mock_reboot:
reboot_request = {'reboot': {'type': 'HARD'}}
self.api.post_server_action(server['id'], reboot_request)
# In this case we wait for the status to change to ERROR using
# the non-admin user so we can assert the fault details. We also
# wait for the task_state to be None since the wrap_instance_fault
# decorator runs before the reverts_task_state decorator so we will
# be sure the fault is set on the server.
server = self._wait_for_server_parameter(
server, {'status': 'ERROR', 'OS-EXT-STS:task_state': None},
api=self.api)
mock_reboot.assert_called_once()
self._set_policy_rules(overwrite=False)
server_id = server['id']
# Calls GET on the server actions and verifies that the reboot
# action expected in the response.
response = self.api.api_get('/servers/%s/os-instance-actions' %
server_id)
server_actions = response.body['instanceActions']
for actions in server_actions:
if actions['action'] == 'reboot':
reboot_request_id = actions['request_id']
# non admin shows instance actions details and verifies the 'details'
# in the action events via 'request_id', since microversion 2.51 that
# we can show events, but in microversion 2.84 that we can show
# 'details' for non-admin.
self.api.microversion = '2.84'
action_events_response = self.api.api_get(
'/servers/%s/os-instance-actions/%s' % (server_id,
reboot_request_id))
reboot_action = action_events_response.body['instanceAction']
# Since reboot action failed, the 'message' property in reboot action
# should be 'Error', otherwise it's None.
self.assertEqual('Error', reboot_action['message'])
reboot_action_events = reboot_action['events']
# The instance action events from the non-admin user API response
# should not have 'traceback' in it.
self.assertNotIn('traceback', reboot_action_events[0])
# And the sensitive details from the non-nova exception should not be
# in the details.
self.assertIn('details', reboot_action_events[0])
self.assertNotIn(error_msg, reboot_action_events[0]['details'])
# The exception type class name should be in the details.
self.assertIn('HypervisorError', reboot_action_events[0]['details'])
# Get the server fault details for the admin user.
self.admin_api.microversion = '2.84'
action_events_response = self.admin_api.api_get(
'/servers/%s/os-instance-actions/%s' % (server_id,
reboot_request_id))
reboot_action = action_events_response.body['instanceAction']
self.assertEqual('Error', reboot_action['message'])
reboot_action_events = reboot_action['events']
# The admin can see the fault details which includes the traceback,
# and make sure the traceback is there by looking for part of it.
self.assertIn('traceback', reboot_action_events[0])
self.assertIn('in reboot_instance',
reboot_action_events[0]['traceback'])
# The exception type class name should be in the details for the admin
# user as well since the fault handling code cannot distinguish who
# is going to see the message so it only sets class name.
self.assertIn('HypervisorError', reboot_action_events[0]['details'])
def test_instance_action_event_details_with_nova_exception(self):
"""Creates a server using the non-admin user, then reboot it which
will generate a nova exception fault and put the instance into
ERROR status. Then checks that fault details are visible.
"""
# Create the server with the non-admin user.
server = self._build_server(
networks=[{'port': nova_fixtures.NeutronFixture.port_1['id']}])
server = self.api.post_server({'server': server})
server = self._wait_for_state_change(server, 'ACTIVE')
# Stop the server before rebooting it so that after the driver.reboot
# method raises an exception, the fake driver does not report the
# instance power state as running - that will make the compute manager
# set the instance vm_state to error.
self.api.post_server_action(server['id'], {'os-stop': None})
server = self._wait_for_state_change(server, 'SHUTOFF')
# Stub out the compute driver reboot method to raise a nova
# exception 'InstanceRebootFailure' to simulate some error.
exc_reason = 'reboot failure'
with mock.patch.object(
self.compute.manager.driver, 'reboot',
side_effect=exception.InstanceRebootFailure(reason=exc_reason)
) as mock_reboot:
reboot_request = {'reboot': {'type': 'HARD'}}
self.api.post_server_action(server['id'], reboot_request)
# In this case we wait for the status to change to ERROR using
# the non-admin user so we can assert the fault details. We also
# wait for the task_state to be None since the wrap_instance_fault
# decorator runs before the reverts_task_state decorator so we will
# be sure the fault is set on the server.
server = self._wait_for_server_parameter(
server, {'status': 'ERROR', 'OS-EXT-STS:task_state': None},
api=self.api)
mock_reboot.assert_called_once()
self._set_policy_rules(overwrite=False)
server_id = server['id']
# Calls GET on the server actions and verifies that the reboot
# action expected in the response.
response = self.api.api_get('/servers/%s/os-instance-actions' %
server_id)
server_actions = response.body['instanceActions']
for actions in server_actions:
if actions['action'] == 'reboot':
reboot_request_id = actions['request_id']
# non admin shows instance actions details and verifies the 'details'
# in the action events via 'request_id', since microversion 2.51 that
# we can show events, but in microversion 2.84 that we can show
# 'details' for non-admin.
self.api.microversion = '2.84'
action_events_response = self.api.api_get(
'/servers/%s/os-instance-actions/%s' % (server_id,
reboot_request_id))
reboot_action = action_events_response.body['instanceAction']
# Since reboot action failed, the 'message' property in reboot action
# should be 'Error', otherwise it's None.
self.assertEqual('Error', reboot_action['message'])
reboot_action_events = reboot_action['events']
# The instance action events from the non-admin user API response
# should not have 'traceback' in it.
self.assertNotIn('traceback', reboot_action_events[0])
# The nova exception format message should be in the details.
self.assertIn('details', reboot_action_events[0])
self.assertIn(exc_reason, reboot_action_events[0]['details'])
# Get the server fault details for the admin user.
self.admin_api.microversion = '2.84'
action_events_response = self.admin_api.api_get(
'/servers/%s/os-instance-actions/%s' % (server_id,
reboot_request_id))
reboot_action = action_events_response.body['instanceAction']
self.assertEqual('Error', reboot_action['message'])
reboot_action_events = reboot_action['events']
# The admin can see the fault details which includes the traceback,
# and make sure the traceback is there by looking for part of it.
self.assertIn('traceback', reboot_action_events[0])
self.assertIn('in reboot_instance',
reboot_action_events[0]['traceback'])
# The nova exception format message should be in the details.
self.assertIn(exc_reason, reboot_action_events[0]['details'])
|
