Coverage for manila/tests/share/drivers/infinidat/test_infinidat.py: 99%

1# Copyright 2022 Infinidat Ltd. 

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"""Unit tests for INFINIDAT InfiniBox share driver.""" 

16 

17import copy 

18import functools 

19import itertools 

20from unittest import mock 

21 

22import ddt 

23from oslo_utils import units 

24 

25from manila.common import constants 

26from manila import exception 

27from manila.share import configuration 

28from manila.share.drivers.infinidat import infinibox 

29from manila import test 

30from manila import version 

31 

32 

33_MOCK_SHARE_ID = 1 

34_MOCK_SNAPSHOT_ID = 2 

35_MOCK_CLONE_ID = 3 

36 

37_MOCK_SHARE_SIZE = 4 

38 

39_MOCK_NETWORK_SPACE_IP_1 = '1.2.3.4' 

40_MOCK_NETWORK_SPACE_IP_2 = '1.2.3.5' 

41 

42 

43def _create_mock__getitem__(mock): 

44 def mock__getitem__(self, key, default=None): 

45 return getattr(mock, key, default) 

46 return mock__getitem__ 

47 

48 

49test_share = mock.Mock(id=_MOCK_SHARE_ID, size=_MOCK_SHARE_SIZE, 

50 share_proto='NFS') 

51test_share.__getitem__ = _create_mock__getitem__(test_share) 

52 

53test_snapshot = mock.Mock(id=_MOCK_SNAPSHOT_ID, size=test_share.size, 

54 share=test_share, share_proto='NFS') 

55test_snapshot.__getitem__ = _create_mock__getitem__(test_snapshot) 

56 

57original_test_clone = mock.Mock(id=_MOCK_CLONE_ID, size=test_share.size, 

58 share=test_snapshot, share_proto='NFS') 

59original_test_clone.__getitem__ = _create_mock__getitem__(original_test_clone) 

60 

61 

62def skip_driver_setup(func): 

63 @functools.wraps(func) 

64 def f(*args, **kwargs): 

65 return func(*args, **kwargs) 

66 f.__skip_driver_setup = True 

67 return f 

68 

69 

70class FakeInfinisdkException(Exception): 

71 pass 

72 

73 

74class FakeInfinisdkPermission(object): 

75 def __init__(self, permission): 

76 self._permission = permission 

77 

78 def __getattr__(self, attr): 

79 return self._permission[attr] 

80 

81 def __getitem__(self, key): 

82 return self._permission[key] 

83 

84 

85class InfiniboxDriverTestCaseBase(test.TestCase): 

86 def _test_skips_driver_setup(self): 

87 test_method_name = self.id().split('.')[-1] 

88 test_method = getattr(self, test_method_name) 

89 return getattr(test_method, '__skip_driver_setup', False) 

90 

91 def setUp(self): 

92 super(InfiniboxDriverTestCaseBase, self).setUp() 

93 

94 self.configuration = configuration.Configuration(None) 

95 self.configuration.append_config_values( 

96 infinibox.infinidat_connection_opts) 

97 self.configuration.append_config_values( 

98 infinibox.infinidat_auth_opts) 

99 self.configuration.append_config_values( 

100 infinibox.infinidat_general_opts) 

101 self.override_config('infinibox_hostname', 'mockbox') 

102 self.override_config('infinidat_pool_name', 'mockpool') 

103 self.override_config('infinidat_nas_network_space_name', 'mockspace') 

104 self.override_config('infinidat_thin_provision', True) 

105 self.override_config('infinibox_login', 'user') 

106 self.override_config('infinibox_password', 'pass') 

107 self.override_config('infinidat_use_ssl', False) 

108 self.override_config('infinidat_suppress_ssl_warnings', False) 

109 self.override_config('network_config_group', 

110 'test_network_config_group') 

111 self.override_config('admin_network_config_group', 

112 'test_admin_network_config_group') 

113 self.override_config('reserved_share_percentage', 0) 

114 self.override_config('reserved_share_from_snapshot_percentage', 0) 

115 self.override_config('reserved_share_extend_percentage', 0) 

116 self.override_config('filter_function', None) 

117 self.override_config('goodness_function', None) 

118 self.override_config('driver_handles_share_servers', False) 

119 self.override_config('max_over_subscription_ratio', 2) 

120 self.override_config('infinidat_snapdir_accessible', True) 

121 self.override_config('infinidat_snapdir_visible', False) 

122 

123 self.driver = infinibox.InfiniboxShareDriver( 

124 configuration=self.configuration) 

125 

126 # mock external library dependencies 

127 infinisdk = self._patch( 

128 "manila.share.drivers.infinidat.infinibox.infinisdk") 

129 self._capacity_module = ( 

130 self._patch("manila.share.drivers.infinidat.infinibox.capacity")) 

131 self._capacity_module.byte = 1 

132 self._capacity_module.GiB = units.Gi 

133 

134 self._system = self._infinibox_mock() 

135 

136 infinisdk.core.exceptions.InfiniSDKException = FakeInfinisdkException 

137 infinisdk.InfiniBox.return_value = self._system 

138 

139 if not self._test_skips_driver_setup(): 

140 self.driver.do_setup(None) 

141 

142 def _infinibox_mock(self): 

143 result = mock.Mock() 

144 

145 self._mock_export_permissions = [] 

146 

147 self._mock_export = mock.Mock() 

148 self._mock_export.get_export_path.return_value = '/mock_export' 

149 self._mock_export.get_permissions = self._fake_get_permissions 

150 self._mock_export.update_permissions = self._fake_update_permissions 

151 

152 self._mock_filesystem = mock.Mock() 

153 self._mock_filesystem.has_children.return_value = False 

154 self._mock_filesystem.create_snapshot.return_value = ( 

155 self._mock_filesystem) 

156 self._mock_filesystem.get_exports.return_value = [self._mock_export, ] 

157 

158 self._mock_filesystem.size = 4 * self._capacity_module.GiB 

159 self._mock_filesystem.get_size.return_value = ( 

160 self._mock_filesystem.size) 

161 

162 self._mock_pool = mock.Mock() 

163 self._mock_pool.get_free_physical_capacity.return_value = units.Gi 

164 self._mock_pool.get_physical_capacity.return_value = units.Gi 

165 self._mock_pool.get_virtual_capacity.return_value = units.Gi 

166 self._mock_pool.get_free_virtual_capacity.return_value = units.Gi 

167 

168 self._mock_network_space = mock.Mock() 

169 self._mock_network_space.get_ips.return_value = ( 

170 [mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_1, enabled=True), 

171 mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_2, enabled=True)]) 

172 

173 result.network_spaces.safe_get.return_value = self._mock_network_space 

174 result.pools.safe_get.return_value = self._mock_pool 

175 result.filesystems.safe_get.return_value = self._mock_filesystem 

176 result.filesystems.create.return_value = self._mock_filesystem 

177 result.components.nodes.get_all.return_value = [] 

178 return result 

179 

180 def _raise_infinisdk(self, *args, **kwargs): 

181 raise FakeInfinisdkException() 

182 

183 def _fake_get_permissions(self): 

184 return self._mock_export_permissions 

185 

186 def _fake_update_permissions(self, new_export_permissions): 

187 self._mock_export_permissions = [ 

188 FakeInfinisdkPermission(permission) for permission in 

189 new_export_permissions] 

190 

191 def _patch(self, path, *args, **kwargs): 

192 patcher = mock.patch(path, *args, **kwargs) 

193 result = patcher.start() 

194 self.addCleanup(patcher.stop) 

195 return result 

196 

197 

198@ddt.ddt 

199class InfiniboxDriverTestCase(InfiniboxDriverTestCaseBase): 

200 def _generate_mock_metadata(self, share): 

201 return {"system": "openstack", 

202 "openstack_version": version.version_info.release_string(), 

203 "manila_id": share['id'], 

204 "manila_name": share['name'], 

205 "host.created_by": infinibox._INFINIDAT_MANILA_IDENTIFIER} 

206 

207 def _validate_metadata(self, share): 

208 self._mock_filesystem.set_metadata_from_dict.assert_called_once_with( 

209 self._generate_mock_metadata(share)) 

210 

211 @mock.patch("manila.share.drivers.infinidat.infinibox.infinisdk", None) 

212 def test_no_infinisdk_module(self): 

213 self.assertRaises(exception.ManilaException, 

214 self.driver.do_setup, None) 

215 

216 @mock.patch("manila.share.drivers.infinidat.infinibox.capacity", None) 

217 def test_no_capacity_module(self): 

218 self.assertRaises(exception.ManilaException, 

219 self.driver.do_setup, None) 

220 

221 def test_no_auth_parameters(self): 

222 self.override_config('infinibox_login', None) 

223 self.override_config('infinibox_password', None) 

224 self.assertRaises(exception.BadConfigurationException, 

225 self.driver.do_setup, None) 

226 

227 def test_empty_auth_parameters(self): 

228 self.override_config('infinibox_login', '') 

229 self.override_config('infinibox_password', '') 

230 self.assertRaises(exception.BadConfigurationException, 

231 self.driver.do_setup, None) 

232 

233 @skip_driver_setup 

234 def test__setup_and_get_system_object(self): 

235 # This test should skip the driver setup, as it generates more calls to 

236 # the add_auto_retry, set_source_identifier and login methods: 

237 auth = (self.configuration.infinibox_login, 

238 self.configuration.infinibox_password) 

239 

240 self.driver._setup_and_get_system_object( 

241 self.configuration.infinibox_hostname, auth, 

242 self.configuration.infinidat_use_ssl) 

243 

244 self._system.api.add_auto_retry.assert_called_once() 

245 self._system.api.set_source_identifier.assert_called_once_with( 

246 infinibox._INFINIDAT_MANILA_IDENTIFIER) 

247 self._system.login.assert_called_once() 

248 

249 @skip_driver_setup 

250 @mock.patch('manila.share.drivers.infinidat.infinibox.' 

251 'infinisdk.InfiniBox') 

252 @mock.patch('requests.packages.urllib3') 

253 def test_do_setup_ssl_enabled(self, urllib3, infinibox): 

254 self.override_config('infinidat_use_ssl', True) 

255 self.override_config('infinidat_suppress_ssl_warnings', True) 

256 auth = (self.configuration.infinibox_login, 

257 self.configuration.infinibox_password) 

258 self.driver.do_setup(None) 

259 expected = [ 

260 mock.call(urllib3.exceptions.InsecureRequestWarning), 

261 mock.call(urllib3.exceptions.InsecurePlatformWarning)] 

262 urllib3.disable_warnings.assert_has_calls(expected) 

263 infinibox.assert_called_once_with( 

264 self.configuration.infinibox_hostname, auth=auth, 

265 use_ssl=self.configuration.infinidat_use_ssl) 

266 

267 def test_get_share_stats_refreshes(self): 

268 self.driver._update_share_stats() 

269 result = self.driver.get_share_stats() 

270 self.assertEqual(1, result["free_capacity_gb"]) 

271 # change the "free space" in the pool 

272 self._mock_pool.get_free_physical_capacity.return_value = 0 

273 # no refresh - free capacity should stay the same 

274 result = self.driver.get_share_stats(refresh=False) 

275 self.assertEqual(1, result["free_capacity_gb"]) 

276 # refresh - free capacity should change to 0 

277 result = self.driver.get_share_stats(refresh=True) 

278 self.assertEqual(0, result["free_capacity_gb"]) 

279 

280 def test_get_share_stats_pool_not_found(self): 

281 self._system.pools.safe_get.return_value = None 

282 self.assertRaises(exception.ManilaException, 

283 self.driver._update_share_stats) 

284 

285 def test__verify_share_protocol(self): 

286 # test_share is NFS by default: 

287 self.driver._verify_share_protocol(test_share) 

288 

289 def test__verify_share_protocol_fails_for_non_nfs(self): 

290 # set test_share protocol for non-NFS (CIFS, for that matter) and see 

291 # that we fail: 

292 cifs_share = copy.deepcopy(test_share) 

293 cifs_share.share_proto = 'CIFS' 

294 # also need to re-define getitem, otherwise we'll get attributes from 

295 # test_share: 

296 cifs_share.__getitem__ = _create_mock__getitem__(cifs_share) 

297 self.assertRaises(exception.InvalidShare, 

298 self.driver._verify_share_protocol, cifs_share) 

299 

300 def test__verify_access_type_ip(self): 

301 self.assertTrue(self.driver._verify_access_type({'access_type': 'ip'})) 

302 

303 def test__verify_access_type_fails_for_type_user(self): 

304 self.assertRaises( 

305 exception.InvalidShareAccess, self.driver._verify_access_type, 

306 {'access_type': 'user'}) 

307 

308 def test__verify_access_type_fails_for_type_cert(self): 

309 self.assertRaises( 

310 exception.InvalidShareAccess, self.driver._verify_access_type, 

311 {'access_type': 'cert'}) 

312 

313 def test__get_ip_address_range_single_ip(self): 

314 ip_address = self.driver._get_ip_address_range('1.2.3.4') 

315 self.assertEqual('1.2.3.4', ip_address) 

316 

317 def test__get_ip_address_range_ip_range(self): 

318 ip_address_range = self.driver._get_ip_address_range('5.6.7.8/28') 

319 self.assertEqual('5.6.7.1-5.6.7.14', ip_address_range) 

320 

321 def test__get_ip_address_range_invalid_address(self): 

322 self.assertRaises(ValueError, self.driver._get_ip_address_range, 

323 'invalid') 

324 

325 def test__get_infinidat_pool(self): 

326 self.driver._get_infinidat_pool() 

327 self._system.pools.safe_get.assert_called_once() 

328 

329 def test__get_infinidat_pool_no_pools(self): 

330 self._system.pools.safe_get.return_value = None 

331 self.assertRaises(exception.ShareBackendException, 

332 self.driver._get_infinidat_pool) 

333 

334 def test__get_infinidat_pool_api_error(self): 

335 self._system.pools.safe_get.side_effect = ( 

336 self._raise_infinisdk) 

337 self.assertRaises(exception.ShareBackendException, 

338 self.driver._get_infinidat_pool) 

339 

340 def test__get_infinidat_nas_network_space_ips(self): 

341 network_space_ips = self.driver._get_infinidat_nas_network_space_ips() 

342 self._system.network_spaces.safe_get.assert_called_once() 

343 self._mock_network_space.get_ips.assert_called_once() 

344 

345 for network_space_ip in \ 

346 (_MOCK_NETWORK_SPACE_IP_1, _MOCK_NETWORK_SPACE_IP_2): 

347 self.assertIn(network_space_ip, network_space_ips) 

348 

349 def test__get_infinidat_nas_network_space_ips_only_one_ip_enabled(self): 

350 self._mock_network_space.get_ips.return_value = ( 

351 [mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_1, enabled=True), 

352 mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_2, enabled=False)]) 

353 self.assertEqual([_MOCK_NETWORK_SPACE_IP_1], 

354 self.driver._get_infinidat_nas_network_space_ips()) 

355 

356 def test__get_infinidat_nas_network_space_ips_no_network_space(self): 

357 self._system.network_spaces.safe_get.return_value = None 

358 self.assertRaises(exception.ShareBackendException, 

359 self.driver._get_infinidat_nas_network_space_ips) 

360 

361 def test__get_infinidat_nas_network_space_ips_no_ips(self): 

362 self._mock_network_space.get_ips.return_value = [] 

363 self.assertRaises(exception.ShareBackendException, 

364 self.driver._get_infinidat_nas_network_space_ips) 

365 

366 def test__get_infinidat_nas_network_space_ips_no_ips_enabled(self): 

367 self._mock_network_space.get_ips.return_value = ( 

368 [mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_1, enabled=False), 

369 mock.Mock(ip_address=_MOCK_NETWORK_SPACE_IP_2, enabled=False)]) 

370 self.assertRaises(exception.ShareBackendException, 

371 self.driver._get_infinidat_nas_network_space_ips) 

372 

373 def test__get_infinidat_nas_network_space_api_error(self): 

374 self._system.network_spaces.safe_get.side_effect = ( 

375 self._raise_infinisdk) 

376 self.assertRaises(exception.ShareBackendException, 

377 self.driver._get_infinidat_nas_network_space_ips) 

378 

379 def test__get_full_nfs_export_paths(self): 

380 export_paths = self.driver._get_full_nfs_export_paths( 

381 self._mock_export.get_export_path()) 

382 for network_space_ip in \ 

383 (_MOCK_NETWORK_SPACE_IP_1, _MOCK_NETWORK_SPACE_IP_2): 

384 self.assertIn( 

385 "{network_space_ip}:{export_path}".format( 

386 network_space_ip=network_space_ip, 

387 export_path=self._mock_export.get_export_path()), 

388 export_paths) 

389 

390 def test__get_export(self): 

391 # The default return value of get_exports is [mock_export, ]: 

392 export = self.driver._get_export(self._mock_filesystem) 

393 self._mock_filesystem.get_exports.assert_called_once() 

394 self.assertEqual(self._mock_export, export) 

395 

396 def test__get_export_no_filesystem_exports(self): 

397 self._mock_filesystem.get_exports.return_value = [] 

398 self.assertRaises(exception.ShareBackendException, 

399 self.driver._get_export, self._mock_filesystem) 

400 

401 def test__get_export_too_many_filesystem_exports(self): 

402 self._mock_filesystem.get_exports.return_value = [ 

403 self._mock_export, self._mock_export] 

404 self.assertRaises(exception.ShareBackendException, 

405 self.driver._get_export, self._mock_filesystem) 

406 

407 def test__get_export_api_error(self): 

408 self._mock_filesystem.get_exports.side_effect = self._raise_infinisdk 

409 self.assertRaises(exception.ShareBackendException, 

410 self.driver._get_export, self._mock_filesystem) 

411 

412 def test__get_infinidat_access_level_rw(self): 

413 access_level = ( 

414 self.driver._get_infinidat_access_level( 

415 {'access_level': constants.ACCESS_LEVEL_RW})) 

416 self.assertEqual('RW', access_level) 

417 

418 def test__get_infinidat_access_level_ro(self): 

419 access_level = ( 

420 self.driver._get_infinidat_access_level( 

421 {'access_level': constants.ACCESS_LEVEL_RO})) 

422 self.assertEqual('RO', access_level) 

423 

424 def test__get_infinidat_access_level_fails_for_invalid_level(self): 

425 self.assertRaises(exception.InvalidShareAccessLevel, 

426 self.driver._get_infinidat_access_level, 

427 {'access_level': 'invalid'}) 

428 

429 @ddt.data(*itertools.product((True, False), (True, False), (True, False))) 

430 @ddt.unpack 

431 def test_create_share(self, thin_provision, snapdir_accessible, 

432 snapdir_visible): 

433 self.override_config('infinidat_thin_provision', thin_provision) 

434 self.override_config('infinidat_snapdir_accessible', 

435 snapdir_accessible) 

436 self.override_config('infinidat_snapdir_visible', snapdir_visible) 

437 if thin_provision: 

438 provtype = 'THIN' 

439 else: 

440 provtype = 'THICK' 

441 self.driver.do_setup(None) 

442 self.driver.create_share(None, test_share) 

443 share_name = self.driver._make_share_name(test_share) 

444 share_size = test_share.size * self._capacity_module.GiB 

445 self._system.filesystems.create.assert_called_once_with( 

446 pool=self._mock_pool, name=share_name, size=share_size, 

447 provtype=provtype, snapdir_accessible=snapdir_accessible) 

448 self._validate_metadata(test_share) 

449 self._mock_filesystem.add_export.assert_called_once_with( 

450 permissions=[], snapdir_visible=snapdir_visible) 

451 

452 def test_create_share_pool_not_found(self): 

453 self._system.pools.safe_get.return_value = None 

454 self.assertRaises(exception.ManilaException, 

455 self.driver.create_share, None, test_share) 

456 

457 def test_create_share_fails_non_nfs(self): 

458 # set test_share protocol for non-NFS (CIFS, for that matter) and see 

459 # that we fail: 

460 cifs_share = copy.deepcopy(test_share) 

461 cifs_share.share_proto = 'CIFS' 

462 # also need to re-define getitem, otherwise we'll get attributes from 

463 # test_share: 

464 cifs_share.__getitem__ = _create_mock__getitem__(cifs_share) 

465 self.assertRaises(exception.InvalidShare, 

466 self.driver.create_share, None, cifs_share) 

467 

468 def test_create_share_pools_api_fail(self): 

469 # will fail when trying to get pool for share creation: 

470 self._system.pools.safe_get.side_effect = self._raise_infinisdk 

471 self.assertRaises(exception.ShareBackendException, 

472 self.driver.create_share, None, test_share) 

473 

474 def test_create_share_network_spaces_api_fail(self): 

475 # will fail when trying to get full export path to the new share: 

476 self._system.network_spaces.safe_get.side_effect = ( 

477 self._raise_infinisdk) 

478 self.assertRaises(exception.ShareBackendException, 

479 self.driver.create_share, None, test_share) 

480 

481 def test_delete_share(self): 

482 self.driver.delete_share(None, test_share) 

483 self._mock_filesystem.safe_delete.assert_called_once() 

484 self._mock_export.safe_delete.assert_called_once() 

485 

486 def test_delete_share_doesnt_exist(self): 

487 self._system.shares.safe_get.return_value = None 

488 # should not raise an exception 

489 self.driver.delete_share(None, test_share) 

490 

491 def test_delete_share_export_doesnt_exist(self): 

492 self._mock_filesystem.get_exports.return_value = [] 

493 # should not raise an exception 

494 self.driver.delete_share(None, test_share) 

495 

496 def test_delete_share_with_snapshots(self): 

497 # deleting a share with snapshots should succeed: 

498 self._mock_filesystem.has_children.return_value = True 

499 self.driver.delete_share(None, test_share) 

500 self._mock_filesystem.safe_delete.assert_called_once() 

501 self._mock_export.safe_delete.assert_called_once() 

502 

503 def test_delete_share_wrong_share_protocol(self): 

504 # set test_share protocol for non-NFS (CIFS, for that matter) and see 

505 # that delete_share doesn't fail: 

506 cifs_share = copy.deepcopy(test_share) 

507 cifs_share.share_proto = 'CIFS' 

508 # also need to re-define getitem, otherwise we'll get attributes from 

509 # test_share: 

510 cifs_share.__getitem__ = _create_mock__getitem__(cifs_share) 

511 self.driver.delete_share(None, cifs_share) 

512 

513 def test_extend_share(self): 

514 self.driver.extend_share(test_share, _MOCK_SHARE_SIZE * 2) 

515 self._mock_filesystem.resize.assert_called_once() 

516 

517 def test_extend_share_api_fail(self): 

518 self._mock_filesystem.resize.side_effect = self._raise_infinisdk 

519 self.assertRaises(exception.ShareBackendException, 

520 self.driver.extend_share, test_share, 8) 

521 

522 @ddt.data(*itertools.product((True, False), (True, False))) 

523 @ddt.unpack 

524 def test_create_snapshot(self, snapdir_accessible, snapdir_visible): 

525 self.override_config('infinidat_snapdir_accessible', 

526 snapdir_accessible) 

527 self.override_config('infinidat_snapdir_visible', snapdir_visible) 

528 snapshot_name = self.driver._make_snapshot_name(test_snapshot) 

529 self.driver.create_snapshot(None, test_snapshot) 

530 self._mock_filesystem.create_snapshot.assert_called_once_with( 

531 name=snapshot_name, snapdir_accessible=snapdir_accessible) 

532 self._validate_metadata(test_snapshot) 

533 self._mock_filesystem.add_export.assert_called_once_with( 

534 permissions=[], snapdir_visible=snapdir_visible) 

535 

536 def test_create_snapshot_metadata(self): 

537 self._mock_filesystem.create_snapshot.return_value = ( 

538 self._mock_filesystem) 

539 self.driver.create_snapshot(None, test_snapshot) 

540 self._validate_metadata(test_snapshot) 

541 

542 def test_create_snapshot_share_doesnt_exist(self): 

543 self._system.filesystems.safe_get.return_value = None 

544 self.assertRaises(exception.ShareResourceNotFound, 

545 self.driver.create_snapshot, None, test_snapshot) 

546 

547 def test_create_snapshot_create_snapshot_api_fail(self): 

548 # will fail when trying to create a child to the original share: 

549 self._mock_filesystem.create_snapshot.side_effect = ( 

550 self._raise_infinisdk) 

551 self.assertRaises(exception.ShareBackendException, 

552 self.driver.create_snapshot, None, test_snapshot) 

553 

554 def test_create_snapshot_network_spaces_api_fail(self): 

555 # will fail when trying to get full export path to the new snapshot: 

556 self._system.network_spaces.safe_get.side_effect = ( 

557 self._raise_infinisdk) 

558 self.assertRaises(exception.ShareBackendException, 

559 self.driver.create_snapshot, None, test_snapshot) 

560 

561 @ddt.data(*itertools.product((True, False), (True, False))) 

562 @ddt.unpack 

563 def test_create_share_from_snapshot(self, snapdir_accessible, 

564 snapdir_visible): 

565 self.override_config('infinidat_snapdir_accessible', 

566 snapdir_accessible) 

567 self.override_config('infinidat_snapdir_visible', snapdir_visible) 

568 share_name = self.driver._make_share_name(original_test_clone) 

569 self.driver.create_share_from_snapshot(None, original_test_clone, 

570 test_snapshot) 

571 self._mock_filesystem.create_snapshot.assert_called_once_with( 

572 name=share_name, write_protected=False, 

573 snapdir_accessible=snapdir_accessible) 

574 self._mock_filesystem.add_export.assert_called_once_with( 

575 permissions=[], snapdir_visible=snapdir_visible) 

576 

577 def test_create_share_from_snapshot_bigger_size(self): 

578 test_clone = copy.copy(original_test_clone) 

579 test_clone.size = test_share.size * 2 

580 # also need to re-define getitem, otherwise we'll get attributes from 

581 # original_get_clone: 

582 test_clone.__getitem__ = _create_mock__getitem__(test_clone) 

583 

584 self.driver.create_share_from_snapshot(None, test_clone, test_snapshot) 

585 

586 def test_create_share_from_snapshot_doesnt_exist(self): 

587 self._system.filesystems.safe_get.return_value = None 

588 self.assertRaises(exception.ShareSnapshotNotFound, 

589 self.driver.create_share_from_snapshot, 

590 None, original_test_clone, test_snapshot) 

591 

592 def test_create_share_from_snapshot_create_fails(self): 

593 self._mock_filesystem.create_snapshot.side_effect = ( 

594 self._raise_infinisdk) 

595 self.assertRaises(exception.ShareBackendException, 

596 self.driver.create_share_from_snapshot, 

597 None, original_test_clone, test_snapshot) 

598 

599 def test_delete_snapshot(self): 

600 self.driver.delete_snapshot(None, test_snapshot) 

601 self._mock_filesystem.safe_delete.assert_called_once() 

602 self._mock_export.safe_delete.assert_called_once() 

603 

604 def test_delete_snapshot_with_snapshots(self): 

605 # deleting a snapshot with snapshots should succeed: 

606 self._mock_filesystem.has_children.return_value = True 

607 self.driver.delete_snapshot(None, test_snapshot) 

608 self._mock_filesystem.safe_delete.assert_called_once() 

609 self._mock_export.safe_delete.assert_called_once() 

610 

611 def test_delete_snapshot_doesnt_exist(self): 

612 self._system.filesystems.safe_get.return_value = None 

613 # should not raise an exception 

614 self.driver.delete_snapshot(None, test_snapshot) 

615 

616 def test_delete_snapshot_api_fail(self): 

617 self._mock_filesystem.safe_delete.side_effect = self._raise_infinisdk 

618 self.assertRaises(exception.ShareBackendException, 

619 self.driver.delete_snapshot, None, test_snapshot) 

620 

621 @ddt.data(*itertools.product((True, False), (True, False), 

622 (True, False), (True, False))) 

623 @ddt.unpack 

624 def test_ensure_share(self, snapdir_accessible_expected, 

625 snapdir_accessible_actual, 

626 snapdir_visible_expected, 

627 snapdir_visible_actual): 

628 self.override_config('infinidat_snapdir_accessible', 

629 snapdir_accessible_expected) 

630 self.override_config('infinidat_snapdir_visible', 

631 snapdir_visible_expected) 

632 self._mock_filesystem.is_snapdir_accessible.return_value = ( 

633 snapdir_accessible_actual) 

634 self._mock_export.is_snapdir_visible.return_value = ( 

635 snapdir_visible_actual) 

636 self.driver.ensure_share(None, test_share) 

637 self._mock_filesystem.get_exports.assert_called_once() 

638 self._mock_export.get_export_path.assert_called_once() 

639 if snapdir_accessible_actual is not snapdir_accessible_expected: 

640 self._mock_filesystem.update_field.assert_called_once_with( 

641 'snapdir_accessible', snapdir_accessible_expected) 

642 else: 

643 self._mock_filesystem.update_field.assert_not_called() 

644 if snapdir_visible_actual is not snapdir_visible_expected: 

645 self._mock_export.update_snapdir_visible.assert_called_once_with( 

646 snapdir_visible_expected) 

647 else: 

648 self._mock_export.update_snapdir_visible.assert_not_called() 

649 

650 @ddt.data(True, False) 

651 def test_ensure_share_export_missing(self, snapdir_visible): 

652 self.override_config('infinidat_snapdir_visible', snapdir_visible) 

653 self._mock_filesystem.get_exports.return_value = [] 

654 self.driver.ensure_share(None, test_share) 

655 self._mock_filesystem.get_exports.assert_called_once() 

656 self._mock_filesystem.add_export.assert_called_once_with( 

657 permissions=[], snapdir_visible=snapdir_visible) 

658 

659 def test_ensure_share_share_doesnt_exist(self): 

660 self._system.filesystems.safe_get.return_value = None 

661 self.assertRaises(exception.ShareResourceNotFound, 

662 self.driver.ensure_share, None, test_share) 

663 

664 def test_ensure_share_get_exports_api_fail(self): 

665 self._mock_filesystem.get_exports.side_effect = self._raise_infinisdk 

666 self._mock_filesystem.add_export.side_effect = self._raise_infinisdk 

667 self.assertRaises(exception.ShareBackendException, 

668 self.driver.ensure_share, None, test_share) 

669 

670 def test_ensure_share_network_spaces_api_fail(self): 

671 self._system.network_spaces.safe_get.side_effect = ( 

672 self._raise_infinisdk) 

673 self.assertRaises(exception.ShareBackendException, 

674 self.driver.ensure_share, None, test_share) 

675 

676 def test_ensure_shares(self): 

677 test_shares = [test_share] 

678 test_updates = self.driver.ensure_shares(None, test_shares) 

679 self.assertEqual(len(test_shares), len(test_updates)) 

680 

681 @ddt.data(*itertools.product((True, False), (True, False))) 

682 @ddt.unpack 

683 def test_get_backend_info(self, snapdir_accessible, snapdir_visible): 

684 self.override_config('infinidat_snapdir_accessible', 

685 snapdir_accessible) 

686 self.override_config('infinidat_snapdir_visible', 

687 snapdir_visible) 

688 expected = { 

689 'snapdir_accessible': snapdir_accessible, 

690 'snapdir_visible': snapdir_visible 

691 } 

692 result = self.driver.get_backend_info(None) 

693 self.assertEqual(expected, result) 

694 

695 def test_get_network_allocations_number(self): 

696 # Mostly to increase test coverage. The return value should always be 0 

697 # for our driver (see method documentation in base class code): 

698 self.assertEqual(0, self.driver.get_network_allocations_number()) 

699 

700 def test_revert_to_snapshot(self): 

701 self.driver.revert_to_snapshot(None, test_snapshot, [], []) 

702 self._mock_filesystem.restore.assert_called_once() 

703 

704 def test_revert_to_snapshot_snapshot_doesnt_exist(self): 

705 self._system.filesystems.safe_get.return_value = None 

706 self.assertRaises(exception.ShareSnapshotNotFound, 

707 self.driver.revert_to_snapshot, None, test_snapshot, 

708 [], []) 

709 

710 def test_revert_to_snapshot_api_fail(self): 

711 self._mock_filesystem.restore.side_effect = self._raise_infinisdk 

712 self.assertRaises(exception.ShareBackendException, 

713 self.driver.revert_to_snapshot, None, test_snapshot, 

714 [], []) 

715 

716 def test_update_access(self): 

717 access_rules = [ 

718 {'access_level': constants.ACCESS_LEVEL_RO, 

719 'access_to': '1.2.3.4', 

720 'access_type': 'ip'}, 

721 {'access_level': constants.ACCESS_LEVEL_RW, 

722 'access_to': '1.2.3.5', 

723 'access_type': 'ip'}, 

724 {'access_level': constants.ACCESS_LEVEL_RO, 

725 'access_to': '5.6.7.8/28', 

726 'access_type': 'ip'}] 

727 self.driver.update_access(None, test_share, access_rules, [], [], []) 

728 

729 permissions = self._mock_filesystem.get_exports()[0].get_permissions() 

730 # now we are supposed to have three permissions: 

731 # 1. for 1.2.3.4 

732 # 2. for 1.2.3.5