3
from cinder import context
5
from cinder import exception
6
from cinder import test
7
from cinder.openstack.common import jsonutils
8
from cinder.tests.api.openstack import fakes
12
# no auth, just let environ['cinder.context'] pass through
13
api = fakes.volume.APIRouter()
14
mapper = fakes.urlmap.URLMap()
19
class AdminActionsTest(test.TestCase):
21
def test_reset_status_as_admin(self):
23
ctx = context.RequestContext('admin', 'fake', True)
24
# current status is available
25
volume = db.volume_create(ctx, {'status': 'available'})
26
req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
28
req.headers['content-type'] = 'application/json'
29
# request status of 'error'
30
req.body = jsonutils.dumps({'os-reset_status': {'status': 'error'}})
31
# attach admin context to request
32
req.environ['cinder.context'] = ctx
33
resp = req.get_response(app())
35
self.assertEquals(resp.status_int, 202)
36
volume = db.volume_get(ctx, volume['id'])
37
# status changed to 'error'
38
self.assertEquals(volume['status'], 'error')
40
def test_reset_status_as_non_admin(self):
41
# current status is 'error'
42
volume = db.volume_create(context.get_admin_context(),
44
req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
46
req.headers['content-type'] = 'application/json'
47
# request changing status to available
48
req.body = jsonutils.dumps({'os-reset_status': {'status':
51
req.environ['cinder.context'] = context.RequestContext('fake', 'fake')
52
resp = req.get_response(app())
53
# request is not authorized
54
self.assertEquals(resp.status_int, 403)
55
volume = db.volume_get(context.get_admin_context(), volume['id'])
56
# status is still 'error'
57
self.assertEquals(volume['status'], 'error')
59
def test_malformed_reset_status_body(self):
61
ctx = context.RequestContext('admin', 'fake', True)
62
# current status is available
63
volume = db.volume_create(ctx, {'status': 'available'})
64
req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
66
req.headers['content-type'] = 'application/json'
67
# malformed request body
68
req.body = jsonutils.dumps({'os-reset_status': {'x-status': 'bad'}})
69
# attach admin context to request
70
req.environ['cinder.context'] = ctx
71
resp = req.get_response(app())
73
self.assertEquals(resp.status_int, 400)
74
volume = db.volume_get(ctx, volume['id'])
75
# status is still 'available'
76
self.assertEquals(volume['status'], 'available')
78
def test_invalid_status_for_volume(self):
80
ctx = context.RequestContext('admin', 'fake', True)
81
# current status is available
82
volume = db.volume_create(ctx, {'status': 'available'})
83
req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
85
req.headers['content-type'] = 'application/json'
86
# 'invalid' is not a valid status
87
req.body = jsonutils.dumps({'os-reset_status': {'status': 'invalid'}})
88
# attach admin context to request
89
req.environ['cinder.context'] = ctx
90
resp = req.get_response(app())
92
self.assertEquals(resp.status_int, 400)
93
volume = db.volume_get(ctx, volume['id'])
94
# status is still 'available'
95
self.assertEquals(volume['status'], 'available')
97
def test_reset_status_for_missing_volume(self):
99
ctx = context.RequestContext('admin', 'fake', True)
101
req = webob.Request.blank('/v1/fake/volumes/%s/action' %
104
req.headers['content-type'] = 'application/json'
105
# malformed request body
106
req.body = jsonutils.dumps({'os-reset_status': {'status':
108
# attach admin context to request
109
req.environ['cinder.context'] = ctx
110
resp = req.get_response(app())
112
self.assertEquals(resp.status_int, 404)
113
self.assertRaises(exception.NotFound, db.volume_get, ctx,
116
def test_snapshot_reset_status(self):
118
ctx = context.RequestContext('admin', 'fake', True)
119
# snapshot in 'error_deleting'
120
volume = db.volume_create(ctx, {})
121
snapshot = db.snapshot_create(ctx, {'status': 'error_deleting',
122
'volume_id': volume['id']})
123
req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
126
req.headers['content-type'] = 'application/json'
127
# request status of 'error'
128
req.body = jsonutils.dumps({'os-reset_status': {'status': 'error'}})
129
# attach admin context to request
130
req.environ['cinder.context'] = ctx
131
resp = req.get_response(app())
132
# request is accepted
133
self.assertEquals(resp.status_int, 202)
134
snapshot = db.snapshot_get(ctx, snapshot['id'])
135
# status changed to 'error'
136
self.assertEquals(snapshot['status'], 'error')
138
def test_invalid_status_for_snapshot(self):
140
ctx = context.RequestContext('admin', 'fake', True)
141
# snapshot in 'available'
142
volume = db.volume_create(ctx, {})
143
snapshot = db.snapshot_create(ctx, {'status': 'available',
144
'volume_id': volume['id']})
145
req = webob.Request.blank('/v1/fake/snapshots/%s/action' %
148
req.headers['content-type'] = 'application/json'
149
# 'attaching' is not a valid status for snapshots
150
req.body = jsonutils.dumps({'os-reset_status': {'status':
152
# attach admin context to request
153
req.environ['cinder.context'] = ctx
154
resp = req.get_response(app())
155
# request is accepted
156
self.assertEquals(resp.status_int, 400)
157
snapshot = db.snapshot_get(ctx, snapshot['id'])
158
# status is still 'available'
159
self.assertEquals(snapshot['status'], 'available')
161
def test_force_delete(self):
163
ctx = context.RequestContext('admin', 'fake', True)
164
# current status is creating
165
volume = db.volume_create(ctx, {'status': 'creating'})
166
req = webob.Request.blank('/v1/fake/volumes/%s/action' % volume['id'])
168
req.headers['content-type'] = 'application/json'
169
req.body = jsonutils.dumps({'os-force_delete': {}})
170
# attach admin context to request
171
req.environ['cinder.context'] = ctx
172
resp = req.get_response(app())
173
# request is accepted
174
self.assertEquals(resp.status_int, 202)
176
self.assertRaises(exception.NotFound, db.volume_get, ctx, volume['id'])