Coverage for manila/tests/api/v1/test_share_snapshots.py: 100%

1# Copyright 2012 NetApp 

2# All Rights Reserved. 

3# 

4# Licensed under the Apache License, Version 2.0 (the "License"); you may 

5# not use this file except in compliance with the License. You may obtain 

6# a copy of the License at 

7# 

8# http://www.apache.org/licenses/LICENSE-2.0 

9# 

10# Unless required by applicable law or agreed to in writing, software 

11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

13# License for the specific language governing permissions and limitations 

14# under the License. 

15 

16from unittest import mock 

17 

18import ddt 

19from oslo_serialization import jsonutils 

20import webob 

21 

22from manila.api.v1 import share_snapshots 

23from manila.common import constants 

24from manila import context 

25from manila import db 

26from manila import exception 

27from manila.share import api as share_api 

28from manila import test 

29from manila.tests.api.contrib import stubs 

30from manila.tests.api import fakes 

31from manila.tests import db_utils 

32from manila.tests import fake_share 

33 

34 

35@ddt.ddt 

36class ShareSnapshotAPITest(test.TestCase): 

37 """Share Snapshot API Test.""" 

38 

39 def setUp(self): 

40 super(ShareSnapshotAPITest, self).setUp() 

41 self.controller = share_snapshots.ShareSnapshotsController() 

42 

43 self.mock_object(share_api.API, 'get', stubs.stub_share_get) 

44 self.mock_object(share_api.API, 'get_all_snapshots', 

45 stubs.stub_snapshot_get_all_by_project) 

46 self.mock_object(share_api.API, 'get_snapshot', 

47 stubs.stub_snapshot_get) 

48 self.mock_object(share_api.API, 'snapshot_update', 

49 stubs.stub_snapshot_update) 

50 self.snp_example = { 

51 'share_id': 100, 

52 'size': 12, 

53 'force': False, 

54 'display_name': 'updated_share_name', 

55 'display_description': 'updated_share_description', 

56 } 

57 self.maxDiff = None 

58 

59 def test_snapshot_show_status_none(self): 

60 return_snapshot = { 

61 'share_id': 100, 

62 'name': 'fake_share_name', 

63 'description': 'fake_share_description', 

64 'status': None, 

65 } 

66 self.mock_object(share_api.API, 'get_snapshot', 

67 mock.Mock(return_value=return_snapshot)) 

68 req = fakes.HTTPRequest.blank('/fake/snapshots/200') 

69 self.assertRaises(webob.exc.HTTPNotFound, 

70 self.controller.show, 

71 req, '200') 

72 

73 @ddt.data('true', 'True', '<is> True', '1') 

74 def test_snapshot_create(self, snapshot_support): 

75 self.mock_object(share_api.API, 'create_snapshot', 

76 stubs.stub_snapshot_create) 

77 body = { 

78 'snapshot': { 

79 'share_id': 'fakeshareid', 

80 'force': False, 

81 'name': 'displaysnapname', 

82 'description': 'displaysnapdesc', 

83 } 

84 } 

85 req = fakes.HTTPRequest.blank('/fake/snapshots') 

86 

87 res_dict = self.controller.create(req, body) 

88 

89 expected = fake_share.expected_snapshot(id=200) 

90 

91 self.assertEqual(expected, res_dict) 

92 

93 @ddt.data( 

94 {'name': 'name1', 'description': 'x' * 256}, 

95 {'name': 'x' * 256, 'description': 'description1'}, 

96 ) 

97 @ddt.unpack 

98 def test_snapshot_create_invalid_input(self, name, description): 

99 self.mock_object(share_api.API, 'create_snapshot') 

100 self.mock_object( 

101 share_api.API, 

102 'get', 

103 mock.Mock(return_value={'snapshot_support': True, 

104 'is_soft_deleted': False})) 

105 body = { 

106 'snapshot': { 

107 'share_id': 200, 

108 'force': False, 

109 'name': name, 

110 'description': description, 

111 } 

112 } 

113 req = fakes.HTTPRequest.blank('/fake/snapshots') 

114 

115 self.assertRaises( 

116 exception.InvalidInput, 

117 self.controller.create, req, body) 

118 

119 @ddt.data(0, False) 

120 def test_snapshot_create_no_support(self, snapshot_support): 

121 self.mock_object(share_api.API, 'create_snapshot') 

122 self.mock_object( 

123 share_api.API, 

124 'get', 

125 mock.Mock(return_value={'snapshot_support': snapshot_support})) 

126 body = { 

127 'snapshot': { 

128 'share_id': 100, 

129 'force': False, 

130 'name': 'fake_share_name', 

131 'description': 'fake_share_description', 

132 } 

133 } 

134 req = fakes.HTTPRequest.blank('/fake/snapshots') 

135 

136 self.assertRaises( 

137 webob.exc.HTTPUnprocessableEntity, 

138 self.controller.create, req, body) 

139 

140 self.assertFalse(share_api.API.create_snapshot.called) 

141 

142 def test_snapshot_create_in_recycle_bin(self): 

143 self.mock_object(share_api.API, 'create_snapshot') 

144 self.mock_object( 

145 share_api.API, 

146 'get', 

147 mock.Mock(return_value={'snapshot_support': True, 

148 'is_soft_deleted': True})) 

149 body = { 

150 'snapshot': { 

151 'share_id': 200, 

152 'force': False, 

153 'name': 'fake_share_name', 

154 'description': 'fake_share_description', 

155 } 

156 } 

157 req = fakes.HTTPRequest.blank('/fake/snapshots') 

158 

159 self.assertRaises( 

160 webob.exc.HTTPForbidden, 

161 self.controller.create, req, body) 

162 

163 self.assertFalse(share_api.API.create_snapshot.called) 

164 

165 def test_snapshot_create_no_body(self): 

166 body = {} 

167 req = fakes.HTTPRequest.blank('/fake/snapshots') 

168 self.assertRaises(webob.exc.HTTPUnprocessableEntity, 

169 self.controller.create, 

170 req, 

171 body) 

172 

173 def test_snapshot_delete(self): 

174 self.mock_object(share_api.API, 'delete_snapshot', 

175 stubs.stub_snapshot_delete) 

176 req = fakes.HTTPRequest.blank('/fake/snapshots/200') 

177 resp = self.controller.delete(req, 200) 

178 self.assertEqual(202, resp.status_int) 

179 

180 def test_snapshot_delete_nofound(self): 

181 self.mock_object(share_api.API, 'get_snapshot', 

182 stubs.stub_snapshot_get_notfound) 

183 req = fakes.HTTPRequest.blank('/fake/snapshots/200') 

184 self.assertRaises(webob.exc.HTTPNotFound, 

185 self.controller.delete, 

186 req, 

187 200) 

188 

189 def test_snapshot_show(self): 

190 req = fakes.HTTPRequest.blank('/fake/snapshots/200') 

191 res_dict = self.controller.show(req, 200) 

192 expected = fake_share.expected_snapshot(id=200) 

193 self.assertEqual(expected, res_dict) 

194 

195 def test_snapshot_show_nofound(self): 

196 self.mock_object(share_api.API, 'get_snapshot', 

197 stubs.stub_snapshot_get_notfound) 

198 req = fakes.HTTPRequest.blank('/fake/snapshots/200') 

199 self.assertRaises(webob.exc.HTTPNotFound, 

200 self.controller.show, 

201 req, '200') 

202 

203 def test_snapshot_list_summary(self): 

204 self.mock_object(share_api.API, 'get_all_snapshots', 

205 stubs.stub_snapshot_get_all_by_project) 

206 req = fakes.HTTPRequest.blank('/fake/snapshots') 

207 res_dict = self.controller.index(req) 

208 expected = { 

209 'snapshots': [ 

210 { 

211 'name': 'displaysnapname', 

212 'id': 2, 

213 'links': [ 

214 { 

215 'href': 'http://localhost/share/v1/fake/' 

216 'snapshots/2', 

217 'rel': 'self' 

218 }, 

219 { 

220 'href': 'http://localhost/share/fake/snapshots/2', 

221 'rel': 'bookmark' 

222 } 

223 ], 

224 } 

225 ] 

226 } 

227 self.assertEqual(expected, res_dict) 

228 

229 def _snapshot_list_summary_with_search_opts(self, use_admin_context): 

230 search_opts = fake_share.search_opts() 

231 # fake_key should be filtered for non-admin 

232 url = '/fake/snapshots?fake_key=fake_value' 

233 for k, v in search_opts.items(): 

234 url = url + '&' + k + '=' + v 

235 req = fakes.HTTPRequest.blank(url, use_admin_context=use_admin_context) 

236 

237 db_snapshots = [ 

238 {'id': 'id1', 'display_name': 'n1', 

239 'status': 'fake_status', 'share_id': 'fake_share_id'}, 

240 {'id': 'id2', 'display_name': 'n2', 

241 'status': 'fake_status', 'share_id': 'fake_share_id'}, 

242 {'id': 'id3', 'display_name': 'n3', 

243 'status': 'fake_status', 'share_id': 'fake_share_id'}, 

244 ] 

245 snapshots = [db_snapshots[1]] 

246 self.mock_object(share_api.API, 'get_all_snapshots', 

247 mock.Mock(return_value=snapshots)) 

248 

249 result = self.controller.index(req) 

250 

251 search_opts_expected = { 

252 'display_name': search_opts['name'], 

253 'status': search_opts['status'], 

254 'share_id': search_opts['share_id'], 

255 } 

256 if use_admin_context: 

257 search_opts_expected.update({'fake_key': 'fake_value'}) 

258 share_api.API.get_all_snapshots.assert_called_once_with( 

259 req.environ['manila.context'], 

260 limit=int(search_opts['limit']), 

261 offset=int(search_opts['offset']), 

262 sort_key=search_opts['sort_key'], 

263 sort_dir=search_opts['sort_dir'], 

264 search_opts=search_opts_expected, 

265 ) 

266 self.assertEqual(1, len(result['snapshots'])) 

267 self.assertEqual(snapshots[0]['id'], result['snapshots'][0]['id']) 

268 self.assertEqual( 

269 snapshots[0]['display_name'], result['snapshots'][0]['name']) 

270 

271 def test_snapshot_list_summary_with_search_opts_by_non_admin(self): 

272 self._snapshot_list_summary_with_search_opts(use_admin_context=False) 

273 

274 def test_snapshot_list_summary_with_search_opts_by_admin(self): 

275 self._snapshot_list_summary_with_search_opts(use_admin_context=True) 

276 

277 def _snapshot_list_detail_with_search_opts(self, use_admin_context): 

278 search_opts = fake_share.search_opts() 

279 # fake_key should be filtered for non-admin 

280 url = '/fake/shares/detail?fake_key=fake_value' 

281 for k, v in search_opts.items(): 

282 url = url + '&' + k + '=' + v 

283 req = fakes.HTTPRequest.blank(url, use_admin_context=use_admin_context) 

284 

285 db_snapshots = [ 

286 { 

287 'id': 'id1', 

288 'display_name': 'n1', 

289 'status': 'fake_status_other', 

290 'aggregate_status': 'fake_status', 

291 'share_id': 'fake_share_id', 

292 }, 

293 { 

294 'id': 'id2', 

295 'display_name': 'n2', 

296 'status': 'fake_status', 

297 'aggregate_status': 'fake_status', 

298 'share_id': 'fake_share_id', 

299 }, 

300 { 

301 'id': 'id3', 

302 'display_name': 'n3', 

303 'status': 'fake_status_other', 

304 'aggregate_status': 'fake_status', 

305 'share_id': 'fake_share_id', 

306 }, 

307 ] 

308 snapshots = [db_snapshots[1]] 

309 

310 self.mock_object(share_api.API, 'get_all_snapshots', 

311 mock.Mock(return_value=snapshots)) 

312 

313 result = self.controller.detail(req) 

314 

315 search_opts_expected = { 

316 'display_name': search_opts['name'], 

317 'status': search_opts['status'], 

318 'share_id': search_opts['share_id'], 

319 } 

320 if use_admin_context: 

321 search_opts_expected.update({'fake_key': 'fake_value'}) 

322 share_api.API.get_all_snapshots.assert_called_once_with( 

323 req.environ['manila.context'], 

324 limit=int(search_opts['limit']), 

325 offset=int(search_opts['offset']), 

326 sort_key=search_opts['sort_key'], 

327 sort_dir=search_opts['sort_dir'], 

328 search_opts=search_opts_expected, 

329 ) 

330 self.assertEqual(1, len(result['snapshots'])) 

331 self.assertEqual(snapshots[0]['id'], result['snapshots'][0]['id']) 

332 self.assertEqual( 

333 snapshots[0]['display_name'], result['snapshots'][0]['name']) 

334 self.assertEqual( 

335 snapshots[0]['status'], result['snapshots'][0]['status']) 

336 self.assertEqual( 

337 snapshots[0]['share_id'], result['snapshots'][0]['share_id']) 

338 

339 def test_snapshot_list_detail_with_search_opts_by_non_admin(self): 

340 self._snapshot_list_detail_with_search_opts(use_admin_context=False) 

341 

342 def test_snapshot_list_detail_with_search_opts_by_admin(self): 

343 self._snapshot_list_detail_with_search_opts(use_admin_context=True) 

344 

345 def test_snapshot_list_detail(self): 

346 env = {'QUERY_STRING': 'name=Share+Test+Name'} 

347 req = fakes.HTTPRequest.blank('/fake/shares/detail', environ=env) 

348 res_dict = self.controller.detail(req) 

349 expected_s = fake_share.expected_snapshot(id=2) 

350 expected = {'snapshots': [expected_s['snapshot']]} 

351 self.assertEqual(expected, res_dict) 

352 

353 def test_snapshot_list_status_none(self): 

354 snapshots = [ 

355 { 

356 'id': 3, 

357 'share_id': 'fakeshareid', 

358 'size': 1, 

359 'status': None, 

360 'name': 'displaysnapname', 

361 'description': 'displaysnapdesc', 

362 } 

363 ] 

364 self.mock_object(share_api.API, 'get_all_snapshots', 

365 mock.Mock(return_value=snapshots)) 

366 req = fakes.HTTPRequest.blank('/fake/snapshots') 

367 result = self.controller.index(req) 

368 self.assertEqual(1, len(result['snapshots'])) 

369 self.assertEqual(snapshots[0]['id'], result['snapshots'][0]['id']) 

370 

371 def test_snapshot_updates_description(self): 

372 snp = self.snp_example 

373 body = {"snapshot": snp} 

374 

375 req = fakes.HTTPRequest.blank('/fake/snapshot/1') 

376 res_dict = self.controller.update(req, 1, body) 

377 self.assertEqual(snp["display_name"], res_dict['snapshot']["name"]) 

378 

379 def test_snapshot_updates_display_descr(self): 

380 snp = self.snp_example 

381 body = {"snapshot": snp} 

382 

383 req = fakes.HTTPRequest.blank('/fake/snapshot/1') 

384 res_dict = self.controller.update(req, 1, body) 

385 

386 self.assertEqual(snp["display_description"], 

387 res_dict['snapshot']["description"]) 

388 

389 def test_share_not_updates_size(self): 

390 snp = self.snp_example 

391 body = {"snapshot": snp} 

392 

393 req = fakes.HTTPRequest.blank('/fake/snapshot/1') 

394 res_dict = self.controller.update(req, 1, body) 

395 

396 self.assertNotEqual(snp["size"], res_dict['snapshot']["size"]) 

397 

398 

399@ddt.ddt 

400class ShareSnapshotAdminActionsAPITest(test.TestCase): 

401 

402 def setUp(self): 

403 super(ShareSnapshotAdminActionsAPITest, self).setUp() 

404 self.controller = share_snapshots.ShareSnapshotsController() 

405 self.flags(transport_url='rabbit://fake:fake@mqhost:5672') 

406 self.admin_context = context.RequestContext('admin', 'fake', True) 

407 self.member_context = context.RequestContext('fake', 'fake') 

408 

409 def _get_context(self, role): 

410 return getattr(self, '%s_context' % role) 

411 

412 def _setup_snapshot_data(self, snapshot=None): 

413 if snapshot is None: 

414 share = db_utils.create_share() 

415 snapshot = db_utils.create_snapshot( 

416 status=constants.STATUS_AVAILABLE, share_id=share['id']) 

417 req = fakes.HTTPRequest.blank('/v1/fake/snapshots/%s/action' % 

418 snapshot['id']) 

419 return snapshot, req 

420 

421 def _reset_status(self, ctxt, model, req, db_access_method, 

422 valid_code, valid_status=None, body=None): 

423 action_name = 'os-reset_status' 

424 if body is None: 

425 body = {action_name: {'status': constants.STATUS_ERROR}} 

426 req.method = 'POST' 

427 req.headers['content-type'] = 'application/json' 

428 req.body = jsonutils.dumps(body).encode("utf-8") 

429 req.environ['manila.context'] = ctxt 

430 

431 resp = req.get_response(fakes.app()) 

432 

433 # validate response code and model status 

434 self.assertEqual(valid_code, resp.status_int) 

435 

436 actual_model = db_access_method(ctxt, model['id']) 

437 self.assertEqual(valid_status, actual_model['status']) 

438 

439 @ddt.data(*fakes.fixture_reset_status_with_different_roles_v1) 

440 @ddt.unpack 

441 def test_snapshot_reset_status_with_different_roles(self, role, valid_code, 

442 valid_status): 

443 ctxt = self._get_context(role) 

444 snapshot, req = self._setup_snapshot_data() 

445 

446 self._reset_status(ctxt, snapshot, req, db.share_snapshot_get, 

447 valid_code, valid_status) 

448 

449 @ddt.data( 

450 {'os-reset_status': {'x-status': 'bad'}}, 

451 {'os-reset_status': {'status': 'invalid'}}, 

452 ) 

453 def test_snapshot_invalid_reset_status_body(self, body): 

454 snapshot, req = self._setup_snapshot_data() 

455 

456 self._reset_status(self.admin_context, snapshot, req, 

457 db.share_snapshot_get, 400, 

458 constants.STATUS_AVAILABLE, body) 

459 

460 def _force_delete(self, ctxt, model, req, db_access_method, valid_code): 

461 action_name = 'os-force_delete' 

462 req.method = 'POST' 

463 req.headers['content-type'] = 'application/json' 

464 req.body = jsonutils.dumps({action_name: {}}).encode("utf-8") 

465 req.environ['manila.context'] = ctxt 

466 

467 resp = req.get_response(fakes.app()) 

468 

469 # Validate response 

470 self.assertEqual(valid_code, resp.status_int) 

471 

472 @ddt.data( 

473 {'role': 'admin', 'resp_code': 202}, 

474 {'role': 'member', 'resp_code': 403}, 

475 ) 

476 @ddt.unpack 

477 def test_snapshot_force_delete_with_different_roles(self, role, resp_code): 

478 ctxt = self._get_context(role) 

479 snapshot, req = self._setup_snapshot_data() 

480 

481 self._force_delete(ctxt, snapshot, req, db.share_snapshot_get, 

482 resp_code) 

483 

484 def test_snapshot_force_delete_missing(self): 

485 ctxt = self._get_context('admin') 

486 snapshot, req = self._setup_snapshot_data(snapshot={'id': 'fake'}) 

487 

488 self._force_delete(ctxt, snapshot, req, db.share_snapshot_get, 404)