Coverage for manila/tests/api/v2/test_share_snapshots.py: 99%

1# Copyright 2015 EMC Corporation 

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 ast 

19import ddt 

20from oslo_serialization import jsonutils 

21import webob 

22 

23from manila.api.openstack import api_version_request as api_version 

24from manila.api.v2 import share_snapshots 

25from manila.common import constants 

26from manila import context 

27from manila import db 

28from manila import exception 

29from manila import policy 

30from manila.share import api as share_api 

31from manila import test 

32from manila.tests.api.contrib import stubs 

33from manila.tests.api import fakes 

34from manila.tests import db_utils 

35from manila.tests import fake_share 

36from manila import utils 

37 

38MIN_MANAGE_SNAPSHOT_API_VERSION = '2.12' 

39 

40 

41def get_fake_manage_body(share_id=None, provider_location=None, 

42 driver_options=None, **kwargs): 

43 fake_snapshot = { 

44 'share_id': share_id, 

45 'provider_location': provider_location, 

46 'driver_options': driver_options, 

47 'user_id': 'fake_user_id', 

48 'project_id': 'fake_project_id', 

49 } 

50 fake_snapshot.update(kwargs) 

51 return {'snapshot': fake_snapshot} 

52 

53 

54@ddt.ddt 

55class ShareSnapshotAPITest(test.TestCase): 

56 """Share Snapshot API Test.""" 

57 

58 def setUp(self): 

59 super(ShareSnapshotAPITest, self).setUp() 

60 self.controller = share_snapshots.ShareSnapshotsController() 

61 

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

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

64 stubs.stub_snapshot_get_all_by_project) 

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

66 stubs.stub_snapshot_get) 

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

68 stubs.stub_snapshot_update) 

69 self.mock_object( 

70 policy, 'check_policy', mock.Mock(return_value=True) 

71 ) 

72 self.snp_example = { 

73 'share_id': 100, 

74 'size': 12, 

75 'force': False, 

76 'display_name': 'updated_snapshot_name', 

77 'display_description': 'updated_snapshot_description', 

78 } 

79 

80 @ddt.data('1.0', '2.16', '2.17') 

81 def test_snapshot_create(self, version): 

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

83 stubs.stub_snapshot_create) 

84 

85 body = { 

86 'snapshot': { 

87 'share_id': 'fakeshareid', 

88 'force': False, 

89 'name': 'displaysnapname', 

90 'description': 'displaysnapdesc', 

91 } 

92 } 

93 url = ('/v2/fake/snapshots' 

94 if version.startswith('2.') 

95 else '/v1/fake/snapshots') 

96 req = fakes.HTTPRequest.blank(url, version=version) 

97 

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

99 

100 expected = fake_share.expected_snapshot(version=version, id=200) 

101 

102 self.assertEqual(expected, res_dict) 

103 

104 @ddt.data(0, False) 

105 def test_snapshot_create_no_support(self, snapshot_support): 

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

107 self.mock_object( 

108 share_api.API, 

109 'get', 

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

111 body = { 

112 'snapshot': { 

113 'share_id': 100, 

114 'force': False, 

115 'name': 'fake_share_name', 

116 'description': 'fake_share_description', 

117 } 

118 } 

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

120 

121 self.assertRaises( 

122 webob.exc.HTTPUnprocessableEntity, 

123 self.controller.create, req, body) 

124 

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

126 

127 def test_snapshot_create_no_body(self): 

128 body = {} 

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

130 self.assertRaises(webob.exc.HTTPUnprocessableEntity, 

131 self.controller.create, 

132 req, 

133 body) 

134 

135 def test_snapshot_delete(self): 

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

137 stubs.stub_snapshot_delete) 

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

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

140 self.assertEqual(202, resp.status_int) 

141 

142 def test_snapshot_delete_nofound(self): 

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

144 stubs.stub_snapshot_get_notfound) 

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

146 self.assertRaises(webob.exc.HTTPNotFound, 

147 self.controller.delete, 

148 req, 

149 200) 

150 

151 @ddt.data('2.0', '2.16', '2.17', '2.73') 

152 def test_snapshot_show(self, version): 

153 req = fakes.HTTPRequest.blank('/v2/fake/snapshots/200', 

154 version=version) 

155 expected = fake_share.expected_snapshot(version=version, id=200) 

156 

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

158 

159 self.assertEqual(expected, res_dict) 

160 

161 def test_snapshot_show_nofound(self): 

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

163 stubs.stub_snapshot_get_notfound) 

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

165 self.assertRaises(webob.exc.HTTPNotFound, 

166 self.controller.show, 

167 req, '200') 

168 

169 def test_snapshot_list_summary(self): 

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

171 stubs.stub_snapshot_get_all_by_project) 

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

173 res_dict = self.controller.index(req) 

174 expected = { 

175 'snapshots': [ 

176 { 

177 'name': 'displaysnapname', 

178 'id': 2, 

179 'links': [ 

180 { 

181 'href': 'http://localhost/share/v2/fake/' 

182 'snapshots/2', 

183 'rel': 'self' 

184 }, 

185 { 

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

187 'rel': 'bookmark' 

188 } 

189 ], 

190 } 

191 ] 

192 } 

193 self.assertEqual(expected, res_dict) 

194 

195 def _snapshot_list_summary_with_search_opts(self, version, 

196 use_admin_context): 

197 search_opts = fake_share.search_opts() 

198 if (api_version.APIVersionRequest(version) >= 

199 api_version.APIVersionRequest('2.36')): 

200 search_opts.pop('name') 

201 search_opts['display_name~'] = 'fake_name' 

202 if (api_version.APIVersionRequest(version) >= 

203 api_version.APIVersionRequest('2.79')): 

204 search_opts.update({'with_count': 'true'}) 

205 

206 # fake_key should be filtered for non-admin 

207 url = '/v2/fake/snapshots?fake_key=fake_value' 

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

209 url = url + '&' + k + '=' + v 

210 req = fakes.HTTPRequest.blank( 

211 url, use_admin_context=use_admin_context, version=version) 

212 

213 method = 'get_all_snapshots' 

214 db_snapshots = [ 

215 {'id': 'id1', 'display_name': 'n1', 'status': 'fake_status', }, 

216 {'id': 'id2', 'display_name': 'n2', 'status': 'fake_status', }, 

217 {'id': 'id3', 'display_name': 'n3', 'status': 'fake_status', }, 

218 ] 

219 

220 mock_action = {'return_value': [db_snapshots[1]]} 

221 if (api_version.APIVersionRequest(version) >= 

222 api_version.APIVersionRequest('2.79')): 

223 method = 'get_all_snapshots_with_count' 

224 mock_action = {'side_effect': [(1, [db_snapshots[1]])]} 

225 

226 mock_get_all_snapshots = ( 

227 self.mock_object(share_api.API, method, mock.Mock(**mock_action))) 

228 

229 result = self.controller.index(req) 

230 

231 search_opts_expected = { 

232 'status': search_opts['status'], 

233 'share_id': search_opts['share_id'], 

234 } 

235 if (api_version.APIVersionRequest(version) >= 

236 api_version.APIVersionRequest('2.36')): 

237 search_opts_expected['display_name~'] = 'fake_name' 

238 else: 

239 search_opts_expected['display_name'] = search_opts['name'] 

240 if use_admin_context: 

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

242 

243 mock_get_all_snapshots.assert_called_once_with( 

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

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

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

247 sort_key=search_opts['sort_key'], 

248 sort_dir=search_opts['sort_dir'], 

249 search_opts=search_opts_expected, 

250 ) 

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

252 self.assertEqual(db_snapshots[1]['id'], result['snapshots'][0]['id']) 

253 self.assertEqual( 

254 db_snapshots[1]['display_name'], result['snapshots'][0]['name']) 

255 if (api_version.APIVersionRequest(version) >= 

256 api_version.APIVersionRequest('2.79')): 

257 self.assertEqual(1, result['count']) 

258 

259 @ddt.data({'version': '2.35', 'use_admin_context': True}, 

260 {'version': '2.36', 'use_admin_context': True}, 

261 {'version': '2.79', 'use_admin_context': True}, 

262 {'version': '2.35', 'use_admin_context': False}, 

263 {'version': '2.36', 'use_admin_context': False}, 

264 {'version': '2.79', 'use_admin_context': False}) 

265 @ddt.unpack 

266 def test_snapshot_list_summary_with_search_opts(self, version, 

267 use_admin_context): 

268 self._snapshot_list_summary_with_search_opts( 

269 version=version, use_admin_context=use_admin_context) 

270 

271 def test_snapshot_list_metadata_filter(self, version='2.73', 

272 use_admin_context=True): 

273 search_opts = { 

274 'sort_key': 'fake_sort_key', 

275 'sort_dir': 'fake_sort_dir', 

276 'offset': '1', 

277 'limit': '1', 

278 'metadata': "{'foo': 'bar'}" 

279 } 

280 # fake_key should be filtered for non-admin 

281 url = '/v2/fake/snapshots?fake_key=fake_value' 

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

283 url = url + '&' + k + '=' + v 

284 req = fakes.HTTPRequest.blank( 

285 url, use_admin_context=use_admin_context, version=version) 

286 

287 snapshots = [ 

288 {'id': 'id1', 'metadata': {'foo': 'bar'}} 

289 ] 

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

291 mock.Mock(return_value=snapshots)) 

292 

293 result = self.controller.index(req) 

294 

295 search_opts_expected = { 

296 'metadata': ast.literal_eval(search_opts['metadata']) 

297 } 

298 if use_admin_context: 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true

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

300 share_api.API.get_all_snapshots.assert_called_once_with( 

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

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

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

304 sort_key=search_opts['sort_key'], 

305 sort_dir=search_opts['sort_dir'], 

306 search_opts=search_opts_expected, 

307 ) 

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

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

310 

311 def _snapshot_list_detail_with_search_opts(self, version, 

312 use_admin_context): 

313 search_opts = fake_share.search_opts() 

314 if (api_version.APIVersionRequest(version) >= 

315 api_version.APIVersionRequest('2.79')): 

316 search_opts.update({'with_count': 'true'}) 

317 

318 # fake_key should be filtered for non-admin 

319 url = '/v2/fake/shares/detail?fake_key=fake_value' 

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

321 url = url + '&' + k + '=' + v 

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

323 

324 method = 'get_all_snapshots' 

325 db_snapshots = [ 

326 { 

327 'id': 'id1', 

328 'display_name': 'n1', 

329 'status': 'fake_status', 

330 'aggregate_status': 'fake_status', 

331 }, 

332 { 

333 'id': 'id2', 

334 'display_name': 'n2', 

335 'status': 'someotherstatus', 

336 'aggregate_status': 'fake_status', 

337 'share_id': 'fake_share_id', 

338 }, 

339 { 

340 'id': 'id3', 

341 'display_name': 'n3', 

342 'status': 'fake_status', 

343 'aggregate_status': 'fake_status', 

344 }, 

345 ] 

346 mock_action = {'return_value': [db_snapshots[1]]} 

347 if (api_version.APIVersionRequest(version) >= 

348 api_version.APIVersionRequest('2.79')): 

349 method = 'get_all_snapshots_with_count' 

350 mock_action = {'side_effect': [(1, [db_snapshots[1]])]} 

351 

352 mock_get_all_snapshots = ( 

353 self.mock_object(share_api.API, method, mock.Mock(**mock_action))) 

354 

355 result = self.controller.detail(req) 

356 

357 search_opts_expected = { 

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

359 'status': search_opts['status'], 

360 'share_id': search_opts['share_id'], 

361 } 

362 if use_admin_context: 

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

364 mock_get_all_snapshots.assert_called_once_with( 

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

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

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

368 sort_key=search_opts['sort_key'], 

369 sort_dir=search_opts['sort_dir'], 

370 search_opts=search_opts_expected, 

371 ) 

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

373 self.assertEqual(db_snapshots[1]['id'], result['snapshots'][0]['id']) 

374 self.assertEqual( 

375 db_snapshots[1]['display_name'], result['snapshots'][0]['name']) 

376 self.assertEqual( 

377 db_snapshots[1]['aggregate_status'], 

378 result['snapshots'][0]['status']) 

379 self.assertEqual( 

380 db_snapshots[1]['share_id'], result['snapshots'][0]['share_id']) 

381 if (api_version.APIVersionRequest(version) >= 

382 api_version.APIVersionRequest('2.79')): 

383 self.assertEqual(1, result['count']) 

384 

385 @ddt.data({'version': '2.78', 'use_admin_context': True}, 

386 {'version': '2.78', 'use_admin_context': False}, 

387 {'version': '2.79', 'use_admin_context': True}, 

388 {'version': '2.79', 'use_admin_context': False}) 

389 @ddt.unpack 

390 def test_snapshot_list_detail_with_search_opts(self, version, 

391 use_admin_context): 

392 self._snapshot_list_detail_with_search_opts( 

393 version=version, use_admin_context=use_admin_context) 

394 

395 @ddt.data('2.0', '2.16', '2.17') 

396 def test_snapshot_list_detail(self, version): 

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

398 req = fakes.HTTPRequest.blank('/v2/fake/snapshots/detail', 

399 environ=env, 

400 version=version) 

401 expected_s = fake_share.expected_snapshot(version=version, id=2) 

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

403 

404 res_dict = self.controller.detail(req) 

405 

406 self.assertEqual(expected, res_dict) 

407 

408 @ddt.data('2.0', '2.16', '2.17') 

409 def test_snapshot_updates_display_name_and_description(self, version): 

410 snp = self.snp_example 

411 body = {"snapshot": snp} 

412 req = fakes.HTTPRequest.blank('/v2/fake/snapshot/1', version=version) 

413 

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

415 

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

417 

418 if (api_version.APIVersionRequest(version) <= 

419 api_version.APIVersionRequest('2.16')): 

420 self.assertNotIn('user_id', res_dict['snapshot']) 

421 self.assertNotIn('project_id', res_dict['snapshot']) 

422 else: 

423 self.assertIn('user_id', res_dict['snapshot']) 

424 self.assertIn('project_id', res_dict['snapshot']) 

425 

426 def test_share_update_invalid_key(self): 

427 snp = self.snp_example 

428 body = {"snapshot": snp} 

429 

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

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

432 

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

434 

435 def test_access_list(self): 

436 share = db_utils.create_share(mount_snapshot_support=True) 

437 snapshot = db_utils.create_snapshot( 

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

439 

440 expected = [] 

441 

442 self.mock_object(share_api.API, 'get', 

443 mock.Mock(return_value=share)) 

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

445 mock.Mock(return_value=snapshot)) 

446 self.mock_object(share_api.API, 'snapshot_access_get_all', 

447 mock.Mock(return_value=expected)) 

448 

449 id = 'fake_snap_id' 

450 req = fakes.HTTPRequest.blank('/v2/fake/snapshots/%s/action' % id, 

451 version='2.32') 

452 

453 actual = self.controller.access_list(req, id) 

454 

455 self.assertEqual(expected, actual['snapshot_access_list']) 

456 

457 @ddt.data(('1.1.1.1', '2.32'), 

458 ('1.1.1.1', '2.38'), 

459 ('1001::1001', '2.38')) 

460 @ddt.unpack 

461 def test_allow_access(self, ip_address, version): 

462 share = db_utils.create_share(mount_snapshot_support=True) 

463 snapshot = db_utils.create_snapshot( 

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

465 

466 access = { 

467 'id': 'fake_id', 

468 'access_type': 'ip', 

469 'access_to': ip_address, 

470 'state': 'new', 

471 } 

472 

473 get = self.mock_object(share_api.API, 'get', 

474 mock.Mock(return_value=share)) 

475 get_snapshot = self.mock_object(share_api.API, 'get_snapshot', 

476 mock.Mock(return_value=snapshot)) 

477 allow_access = self.mock_object(share_api.API, 'snapshot_allow_access', 

478 mock.Mock(return_value=access)) 

479 body = {'allow_access': access} 

480 req = fakes.HTTPRequest.blank( 

481 '/v2/fake/snapshots/%s/action' % snapshot['id'], version=version) 

482 

483 actual = self.controller.allow_access(req, snapshot['id'], body) 

484 

485 self.assertEqual(access, actual['snapshot_access']) 

486 get.assert_called_once_with(utils.IsAMatcher(context.RequestContext), 

487 share['id']) 

488 get_snapshot.assert_called_once_with( 

489 utils.IsAMatcher(context.RequestContext), snapshot['id']) 

490 allow_access.assert_called_once_with( 

491 utils.IsAMatcher(context.RequestContext), snapshot, 

492 access['access_type'], access['access_to']) 

493 

494 def test_allow_access_data_not_found_exception(self): 

495 share = db_utils.create_share(mount_snapshot_support=True) 

496 snapshot = db_utils.create_snapshot( 

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

498 req = fakes.HTTPRequest.blank( 

499 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

500 body = {} 

501 

502 self.assertRaises(webob.exc.HTTPBadRequest, 

503 self.controller.allow_access, req, 

504 snapshot['id'], body) 

505 

506 def test_allow_access_exists_exception(self): 

507 share = db_utils.create_share(mount_snapshot_support=True) 

508 snapshot = db_utils.create_snapshot( 

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

510 req = fakes.HTTPRequest.blank( 

511 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

512 access = { 

513 'id': 'fake_id', 

514 'access_type': 'ip', 

515 'access_to': '1.1.1.1', 

516 'state': 'new', 

517 } 

518 msg = "Share snapshot access exists." 

519 

520 get = self.mock_object(share_api.API, 'get', mock.Mock( 

521 return_value=share)) 

522 get_snapshot = self.mock_object(share_api.API, 'get_snapshot', 

523 mock.Mock(return_value=snapshot)) 

524 allow_access = self.mock_object( 

525 share_api.API, 'snapshot_allow_access', mock.Mock( 

526 side_effect=exception.ShareSnapshotAccessExists(msg))) 

527 

528 body = {'allow_access': access} 

529 

530 self.assertRaises(webob.exc.HTTPBadRequest, 

531 self.controller.allow_access, req, 

532 snapshot['id'], body) 

533 

534 get.assert_called_once_with(utils.IsAMatcher(context.RequestContext), 

535 share['id']) 

536 get_snapshot.assert_called_once_with( 

537 utils.IsAMatcher(context.RequestContext), snapshot['id']) 

538 allow_access.assert_called_once_with( 

539 utils.IsAMatcher(context.RequestContext), snapshot, 

540 access['access_type'], access['access_to']) 

541 

542 def test_allow_access_share_without_mount_snap_support(self): 

543 share = db_utils.create_share(mount_snapshot_support=False) 

544 snapshot = db_utils.create_snapshot( 

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

546 

547 access = { 

548 'id': 'fake_id', 

549 'access_type': 'ip', 

550 'access_to': '1.1.1.1', 

551 'state': 'new', 

552 } 

553 

554 get_snapshot = self.mock_object(share_api.API, 'get_snapshot', 

555 mock.Mock(return_value=snapshot)) 

556 get = self.mock_object(share_api.API, 'get', 

557 mock.Mock(return_value=share)) 

558 

559 body = {'allow_access': access} 

560 req = fakes.HTTPRequest.blank( 

561 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

562 

563 self.assertRaises(webob.exc.HTTPBadRequest, 

564 self.controller.allow_access, req, 

565 snapshot['id'], body) 

566 

567 get.assert_called_once_with(utils.IsAMatcher(context.RequestContext), 

568 share['id']) 

569 get_snapshot.assert_called_once_with( 

570 utils.IsAMatcher(context.RequestContext), snapshot['id']) 

571 

572 def test_allow_access_empty_parameters(self): 

573 share = db_utils.create_share(mount_snapshot_support=True) 

574 snapshot = db_utils.create_snapshot( 

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

576 

577 access = {'id': 'fake_id', 

578 'access_type': '', 

579 'access_to': ''} 

580 

581 body = {'allow_access': access} 

582 req = fakes.HTTPRequest.blank( 

583 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

584 

585 self.assertRaises(webob.exc.HTTPBadRequest, 

586 self.controller.allow_access, req, 

587 snapshot['id'], body) 

588 

589 def test_deny_access(self): 

590 share = db_utils.create_share(mount_snapshot_support=True) 

591 snapshot = db_utils.create_snapshot( 

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

593 access = db_utils.create_snapshot_access( 

594 share_snapshot_id=snapshot['id']) 

595 

596 get = self.mock_object(share_api.API, 'get', 

597 mock.Mock(return_value=share)) 

598 get_snapshot = self.mock_object(share_api.API, 'get_snapshot', 

599 mock.Mock(return_value=snapshot)) 

600 access_get = self.mock_object(share_api.API, 'snapshot_access_get', 

601 mock.Mock(return_value=access)) 

602 deny_access = self.mock_object(share_api.API, 'snapshot_deny_access') 

603 

604 body = {'deny_access': {'access_id': access.id}} 

605 req = fakes.HTTPRequest.blank( 

606 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

607 

608 resp = self.controller.deny_access(req, snapshot['id'], body) 

609 

610 self.assertEqual(202, resp.status_int) 

611 get.assert_called_once_with(utils.IsAMatcher(context.RequestContext), 

612 share['id']) 

613 get_snapshot.assert_called_once_with( 

614 utils.IsAMatcher(context.RequestContext), snapshot['id']) 

615 access_get.assert_called_once_with( 

616 utils.IsAMatcher(context.RequestContext), 

617 body['deny_access']['access_id']) 

618 deny_access.assert_called_once_with( 

619 utils.IsAMatcher(context.RequestContext), snapshot, access) 

620 

621 def test_deny_access_data_not_found_exception(self): 

622 share = db_utils.create_share(mount_snapshot_support=True) 

623 snapshot = db_utils.create_snapshot( 

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

625 req = fakes.HTTPRequest.blank( 

626 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

627 body = {} 

628 

629 self.assertRaises(webob.exc.HTTPBadRequest, 

630 self.controller.deny_access, req, 

631 snapshot['id'], body) 

632 

633 def test_deny_access_access_rule_not_found(self): 

634 share = db_utils.create_share(mount_snapshot_support=True) 

635 snapshot = db_utils.create_snapshot( 

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

637 access = db_utils.create_snapshot_access( 

638 share_snapshot_id=snapshot['id']) 

639 wrong_access = { 

640 'access_type': 'fake_type', 

641 'access_to': 'fake_IP', 

642 'share_snapshot_id': 'fake_id' 

643 } 

644 

645 get = self.mock_object(share_api.API, 'get', 

646 mock.Mock(return_value=share)) 

647 get_snapshot = self.mock_object(share_api.API, 'get_snapshot', 

648 mock.Mock(return_value=snapshot)) 

649 access_get = self.mock_object(share_api.API, 'snapshot_access_get', 

650 mock.Mock(return_value=wrong_access)) 

651 

652 body = {'deny_access': {'access_id': access.id}} 

653 req = fakes.HTTPRequest.blank( 

654 '/v2/fake/snapshots/%s/action' % snapshot['id'], version='2.32') 

655 

656 self.assertRaises(webob.exc.HTTPBadRequest, 

657 self.controller.deny_access, req, snapshot['id'], 

658 body) 

659 get.assert_called_once_with(utils.IsAMatcher(context.RequestContext), 

660 share['id']) 

661 get_snapshot.assert_called_once_with( 

662 utils.IsAMatcher(context.RequestContext), snapshot['id']) 

663 access_get.assert_called_once_with( 

664 utils.IsAMatcher(context.RequestContext), 

665 body['deny_access']['access_id']) 

666 

667 

668@ddt.ddt 

669class ShareSnapshotAdminActionsAPITest(test.TestCase): 

670 

671 def setUp(self): 

672 super(ShareSnapshotAdminActionsAPITest, self).setUp() 

673 self.controller = share_snapshots.ShareSnapshotsController() 

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

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

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

677 

678 self.resource_name = self.controller.resource_name 

679 self.manage_request = fakes.HTTPRequest.blank( 

680 '/v2/fake/snapshots/manage', use_admin_context=True, 

681 version=MIN_MANAGE_SNAPSHOT_API_VERSION) 

682 self.snapshot_id = 'fake' 

683 self.unmanage_request = fakes.HTTPRequest.blank( 

684 '/v2/fake/snapshots/%s/unmanage' % self.snapshot_id, 

685 use_admin_context=True, 

686 version=MIN_MANAGE_SNAPSHOT_API_VERSION) 

687 

688 def _get_context(self, role): 

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

690 

691 def _setup_snapshot_data(self, snapshot=None, version='2.7'): 

692 if snapshot is None: 

693 share = db_utils.create_share() 

694 snapshot = db_utils.create_snapshot( 

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

696 path = '/v2/fake/snapshots/%s/action' % snapshot['id'] 

697 req = fakes.HTTPRequest.blank(path, script_name=path, version=version) 

698 return snapshot, req 

699 

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

701 valid_code, valid_status=None, body=None, version='2.7'): 

702 if float(version) > 2.6: 

703 action_name = 'reset_status' 

704 else: 

705 action_name = 'os-reset_status' 

706 if body is None: 

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

708 req.method = 'POST' 

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

710 req.headers['X-Openstack-Manila-Api-Version'] = version 

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

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

713 

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

715