Coverage for manila/tests/share/test_driver.py: 100%

1# Copyright 2012 NetApp 

2# Copyright 2014 Mirantis Inc. 

3# All Rights Reserved. 

4# 

5# Licensed under the Apache License, Version 2.0 (the "License"); you may 

6# not use this file except in compliance with the License. You may obtain 

7# a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, software 

12# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

13# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

14# License for the specific language governing permissions and limitations 

15# under the License. 

16"""Unit tests for the Share driver module.""" 

17 

18import time 

19from unittest import mock 

20 

21 

22import ddt 

23 

24from manila.common import constants 

25from manila import exception 

26from manila import network 

27from manila.share import configuration 

28from manila.share import driver 

29from manila import test 

30from manila.tests import utils as test_utils 

31from manila import utils 

32 

33 

34def fake_execute_with_raise(*cmd, **kwargs): 

35 raise exception.ProcessExecutionError 

36 

37 

38def fake_sleep(duration): 

39 pass 

40 

41 

42class ShareDriverWithExecuteMixin(driver.ShareDriver, driver.ExecuteMixin): 

43 pass 

44 

45 

46@ddt.ddt 

47class ShareDriverTestCase(test.TestCase): 

48 _SNAPSHOT_METHOD_NAMES = ["create_snapshot", "delete_snapshot"] 

49 

50 def setUp(self): 

51 super(ShareDriverTestCase, self).setUp() 

52 self.utils = utils 

53 self.mock_object(self.utils, 'execute', fake_execute_with_raise) 

54 self.time = time 

55 self.mock_object(self.time, 'sleep', fake_sleep) 

56 driver.CONF.set_default('driver_handles_share_servers', True) 

57 

58 def test__try_execute(self): 

59 execute_mixin = ShareDriverWithExecuteMixin( 

60 True, configuration=configuration.Configuration(None)) 

61 self.assertRaises(exception.ProcessExecutionError, 

62 execute_mixin._try_execute) 

63 

64 def test_verify_share_driver_mode_option_type(self): 

65 data = {'DEFAULT': {'driver_handles_share_servers': 'True'}} 

66 with test_utils.create_temp_config_with_opts(data): 

67 share_driver = driver.ShareDriver([True, False]) 

68 self.assertTrue(share_driver.driver_handles_share_servers) 

69 

70 def _instantiate_share_driver(self, network_config_group, 

71 driver_handles_share_servers, 

72 admin_network_config_group=None): 

73 self.mock_object(network, 'API') 

74 config = mock.Mock() 

75 config.append_config_values = mock.Mock() 

76 config.config_group = 'fake_config_group' 

77 config.network_config_group = network_config_group 

78 if admin_network_config_group: 

79 config.admin_network_config_group = admin_network_config_group 

80 config.safe_get = mock.Mock(return_value=driver_handles_share_servers) 

81 

82 share_driver = driver.ShareDriver([True, False], configuration=config) 

83 

84 self.assertTrue(hasattr(share_driver, 'configuration')) 

85 config.append_config_values.assert_called_once_with(driver.share_opts) 

86 if driver_handles_share_servers: 

87 calls = [] 

88 if network_config_group: 

89 calls.append(mock.call( 

90 config_group_name=config.network_config_group)) 

91 else: 

92 calls.append(mock.call( 

93 config_group_name=config.config_group)) 

94 if admin_network_config_group: 

95 calls.append(mock.call( 

96 config_group_name=config.admin_network_config_group, 

97 label='admin')) 

98 network.API.assert_has_calls(calls) 

99 self.assertTrue(hasattr(share_driver, 'network_api')) 

100 self.assertTrue(hasattr(share_driver, 'admin_network_api')) 

101 self.assertIsNotNone(share_driver.network_api) 

102 self.assertIsNotNone(share_driver.admin_network_api) 

103 else: 

104 self.assertFalse(hasattr(share_driver, 'network_api')) 

105 self.assertTrue(hasattr(share_driver, 'admin_network_api')) 

106 self.assertIsNone(share_driver.admin_network_api) 

107 self.assertFalse(network.API.called) 

108 return share_driver 

109 

110 def test_instantiate_share_driver(self): 

111 self._instantiate_share_driver(None, True) 

112 

113 def test_instantiate_share_driver_another_config_group(self): 

114 self._instantiate_share_driver("fake_network_config_group", True) 

115 

116 def test_instantiate_share_driver_with_admin_network(self): 

117 self._instantiate_share_driver( 

118 "fake_network_config_group", True, 

119 "fake_admin_network_config_group") 

120 

121 def test_instantiate_share_driver_no_configuration(self): 

122 self.mock_object(network, 'API') 

123 

124 share_driver = driver.ShareDriver(True, configuration=None) 

125 

126 self.assertIsNone(share_driver.configuration) 

127 network.API.assert_called_once_with(config_group_name=None) 

128 

129 def test_get_share_stats_refresh_false(self): 

130 share_driver = driver.ShareDriver(True, configuration=None) 

131 share_driver._stats = {'fake_key': 'fake_value'} 

132 

133 result = share_driver.get_share_stats(False) 

134 

135 self.assertEqual(share_driver._stats, result) 

136 

137 def test_get_share_stats_refresh_true(self): 

138 conf = configuration.Configuration(None) 

139 expected_keys = [ 

140 'qos', 'driver_version', 'share_backend_name', 

141 'free_capacity_gb', 'total_capacity_gb', 

142 'driver_handles_share_servers', 

143 'reserved_percentage', 'reserved_snapshot_percentage', 

144 'reserved_share_extend_percentage', 

145 'vendor_name', 'storage_protocol', 

146 'snapshot_support', 'mount_snapshot_support', 

147 'mount_point_name_support', 

148 ] 

149 share_driver = driver.ShareDriver(True, configuration=conf) 

150 fake_stats = {'fake_key': 'fake_value'} 

151 share_driver._stats = fake_stats 

152 

153 result = share_driver.get_share_stats(True) 

154 

155 self.assertNotEqual(fake_stats, result) 

156 for key in expected_keys: 

157 self.assertIn(key, result) 

158 self.assertEqual('Open Source', result['vendor_name']) 

159 

160 def test_get_share_status(self): 

161 share_driver = self._instantiate_share_driver(None, False) 

162 self.assertRaises(NotImplementedError, 

163 share_driver.get_share_status, 

164 None, None) 

165 

166 @ddt.data( 

167 {'opt': True, 'allowed': True}, 

168 {'opt': True, 'allowed': (True, False)}, 

169 {'opt': True, 'allowed': [True, False]}, 

170 {'opt': True, 'allowed': set([True, False])}, 

171 {'opt': False, 'allowed': False}, 

172 {'opt': False, 'allowed': (True, False)}, 

173 {'opt': False, 'allowed': [True, False]}, 

174 {'opt': False, 'allowed': set([True, False])}) 

175 @ddt.unpack 

176 def test__verify_share_server_handling_valid_cases(self, opt, allowed): 

177 conf = configuration.Configuration(None) 

178 self.mock_object(conf, 'safe_get', mock.Mock(return_value=opt)) 

179 share_driver = driver.ShareDriver(allowed, configuration=conf) 

180 self.assertTrue(conf.safe_get.called) 

181 self.assertEqual(opt, share_driver.driver_handles_share_servers) 

182 

183 @ddt.data( 

184 {'opt': False, 'allowed': True}, 

185 {'opt': True, 'allowed': False}, 

186 {'opt': None, 'allowed': True}, 

187 {'opt': 'True', 'allowed': True}, 

188 {'opt': 'False', 'allowed': False}, 

189 {'opt': [], 'allowed': True}, 

190 {'opt': True, 'allowed': []}, 

191 {'opt': True, 'allowed': ['True']}, 

192 {'opt': False, 'allowed': ['False']}) 

193 @ddt.unpack 

194 def test__verify_share_server_handling_invalid_cases(self, opt, allowed): 

195 conf = configuration.Configuration(None) 

196 self.mock_object(conf, 'safe_get', mock.Mock(return_value=opt)) 

197 self.assertRaises( 

198 exception.ManilaException, 

199 driver.ShareDriver, allowed, configuration=conf) 

200 self.assertTrue(conf.safe_get.called) 

201 

202 def test_setup_server_handling_disabled(self): 

203 share_driver = self._instantiate_share_driver(None, False) 

204 # We expect successful execution, nothing to assert 

205 share_driver.setup_server('Nothing is expected to happen.') 

206 

207 def test_setup_server_handling_enabled(self): 

208 share_driver = self._instantiate_share_driver(None, True) 

209 self.assertRaises( 

210 NotImplementedError, 

211 share_driver.setup_server, 

212 'fake_network_info') 

213 

214 def test_teardown_server_handling_disabled(self): 

215 share_driver = self._instantiate_share_driver(None, False) 

216 # We expect successful execution, nothing to assert 

217 share_driver.teardown_server('Nothing is expected to happen.') 

218 

219 def test_teardown_server_handling_enabled(self): 

220 share_driver = self._instantiate_share_driver(None, True) 

221 self.assertRaises( 

222 NotImplementedError, 

223 share_driver.teardown_server, 

224 'fake_share_server_details') 

225 

226 def _assert_is_callable(self, obj, attr): 

227 self.assertTrue(callable(getattr(obj, attr))) 

228 

229 @ddt.data('manage_existing', 

230 'unmanage') 

231 def test_drivers_methods_needed_by_manage_functionality(self, method): 

232 share_driver = self._instantiate_share_driver(None, False) 

233 

234 self._assert_is_callable(share_driver, method) 

235 

236 @ddt.data('manage_existing_snapshot', 

237 'unmanage_snapshot') 

238 def test_drivers_methods_needed_by_manage_snapshot_functionality( 

239 self, method): 

240 share_driver = self._instantiate_share_driver(None, False) 

241 

242 self._assert_is_callable(share_driver, method) 

243 

244 @ddt.data('revert_to_snapshot', 

245 'revert_to_replicated_snapshot') 

246 def test_drivers_methods_needed_by_share_revert_to_snapshot_functionality( 

247 self, method): 

248 share_driver = self._instantiate_share_driver(None, False) 

249 

250 self._assert_is_callable(share_driver, method) 

251 

252 @ddt.data(True, False) 

253 def test_get_share_server_pools(self, value): 

254 driver.CONF.set_default('driver_handles_share_servers', value) 

255 share_driver = driver.ShareDriver(value) 

256 self.assertEqual([], 

257 share_driver.get_share_server_pools('fake_server')) 

258 

259 def test_check_for_setup_error(self): 

260 driver.CONF.set_default('driver_handles_share_servers', False) 

261 share_driver = driver.ShareDriver(False) 

262 share_driver.configuration = configuration.Configuration(None) 

263 share_driver.check_for_setup_error() 

264 

265 def test_snapshot_support_exists(self): 

266 driver.CONF.set_default('driver_handles_share_servers', True) 

267 fake_method = lambda *args, **kwargs: None # noqa: E731 

268 child_methods = { 

269 "create_snapshot": fake_method, 

270 "delete_snapshot": fake_method, 

271 } 

272 child_class_instance = type( 

273 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

274 self.mock_object(child_class_instance, "configuration") 

275 

276 child_class_instance._update_share_stats() 

277 

278 self.assertTrue(child_class_instance._stats["snapshot_support"]) 

279 self.assertTrue(child_class_instance.configuration.safe_get.called) 

280 

281 @ddt.data( 

282 ([], [], False), 

283 (_SNAPSHOT_METHOD_NAMES, [], True), 

284 (_SNAPSHOT_METHOD_NAMES, _SNAPSHOT_METHOD_NAMES, True), 

285 (_SNAPSHOT_METHOD_NAMES[0:1], _SNAPSHOT_METHOD_NAMES[1:], True), 

286 ([], _SNAPSHOT_METHOD_NAMES, True), 

287 ) 

288 @ddt.unpack 

289 def test_check_redefined_driver_methods(self, common_drv_meth_names, 

290 child_drv_meth_names, 

291 expected_result): 

292 # This test covers the case of drivers inheriting other drivers or 

293 # common classes. 

294 

295 driver.CONF.set_default('driver_handles_share_servers', True) 

296 

297 common_drv_methods, child_drv_methods = [ 

298 {method_name: lambda *args, **kwargs: None # noqa: E731 

299 for method_name in method_names} 

300 for method_names in (common_drv_meth_names, 

301 child_drv_meth_names)] 

302 

303 common_drv = type( 

304 "NotRedefinedCommon", (driver.ShareDriver, ), common_drv_methods) 

305 child_drv_instance = type("NotRedefined", (common_drv, ), 

306 child_drv_methods)(True) 

307 

308 has_redefined_methods = ( 

309 child_drv_instance._has_redefined_driver_methods( 

310 self._SNAPSHOT_METHOD_NAMES)) 

311 

312 self.assertEqual(expected_result, has_redefined_methods) 

313 

314 @ddt.data( 

315 (), 

316 ("create_snapshot"), 

317 ("delete_snapshot"), 

318 ("create_snapshot", "delete_snapshotFOO"), 

319 ) 

320 def test_snapshot_support_absent(self, methods): 

321 driver.CONF.set_default('driver_handles_share_servers', True) 

322 fake_method = lambda *args, **kwargs: None # noqa: E731 

323 child_methods = {} 

324 for method in methods: 

325 child_methods[method] = fake_method 

326 child_class_instance = type( 

327 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

328 self.mock_object(child_class_instance, "configuration") 

329 

330 child_class_instance._update_share_stats() 

331 

332 self.assertFalse(child_class_instance._stats["snapshot_support"]) 

333 self.assertTrue(child_class_instance.configuration.safe_get.called) 

334 

335 @ddt.data(True, False) 

336 def test_snapshot_support_not_exists_and_set_explicitly( 

337 self, snapshots_are_supported): 

338 driver.CONF.set_default('driver_handles_share_servers', True) 

339 child_class_instance = type( 

340 "NotRedefined", (driver.ShareDriver, ), {})(True) 

341 self.mock_object(child_class_instance, "configuration") 

342 

343 child_class_instance._update_share_stats( 

344 {"snapshot_support": snapshots_are_supported}) 

345 

346 self.assertEqual( 

347 snapshots_are_supported, 

348 child_class_instance._stats["snapshot_support"]) 

349 self.assertTrue(child_class_instance.configuration.safe_get.called) 

350 

351 @ddt.data(True, False) 

352 def test_snapshot_support_exists_and_set_explicitly( 

353 self, snapshots_are_supported): 

354 driver.CONF.set_default('driver_handles_share_servers', True) 

355 fake_method = lambda *args, **kwargs: None # noqa: E731 

356 child_methods = { 

357 "create_snapshot": fake_method, 

358 "delete_snapshot": fake_method, 

359 } 

360 child_class_instance = type( 

361 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

362 self.mock_object(child_class_instance, "configuration") 

363 

364 child_class_instance._update_share_stats( 

365 {"snapshot_support": snapshots_are_supported}) 

366 

367 self.assertEqual( 

368 snapshots_are_supported, 

369 child_class_instance._stats["snapshot_support"]) 

370 self.assertTrue(child_class_instance.configuration.safe_get.called) 

371 

372 def test_create_share_from_snapshot_support_exists(self): 

373 driver.CONF.set_default('driver_handles_share_servers', True) 

374 fake_method = lambda *args, **kwargs: None # noqa: E731 

375 child_methods = { 

376 "create_share_from_snapshot": fake_method, 

377 "create_snapshot": fake_method, 

378 "delete_snapshot": fake_method, 

379 } 

380 child_class_instance = type( 

381 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

382 self.mock_object(child_class_instance, "configuration") 

383 

384 child_class_instance._update_share_stats() 

385 

386 self.assertTrue( 

387 child_class_instance._stats["create_share_from_snapshot_support"]) 

388 self.assertTrue(child_class_instance.configuration.safe_get.called) 

389 

390 @ddt.data( 

391 (), 

392 ("create_snapshot"), 

393 ("create_share_from_snapshotFOO"), 

394 ) 

395 def test_create_share_from_snapshot_support_absent(self, methods): 

396 driver.CONF.set_default('driver_handles_share_servers', True) 

397 fake_method = lambda *args, **kwargs: None # noqa: E731 

398 child_methods = {} 

399 for method in methods: 

400 child_methods[method] = fake_method 

401 child_class_instance = type( 

402 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

403 self.mock_object(child_class_instance, "configuration") 

404 

405 child_class_instance._update_share_stats() 

406 

407 self.assertFalse( 

408 child_class_instance._stats["create_share_from_snapshot_support"]) 

409 self.assertTrue(child_class_instance.configuration.safe_get.called) 

410 

411 @ddt.data(True, False) 

412 def test_create_share_from_snapshot_not_exists_and_set_explicitly( 

413 self, creating_shares_from_snapshot_is_supported): 

414 driver.CONF.set_default('driver_handles_share_servers', True) 

415 child_class_instance = type( 

416 "NotRedefined", (driver.ShareDriver, ), {})(True) 

417 self.mock_object(child_class_instance, "configuration") 

418 

419 child_class_instance._update_share_stats({ 

420 "create_share_from_snapshot_support": 

421 creating_shares_from_snapshot_is_supported, 

422 }) 

423 

424 self.assertEqual( 

425 creating_shares_from_snapshot_is_supported, 

426 child_class_instance._stats["create_share_from_snapshot_support"]) 

427 self.assertTrue(child_class_instance.configuration.safe_get.called) 

428 

429 @ddt.data(True, False) 

430 def test_create_share_from_snapshot_exists_and_set_explicitly( 

431 self, create_share_from_snapshot_supported): 

432 driver.CONF.set_default('driver_handles_share_servers', True) 

433 fake_method = lambda *args, **kwargs: None # noqa: E731 

434 child_methods = {"create_share_from_snapshot": fake_method} 

435 child_class_instance = type( 

436 "NotRedefined", (driver.ShareDriver, ), child_methods)(True) 

437 self.mock_object(child_class_instance, "configuration") 

438 

439 child_class_instance._update_share_stats({ 

440 "create_share_from_snapshot_support": 

441 create_share_from_snapshot_supported, 

442 }) 

443 

444 self.assertEqual( 

445 create_share_from_snapshot_supported, 

446 child_class_instance._stats["create_share_from_snapshot_support"]) 

447 self.assertTrue(child_class_instance.configuration.safe_get.called) 

448 

449 def test_get_periodic_hook_data(self): 

450 share_driver = self._instantiate_share_driver(None, False) 

451 share_instances = ["list", "of", "share", "instances"] 

452 

453 result = share_driver.get_periodic_hook_data( 

454 "fake_context", share_instances) 

455 

456 self.assertEqual(share_instances, result) 

457 

458 def test_get_admin_network_allocations_number(self): 

459 share_driver = self._instantiate_share_driver(None, True) 

460 

461 self.assertEqual( 

462 0, share_driver.get_admin_network_allocations_number()) 

463 

464 def test_allocate_admin_network_count_None(self): 

465 share_driver = self._instantiate_share_driver(None, True) 

466 ctxt = 'fake_context' 

467 share_server = 'fake_share_server' 

468 mock_get_admin_network_allocations_number = self.mock_object( 

469 share_driver, 

470 'get_admin_network_allocations_number', 

471 mock.Mock(return_value=0)) 

472 self.mock_object( 

473 share_driver.admin_network_api, 

474 'allocate_network', 

475 mock.Mock(side_effect=Exception('ShouldNotBeRaised'))) 

476 

477 share_driver.allocate_admin_network(ctxt, share_server) 

478 

479 mock_get_admin_network_allocations_number.assert_called_once_with() 

480 self.assertFalse( 

481 share_driver.admin_network_api.allocate_network.called) 

482 

483 def test_allocate_admin_network_count_0(self): 

484 share_driver = self._instantiate_share_driver(None, True) 

485 ctxt = 'fake_context' 

486 share_server = 'fake_share_server' 

487 self.mock_object( 

488 share_driver, 

489 'get_admin_network_allocations_number', 

490 mock.Mock(return_value=0)) 

491 self.mock_object( 

492 share_driver.admin_network_api, 

493 'allocate_network', 

494 mock.Mock(side_effect=Exception('ShouldNotBeRaised'))) 

495 

496 share_driver.allocate_admin_network(ctxt, share_server, count=0) 

497 

498 self.assertFalse( 

499 share_driver.get_admin_network_allocations_number.called) 

500 self.assertFalse( 

501 share_driver.admin_network_api.allocate_network.called) 

502 

503 def test_allocate_admin_network_count_1_api_initialized(self): 

504 share_driver = self._instantiate_share_driver(None, True) 

505 ctxt = 'fake_context' 

506 share_server = 'fake_share_server' 

507 mock_get_admin_network_allocations_number = self.mock_object( 

508 share_driver, 

509 'get_admin_network_allocations_number', 

510 mock.Mock(return_value=1)) 

511 self.mock_object( 

512 share_driver.admin_network_api, 

513 'allocate_network', 

514 mock.Mock()) 

515 

516 share_driver.allocate_admin_network(ctxt, share_server) 

517 

518 mock_get_admin_network_allocations_number.assert_called_once_with() 

519 (share_driver.admin_network_api.allocate_network. 

520 assert_called_once_with(ctxt, share_server, count=1)) 

521 

522 def test_allocate_admin_network_count_1_api_not_initialized(self): 

523 share_driver = self._instantiate_share_driver(None, True, None) 

524 ctxt = 'fake_context' 

525 share_server = 'fake_share_server' 

526 share_driver._admin_network_api = None 

527 mock_get_admin_network_allocations_number = self.mock_object( 

528 share_driver, 

529 'get_admin_network_allocations_number', 

530 mock.Mock(return_value=1)) 

531 

532 self.assertRaises( 

533 exception.NetworkBadConfigurationException, 

534 share_driver.allocate_admin_network, 

535 ctxt, share_server, 

536 ) 

537 mock_get_admin_network_allocations_number.assert_called_once_with() 

538 

539 def test_migration_start(self): 

540 

541 driver.CONF.set_default('driver_handles_share_servers', False) 

542 share_driver = driver.ShareDriver(False) 

543 

544 self.assertRaises(NotImplementedError, share_driver.migration_start, 

545 None, None, None, None, None, None, None) 

546 

547 def test_migration_continue(self): 

548 

549 driver.CONF.set_default('driver_handles_share_servers', False) 

550 share_driver = driver.ShareDriver(False) 

551 

552 self.assertRaises(NotImplementedError, share_driver.migration_continue, 

553 None, None, None, None, None, None, None) 

554 

555 def test_migration_complete(self): 

556 

557 driver.CONF.set_default('driver_handles_share_servers', False) 

558 share_driver = driver.ShareDriver(False) 

559 

560 self.assertRaises(NotImplementedError, share_driver.migration_complete, 

561 None, None, None, None, None, None, None) 

562 

563 def test_migration_cancel(self): 

564 

565 driver.CONF.set_default('driver_handles_share_servers', False) 

566 share_driver = driver.ShareDriver(False) 

567 

568 self.assertRaises(NotImplementedError, share_driver.migration_cancel, 

569 None, None, None, None, None, None, None) 

570 

571 def test_migration_get_progress(self): 

572 

573 driver.CONF.set_default('driver_handles_share_servers', False) 

574 share_driver = driver.ShareDriver(False) 

575 

576 self.assertRaises(NotImplementedError, 

577 share_driver.migration_get_progress, 

578 None, None, None, None, None, None, None) 

579 

580 def test_share_server_migration_start(self): 

581 driver.CONF.set_default('driver_handles_share_servers', True) 

582 share_driver = driver.ShareDriver(True) 

583 

584 self.assertRaises(NotImplementedError, 

585 share_driver.share_server_migration_start, 

586 None, None, None, None, None) 

587 

588 def test_share_server_migration_continue(self): 

589 driver.CONF.set_default('driver_handles_share_servers', True) 

590 share_driver = driver.ShareDriver(True) 

591 

592 self.assertRaises(NotImplementedError, 

593 share_driver.share_server_migration_continue, 

594 None, None, None, None, None) 

595 

596 def test_share_server_migration_get_progress(self): 

597 driver.CONF.set_default('driver_handles_share_servers', True) 

598 share_driver = driver.ShareDriver(True) 

599 

600 self.assertRaises(NotImplementedError, 

601 share_driver.share_server_migration_get_progress, 

602 None, None, None, None, None) 

603 

604 def test_share_server_migration_cancel(self): 

605 driver.CONF.set_default('driver_handles_share_servers', True) 

606 share_driver = driver.ShareDriver(True) 

607 

608 self.assertRaises(NotImplementedError, 

609 share_driver.share_server_migration_cancel, 

610 None, None, None, None, None) 

611 

612 def test_share_server_migration_check_compatibility(self): 

613 driver.CONF.set_default('driver_handles_share_servers', True) 

614 share_driver = driver.ShareDriver(True) 

615 expected_compatibility = { 

616 'compatible': False, 

617 'writable': False, 

618 'nondisruptive': False, 

619 'preserve_snapshots': False, 

620 'migration_cancel': False, 

621 'migration_get_progress': False, 

622 } 

623 

624 driver_compatibility = ( 

625 share_driver.share_server_migration_check_compatibility( 

626 None, None, None, None, None, None)) 

627 self.assertEqual(expected_compatibility, driver_compatibility) 

628 

629 def test_share_server_migration_complete(self): 

630 driver.CONF.set_default('driver_handles_share_servers', True) 

631 share_driver = driver.ShareDriver(True) 

632 

633 self.assertRaises( 

634 NotImplementedError, 

635 share_driver.share_server_migration_complete, 

636 None, None, None, None, None, None) 

637 

638 @ddt.data(True, False) 

639 def test_connection_get_info(self, admin): 

640 

641 expected = { 

642 'mount': 'mount -vt nfs %(options)s /fake/fake_id %(path)s', 

643 'unmount': 'umount -v %(path)s', 

644 'access_mapping': { 

645 'ip': ['nfs'] 

646 } 

647 } 

648 

649 fake_share = { 

650 'id': 'fake_id', 

651 'share_proto': 'nfs', 

652 'export_locations': [{ 

653 'path': '/fake/fake_id', 

654 'is_admin_only': admin 

655 }] 

656 } 

657 

658 driver.CONF.set_default('driver_handles_share_servers', False) 

659 share_driver = driver.ShareDriver(False) 

660 share_driver.configuration = configuration.Configuration(None) 

661 

662 connection_info = share_driver.connection_get_info( 

663 None, fake_share, "fake_server") 

664 

665 self.assertEqual(expected, connection_info) 

666 

667 def test_migration_check_compatibility(self): 

668 

669 driver.CONF.set_default('driver_handles_share_servers', False) 

670 share_driver = driver.ShareDriver(False) 

671 share_driver.configuration = configuration.Configuration(None) 

672 expected = { 

673 'compatible': False, 

674 'writable': False, 

675 'preserve_metadata': False, 

676 'nondisruptive': False, 

677 'preserve_snapshots': False, 

678 } 

679 

680 result = share_driver.migration_check_compatibility( 

681 None, None, None, None, None) 

682 

683 self.assertEqual(expected, result) 

684 

685 def test_update_access(self): 

686 share_driver = driver.ShareDriver(True, configuration=None) 

687 self.assertRaises( 

688 NotImplementedError, 

689 share_driver.update_access, 

690 'ctx', 

691 'fake_share', 

692 'fake_access_rules', 

693 'fake_add_rules', 

694 'fake_delete_rules', 

695 'fake_update_rules' 

696 ) 

697 

698 def test_create_replica(self): 

699 share_driver = self._instantiate_share_driver(None, True) 

700 self.assertRaises(NotImplementedError, 

701 share_driver.create_replica, 

702 'fake_context', ['r1', 'r2'], 

703 'fake_new_replica', [], []) 

704 

705 def test_delete_replica(self): 

706 share_driver = self._instantiate_share_driver(None, True) 

707 self.assertRaises(NotImplementedError, 

708 share_driver.delete_replica, 

709 'fake_context', ['r1', 'r2'], 

710 'fake_replica', []) 

711 

712 def test_promote_replica(self): 

713 share_driver = self._instantiate_share_driver(None, True) 

714 self.assertRaises(NotImplementedError, 

715 share_driver.promote_replica, 

716 'fake_context', [], 'fake_replica', []) 

717 

718 def test_update_replica_state(self): 

719 share_driver = self._instantiate_share_driver(None, True) 

720 self.assertRaises(NotImplementedError, 

721 share_driver.update_replica_state, 

722 'fake_context', ['r1', 'r2'], 'fake_replica', [], []) 

723 

724 def test_create_replicated_snapshot(self): 

725 share_driver = self._instantiate_share_driver(None, False) 

726 self.assertRaises(NotImplementedError, 

727 share_driver.create_replicated_snapshot, 

728 'fake_context', ['r1', 'r2'], ['s1', 's2']) 

729 

730 def test_delete_replicated_snapshot(self): 

731 share_driver = self._instantiate_share_driver(None, False) 

732 self.assertRaises(NotImplementedError, 

733 share_driver.delete_replicated_snapshot, 

734 'fake_context', ['r1', 'r2'], ['s1', 's2']) 

735 

736 def test_update_replicated_snapshot(self): 

737 share_driver = self._instantiate_share_driver(None, False) 

738 self.assertRaises(NotImplementedError, 

739 share_driver.update_replicated_snapshot, 

740 'fake_context', ['r1', 'r2'], 'r1', 

741 ['s1', 's2'], 's1') 

742 

743 @ddt.data(True, False) 

744 def test_share_group_snapshot_support_exists_and_equals_snapshot_support( 

745 self, snapshots_are_supported): 

746 driver.CONF.set_default('driver_handles_share_servers', True) 

747 child_class_instance = driver.ShareDriver(True) 

748 child_class_instance._snapshots_are_supported = snapshots_are_supported 

749 self.mock_object(child_class_instance, "configuration") 

750 

751 child_class_instance._update_share_stats() 

752 

753 self.assertEqual( 

754 snapshots_are_supported, 

755 child_class_instance._stats["snapshot_support"]) 

756 self.assertTrue(child_class_instance.configuration.safe_get.called) 

757 

758 def test_create_share_group_from_share_group_snapshot(self): 

759 share_driver = self._instantiate_share_driver(None, False) 

760 fake_shares = [ 

761 {'id': 'fake_share_%d' % i, 

762 'source_share_group_snapshot_member_id': 'fake_member_%d' % i} 

763 for i in (1, 2)] 

764 fake_share_group_dict = { 

765 'source_share_group_snapshot_id': 'some_fake_uuid_abc', 

766 'shares': fake_shares, 

767 'id': 'some_fake_uuid_def', 

768 } 

769 fake_share_group_snapshot_dict = { 

770 'share_group_snapshot_members': [ 

771 {'id': 'fake_member_1'}, {'id': 'fake_member_2'}], 

772 'id': 'fake_share_group_snapshot_id', 

773 } 

774 mock_create = self.mock_object( 

775 share_driver, 'create_share_from_snapshot', 

776 mock.Mock(side_effect=['fake_export1', 'fake_export2'])) 

777 expected_share_updates = [ 

778 {