Coverage for manila/tests/share/drivers/dell_emc/plugins/powerscale/test_powerscale_api.py: 99%

1# Copyright (c) 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 ddt 

19from oslo_serialization import jsonutils as json 

20import requests 

21import requests_mock 

22from urllib.parse import quote 

23 

24from manila import exception 

25from manila.share.drivers.dell_emc.plugins.powerscale import powerscale_api 

26from manila import test 

27 

28 

29@ddt.ddt 

30class PowerScaleApiTest(test.TestCase): 

31 

32 @mock.patch('manila.share.drivers.dell_emc.plugins.powerscale.' 

33 'powerscale_api.PowerScaleApi.create_session') 

34 def setUp(self, mockup_create_session): 

35 super(PowerScaleApiTest, self).setUp() 

36 

37 mockup_create_session.return_value = True 

38 self._mock_url = 'https://localhost:8080' 

39 self.username = 'admin' 

40 self.password = 'pwd' 

41 self.dir_permission = '0777' 

42 self.powerscale_api = powerscale_api.PowerScaleApi( 

43 self._mock_url, self.username, self.password, 

44 dir_permission=self.dir_permission 

45 ) 

46 self.powerscale_api_threshold = powerscale_api.PowerScaleApi( 

47 self._mock_url, self.username, self.password, 

48 dir_permission=self.dir_permission, 

49 threshold_limit=80 

50 ) 

51 

52 @mock.patch('manila.share.drivers.dell_emc.plugins.powerscale.' 

53 'powerscale_api.PowerScaleApi.create_session') 

54 def test__init__login_failure(self, mockup_create_session): 

55 mockup_create_session.return_value = False 

56 self.assertRaises( 

57 exception.BadConfigurationException, 

58 self.powerscale_api.__init__, 

59 self._mock_url, 

60 self.username, 

61 self.password, 

62 False, 

63 None, 

64 self.dir_permission 

65 ) 

66 

67 def test__verify_cert(self): 

68 verify_cert = self.powerscale_api.verify_ssl_cert 

69 certificate_path = self.powerscale_api.certificate_path 

70 self.powerscale_api.verify_ssl_cert = True 

71 self.powerscale_api.certificate_path = "fake_certificate_path" 

72 self.assertEqual(self.powerscale_api._verify_cert, 

73 self.powerscale_api.certificate_path) 

74 self.powerscale_api.verify_ssl_cert = verify_cert 

75 self.powerscale_api.certificate_path = certificate_path 

76 

77 @mock.patch('requests.Session.request') 

78 def test_create_session_success(self, mock_request): 

79 mock_response = mock.Mock() 

80 mock_response.status_code = 201 

81 mock_response.cookies = {'isisessid': 'test_session_token', 

82 'isicsrf': 'test_csrf_token'} 

83 mock_request.return_value = mock_response 

84 result = self.powerscale_api.create_session( 

85 self.username, self.password) 

86 mock_request.assert_called_once_with( 

87 'POST', self._mock_url + '/session/1/session', 

88 headers={"Content-type": "application/json"}, 

89 data=json.dumps({"username": self.username, 

90 "password": self.password, 

91 "services": ["platform", "namespace"]}), 

92 verify=False 

93 ) 

94 self.assertTrue(result) 

95 self.assertEqual(self.powerscale_api.session_token, 

96 'test_session_token') 

97 self.assertEqual(self.powerscale_api.csrf_token, 'test_csrf_token') 

98 

99 @mock.patch('requests.Session.request') 

100 def test_create_session_failure(self, mock_request): 

101 mock_response = mock.Mock() 

102 mock_response.status_code = 401 

103 mock_response.json.return_value = { 

104 'message': 'Username or password is incorrect.'} 

105 mock_request.return_value = mock_response 

106 result = self.powerscale_api.create_session( 

107 self.username, self.password) 

108 self.assertFalse(result) 

109 self.assertIsNone(self.powerscale_api.session_token) 

110 self.assertIsNone(self.powerscale_api.csrf_token) 

111 

112 @ddt.data(False, True) 

113 def test_create_directory(self, is_recursive): 

114 with requests_mock.Mocker() as m: 

115 path = '/ifs/test' 

116 self.assertEqual(0, len(m.request_history)) 

117 self._add_create_directory_response(m, path, is_recursive) 

118 

119 r = self.powerscale_api.create_directory(path, 

120 recursive=is_recursive) 

121 

122 self.assertTrue(r) 

123 self.assertEqual(1, len(m.request_history)) 

124 request = m.request_history[0] 

125 self._verify_dir_creation_request(request, path, is_recursive) 

126 

127 def test_create_directory_no_permission(self): 

128 with requests_mock.Mocker() as m: 

129 path = '/ifs/test' 

130 self.powerscale_api.dir_permission = None 

131 self.assertEqual(0, len(m.request_history)) 

132 self._add_create_directory_response(m, path, True) 

133 

134 r = self.powerscale_api.create_directory(path, 

135 recursive=True) 

136 

137 self.powerscale_api.dir_permission = '0777' 

138 self.assertTrue(r) 

139 self.assertEqual(1, len(m.request_history)) 

140 request = m.request_history[0] 

141 self.assertNotIn("x-isi-ifs-access-control", request.headers) 

142 

143 @requests_mock.mock() 

144 def test_clone_snapshot(self, m): 

145 snapshot_name = 'snapshot01' 

146 fq_target_dir = '/ifs/admin/target' 

147 

148 self.assertEqual(0, len(m.request_history)) 

149 self._add_create_directory_response(m, fq_target_dir, False) 

150 snapshots_json = ( 

151 '{"snapshots": ' 

152 '[{"name": "snapshot01", "path": "/ifs/admin/source"}]' 

153 '}' 

154 ) 

155 self._add_get_snapshot_response(m, snapshot_name, snapshots_json) 

156 

157 # In order to test cloning a snapshot, we build out a mock 

158 # source directory tree. After the method under test is called we 

159 # will verify the necessary calls are made to clone a snapshot. 

160 source_dir_listing_json = ( 

161 '{"children": [' 

162 '{"name": "dir1", "type": "container"},' 

163 '{"name": "dir2", "type": "container"},' 

164 '{"name": "file1", "type": "object"},' 

165 '{"name": "file2", "type": "object"}' 

166 ']}' 

167 ) 

168 self._add_get_directory_listing_response( 

169 m, '/ifs/.snapshot/{0}/admin/source'.format(snapshot_name), 

170 source_dir_listing_json) 

171 

172 # Add request responses for creating directories and cloning files 

173 # to the destination tree 

174 self._add_file_clone_response(m, '/ifs/admin/target/file1', 

175 snapshot_name) 

176 self._add_file_clone_response(m, '/ifs/admin/target/file2', 

177 snapshot_name) 

178 self._add_create_directory_response(m, fq_target_dir + '/dir1', False) 

179 self._add_get_directory_listing_response( 

180 m, '/ifs/.snapshot/{0}/admin/source/dir1'.format(snapshot_name), 

181 '{"children": [' 

182 '{"name": "file11", "type": "object"}, ' 

183 '{"name": "file12", "type": "object"}' 

184 ']}') 

185 self._add_file_clone_response(m, '/ifs/admin/target/dir1/file11', 

186 snapshot_name) 

187 self._add_file_clone_response(m, '/ifs/admin/target/dir1/file12', 

188 snapshot_name) 

189 self._add_create_directory_response(m, fq_target_dir + '/dir2', False) 

190 self._add_get_directory_listing_response( 

191 m, '/ifs/.snapshot/{0}/admin/source/dir2'.format(snapshot_name), 

192 '{"children": [' 

193 '{"name": "file21", "type": "object"}, ' 

194 '{"name": "file22", "type": "object"}' 

195 ']}') 

196 self._add_file_clone_response(m, '/ifs/admin/target/dir2/file21', 

197 snapshot_name) 

198 self._add_file_clone_response(m, '/ifs/admin/target/dir2/file22', 

199 snapshot_name) 

200 

201 # Call method under test 

202 self.powerscale_api.clone_snapshot(snapshot_name, fq_target_dir, None) 

203 

204 # Verify calls needed to clone the source snapshot to the target dir 

205 expected_calls = [] 

206 clone_path_list = [ 

207 'file1', 'file2', 'dir1/file11', 'dir1/file12', 

208 'dir2/file21', 'dir2/file22'] 

209 for path in clone_path_list: 

210 expected_call = PowerScaleApiTest.ExpectedCall( 

211 PowerScaleApiTest.ExpectedCall.FILE_CLONE, 

212 self._mock_url + '/namespace/ifs/admin/target/' + path, 

213 ['/ifs/admin/target/' + path, '/ifs/admin/source/' + path, 

214 snapshot_name]) 

215 expected_calls.append(expected_call) 

216 dir_path_list = [ 

217 ('/dir1?recursive', '/dir1'), 

218 ('/dir2?recursive', '/dir2'), 

219 ('?recursive=', '')] 

220 for url, path in dir_path_list: 

221 expected_call = PowerScaleApiTest.ExpectedCall( 

222 PowerScaleApiTest.ExpectedCall.DIR_CREATION, 

223 self._mock_url + '/namespace/ifs/admin/target' + url, 

224 ['/ifs/admin/target' + path, False]) 

225 expected_calls.append(expected_call) 

226 

227 self._verify_clone_snapshot_calls(expected_calls, m.request_history) 

228 

229 class ExpectedCall(object): 

230 DIR_CREATION = 'dir_creation' 

231 FILE_CLONE = 'file_clone' 

232 

233 def __init__(self, request_type, match_url, verify_args): 

234 self.request_type = request_type 

235 self.match_url = match_url 

236 self.verify_args = verify_args 

237 

238 def _verify_clone_snapshot_calls(self, expected_calls, response_calls): 

239 actual_calls = [] 

240 for call in response_calls: 

241 actual_calls.append(call) 

242 for expected_call in expected_calls: 

243 # Match the expected call to the actual call, then verify 

244 match_found = False 

245 for call in actual_calls: 

246 if call.url.startswith(expected_call.match_url): 

247 match_found = True 

248 if expected_call.request_type == 'dir_creation': 

249 self._verify_dir_creation_request( 

250 call, *expected_call.verify_args) 

251 elif expected_call.request_type == 'file_clone': 251 ↛ 254line 251 didn't jump to line 254 because the condition on line 251 was always true

252 pass 

253 else: 

254 self.fail('Invalid request type') 

255 actual_calls.remove(call) 

256 self.assertTrue(match_found) 

257 

258 @requests_mock.mock() 

259 def test_get_directory_listing(self, m): 

260 self.assertEqual(0, len(m.request_history)) 

261 fq_dir_path = 'ifs/admin/test' 

262 json_str = '{"my_json": "test123"}' 

263 self._add_get_directory_listing_response(m, fq_dir_path, json_str) 

264 

265 actual_json = self.powerscale_api.get_directory_listing(fq_dir_path) 

266 

267 self.assertEqual(1, len(m.request_history)) 

268 self.assertEqual(json.loads(json_str), actual_json) 

269 

270 @ddt.data((200, True), (404, False)) 

271 def test_is_path_existent(self, data): 

272 status_code, expected_return_value = data 

273 with requests_mock.mock() as m: 

274 self.assertEqual(0, len(m.request_history)) 

275 path = '/ifs/home/admin' 

276 m.head('{0}/namespace{1}'.format(self._mock_url, path), 

277 status_code=status_code) 

278 

279 r = self.powerscale_api.is_path_existent(path) 

280 

281 self.assertEqual(expected_return_value, r) 

282 self.assertEqual(1, len(m.request_history)) 

283 

284 @requests_mock.mock() 

285 def test_is_path_existent_unexpected_error(self, m): 

286 path = '/ifs/home/admin' 

287 m.head('{0}/namespace{1}'.format(self._mock_url, path), 

288 status_code=400) 

289 

290 self.assertRaises( 

291 requests.exceptions.HTTPError, 

292 self.powerscale_api.is_path_existent, 

293 '/ifs/home/admin') 

294 

295 @ddt.data( 

296 (200, '{"snapshots": [{"path": "/ifs/home/test"}]}', 

297 {'path': '/ifs/home/test'}), 

298 (404, '{"errors": []}', None) 

299 ) 

300 def test_get_snapshot(self, data): 

301 status_code, json_body, expected_return_value = data 

302 with requests_mock.mock() as m: 

303 self.assertEqual(0, len(m.request_history)) 

304 snapshot_name = 'foo1' 

305 self._add_get_snapshot_response(m, snapshot_name, json_body, 

306 status=status_code) 

307 

308 r = self.powerscale_api.get_snapshot(snapshot_name) 

309 

310 self.assertEqual(1, len(m.request_history)) 

311 self.assertEqual(expected_return_value, r) 

312 

313 def test_clone_snapshot_with_provider_location(self): 

314 """clone_snapshot should use provider_location to get snapshot""" 

315 self.powerscale_api.create_directory = mock.Mock() 

316 self.powerscale_api.get_snapshot_id = mock.Mock() 

317 self.powerscale_api.get_snapshot = mock.Mock() 

318 self.powerscale_api._clone_directory_contents = mock.Mock() 

319 fq_target_dir = '/ifs/data/clone1' 

320 provider_location = '1234' 

321 snapshot = { 

322 'name': 'snap-123', 

323 'path': '/ifs/data/share1' 

324 } 

325 self.powerscale_api.get_snapshot_id.return_value = snapshot 

326 self.powerscale_api.clone_snapshot( 

327 snapshot_name='ignored-snapshot-name', 

328 fq_target_dir=fq_target_dir, 

329 provider_location=provider_location 

330 ) 

331 (self.powerscale_api. 

332 create_directory.assert_called_once_with(fq_target_dir)) 

333 (self.powerscale_api. 

334 get_snapshot_id.assert_called_once_with(provider_location)) 

335 self.powerscale_api.get_snapshot.assert_not_called() 

336 

337 @requests_mock.mock() 

338 def test_get_snapshot_unexpected_error(self, m): 

339 snapshot_name = 'foo1' 

340 json_body = '{"snapshots": [{"path": "/ifs/home/test"}]}' 

341 self._add_get_snapshot_response( 

342 m, snapshot_name, json_body, status=400) 

343 

344 self.assertRaises( 

345 requests.exceptions.HTTPError, self.powerscale_api.get_snapshot, 

346 snapshot_name) 

347 

348 @requests_mock.mock() 

349 def test_get_snapshots(self, m): 

350 self.assertEqual(0, len(m.request_history)) 

351 snapshot_json = '{"snapshots": [{"path": "/ifs/home/test"}]}' 

352 m.get('{0}/platform/1/snapshot/snapshots'.format(self._mock_url), 

353 status_code=200, json=json.loads(snapshot_json)) 

354 

355 r = self.powerscale_api.get_snapshots() 

356 

357 self.assertEqual(1, len(m.request_history)) 

358 self.assertEqual(json.loads(snapshot_json), r) 

359 

360 @requests_mock.mock() 

361 def test_get_snapshots_error_occurred(self, m): 

362 self.assertEqual(0, len(m.request_history)) 

363 m.get('{0}/platform/1/snapshot/snapshots'.format(self._mock_url), 

364 status_code=404) 

365 

366 self.assertRaises(requests.exceptions.HTTPError, 

367 self.powerscale_api.get_snapshots) 

368 

369 self.assertEqual(1, len(m.request_history)) 

370 

371 @ddt.data( 

372 ('/ifs/home/admin', 

373 '{"exports": [{"id": 42, "paths": ["/ifs/home/admin"]}], "total": 1}', 

374 42), 

375 ('/ifs/home/test', 

376 '{"exports": [], "total": 0}', None) 

377 ) 

378 def test_lookup_nfs_export(self, data): 

379 share_path, response_json, expected_return = data 

380 with requests_mock.mock() as m: 

381 self.assertEqual(0, len(m.request_history)) 

382 m.get('{0}/platform/12/protocols/nfs/exports?path={1}' 

383 .format(self._mock_url, 

384 share_path.replace('/', '%2F')), 

385 json=json.loads(response_json)) 

386 

387 r = self.powerscale_api.lookup_nfs_export(share_path) 

388 

389 self.assertEqual(1, len(m.request_history)) 

390 self.assertEqual(expected_return, r) 

391 

392 @requests_mock.mock() 

393 def test_get_nfs_export(self, m): 

394 self.assertEqual(0, len(m.request_history)) 

395 export_id = 42 

396 response_json = '{"exports": [{"id": 1}]}' 

397 status_code = 200 

398 m.get('{0}/platform/1/protocols/nfs/exports/{1}' 

399 .format(self._mock_url, export_id), 

400 json=json.loads(response_json), status_code=status_code) 

401 

402 r = self.powerscale_api.get_nfs_export(export_id) 

403 

404 self.assertEqual(1, len(m.request_history)) 

405 self.assertEqual(json.loads('{"id": 1}'), r) 

406 

407 @requests_mock.mock() 

408 def test_get_nfs_export_error(self, m): 

409 self.assertEqual(0, len(m.request_history)) 

410 export_id = 3 

411 response_json = '{}' 

412 status_code = 404 

413 m.get('{0}/platform/1/protocols/nfs/exports/{1}' 

414 .format(self._mock_url, export_id), 

415 json=json.loads(response_json), status_code=status_code) 

416 

417 r = self.powerscale_api.get_nfs_export(export_id) 

418 

419 self.assertEqual(1, len(m.request_history)) 

420 self.assertIsNone(r) 

421 

422 @requests_mock.mock() 

423 def test_lookup_smb_share(self, m): 

424 self.assertEqual(0, len(m.request_history)) 

425 share_name = 'my_smb_share' 

426 share_json = '{"id": "my_smb_share"}' 

427 response_json = '{{"shares": [{0}]}}'.format(share_json) 

428 m.get('{0}/platform/1/protocols/smb/shares/{1}' 

429 .format(self._mock_url, share_name), status_code=200, 

430 json=json.loads(response_json)) 

431 

432 r = self.powerscale_api.lookup_smb_share(share_name) 

433 

434 self.assertEqual(1, len(m.request_history)) 

435 self.assertEqual(json.loads(share_json), r) 

436 

437 @requests_mock.mock() 

438 def test_lookup_smb_share_error(self, m): 

439 self.assertEqual(0, len(m.request_history)) 

440 share_name = 'my_smb_share' 

441 m.get('{0}/platform/1/protocols/smb/shares/{1}'.format( 

442 self._mock_url, share_name), status_code=404) 

443 

444 r = self.powerscale_api.lookup_smb_share(share_name) 

445 

446 self.assertEqual(1, len(m.request_history)) 

447 self.assertIsNone(r) 

448 

449 @ddt.data((201, True), (404, False)) 

450 def test_create_nfs_export(self, data): 

451 status_code, expected_return_value = data 

452 with requests_mock.mock() as m: 

453 self.assertEqual(0, len(m.request_history)) 

454 export_path = '/ifs/home/test' 

455 m.post(self._mock_url + '/platform/1/protocols/nfs/exports', 

456 status_code=status_code) 

457 

458 r = self.powerscale_api.create_nfs_export(export_path) 

459 

460 self.assertEqual(1, len(m.request_history)) 

461 call = m.request_history[0] 

462 expected_request_body = '{"paths": ["/ifs/home/test"]}' 

463 self.assertEqual(json.loads(expected_request_body), 

464 json.loads(call.body)) 

465 self.assertEqual(expected_return_value, r) 

466 

467 @ddt.data((201, True), (404, False)) 

468 def test_create_smb_share(self, data): 

469 status_code, expected_return_value = data 

470 with requests_mock.mock() as m: 

471 self.assertEqual(0, len(m.request_history)) 

472 share_name = 'my_smb_share' 

473 share_path = '/ifs/home/admin/smb_share' 

474 m.post(self._mock_url + '/platform/1/protocols/smb/shares', 

475 status_code=status_code) 

476 

477 r = self.powerscale_api.create_smb_share(share_name, share_path) 

478 

479 self.assertEqual(expected_return_value, r) 

480 self.assertEqual(1, len(m.request_history)) 

481 expected_request_data = { 

482 'name': share_name, 

483 'path': share_path, 

484 'permissions': [] 

485 } 

486 self.assertEqual(expected_request_data, 

487 json.loads(m.request_history[0].body)) 

488 

489 @requests_mock.mock() 

490 def test_create_snapshot(self, m): 

491 self.assertEqual(0, len(m.request_history)) 

492 snapshot_name = 'my_snapshot_01' 

493 snapshot_path = '/ifs/home/admin' 

494 m.post( 

495 self._mock_url + '/platform/1/snapshot/snapshots', 

496 status_code=201, 

497 json={"id": 123, "name": snapshot_name, "path": snapshot_path} 

498 ) 

499 

500 r = self.powerscale_api.create_snapshot(snapshot_name, snapshot_path) 

501 

502 self.assertEqual(1, len(m.request_history)) 

503 self.assertTrue(r) 

504 expected_request_body = json.loads( 

505 '{{"name": "{0}", "path": "{1}"}}' 

506 .format(snapshot_name, snapshot_path) 

507 ) 

508 self.assertEqual(expected_request_body, 

509 json.loads(m.request_history[0].body)) 

510 

511 @requests_mock.mock() 

512 def test_create_snapshot_error_case(self, m): 

513 self.assertEqual(0, len(m.request_history)) 

514 snapshot_name = 'my_snapshot_01' 

515 snapshot_path = '/ifs/home/admin' 

516 m.post(self._mock_url + '/platform/1/snapshot/snapshots', 

517 status_code=404) 

518 

519 self.assertIsNone(self.powerscale_api.create_snapshot(snapshot_name, 

520 snapshot_path)) 

521 

522 @ddt.data(True, False) 

523 def test_delete_path(self, is_recursive_delete): 

524 with requests_mock.mock() as m: 

525 self.assertEqual(0, len(m.request_history)) 

526 fq_path = '/ifs/home/admin/test' 

527 m.delete(self._mock_url + '/namespace' + fq_path + '?recursive=' 

528 + str(is_recursive_delete), status_code=204) 

529 

530 self.powerscale_api.delete_path( 

531 fq_path, recursive=is_recursive_delete) 

532 

533 self.assertEqual(1, len(m.request_history)) 

534 

535 @requests_mock.mock() 

536 def test_delete_path_error_case(self, m): 

537 fq_path = '/ifs/home/admin/test' 

538 m.delete(self._mock_url + '/namespace' + fq_path + '?recursive=False', 

539 status_code=403) 

540 

541 self.assertEqual( 

542 self.powerscale_api.delete_path( 

543 fq_path, recursive=False), False) 

544 

545 @ddt.data((204, True), (404, False)) 

546 def test_delete_nfs_share(self, data): 

547 status_code, expected_return_value = data 

548 with requests_mock.mock() as m: 

549 self.assertEqual(0, len(m.request_history)) 

550 share_number = 42 

551 m.delete('{0}/platform/1/protocols/nfs/exports/{1}' 

552 .format(self._mock_url, share_number), 

553 status_code=status_code) 

554 

555 r = self.powerscale_api.delete_nfs_share(share_number) 

556 

557 self.assertEqual(1, len(m.request_history)) 

558 self.assertEqual(expected_return_value, r) 

559 

560 @ddt.data((204, True), (404, False)) 

561 def test_delete_smb_shares(self, data): 

562 status_code, expected_return_value = data 

563 with requests_mock.mock() as m: 

564 self.assertEqual(0, len(m.request_history)) 

565 

566 share_name = 'smb_share_42' 

567 m.delete('{0}/platform/1/protocols/smb/shares/{1}' 

568 .format(self._mock_url, share_name), 

569 status_code=status_code) 

570 

571 r = self.powerscale_api.delete_smb_share(share_name) 

572 

573 self.assertEqual(1, len(m.request_history)) 

574 self.assertEqual(expected_return_value, r) 

575 

576 @requests_mock.mock() 

577 def test_delete_snapshot(self, m): 

578 self.assertEqual(0, len(m.request_history)) 

579 m.delete(self._mock_url + '/platform/1/snapshot/snapshots/my_snapshot', 

580 status_code=204) 

581 

582 self.powerscale_api.delete_snapshot("my_snapshot") 

583 

584 self.assertEqual(1, len(m.request_history)) 

585 

586 @requests_mock.mock() 

587 def test_delete_snapshot_error_case(self, m): 

588 m.delete(self._mock_url + '/platform/1/snapshot/snapshots/my_snapshot', 

589 status_code=403) 

590 

591 self.assertEqual( 

592 self.powerscale_api.delete_snapshot("my_snapshot"), False) 

593 

594 @requests_mock.mock() 

595 def test_quota_create(self, m): 

596 quota_path = '/ifs/manila/test' 

597 quota_size = 256 

598 self.assertEqual(0, len(m.request_history)) 

599 m.post(self._mock_url + '/platform/1/quota/quotas', status_code=201) 

600 

601 self.powerscale_api.quota_create(quota_path, 'directory', quota_size) 

602 

603 self.assertEqual(1, len(m.request_history)) 

604 expected_request_json = { 

605 'path': quota_path, 

606 'type': 'directory', 

607 'include_snapshots': False, 

608 'thresholds_include_overhead': False, 

609 'enforced': True, 

610 'thresholds': {'hard': quota_size}, 

611 } 

612 call_body = m.request_history[0].body 

613 self.assertEqual(expected_request_json, json.loads(call_body)) 

614 

615 @requests_mock.mock() 

616 def test_quota_create_with_threshold(self, m): 

617 quota_path = '/ifs/manila/test' 

618 quota_size = 100 

619 self.assertEqual(0, len(m.request_history)) 

620 m.post(self._mock_url + '/platform/1/quota/quotas', status_code=201) 

621 self.powerscale_api_threshold.quota_create( 

622 quota_path, 

623 'directory', 

624 quota_size 

625 ) 

626 

627 advisory_size = round( 

628 (quota_size * self.powerscale_api_threshold.threshold_limit) / 100) 

629 self.assertEqual(1, len(m.request_history)) 

630 expected_request_json = { 

631 'path': quota_path, 

632 'type': 'directory', 

633 'include_snapshots': False, 

634 'thresholds_include_overhead': False, 

635 'enforced': True, 

636 'thresholds': {'hard': quota_size, 

637 'advisory': advisory_size}, 

638 } 

639 call_body = m.request_history[0].body 

640 self.assertEqual(expected_request_json, json.loads(call_body)) 

641 

642 @requests_mock.mock() 

643 def test_quota_create__path_does_not_exist(self, m): 

644 quota_path = '/ifs/test2' 

645 self.assertEqual(0, len(m.request_history)) 

646 m.post(self._mock_url + '/platform/1/quota/quotas', status_code=400) 

647 

648 self.assertRaises( 

649 requests.exceptions.HTTPError, 

650 self.powerscale_api.quota_create, 

651 quota_path, 'directory', 2 

652 ) 

653 

654 @requests_mock.mock() 

655 def test_quota_get(self, m): 

656 self.assertEqual(0, len(m.request_history)) 

657 response_json = {'quotas': [{}]} 

658 m.get(self._mock_url + '/platform/1/quota/quotas', json=response_json, 

659 status_code=200) 

660 quota_path = "/ifs/manila/test" 

661 quota_type = "directory" 

662 

663 self.powerscale_api.quota_get(quota_path, quota_type) 

664 

665 self.assertEqual(1, len(m.request_history)) 

666 request_query_string = m.request_history[0].qs 

667 expected_query_string = {'path': [quota_path]} 

668 self.assertEqual(expected_query_string, request_query_string) 

669 

670 @requests_mock.mock() 

671 def test_quota_get__path_does_not_exist(self, m): 

672 self.assertEqual(0, len(m.request_history)) 

673 m.get(self._mock_url + '/platform/1/quota/quotas', status_code=404) 

674 

675 response = self.powerscale_api.quota_get( 

676 '/ifs/does_not_exist', 'directory') 

677 

678 self.assertIsNone(response) 

679 

680 @requests_mock.mock() 

681 def test_quota_modify(self, m): 

682 self.assertEqual(0, len(m.request_history)) 

683 quota_id = "ADEF1G" 

684 new_size = 1024 

685 m.put('{0}/platform/1/quota/quotas/{1}'.format( 

686 self._mock_url, quota_id), status_code=204) 

687 

688 self.powerscale_api.quota_modify_size(quota_id, new_size) 

689 

690 self.assertEqual(1, len(m.request_history)) 

691 expected_request_body = {'thresholds': {'hard': new_size}} 

692 request_body = m.request_history[0].body 

693 self.assertEqual(expected_request_body, json.loads(request_body)) 

694 

695 @requests_mock.mock() 

696 def test_quota_modify_with_threshold(self, m): 

697 self.assertEqual(0, len(m.request_history)) 

698 quota_id = "ADEF1G" 

699 new_size = 1024 

700 advisory_size = round( 

701 (new_size * self.powerscale_api_threshold.threshold_limit) / 100) 

702 m.put('{0}/platform/1/quota/quotas/{1}'.format( 

703 self._mock_url, quota_id), status_code=204) 

704 self.powerscale_api_threshold.quota_modify_size(quota_id, new_size) 

705 self.assertEqual(1, len(m.request_history)) 

706 expected_request_body = {'thresholds': {'hard': new_size, 

707 'advisory': advisory_size}} 

708 request_body = m.request_history[0].body 

709 self.assertEqual(expected_request_body, json.loads(request_body)) 

710 

711 @requests_mock.mock() 

712 def test_quota_modify__given_id_does_not_exist(self, m): 

713 quota_id = 'ADE2F' 

714 m.put('{0}/platform/1/quota/quotas/{1}'.format( 

715 self._mock_url, quota_id), status_code=404) 

716 

717 self.assertRaises( 

718 requests.exceptions.HTTPError, 

719 self.powerscale_api.quota_modify_size, 

720 quota_id, 1024 

721 ) 

722 

723 @requests_mock.mock() 

724 def test_quota_set__quota_already_exists(self, m): 

725 self.assertEqual(0, len(m.request_history)) 

726 quota_path = '/ifs/manila/test' 

727 quota_type = 'directory' 

728 quota_size = 256 

729 quota_id = 'AFE2C' 

730 m.get('{0}/platform/1/quota/quotas'.format( 

731 self._mock_url), json={'quotas': [{'id': quota_id}]}, 

732 status_code=200) 

733 m.put( 

734 '{0}/platform/1/quota/quotas/{1}'.format(self._mock_url, quota_id), 

735 status_code=204 

736 ) 

737 

738 self.powerscale_api.quota_set(quota_path, quota_type, quota_size) 

739 

740 expected_quota_modify_json = {'thresholds': {'hard': quota_size}} 

741 quota_put_json = json.loads(m.request_history[1].body) 

742 self.assertEqual(expected_quota_modify_json, quota_put_json) 

743 

744 @requests_mock.mock()