← Back to branch summary

~ubuntu-cloud-archive/ubuntu/precise/cinder/trunk

« back to all changes in this revision

Viewing changes to cinder/tests/api/contrib/test_admin_actions.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short
  • Date: 2012-11-23 08:39:28 UTC
  • mfrom: (1.1.9)
  • Revision ID: package-import@ubuntu.com-20121123083928-xvzet603cjfj9p1t
Tags: 2013.1~g1-0ubuntu1
https://launchpad.net/bugs/1075948
* New upstream release.
* debian/patches/avoid_setuptools_git_dependency.patch:
  Avoid git installation. (LP: #1075948) 

Show diffs side-by-side

added added

removed removed

Lines of Context:
cinder/tests/api/contrib/test_admin_actions.py
 
1
import webob
 
2
 
 
3
from cinder import context
 
4
from cinder import db
 
5
from cinder import exception
 
6
from cinder.openstack.common import jsonutils
 
7
from cinder import test
 
8
from cinder.tests.api.openstack import fakes
 
9
from cinder.volume import api as volume_api
 
10
 
 
11
 
 
12
def app():
 
13
    # no auth, just let environ['cinder.context'] pass through
 
14
    api = fakes.router.APIRouter()
 
15
    mapper = fakes.urlmap.URLMap()
 
16
    mapper['/v1'] = api
 
17
    return mapper
 
18
 
 
19
 
 
20
class AdminActionsTest(test.TestCase):
 
21
 
 
22
    def setUp(self):
 
23
        super(AdminActionsTest, self).setUp()
 
24
        self.flags(rpc_backend='cinder.openstack.common.rpc.impl_fake')
 
25
        self.volume_api = volume_api.API()
 
26
 
 
27
    def test_reset_status_as_admin(self):
 
28
        # admin context
 
29
        ctx = context.RequestContext('admin', 'fake', True)
 
30
        # current status is available
 
31
        volume = db.volume_create(ctx, {'status': 'available'})
 
32
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
33
        req.method = 'POST'
 
34
        req.headers['content-type'] = 'application/json'
 
35
        # request status of 'error'
 
36
        req.body = jsonutils.dumps({'os-reset_status': {'status': 'error'}})
 
37
        # attach admin context to request
 
38
        req.environ['cinder.context'] = ctx
 
39
        resp = req.get_response(app())
 
40
        # request is accepted
 
41
        self.assertEquals(resp.status_int, 202)
 
42
        volume = db.volume_get(ctx, volume['id'])
 
43
        # status changed to 'error'
 
44
        self.assertEquals(volume['status'], 'error')
 
45
 
 
46
    def test_reset_status_as_non_admin(self):
 
47
        # current status is 'error'
 
48
        volume = db.volume_create(context.get_admin_context(),
 
49
                                  {'status': 'error'})
 
50
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
51
        req.method = 'POST'
 
52
        req.headers['content-type'] = 'application/json'
 
53
        # request changing status to available
 
54
        req.body = jsonutils.dumps({'os-reset_status': {'status':
 
55
                                                        'available'}})
 
56
        # non-admin context
 
57
        req.environ['cinder.context'] = context.RequestContext('fake', 'fake')
 
58
        resp = req.get_response(app())
 
59
        # request is not authorized
 
60
        self.assertEquals(resp.status_int, 403)
 
61
        volume = db.volume_get(context.get_admin_context(), volume['id'])
 
62
        # status is still 'error'
 
63
        self.assertEquals(volume['status'], 'error')
 
64
 
 
65
    def test_malformed_reset_status_body(self):
 
66
        # admin context
 
67
        ctx = context.RequestContext('admin', 'fake', True)
 
68
        # current status is available
 
69
        volume = db.volume_create(ctx, {'status': 'available'})
 
70
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
71
        req.method = 'POST'
 
72
        req.headers['content-type'] = 'application/json'
 
73
        # malformed request body
 
74
        req.body = jsonutils.dumps({'os-reset_status': {'x-status': 'bad'}})
 
75
        # attach admin context to request
 
76
        req.environ['cinder.context'] = ctx
 
77
        resp = req.get_response(app())
 
78
        # bad request
 
79
        self.assertEquals(resp.status_int, 400)
 
80
        volume = db.volume_get(ctx, volume['id'])
 
81
        # status is still 'available'
 
82
        self.assertEquals(volume['status'], 'available')
 
83
 
 
84
    def test_invalid_status_for_volume(self):
 
85
        # admin context
 
86
        ctx = context.RequestContext('admin', 'fake', True)
 
87
        # current status is available
 
88
        volume = db.volume_create(ctx, {'status': 'available'})
 
89
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
90
        req.method = 'POST'
 
91
        req.headers['content-type'] = 'application/json'
 
92
        # 'invalid' is not a valid status
 
93
        req.body = jsonutils.dumps({'os-reset_status': {'status': 'invalid'}})
 
94
        # attach admin context to request
 
95
        req.environ['cinder.context'] = ctx
 
96
        resp = req.get_response(app())
 
97
        # bad request
 
98
        self.assertEquals(resp.status_int, 400)
 
99
        volume = db.volume_get(ctx, volume['id'])
 
100
        # status is still 'available'
 
101
        self.assertEquals(volume['status'], 'available')
 
102
 
 
103
    def test_reset_status_for_missing_volume(self):
 
104
        # admin context
 
105
        ctx = context.RequestContext('admin', 'fake', True)
 
106
        # missing-volume-id
 
107
        req = webob.Request.blank('/v1/fake/volumes/%s/action' %
 
108
                                  'missing-volume-id')
 
109
        req.method = 'POST'
 
110
        req.headers['content-type'] = 'application/json'
 
111
        # malformed request body
 
112
        req.body = jsonutils.dumps({'os-reset_status': {'status':
 
113
                                                        'available'}})
 
114
        # attach admin context to request
 
115
        req.environ['cinder.context'] = ctx
 
116
        resp = req.get_response(app())
 
117
        # not found
 
118
        self.assertEquals(resp.status_int, 404)
 
119
        self.assertRaises(exception.NotFound, db.volume_get, ctx,
 
120
                          'missing-volume-id')
 
121
 
 
122
    def test_reset_attached_status(self):
 
123
        # admin context
 
124
        ctx = context.RequestContext('admin', 'fake', True)
 
125
        # current status is available
 
126
        volume = db.volume_create(ctx, {'status': 'available',
 
127
                                        'attach_status': 'attached'})
 
128
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
129
        req.method = 'POST'
 
130
        req.headers['content-type'] = 'application/json'
 
131
        # request update attach_status to detached
 
132
        body = {'os-reset_status': {'status': 'available',
 
133
                                    'attach_status': 'detached'}}
 
134
        req.body = jsonutils.dumps(body)
 
135
        # attach admin context to request
 
136
        req.environ['cinder.context'] = ctx
 
137
        resp = req.get_response(app())
 
138
        # request is accepted
 
139
        self.assertEquals(resp.status_int, 202)
 
140
        volume = db.volume_get(ctx, volume['id'])
 
141
        # attach_status changed to 'detached'
 
142
        self.assertEquals(volume['attach_status'], 'detached')
 
143
        # status un-modified
 
144
        self.assertEquals(volume['status'], 'available')
 
145
 
 
146
    def test_invalid_reset_attached_status(self):
 
147
        # admin context
 
148
        ctx = context.RequestContext('admin', 'fake', True)
 
149
        # current status is available
 
150
        volume = db.volume_create(ctx, {'status': 'available',
 
151
                                        'attach_status': 'detached'})
 
152
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
153
        req.method = 'POST'
 
154
        req.headers['content-type'] = 'application/json'
 
155
        # 'invalid' is not a valid attach_status
 
156
        body = {'os-reset_status': {'status': 'available',
 
157
                                    'attach_status': 'invalid'}}
 
158
        req.body = jsonutils.dumps(body)
 
159
        # attach admin context to request
 
160
        req.environ['cinder.context'] = ctx
 
161
        resp = req.get_response(app())
 
162
        # bad request
 
163
        self.assertEquals(resp.status_int, 400)
 
164
        volume = db.volume_get(ctx, volume['id'])
 
165
        # status and attach_status un-modified
 
166
        self.assertEquals(volume['status'], 'available')
 
167
        self.assertEquals(volume['attach_status'], 'detached')
 
168
 
 
169
    def test_snapshot_reset_status(self):
 
170
        # admin context
 
171
        ctx = context.RequestContext('admin', 'fake', True)
 
172
        # snapshot in 'error_deleting'
 
173
        volume = db.volume_create(ctx, {})
 
174
        snapshot = db.snapshot_create(ctx, {'status': 'error_deleting',
 
175
                                            'volume_id': volume['id']})
 
176
        req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
 
177
                                  snapshot['id'])
 
178
        req.method = 'POST'
 
179
        req.headers['content-type'] = 'application/json'
 
180
        # request status of 'error'
 
181
        req.body = jsonutils.dumps({'os-reset_status': {'status': 'error'}})
 
182
        # attach admin context to request
 
183
        req.environ['cinder.context'] = ctx
 
184
        resp = req.get_response(app())
 
185
        # request is accepted
 
186
        self.assertEquals(resp.status_int, 202)
 
187
        snapshot = db.snapshot_get(ctx, snapshot['id'])
 
188
        # status changed to 'error'
 
189
        self.assertEquals(snapshot['status'], 'error')
 
190
 
 
191
    def test_invalid_status_for_snapshot(self):
 
192
        # admin context
 
193
        ctx = context.RequestContext('admin', 'fake', True)
 
194
        # snapshot in 'available'
 
195
        volume = db.volume_create(ctx, {})
 
196
        snapshot = db.snapshot_create(ctx, {'status': 'available',
 
197
                                            'volume_id': volume['id']})
 
198
        req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
 
199
                                  snapshot['id'])
 
200
        req.method = 'POST'
 
201
        req.headers['content-type'] = 'application/json'
 
202
        # 'attaching' is not a valid status for snapshots
 
203
        req.body = jsonutils.dumps({'os-reset_status': {'status':
 
204
                                                        'attaching'}})
 
205
        # attach admin context to request
 
206
        req.environ['cinder.context'] = ctx
 
207
        resp = req.get_response(app())
 
208
        # request is accepted
 
209
        self.assertEquals(resp.status_int, 400)
 
210
        snapshot = db.snapshot_get(ctx, snapshot['id'])
 
211
        # status is still 'available'
 
212
        self.assertEquals(snapshot['status'], 'available')
 
213
 
 
214
    def test_force_delete(self):
 
215
        # admin context
 
216
        ctx = context.RequestContext('admin', 'fake', True)
 
217
        # current status is creating
 
218
        volume = db.volume_create(ctx, {'status': 'creating'})
 
219
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
220
        req.method = 'POST'
 
221
        req.headers['content-type'] = 'application/json'
 
222
        req.body = jsonutils.dumps({'os-force_delete': {}})
 
223
        # attach admin context to request
 
224
        req.environ['cinder.context'] = ctx
 
225
        resp = req.get_response(app())
 
226
        # request is accepted
 
227
        self.assertEquals(resp.status_int, 202)
 
228
        # volume is deleted
 
229
        self.assertRaises(exception.NotFound, db.volume_get, ctx, volume['id'])
 
230
 
 
231
    def test_force_delete_snapshot(self):
 
232
        # admin context
 
233
        ctx = context.RequestContext('admin', 'fake', True)
 
234
        # current status is creating
 
235
        volume = db.volume_create(ctx, {'host': 'test'})
 
236
        snapshot = db.snapshot_create(ctx, {'status': 'creating',
 
237
                                            'volume_size': 1,
 
238
                                            'volume_id': volume['id']})
 
239
        path = '/v1/fake/snapshots/%s/action' % snapshot['id']
 
240
        req = webob.Request.blank(path)
 
241
        req.method = 'POST'
 
242
        req.headers['content-type'] = 'application/json'
 
243
        req.body = jsonutils.dumps({'os-force_delete': {}})
 
244
        # attach admin context to request
 
245
        req.environ['cinder.context'] = ctx
 
246
        # start service to handle rpc.cast for 'delete snapshot'
 
247
        self.start_service('volume', host='test')
 
248
        # make request
 
249
        resp = req.get_response(app())
 
250
        # request is accepted
 
251
        self.assertEquals(resp.status_int, 202)
 
252
        # snapshot is deleted
 
253
        self.assertRaises(exception.NotFound, db.snapshot_get, ctx,
 
254
                          snapshot['id'])
 
255
 
 
256
    def test_force_detach_volume(self):
 
257
        # admin context
 
258
        ctx = context.RequestContext('admin', 'fake', True)
 
259
        # current status is available
 
260
        volume = db.volume_create(ctx, {'status': 'available', 'host': 'test',
 
261
                                        'provider_location': ''})
 
262
        # start service to handle rpc messages for attach requests
 
263
        self.start_service('volume', host='test')
 
264
        self.volume_api.reserve_volume(ctx, volume)
 
265
        self.volume_api.initialize_connection(ctx, volume, {})
 
266
        mountpoint = '/dev/vbd'
 
267
        self.volume_api.attach(ctx, volume, fakes.FAKE_UUID, mountpoint)
 
268
        # volume is attached
 
269
        volume = db.volume_get(ctx, volume['id'])
 
270
        self.assertEquals(volume['status'], 'in-use')
 
271
        self.assertEquals(volume['instance_uuid'], fakes.FAKE_UUID)
 
272
        self.assertEquals(volume['mountpoint'], mountpoint)
 
273
        self.assertEquals(volume['attach_status'], 'attached')
 
274
        # build request to force detach
 
275
        req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
 
276
        req.method = 'POST'
 
277
        req.headers['content-type'] = 'application/json'
 
278
        # request status of 'error'
 
279
        req.body = jsonutils.dumps({'os-force_detach': None})
 
280
        # attach admin context to request
 
281
        req.environ['cinder.context'] = ctx
 
282
        # make request
 
283
        resp = req.get_response(app())
 
284
        # request is accepted
 
285
        self.assertEquals(resp.status_int, 202)
 
286
        volume = db.volume_get(ctx, volume['id'])
 
287
        # status changed to 'available'
 
288
        self.assertEquals(volume['status'], 'available')
 
289
        self.assertEquals(volume['instance_uuid'], None)
 
290
        self.assertEquals(volume['mountpoint'], None)
 
291
        self.assertEquals(volume['attach_status'], 'detached')