Coverage for manila/tests/share/drivers/inspur/as13000/test_as13000_nas.py: 99%

1# Copyright 2018 Inspur Corp. 

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 

16""" 

17Share driver test for Inspur AS13000 

18""" 

19 

20import json 

21import time 

22from unittest import mock 

23 

24import ddt 

25from oslo_config import cfg 

26import requests 

27 

28from manila import context 

29from manila import exception 

30from manila.share import driver 

31from manila.share.drivers.inspur.as13000 import as13000_nas 

32from manila import test 

33from manila.tests import fake_share 

34 

35CONF = cfg.CONF 

36 

37 

38class FakeConfig(object): 

39 def __init__(self, *args, **kwargs): 

40 self.driver_handles_share_servers = False 

41 self.share_driver = 'fake_share_driver_name' 

42 self.share_backend_name = 'fake_as13000' 

43 self.as13000_nas_ip = kwargs.get( 

44 'as13000_nas_ip', 'some_ip') 

45 self.as13000_nas_port = kwargs.get( 

46 'as13000_nas_port', 'some_port') 

47 self.as13000_nas_login = kwargs.get( 

48 'as13000_nas_login', 'username') 

49 self.as13000_nas_password = kwargs.get( 

50 'as13000_nas_password', 'password') 

51 self.as13000_share_pools = kwargs.get( 

52 'as13000_share_pools', ['fakepool']) 

53 self.as13000_token_available_time = kwargs.get( 

54 'as13000_token_available_time', 3600) 

55 self.network_config_group = kwargs.get( 

56 "network_config_group", "fake_network_config_group") 

57 self.admin_network_config_group = kwargs.get( 

58 "admin_network_config_group", "fake_admin_network_config_group") 

59 self.config_group = kwargs.get("config_group", "fake_config_group") 

60 self.reserved_share_percentage = kwargs.get( 

61 "reserved_share_percentage", 0) 

62 self.reserved_share_from_snapshot_percentage = kwargs.get( 

63 "reserved_share_from_snapshot_percentage", 0) 

64 self.reserved_share_extend_percentage = kwargs.get( 

65 "reserved_share_extend_percentage", 0) 

66 self.max_over_subscription_ratio = kwargs.get( 

67 "max_over_subscription_ratio", 20.0) 

68 self.filter_function = kwargs.get("filter_function", None) 

69 self.goodness_function = kwargs.get("goodness_function", None) 

70 

71 def safe_get(self, key): 

72 return getattr(self, key) 

73 

74 def append_config_values(self, *args, **kwargs): 

75 pass 

76 

77 

78test_config = FakeConfig() 

79 

80 

81class FakeResponse(object): 

82 def __init__(self, status, output): 

83 self.status_code = status 

84 self.text = 'return message' 

85 self._json = output 

86 

87 def json(self): 

88 return self._json 

89 

90 def close(self): 

91 pass 

92 

93 

94@ddt.ddt 

95class RestAPIExecutorTestCase(test.TestCase): 

96 def setUp(self): 

97 self.rest_api = as13000_nas.RestAPIExecutor( 

98 test_config.as13000_nas_ip, 

99 test_config.as13000_nas_port, 

100 test_config.as13000_nas_login, 

101 test_config.as13000_nas_password) 

102 super(RestAPIExecutorTestCase, self).setUp() 

103 

104 def test_logins(self): 

105 mock_login = self.mock_object(self.rest_api, 'login', 

106 mock.Mock(return_value='fake_token')) 

107 self.rest_api.logins() 

108 mock_login.assert_called_once() 

109 

110 def test_login(self): 

111 fake_response = { 

112 'token': 'fake_token', 

113 'expireTime': '7200', 

114 'type': 0} 

115 mock_sra = self.mock_object(self.rest_api, 'send_rest_api', 

116 mock.Mock(return_value=fake_response)) 

117 result = self.rest_api.login() 

118 

119 self.assertEqual('fake_token', result) 

120 

121 login_params = {'name': test_config.as13000_nas_login, 

122 'password': test_config.as13000_nas_password} 

123 mock_sra.assert_called_once_with(method='security/token', 

124 params=login_params, 

125 request_type='post') 

126 

127 def test_logout(self): 

128 mock_sra = self.mock_object(self.rest_api, 'send_rest_api', 

129 mock.Mock(return_value=None)) 

130 self.rest_api.logout() 

131 mock_sra.assert_called_once_with( 

132 method='security/token', request_type='delete') 

133 

134 @ddt.data(True, False) 

135 def test_refresh_token(self, force): 

136 mock_login = self.mock_object(self.rest_api, 'login', 

137 mock.Mock(return_value='fake_token')) 

138 mock_logout = self.mock_object(self.rest_api, 'logout', 

139 mock.Mock()) 

140 self.rest_api.refresh_token(force) 

141 if force is not True: 

142 mock_logout.assert_called_once_with() 

143 mock_login.assert_called_once_with() 

144 

145 def test_send_rest_api(self): 

146 expected = {'value': 'abc'} 

147 mock_sa = self.mock_object(self.rest_api, 'send_api', 

148 mock.Mock(return_value=expected)) 

149 result = self.rest_api.send_rest_api( 

150 method='fake_method', 

151 params='fake_params', 

152 request_type='fake_type') 

153 self.assertEqual(expected, result) 

154 mock_sa.assert_called_once_with( 

155 'fake_method', 

156 'fake_params', 

157 'fake_type') 

158 

159 @mock.patch('time.sleep') 

160 def test_send_rest_api_retry(self, mock_sleep): 

161 expected = {'value': 'abc'} 

162 mock_sa = self.mock_object( 

163 self.rest_api, 

164 'send_api', 

165 mock.Mock( 

166 side_effect=( 

167 exception.NetworkException, 

168 expected))) 

169 # mock.Mock(side_effect=exception.NetworkException)) 

170 mock_rt = self.mock_object(self.rest_api, 'refresh_token', mock.Mock()) 

171 result = self.rest_api.send_rest_api( 

172 method='fake_method', 

173 params='fake_params', 

174 request_type='fake_type' 

175 ) 

176 self.assertEqual(expected, result) 

177 

178 mock_sa.assert_called_with( 

179 'fake_method', 

180 'fake_params', 

181 'fake_type') 

182 mock_rt.assert_called_with(force=True) 

183 mock_sleep.assert_called_once_with(1) 

184 

185 @mock.patch('time.sleep') 

186 def test_send_rest_api_3times_fail(self, mock_sleep): 

187 mock_sa = self.mock_object( 

188 self.rest_api, 'send_api', mock.Mock( 

189 side_effect=(exception.NetworkException))) 

190 mock_rt = self.mock_object(self.rest_api, 'refresh_token', mock.Mock()) 

191 self.assertRaises( 

192 exception.ShareBackendException, 

193 self.rest_api.send_rest_api, 

194 method='fake_method', 

195 params='fake_params', 

196 request_type='fake_type') 

197 mock_sa.assert_called_with('fake_method', 

198 'fake_params', 

199 'fake_type') 

200 mock_rt.assert_called_with(force=True) 

201 # Should sleep 3 times (once for each retry) 

202 self.assertEqual(3, mock_sleep.call_count) 

203 mock_sleep.assert_has_calls([mock.call(1), mock.call(1), mock.call(1)]) 

204 

205 def test_send_rest_api_backend_error_fail(self): 

206 mock_sa = self.mock_object(self.rest_api, 'send_api', mock.Mock( 

207 side_effect=(exception.ShareBackendException( 

208 'fake_error_message')))) 

209 mock_rt = self.mock_object(self.rest_api, 'refresh_token') 

210 self.assertRaises( 

211 exception.ShareBackendException, 

212 self.rest_api.send_rest_api, 

213 method='fake_method', 

214 params='fake_params', 

215 request_type='fake_type') 

216 mock_sa.assert_called_with('fake_method', 

217 'fake_params', 

218 'fake_type') 

219 mock_rt.assert_not_called() 

220 

221 @ddt.data( 

222 {'method': 'fake_method', 'request_type': 'post', 'params': 

223 {'fake_param': 'fake_value'}}, 

224 {'method': 'fake_method', 'request_type': 'get', 'params': 

225 {'fake_param': 'fake_value'}}, 

226 {'method': 'fake_method', 'request_type': 'delete', 'params': 

227 {'fake_param': 'fake_value'}}, 

228 {'method': 'fake_method', 'request_type': 'put', 'params': 

229 {'fake_param': 'fake_value'}}, ) 

230 @ddt.unpack 

231 def test_send_api(self, method, params, request_type): 

232 self.rest_api._token_pool = ['fake_token'] 

233 if request_type in ('post', 'delete', 'put'): 

234 fake_output = {'code': 0, 'message': 'success'} 

235 elif request_type == 'get': 235 ↛ 238line 235 didn't jump to line 238 because the condition on line 235 was always true

236 fake_output = {'code': 0, 'data': 'fake_date'} 

237 

238 fake_response = FakeResponse(200, fake_output) 

239 mock_request = self.mock_object(requests, 

240 request_type, 

241 mock.Mock(return_value=fake_response)) 

242 self.rest_api.send_api(method, 

243 params=params, 

244 request_type=request_type) 

245 

246 url = 'http://%s:%s/rest/%s' % (test_config.as13000_nas_ip, 

247 test_config.as13000_nas_port, 

248 method) 

249 headers = {'X-Auth-Token': 'fake_token'} 

250 mock_request.assert_called_once_with(url, 

251 data=json.dumps(params), 

252 headers=headers) 

253 

254 @ddt.data({'method': r'security/token', 

255 'params': {'name': test_config.as13000_nas_login, 

256 'password': test_config.as13000_nas_password}, 

257 'request_type': 'post'}, 

258 {'method': r'security/token', 

259 'params': None, 

260 'request_type': 'delete'}) 

261 @ddt.unpack 

262 def test_send_api_access_success(self, method, params, request_type): 

263 if request_type == 'post': 

264 fake_value = {'code': 0, 'data': { 

265 'token': 'fake_token', 

266 'expireTime': '7200', 

267 'type': 0}} 

268 mock_requests = self.mock_object( 

269 requests, 'post', mock.Mock( 

270 return_value=FakeResponse( 

271 200, fake_value))) 

272 result = self.rest_api.send_api(method, params, request_type) 

273 self.assertEqual(fake_value['data'], result) 

274 mock_requests.assert_called_once_with( 

275 'http://%s:%s/rest/%s' % 

276 (test_config.as13000_nas_ip, 

277 test_config.as13000_nas_port, 

278 method), 

279 data=json.dumps(params), 

280 headers=None) 

281 if request_type == 'delete': 

282 fake_value = {'code': 0, 'message': 'Success!'} 

283 self.rest_api._token_pool = ['fake_token'] 

284 mock_requests = self.mock_object( 

285 requests, 'delete', mock.Mock( 

286 return_value=FakeResponse( 

287 200, fake_value))) 

288 self.rest_api.send_api(method, params, request_type) 

289 mock_requests.assert_called_once_with( 

290 'http://%s:%s/rest/%s' % 

291 (test_config.as13000_nas_ip, 

292 test_config.as13000_nas_port, 

293 method), 

294 data=None, 

295 headers={'X-Auth-Token': 'fake_token'}) 

296 

297 def test_send_api_wrong_access_fail(self): 

298 req_params = {'method': r'security/token', 

299 'params': {'name': test_config.as13000_nas_login, 

300 'password': 'fake_password'}, 

301 'request_type': 'post'} 

302 fake_value = {'message': ' User name or password error.', 'code': 400} 

303 mock_request = self.mock_object( 

304 requests, 'post', mock.Mock( 

305 return_value=FakeResponse( 

306 200, fake_value))) 

307 self.assertRaises( 

308 exception.ShareBackendException, 

309 self.rest_api.send_api, 

310 method=req_params['method'], 

311 params=req_params['params'], 

312 request_type=req_params['request_type']) 

313 mock_request.assert_called_once_with( 

314 'http://%s:%s/rest/%s' % 

315 (test_config.as13000_nas_ip, 

316 test_config.as13000_nas_port, 

317 req_params['method']), 

318 data=json.dumps( 

319 req_params['params']), 

320 headers=None) 

321 

322 def test_send_api_token_overtime_fail(self): 

323 self.rest_api._token_pool = ['fake_token'] 

324 fake_value = {'method': 'fake_url', 

325 'params': 'fake_params', 

326 'reuest_type': 'post'} 

327 fake_out_put = {'message': 'Unauthorized access!', 'code': 301} 

328 mock_requests = self.mock_object( 

329 requests, 'post', mock.Mock( 

330 return_value=FakeResponse( 

331 200, fake_out_put))) 

332 self.assertRaises(exception.NetworkException, 

333 self.rest_api.send_api, 

334 method='fake_url', 

335 params='fake_params', 

336 request_type='post') 

337 mock_requests.assert_called_once_with( 

338 'http://%s:%s/rest/%s' % 

339 (test_config.as13000_nas_ip, 

340 test_config.as13000_nas_port, 

341 fake_value['method']), 

342 data=json.dumps('fake_params'), 

343 headers={ 

344 'X-Auth-Token': 'fake_token'}) 

345 

346 def test_send_api_fail(self): 

347 self.rest_api._token_pool = ['fake_token'] 

348 fake_output = {'code': 100, 'message': 'fake_message'} 

349 mock_request = self.mock_object( 

350 requests, 'post', mock.Mock( 

351 return_value=FakeResponse( 

352 200, fake_output))) 

353 self.assertRaises( 

354 exception.ShareBackendException, 

355 self.rest_api.send_api, 

356 method='fake_method', 

357 params='fake_params', 

358 request_type='post') 

359 mock_request.assert_called_once_with( 

360 'http://%s:%s/rest/%s' % 

361 (test_config.as13000_nas_ip, 

362 test_config.as13000_nas_port, 

363 'fake_method'), 

364 data=json.dumps('fake_params'), 

365 headers={'X-Auth-Token': 'fake_token'} 

366 ) 

367 

368 

369@ddt.ddt 

370class AS13000ShareDriverTestCase(test.TestCase): 

371 def __init__(self, *args, **kwds): 

372 super(AS13000ShareDriverTestCase, self).__init__(*args, **kwds) 

373 self._ctxt = context.get_admin_context() 

374 self.configuration = FakeConfig() 

375 

376 def setUp(self): 

377 self.mock_object(as13000_nas.CONF, '_check_required_opts') 

378 self.driver = as13000_nas.AS13000ShareDriver( 

379 configuration=self.configuration) 

380 super(AS13000ShareDriverTestCase, self).setUp() 

381 

382 def test_do_setup(self): 

383 mock_login = self.mock_object( 

384 as13000_nas.RestAPIExecutor, 'logins', mock.Mock()) 

385 mock_vpe = self.mock_object( 

386 self.driver, 

387 '_validate_pools_exist', 

388 mock.Mock()) 

389 mock_gdd = self.mock_object( 

390 self.driver, '_get_directory_detail', mock.Mock( 

391 return_value='{}')) 

392 mock_gni = self.mock_object( 

393 self.driver, '_get_nodes_ips', mock.Mock( 

394 return_value=['fake_ips'])) 

395 self.driver.do_setup(self._ctxt) 

396 mock_login.assert_called_once() 

397 mock_vpe.assert_called_once() 

398 mock_gdd.assert_called_once_with( 

399 test_config.as13000_share_pools[0]) 

400 mock_gni.assert_called_once() 

401 

402 def test_do_setup_login_fail(self): 

403 mock_login = self.mock_object( 

404 as13000_nas.RestAPIExecutor, 'logins', mock.Mock( 

405 side_effect=exception.ShareBackendException('fake_exception'))) 

406 self.assertRaises( 

407 exception.ShareBackendException, 

408 self.driver.do_setup, 

409 self._ctxt) 

410 mock_login.assert_called_once() 

411 

412 def test_do_setup_vpe_failed(self): 

413 mock_login = self.mock_object(as13000_nas.RestAPIExecutor, 

414 'logins', mock.Mock()) 

415 side_effect = exception.InvalidInput(reason='fake_exception') 

416 mock_vpe = self.mock_object(self.driver, 

417 '_validate_pools_exist', 

418 mock.Mock(side_effect=side_effect)) 

419 self.assertRaises(exception.InvalidInput, 

420 self.driver.do_setup, 

421 self._ctxt) 

422 mock_login.assert_called_once() 

423 mock_vpe.assert_called_once() 

424 

425 def test_check_for_setup_error_base_dir_detail_failed(self): 

426 self.driver.base_dir_detail = None 

427 self.driver.ips = ['fake_ip'] 

428 self.assertRaises( 

429 exception.ShareBackendException, 

430 self.driver.check_for_setup_error) 

431 

432 def test_check_for_setup_error_node_status_fail(self): 

433 self.driver.base_dir_detail = 'fakepool' 

434 self.driver.ips = [] 

435 self.assertRaises(exception.ShareBackendException, 

436 self.driver.check_for_setup_error) 

437 

438 @ddt.data('nfs', 'cifs') 

439 def test_create_share(self, share_proto): 

440 share = fake_share.fake_share(share_proto=share_proto) 

441 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

442 mock_cd = self.mock_object(self.driver, '_create_directory', 

443 mock.Mock(return_value='/fake/path')) 

444 mock_cns = self.mock_object(self.driver, '_create_nfs_share') 

445 mock_ccs = self.mock_object(self.driver, '_create_cifs_share') 

446 mock_sdq = self.mock_object(self.driver, '_set_directory_quota') 

447 

448 self.driver.ips = ['127.0.0.1'] 

449 locations = self.driver.create_share(self._ctxt, share_instance) 

450 if share_proto == 'nfs': 

451 expect_locations = [{'path': r'127.0.0.1:/fake/path'}] 

452 self.assertEqual(locations, expect_locations) 

453 else: 

454 expect_locations = [{'path': r'\\127.0.0.1\share_fakeinstanceid'}] 

455 self.assertEqual(locations, expect_locations) 

456 

457 mock_cd.assert_called_once_with(share_name='share_fakeinstanceid', 

458 pool_name='P') 

459 

460 if share_proto == 'nfs': 

461 mock_cns.assert_called_once_with(share_path='/fake/path') 

462 elif share['share_proto'] == 'cifs': 462 ↛ 466line 462 didn't jump to line 466 because the condition on line 462 was always true

463 mock_ccs.assert_called_once_with(share_path='/fake/path', 

464 share_name='share_fakeinstanceid') 

465 

466 mock_sdq.assert_called_once_with('/fake/path', share['size']) 

467 

468 @ddt.data('nfs', 'cifs') 

469 def test_create_share_from_snapshot(self, share_proto): 

470 share = fake_share.fake_share(share_proto=share_proto) 

471 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

472 mock_cd = self.mock_object(self.driver, '_create_directory', 

473 mock.Mock(return_value='/fake/path')) 

474 mock_cns = self.mock_object(self.driver, '_create_nfs_share') 

475 mock_ccs = self.mock_object(self.driver, '_create_cifs_share') 

476 mock_sdq = self.mock_object(self.driver, '_set_directory_quota') 

477 mock_cdtd = self.mock_object(self.driver, '_clone_directory_to_dest') 

478 

479 self.driver.ips = ['127.0.0.1'] 

480 locations = self.driver.create_share_from_snapshot( 

481 self._ctxt, share_instance, None) 

482 if share_proto == 'nfs': 

483 expect_locations = [{'path': r'127.0.0.1:/fake/path'}] 

484 self.assertEqual(locations, expect_locations) 

485 else: 

486 expect_locations = [{'path': r'\\127.0.0.1\share_fakeinstanceid'}] 

487 self.assertEqual(locations, expect_locations) 

488 

489 mock_cd.assert_called_once_with(share_name='share_fakeinstanceid', 

490 pool_name='P') 

491 

492 if share_proto == 'nfs': 

493 mock_cns.assert_called_once_with(share_path='/fake/path') 

494 elif share['share_proto'] == 'cifs': 494 ↛ 498line 494 didn't jump to line 498 because the condition on line 494 was always true

495 mock_ccs.assert_called_once_with(share_path='/fake/path', 

496 share_name='share_fakeinstanceid') 

497 

498 mock_sdq.assert_called_once_with('/fake/path', share['size']) 

499 mock_cdtd.assert_called_once_with(snapshot=None, 

500 dest_path='/fake/path') 

501 

502 @ddt.data('nfs', 'cifs') 

503 def test_delete_share(self, share_proto): 

504 share = fake_share.fake_share(share_proto=share_proto) 

505 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

506 expect_share_path = r'/P/share_fakeinstanceid' 

507 

508 mock_gns = self.mock_object(self.driver, '_get_nfs_share', 

509 mock.Mock(return_value=['fake_share'])) 

510 mock_dns = self.mock_object(self.driver, '_delete_nfs_share') 

511 mock_gcs = self.mock_object(self.driver, '_get_cifs_share', 

512 mock.Mock(return_value=['fake_share'])) 

513 mock_dcs = self.mock_object(self.driver, '_delete_cifs_share') 

514 mock_dd = self.mock_object(self.driver, '_delete_directory') 

515 

516 self.driver.delete_share(self._ctxt, share_instance) 

517 if share_proto == 'nfs': 

518 mock_gns.assert_called_once_with(expect_share_path) 

519 mock_dns.assert_called_once_with(expect_share_path) 

520 else: 

521 mock_gcs.assert_called_once_with('share_fakeinstanceid') 

522 mock_dcs.assert_called_once_with('share_fakeinstanceid') 

523 

524 mock_dd.assert_called_once_with(expect_share_path) 

525 

526 @ddt.data('nfs', 'cifs') 

527 def test_delete_share_not_exist(self, share_proto): 

528 share = fake_share.fake_share(share_proto=share_proto) 

529 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

530 expect_share_path = r'/P/share_fakeinstanceid' 

531 

532 mock_gns = self.mock_object(self.driver, '_get_nfs_share', 

533 mock.Mock(return_value=[])) 

534 mock_gcs = self.mock_object(self.driver, '_get_cifs_share', 

535 mock.Mock(return_value=[])) 

536 self.driver.delete_share(self._ctxt, share_instance) 

537 if share_proto == 'nfs': 

538 mock_gns.assert_called_once_with(expect_share_path) 

539 elif share_proto == 'cifs': 539 ↛ exitline 539 didn't return from function 'test_delete_share_not_exist' because the condition on line 539 was always true

540 mock_gcs.assert_called_once_with('share_fakeinstanceid') 

541 

542 def test_extend_share(self): 

543 share = fake_share.fake_share() 

544 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

545 expect_share_path = r'/P/share_fakeinstanceid' 

546 

547 mock_sdq = self.mock_object(self.driver, '_set_directory_quota') 

548 

549 self.driver.extend_share(share_instance, 2) 

550 

551 mock_sdq.assert_called_once_with(expect_share_path, 2) 

552 

553 @ddt.data('nfs', 'cifs') 

554 def test_ensure_share(self, share_proto): 

555 share = fake_share.fake_share(share_proto=share_proto) 

556 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

557 

558 mock_gns = self.mock_object(self.driver, '_get_nfs_share', 

559 mock.Mock(return_value=['fake_share'])) 

560 mock_gcs = self.mock_object(self.driver, '_get_cifs_share', 

561 mock.Mock(return_value=['fake_share'])) 

562 

563 self.driver.ips = ['127.0.0.1'] 

564 locations = self.driver.ensure_share(self._ctxt, share_instance) 

565 if share_proto == 'nfs': 

566 expect_locations = [{'path': r'127.0.0.1:/P/share_fakeinstanceid'}] 

567 self.assertEqual(locations, expect_locations) 

568 mock_gns.assert_called_once_with(r'/P/share_fakeinstanceid') 

569 else: 

570 expect_locations = [{'path': r'\\127.0.0.1\share_fakeinstanceid'}] 

571 self.assertEqual(locations, expect_locations) 

572 mock_gcs.assert_called_once_with(r'share_fakeinstanceid') 

573 

574 def test_ensure_share_fail_1(self): 

575 share = fake_share.fake_share() 

576 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

577 

578 self.assertRaises(exception.InvalidInput, self.driver.ensure_share, 

579 self._ctxt, share_instance) 

580 

581 @ddt.data('nfs', 'cifs') 

582 def test_ensure_share_None_share_fail(self, share_proto): 

583 share = fake_share.fake_share(share_proto=share_proto) 

584 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

585 

586 mock_gns = self.mock_object(self.driver, '_get_nfs_share', 

587 mock.Mock(return_value=[])) 

588 mock_gcs = self.mock_object(self.driver, '_get_cifs_share', 

589 mock.Mock(return_value=[])) 

590 self.assertRaises(exception.ShareResourceNotFound, 

591 self.driver.ensure_share, 

592 self._ctxt, share_instance) 

593 

594 if share_proto == 'nfs': 

595 mock_gns.assert_called_once_with(r'/P/share_fakeinstanceid') 

596 elif share['share_proto'] == 'cifs': 596 ↛ exitline 596 didn't return from function 'test_ensure_share_None_share_fail' because the condition on line 596 was always true

597 mock_gcs.assert_called_once_with(r'share_fakeinstanceid') 

598 

599 def test_create_snapshot(self): 

600 share = fake_share.fake_share() 

601 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

602 snapshot_instance_pseudo = { 

603 'share': share_instance, 

604 'id': 'fakesnapid' 

605 } 

606 

607 mock_rest = self.mock_object(as13000_nas.RestAPIExecutor, 

608 'send_rest_api') 

609 

610 self.driver.create_snapshot(self._ctxt, snapshot_instance_pseudo) 

611 

612 method = 'snapshot/directory' 

613 request_type = 'post' 

614 params = {'path': r'/P/share_fakeinstanceid', 

615 'snapName': 'snap_fakesnapid'} 

616 mock_rest.assert_called_once_with(method=method, 

617 request_type=request_type, 

618 params=params) 

619 

620 def test_delete_snapshot_normal(self): 

621 share = fake_share.fake_share() 

622 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

623 snapshot_instance_pseudo = { 

624 'share': share_instance, 

625 'id': 'fakesnapid' 

626 } 

627 

628 mock_gsfs = self.mock_object(self.driver, '_get_snapshots_from_share', 

629 mock.Mock(return_value=['fakesnapshot'])) 

630 mock_rest = self.mock_object(as13000_nas.RestAPIExecutor, 

631 'send_rest_api') 

632 

633 self.driver.delete_snapshot(self._ctxt, snapshot_instance_pseudo) 

634 

635 mock_gsfs.assert_called_once_with('/P/share_fakeinstanceid') 

636 method = ('snapshot/directory?' 

637 'path=/P/share_fakeinstanceid&snapName=snap_fakesnapid') 

638 request_type = 'delete' 

639 mock_rest.assert_called_once_with(method=method, 

640 request_type=request_type) 

641 

642 def test_delete_snapshot_not_exist(self): 

643 share = fake_share.fake_share() 

644 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

645 snapshot_instance_pseudo = { 

646 'share': share_instance, 

647 'snapshot_id': 'fakesnapid' 

648 } 

649 

650 mock_gsfs = self.mock_object(self.driver, '_get_snapshots_from_share', 

651 mock.Mock(return_value=[])) 

652 mock_rest = self.mock_object(as13000_nas.RestAPIExecutor, 

653 'send_rest_api') 

654 

655 self.driver.delete_snapshot(self._ctxt, snapshot_instance_pseudo) 

656 

657 mock_gsfs.assert_called_once_with('/P/share_fakeinstanceid') 

658 mock_rest.assert_not_called() 

659 

660 @ddt.data('nfs', 'icfs', 'cifs') 

661 def test_transfer_rule_to_client(self, proto): 

662 rule = {'access_to': '1.1.1.1', 'access_level': 'rw'} 

663 

664 result = self.driver.transfer_rule_to_client(proto, rule) 

665 

666 client = {'name': '1.1.1.1', 

667 'authority': 'rwx' if proto == 'cifs' else 'rw'} 

668 

669 if proto == 'nfs': 

670 client.update({'type': 0}) 

671 else: 

672 client.update({'type': 1}) 

673 

674 self.assertEqual(client, result) 

675 

676 @ddt.data({'share_proto': 'nfs', 'use_access': True}, 

677 {'share_proto': 'nfs', 'use_access': False}, 

678 {'share_proto': 'cifs', 'use_access': True}, 

679 {'share_proto': 'cifs', 'use_access': False}) 

680 @ddt.unpack 

681 def test_update_access(self, share_proto, use_access): 

682 share = fake_share.fake_share(share_proto=share_proto) 

683 share_instance = fake_share.fake_share_instance(share, host="H@B#P") 

684 

685 access_rules = [{'access_to': 'fakename1', 

686 'access_level': 'fakelevel1'}, 

687 {'access_to': 'fakename2', 

688 'access_level': 'fakelevel2'}] 

689 add_rules = [{'access_to': 'fakename1', 'access_level': 'fakelevel1'}] 

690 del_rules = [{'access_to': 'fakename2', 'access_level': 'fakelevel2'}] 

691 

692 mock_ca = self.mock_object(self.driver, '_clear_access') 

693 

694 fake_share_backend = {'pathAuthority': 'fakepathAuthority'} 

695 mock_gns = self.mock_object(self.driver, '_get_nfs_share', 

696 mock.Mock(return_value=fake_share_backend)) 

697 mock_rest = self.mock_object(as13000_nas.RestAPIExecutor, 

698 'send_rest_api') 

699 if use_access: 

700 self.driver.update_access(self._ctxt, share_instance, 

701 access_rules, [], [], []) 

702 else: 

703 self.driver.update_access(self._ctxt, share_instance, 

704 [], add_rules, del_rules, []) 

705 

706 access_clients = [{'name': rule['access_to'], 

707 'type': 0 if share_proto == 'nfs' else 1, 

708 'authority': rule['access_level'] 

709 } for rule in access_rules] 

710 add_clients = [{'name': rule['access_to'], 

711 'type': 0 if share_proto == 'nfs' else 1, 

712 'authority': rule['access_level'] 

713 } for rule in add_rules] 

714 del_clients = [{'name': rule['access_to'], 

715 'type': 0 if share_proto == 'nfs' else 1, 

716 'authority': rule['access_level'] 

717 } for rule in del_rules] 

718 

719 params = { 

720 'path': r'/P/share_fakeinstanceid', 

721 'addedClientList': [], 

722 'deletedClientList': [], 

723 'editedClientList': [] 

724 } 

725 

726 if share_proto == 'nfs': 

727 mock_gns.assert_called_once_with(r'/P/share_fakeinstanceid') 

728 params['pathAuthority'] = fake_share_backend['pathAuthority'] 

729 else: 

730 params['name'] = 'share_fakeinstanceid' 

731 

732 if use_access: 

733 mock_ca.assert_called_once_with(share_instance) 

734 params['addedClientList'] = access_clients 

735 else: