Coverage for manila/tests/share/drivers/dell_emc/plugins/powerstore/test_client.py: 100%

1# Copyright (c) 2023 Dell Inc. or its subsidiaries. 

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 

16import json 

17import os 

18 

19import requests_mock 

20 

21from manila.share.drivers.dell_emc.plugins.powerstore import client 

22from manila import test 

23 

24 

25class TestClient(test.TestCase): 

26 

27 REST_IP = "192.168.0.110" 

28 NAS_SERVER_NAME = "powerstore-nasserver" 

29 NAS_SERVER_ID = "6423d56e-eaf3-7424-be0b-1a9efb93188b" 

30 NAS_SERVER_IP = "192.168.11.23" 

31 NFS_EXPORT_NAME = "powerstore-nfs-share" 

32 NFS_EXPORT_SIZE = 3221225472 

33 NFS_EXPORT_NEW_SIZE = 6221225472 

34 FILESYSTEM_ID = "6454e9a9-a698-e9bc-ca61-1a9efb93188b" 

35 NFS_EXPORT_ID = "6454ec18-7b8d-1532-1b8a-1a9efb93188b" 

36 RW_HOSTS = "192.168.1.10" 

37 RO_HOSTS = "192.168.1.11" 

38 SMB_SHARE_NAME = "powerstore-smb-share" 

39 SMB_SHARE_ID = "64927ae9-3403-6930-a784-f227b9987c54" 

40 RW_USERS = "user1" 

41 RO_USERS = "user2" 

42 SNAPSHOT_NAME = "powerstore-nfs-share-snap" 

43 SNAPSHOT_ID = "6454ea29-09c3-030e-cfc3-1a9efb93188b" 

44 CLONE_ID = "64560f05-e677-ec2a-7fcf-1a9efb93188b" 

45 CLONE_NAME = "powerstore-nfs-share-snap-clone" 

46 CLUSTER_ID = "0" 

47 

48 CLIENT_OPTIONS = { 

49 "rest_ip": REST_IP, 

50 "rest_username": "admin", 

51 "rest_password": "pwd", 

52 "verify_certificate": False, 

53 "certificate_path": None 

54 } 

55 

56 def setUp(self): 

57 super(TestClient, self).setUp() 

58 

59 self._mock_url = "https://%s/api/rest" % self.REST_IP 

60 self.client = client.PowerStoreClient(**self.CLIENT_OPTIONS) 

61 self.mockup_file_base = os.path.join( 

62 os.path.dirname(os.path.realpath(__file__)), 'mockup') 

63 

64 def _getJsonFile(self, filename): 

65 f = open(os.path.join(self.mockup_file_base, filename)) 

66 data = json.load(f) 

67 f.close() 

68 return data 

69 

70 def test__verify_cert(self): 

71 verify_cert = self.client.verify_certificate 

72 certificate_path = self.client.certificate_path 

73 self.client.verify_certificate = True 

74 self.client.certificate_path = "fake_certificate_path" 

75 self.assertEqual(self.client._verify_cert, 

76 self.client.certificate_path) 

77 self.client.verify_certificate = verify_cert 

78 self.client.certificate_path = certificate_path 

79 

80 @requests_mock.mock() 

81 def test__send_request(self, m): 

82 url = "{0}/fake_res".format(self._mock_url) 

83 m.get(url, status_code=200) 

84 self.client._send_get_request("/fake_res", None, None, False) 

85 

86 @requests_mock.mock() 

87 def test_get_nas_server_id(self, m): 

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

89 self._add_get_nas_server_id_response( 

90 m, self.NAS_SERVER_NAME, 

91 self._getJsonFile("get_nas_server_id_response.json") 

92 ) 

93 id = self.client.get_nas_server_id(self.NAS_SERVER_NAME) 

94 self.assertEqual(id, self.NAS_SERVER_ID) 

95 

96 def _add_get_nas_server_id_response(self, m, nas_server, json_str): 

97 url = "{0}/nas_server?name=eq.{1}".format( 

98 self._mock_url, nas_server 

99 ) 

100 m.get(url, status_code=200, json=json_str) 

101 

102 @requests_mock.mock() 

103 def test_get_nas_server_id_failure(self, m): 

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

105 self._add_get_nas_server_id_response_failure( 

106 m, self.NAS_SERVER_NAME 

107 ) 

108 id = self.client.get_nas_server_id(self.NAS_SERVER_NAME) 

109 self.assertIsNone(id) 

110 

111 def _add_get_nas_server_id_response_failure(self, m, nas_server): 

112 url = "{0}/nas_server?name=eq.{1}".format( 

113 self._mock_url, nas_server 

114 ) 

115 m.get(url, status_code=400) 

116 

117 @requests_mock.mock() 

118 def test_get_nas_server_interfaces(self, m): 

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

120 self._add_get_nas_server_interfaces_response( 

121 m, self.NAS_SERVER_ID, 

122 self._getJsonFile("get_nas_server_interfaces_response.json") 

123 ) 

124 interfaces = self.client.get_nas_server_interfaces(self.NAS_SERVER_ID) 

125 self.assertEqual(interfaces[0]['ip'], self.NAS_SERVER_IP) 

126 self.assertEqual(interfaces[0]['preferred'], True) 

127 

128 def _add_get_nas_server_interfaces_response(self, m, nas_server_id, 

129 json_str): 

130 url = "{0}/nas_server/{1}?select=" \ 

131 "current_preferred_IPv4_interface_id," \ 

132 "current_preferred_IPv6_interface_id," \ 

133 "file_interfaces(id,ip_address)".format( 

134 self._mock_url, nas_server_id 

135 ) 

136 m.get(url, status_code=200, json=json_str) 

137 

138 @requests_mock.mock() 

139 def test_get_nas_server_interfaces_failure(self, m): 

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

141 self._add_get_nas_server_interfaces_response_failure( 

142 m, self.NAS_SERVER_ID 

143 ) 

144 interfaces = self.client.get_nas_server_interfaces(self.NAS_SERVER_ID) 

145 self.assertIsNone(interfaces) 

146 

147 def _add_get_nas_server_interfaces_response_failure(self, m, 

148 nas_server_id): 

149 url = "{0}/nas_server/{1}?select=" \ 

150 "current_preferred_IPv4_interface_id," \ 

151 "current_preferred_IPv6_interface_id," \ 

152 "file_interfaces(id,ip_address)".format( 

153 self._mock_url, nas_server_id 

154 ) 

155 m.get(url, status_code=400) 

156 

157 @requests_mock.mock() 

158 def test_create_filesystem(self, m): 

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

160 self._add_create_filesystem_response( 

161 m, self._getJsonFile("create_filesystem_response.json") 

162 ) 

163 id = self.client.create_filesystem( 

164 self.NAS_SERVER_ID, 

165 self.NFS_EXPORT_NAME, 

166 self.NFS_EXPORT_SIZE 

167 ) 

168 self.assertEqual(id, self.FILESYSTEM_ID) 

169 

170 def _add_create_filesystem_response(self, m, json_str): 

171 url = "{0}/file_system".format(self._mock_url) 

172 m.post(url, status_code=201, json=json_str) 

173 

174 @requests_mock.mock() 

175 def test_create_filesystem_failure(self, m): 

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

177 self._add_create_filesystem_response_failure(m) 

178 id = self.client.create_filesystem( 

179 self.NAS_SERVER_ID, 

180 self.NFS_EXPORT_NAME, 

181 self.NFS_EXPORT_SIZE 

182 ) 

183 self.assertIsNone(id) 

184 

185 def _add_create_filesystem_response_failure(self, m): 

186 url = "{0}/file_system".format(self._mock_url) 

187 m.post(url, status_code=400) 

188 

189 @requests_mock.mock() 

190 def test_create_nfs_export(self, m): 

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

192 self._add_create_nfs_export_response( 

193 m, self._getJsonFile("create_nfs_export_response.json") 

194 ) 

195 id = self.client.create_nfs_export(self.FILESYSTEM_ID, 

196 self.NFS_EXPORT_NAME) 

197 self.assertEqual(id, self.NFS_EXPORT_ID) 

198 

199 def _add_create_nfs_export_response(self, m, json_str): 

200 url = "{0}/nfs_export".format(self._mock_url) 

201 m.post(url, status_code=201, json=json_str) 

202 

203 @requests_mock.mock() 

204 def test_create_nfs_export_failure(self, m): 

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

206 self._add_create_nfs_export_response_failure(m) 

207 id = self.client.create_nfs_export(self.FILESYSTEM_ID, 

208 self.NFS_EXPORT_NAME) 

209 self.assertIsNone(id) 

210 

211 def _add_create_nfs_export_response_failure(self, m): 

212 url = "{0}/nfs_export".format(self._mock_url) 

213 m.post(url, status_code=400) 

214 

215 @requests_mock.mock() 

216 def test_delete_filesystem(self, m): 

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

218 self._add_delete_filesystem_response(m, self.FILESYSTEM_ID) 

219 result = self.client.delete_filesystem(self.FILESYSTEM_ID) 

220 self.assertEqual(result, True) 

221 

222 def _add_delete_filesystem_response(self, m, filesystem_id): 

223 url = "{0}/file_system/{1}".format( 

224 self._mock_url, filesystem_id 

225 ) 

226 m.delete(url, status_code=204) 

227 

228 @requests_mock.mock() 

229 def test_get_nfs_export_name(self, m): 

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

231 self._add_get_nfs_export_name_response( 

232 m, 

233 self.NFS_EXPORT_ID, 

234 self._getJsonFile("get_nfs_export_name_response.json"), 

235 ) 

236 name = self.client.get_nfs_export_name(self.NFS_EXPORT_ID) 

237 self.assertEqual(name, self.NFS_EXPORT_NAME) 

238 

239 def _add_get_nfs_export_name_response(self, m, export_id, json_str): 

240 url = "{0}/nfs_export/{1}?select=name".format( 

241 self._mock_url, export_id 

242 ) 

243 m.get(url, status_code=200, json=json_str) 

244 

245 @requests_mock.mock() 

246 def test_get_nfs_export_name_failure(self, m): 

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

248 self._add_get_nfs_export_name_response_failure(m, self.NFS_EXPORT_ID) 

249 name = self.client.get_nfs_export_name(self.NFS_EXPORT_ID) 

250 self.assertIsNone(name) 

251 

252 def _add_get_nfs_export_name_response_failure(self, 

253 m, export_id): 

254 url = "{0}/nfs_export/{1}?select=name".format( 

255 self._mock_url, export_id 

256 ) 

257 m.get(url, status_code=400) 

258 

259 @requests_mock.mock() 

260 def test_get_nfs_export_id(self, m): 

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

262 self._add_get_nfs_export_id_response( 

263 m, self.NFS_EXPORT_NAME, 

264 self._getJsonFile("get_nfs_export_id_response.json") 

265 ) 

266 id = self.client.get_nfs_export_id(self.NFS_EXPORT_NAME) 

267 self.assertEqual(id, self.NFS_EXPORT_ID) 

268 

269 def _add_get_nfs_export_id_response(self, m, name, json_str): 

270 url = "{0}/nfs_export?select=id&name=eq.{1}".format( 

271 self._mock_url, name 

272 ) 

273 m.get(url, status_code=200, json=json_str) 

274 

275 @requests_mock.mock() 

276 def test_get_nfs_export_id_failure(self, m): 

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

278 self._add_get_nfs_export_id_response_failure(m, self.NFS_EXPORT_NAME) 

279 id = self.client.get_nfs_export_id(self.NFS_EXPORT_NAME) 

280 self.assertIsNone(id) 

281 

282 def _add_get_nfs_export_id_response_failure(self, m, name): 

283 url = "{0}/nfs_export?select=id&name=eq.{1}".format( 

284 self._mock_url, name 

285 ) 

286 m.get(url, status_code=400) 

287 

288 @requests_mock.mock() 

289 def test_get_filesystem_id(self, m): 

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

291 self._add_get_filesystem_id_response( 

292 m, self.NFS_EXPORT_NAME, 

293 self._getJsonFile("get_fileystem_id_response.json") 

294 ) 

295 id = self.client.get_filesystem_id(self.NFS_EXPORT_NAME) 

296 self.assertEqual(id, self.FILESYSTEM_ID) 

297 

298 def _add_get_filesystem_id_response(self, m, name, json_str): 

299 url = "{0}/file_system?name=eq.{1}".format( 

300 self._mock_url, name 

301 ) 

302 m.get(url, status_code=200, json=json_str) 

303 

304 @requests_mock.mock() 

305 def test_get_filesystem_id_failure(self, m): 

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

307 self._add_get_filesystem_id_response_failure(m, self.NFS_EXPORT_NAME) 

308 id = self.client.get_filesystem_id(self.NFS_EXPORT_NAME) 

309 self.assertIsNone(id) 

310 

311 def _add_get_filesystem_id_response_failure(self, m, name): 

312 url = "{0}/file_system?name=eq.{1}".format( 

313 self._mock_url, name 

314 ) 

315 m.get(url, status_code=400) 

316 

317 @requests_mock.mock() 

318 def test_set_export_access(self, m): 

319 

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

321 self._add_set_export_access_response(m, self.NFS_EXPORT_ID) 

322 result = self.client.set_export_access(self.NFS_EXPORT_ID, 

323 self.RW_HOSTS, 

324 self.RO_HOSTS) 

325 self.assertEqual(result, True) 

326 

327 def _add_set_export_access_response(self, m, export_id): 

328 url = "{0}/nfs_export/{1}".format(self._mock_url, export_id) 

329 m.patch(url, status_code=204) 

330 

331 @requests_mock.mock() 

332 def test_resize_filesystem(self, m): 

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

334 self._add_resize_filesystem_response(m, self.FILESYSTEM_ID) 

335 result, detail = self.client.resize_filesystem( 

336 self.FILESYSTEM_ID, self.NFS_EXPORT_NEW_SIZE) 

337 self.assertTrue(result) 

338 self.assertIsNone(detail) 

339 

340 def _add_resize_filesystem_response(self, m, filesystem_id): 

341 url = "{0}/file_system/{1}".format( 

342 self._mock_url, filesystem_id 

343 ) 

344 m.patch(url, status_code=204) 

345 

346 @requests_mock.mock() 

347 def test_resize_filesystem_shrink_failure(self, m): 

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

349 self._add_resize_filesystem_shrink_failure_response( 

350 m, self.FILESYSTEM_ID, 

351 self._getJsonFile( 

352 "resize_filesystem_shrink_failure_response.json")) 

353 result, detail = self.client.resize_filesystem( 

354 self.FILESYSTEM_ID, self.NFS_EXPORT_NEW_SIZE) 

355 self.assertFalse(result) 

356 self.assertIsNotNone(detail) 

357 

358 def _add_resize_filesystem_shrink_failure_response( 

359 self, m, filesystem_id, json_str): 

360 url = "{0}/file_system/{1}".format( 

361 self._mock_url, filesystem_id 

362 ) 

363 m.patch(url, status_code=422, json=json_str) 

364 

365 @requests_mock.mock() 

366 def test_get_fsid_from_export_name(self, m): 

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

368 self._add_get_fsid_from_export_name_response( 

369 m, self.NFS_EXPORT_NAME, 

370 self._getJsonFile("get_fsid_from_export_name_response.json") 

371 ) 

372 id = self.client.get_fsid_from_export_name(self.NFS_EXPORT_NAME) 

373 self.assertEqual(id, self.FILESYSTEM_ID) 

374 

375 def _add_get_fsid_from_export_name_response(self, m, name, json_str): 

376 url = "{0}/nfs_export?select=file_system_id&name=eq.{1}".format( 

377 self._mock_url, name 

378 ) 

379 m.get(url, status_code=200, json=json_str) 

380 

381 @requests_mock.mock() 

382 def test_get_fsid_from_export_name_failure(self, m): 

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

384 self._add_get_fsid_from_export_name_response_failure( 

385 m, self.NFS_EXPORT_NAME 

386 ) 

387 id = self.client.get_fsid_from_export_name(self.NFS_EXPORT_NAME) 

388 self.assertIsNone(id) 

389 

390 def _add_get_fsid_from_export_name_response_failure(self, m, name): 

391 url = "{0}/nfs_export?select=file_system_id&name=eq.{1}".format( 

392 self._mock_url, name 

393 ) 

394 m.get(url, status_code=400) 

395 

396 @requests_mock.mock() 

397 def test_create_snapshot(self, m): 

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

399 self._add_create_snapshot_response( 

400 m, self.FILESYSTEM_ID, 

401 self._getJsonFile("create_snapshot_response.json") 

402 ) 

403 id = self.client.create_snapshot(self.FILESYSTEM_ID, 

404 self.SNAPSHOT_NAME) 

405 self.assertEqual(id, self.SNAPSHOT_ID) 

406 

407 def _add_create_snapshot_response(self, m, filesystem_id, json_str): 

408 url = "{0}/file_system/{1}/snapshot".format(self._mock_url, 

409 filesystem_id) 

410 m.post(url, status_code=201, json=json_str) 

411 

412 @requests_mock.mock() 

413 def test_create_snapshot_failure(self, m): 

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

415 self._add_create_snapshot_response_failure(m, self.FILESYSTEM_ID) 

416 id = self.client.create_snapshot(self.FILESYSTEM_ID, 

417 self.SNAPSHOT_NAME) 

418 self.assertIsNone(id) 

419 

420 def _add_create_snapshot_response_failure(self, m, filesystem_id): 

421 url = "{0}/file_system/{1}/snapshot".format(self._mock_url, 

422 filesystem_id) 

423 m.post(url, status_code=400) 

424 

425 @requests_mock.mock() 

426 def test_restore_snapshot(self, m): 

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

428 self._add_restore_snapshot_response( 

429 m, self.SNAPSHOT_ID 

430 ) 

431 result = self.client.restore_snapshot(self.SNAPSHOT_ID) 

432 self.assertEqual(result, True) 

433 

434 def _add_restore_snapshot_response(self, m, snapshot_id): 

435 url = "{0}/file_system/{1}/restore".format(self._mock_url, 

436 snapshot_id) 

437 m.post(url, status_code=204) 

438 

439 @requests_mock.mock() 

440 def test_restore_snapshot_failure(self, m): 

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

442 self._add_restore_snapshot_response_failure( 

443 m, self.SNAPSHOT_ID 

444 ) 

445 result = self.client.restore_snapshot(self.SNAPSHOT_ID) 

446 self.assertEqual(result, False) 

447 

448 def _add_restore_snapshot_response_failure(self, m, snapshot_id): 

449 url = "{0}/file_system/{1}/restore".format(self._mock_url, 

450 snapshot_id) 

451 m.post(url, status_code=400) 

452 

453 @requests_mock.mock() 

454 def test_clone_snapshot(self, m): 

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

456 self._add_clone_snapshot_response( 

457 m, self.SNAPSHOT_ID, 

458 self._getJsonFile("clone_snapshot_response.json") 

459 ) 

460 id = self.client.clone_snapshot(self.SNAPSHOT_ID, 

461 self.CLONE_NAME) 

462 self.assertEqual(id, self.CLONE_ID) 

463 

464 def _add_clone_snapshot_response(self, m, snapshot_id, json_str): 

465 url = "{0}/file_system/{1}/clone".format(self._mock_url, 

466 snapshot_id) 

467 m.post(url, status_code=201, json=json_str) 

468 

469 @requests_mock.mock() 

470 def test_clone_snapshot_failure(self, m): 

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

472 self._add_clone_snapshot_response_failure( 

473 m, self.SNAPSHOT_ID 

474 ) 

475 id = self.client.clone_snapshot(self.SNAPSHOT_ID, 

476 self.CLONE_NAME) 

477 self.assertIsNone(id) 

478 

479 def _add_clone_snapshot_response_failure(self, m, snapshot_id): 

480 url = "{0}/file_system/{1}/clone".format(self._mock_url, 

481 snapshot_id) 

482 m.post(url, status_code=400) 

483 

484 @requests_mock.mock() 

485 def test_get_cluster_id(self, m): 

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

487 self._add_get_cluster_id_response( 

488 m, 

489 self._getJsonFile("get_cluster_id_response.json") 

490 ) 

491 id = self.client.get_cluster_id() 

492 self.assertEqual(id, self.CLUSTER_ID) 

493 

494 def _add_get_cluster_id_response(self, m, json_str): 

495 url = "{0}/cluster".format(self._mock_url) 

496 m.get(url, status_code=200, json=json_str) 

497 

498 @requests_mock.mock() 

499 def test_get_cluster_id_failure(self, m): 

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

501 self._add_get_cluster_id_response_failure(m) 

502 id = self.client.get_cluster_id() 

503 self.assertIsNone(id) 

504 

505 def _add_get_cluster_id_response_failure(self, m): 

506 url = "{0}/cluster".format(self._mock_url) 

507 m.get(url, status_code=400) 

508 

509 @requests_mock.mock() 

510 def test_retreive_cluster_capacity_metrics(self, m): 

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

512 self._add_retreive_cluster_capacity_metrics_response( 

513 m, 

514 self._getJsonFile( 

515 "retreive_cluster_capacity_metrics_response.json") 

516 ) 

517 total, used = self.client.retreive_cluster_capacity_metrics( 

518 self.CLUSTER_ID) 

519 self.assertEqual(total, 47345047046144) 

520 self.assertEqual(used, 366003363027) 

521 

522 def _add_retreive_cluster_capacity_metrics_response(self, m, json_str): 

523 url = "{0}/metrics/generate?order=timestamp".format(self._mock_url) 

524 m.post(url, status_code=200, json=json_str) 

525 

526 @requests_mock.mock() 

527 def test_retreive_cluster_capacity_metrics_failure(self, m): 

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

529 self._add_retreive_cluster_capacity_metrics_response_failure(m) 

530 total, used = self.client.retreive_cluster_capacity_metrics( 

531 self.CLUSTER_ID) 

532 self.assertIsNone(total) 

533 self.assertIsNone(used) 

534 

535 def _add_retreive_cluster_capacity_metrics_response_failure(self, m): 

536 url = "{0}/metrics/generate?order=timestamp".format(self._mock_url) 

537 m.post(url, status_code=400) 

538 

539 @requests_mock.mock() 

540 def test_create_smb_share(self, m): 

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

542 self._add_create_smb_share_response( 

543 m, self._getJsonFile("create_smb_share_response.json") 

544 ) 

545 id = self.client.create_smb_share(self.FILESYSTEM_ID, 

546 self.SMB_SHARE_NAME) 

547 self.assertEqual(id, self.SMB_SHARE_ID) 

548 

549 def _add_create_smb_share_response(self, m, json_str): 

550 url = "{0}/smb_share".format(self._mock_url) 

551 m.post(url, status_code=201, json=json_str) 

552 

553 @requests_mock.mock() 

554 def test_create_smb_share_failure(self, m): 

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

556 self._add_create_smb_share_response_failure(m) 

557 id = self.client.create_smb_share(self.FILESYSTEM_ID, 

558 self.SMB_SHARE_NAME) 

559 self.assertIsNone(id) 

560 

561 def _add_create_smb_share_response_failure(self, m): 

562 url = "{0}/smb_share".format(self._mock_url) 

563 m.post(url, status_code=400) 

564 

565 @requests_mock.mock() 

566 def test_get_fsid_from_share_name(self, m): 

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

568 self._add_get_fsid_from_share_name_response( 

569 m, self.NFS_EXPORT_NAME, 

570 self._getJsonFile("get_fsid_from_share_name_response.json") 

571 ) 

572 id = self.client.get_fsid_from_share_name(self.NFS_EXPORT_NAME) 

573 self.assertEqual(id, self.FILESYSTEM_ID) 

574 

575 def _add_get_fsid_from_share_name_response(self, m, name, json_str): 

576 url = "{0}/smb_share?select=file_system_id&name=eq.{1}".format( 

577 self._mock_url, name 

578 ) 

579 m.get(url, status_code=200, json=json_str) 

580 

581 @requests_mock.mock() 

582 def test_get_fsid_from_share_name_failure(self, m): 

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

584 self._add_get_fsid_from_share_name_response_failure( 

585 m, self.SMB_SHARE_NAME 

586 ) 

587 id = self.client.get_fsid_from_share_name(self.SMB_SHARE_NAME) 

588 self.assertIsNone(id) 

589 

590 def _add_get_fsid_from_share_name_response_failure(self, m, name): 

591 url = "{0}/smb_share?select=file_system_id&name=eq.{1}".format( 

592 self._mock_url, name 

593 ) 

594 m.get(url, status_code=400) 

595 

596 @requests_mock.mock() 

597 def test_get_smb_share_id(self, m): 

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

599 self._add_get_smb_share_id_response( 

600 m, self.SMB_SHARE_NAME, 

601 self._getJsonFile("get_smb_share_id_response.json") 

602 ) 

603 id = self.client.get_smb_share_id(self.SMB_SHARE_NAME) 

604 self.assertEqual(id, self.SMB_SHARE_ID) 

605 

606 def _add_get_smb_share_id_response(self, m, name, json_str): 

607 url = "{0}/smb_share?select=id&name=eq.{1}".format( 

608 self._mock_url, name 

609 ) 

610 m.get(url, status_code=200, json=json_str) 

611 

612 @requests_mock.mock() 

613 def test_get_smb_share_id_failure(self, m): 

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

615 self._add_get_smb_share_id_response_failure( 

616 m, self.SMB_SHARE_NAME 

617 ) 

618 id = self.client.get_smb_share_id(self.SMB_SHARE_NAME) 

619 self.assertIsNone(id) 

620 

621 def _add_get_smb_share_id_response_failure(self, m, name): 

622 url = "{0}/smb_share?select=id&name=eq.{1}".format( 

623 self._mock_url, name 

624 ) 

625 m.get(url, status_code=400) 

626 

627 @requests_mock.mock() 

628 def test_get_nas_server_smb_netbios(self, m): 

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

630 self._add_get_nas_server_smb_netbios_response( 

631 m, self.SMB_SHARE_NAME, 

632 self._getJsonFile("get_nas_server_smb_netbios_response.json") 

633 ) 

634 id = self.client.get_nas_server_smb_netbios(self.SMB_SHARE_NAME) 

635 self.assertEqual(id, "OPENSTACK") 

636 

637 def _add_get_nas_server_smb_netbios_response(self, m, name, json_str): 

638 url = "{0}/nas_server?select=smb_servers" \ 

639 "(is_standalone,netbios_name)&name=eq.{1}".format( 

640 self._mock_url, name 

641 ) 

642 m.get(url, status_code=200, json=json_str) 

643 

644 @requests_mock.mock() 

645 def test_get_nas_server_smb_netbios_failure(self, m): 

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

647 self._add_get_nas_server_smb_netbios_response_failure( 

648 m, self.SMB_SHARE_NAME 

649 ) 

650 id = self.client.get_nas_server_smb_netbios(self.SMB_SHARE_NAME) 

651 self.assertIsNone(id) 

652 

653 def _add_get_nas_server_smb_netbios_response_failure(self, m, name): 

654 url = "{0}/nas_server?select=smb_servers" \ 

655 "(is_standalone,netbios_name)&name=eq.{1}".format( 

656 self._mock_url, name 

657 ) 

658 m.get(url, status_code=400) 

659 

660 @requests_mock.mock() 

661 def test_set_acl(self, m): 

662 

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

664 self._add_set_acl_response(m, self.SMB_SHARE_ID) 

665 result = self.client.set_acl(self.SMB_SHARE_ID, 

666 self.RW_USERS, 

667 self.RO_USERS) 

668 self.assertEqual(result, True) 

669 

670 def _add_set_acl_response(self, m, share_id): 

671 url = "{0}/smb_share/{1}/set_acl".format(self._mock_url, share_id) 

672 m.post(url, status_code=204)