Coverage for manila/tests/share/drivers/vastdata/test

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

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

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.

16import itertools

17import unittest

18from unittest import mock

20import ddt

21import netaddr

23import manila.context as manila_context

24import manila.exception as exception

25from manila.share import configuration

26from manila.share.drivers.vastdata import driver

27from manila.share.drivers.vastdata import driver_util

28from manila.tests import fake_share

29from manila.tests.share.drivers.vastdata.test_rest import fake_metrics

32@mock.patch(

33 "manila.share.drivers.vastdata.rest.Session.refresh_auth_token",

34 mock.MagicMock()

35)

36@ddt.ddt

37class VASTShareDriverTestCase(unittest.TestCase):

38 def _create_mocked_rest_api(self):

39 # Create a mock RestApi instance

40 mock_rest_api = mock.MagicMock()

42 # Create mock sub resources with their methods

43 subresources = [

44 "views",

45 "view_policies",

46 "capacity_metrics",

47 "quotas",

48 "vip_pools",

49 "snapshots",

50 "folders",

51 ]

52 methods = [

53 "list", "create", "update",

54 "delete", "one", "ensure", "vips"

55 ]

57 for subresource in subresources:

58 mock_subresource = mock.MagicMock()

59 setattr(mock_rest_api, subresource, mock_subresource)

61 for method in methods:

62 mock_method = mock.MagicMock()

63 setattr(mock_subresource, method, mock_method)

65 return mock_rest_api

67 @mock.patch(

68 "manila.share.drivers.vastdata.rest.Session.refresh_auth_token"

69 )

70 def setUp(self, m_auth_token):

71 super().setUp()

72 self.fake_conf = configuration.Configuration(None)

73 self._context = manila_context.get_admin_context()

74 self._snapshot = fake_share.fake_snapshot_instance()

76 self.fake_conf.set_default("driver_handles_share_servers", False)

77 self.fake_conf.set_default("share_backend_name", "vast")

78 self.fake_conf.set_default("vast_mgmt_host", "test")

79 self.fake_conf.set_default("vast_root_export", "/fake")

80 self.fake_conf.set_default("vast_vippool_name", "vippool")

81 self.fake_conf.set_default("vast_mgmt_user", "user")

82 self.fake_conf.set_default("vast_mgmt_password", "password")

83 self._driver = driver.VASTShareDriver(

84 execute=mock.MagicMock(), configuration=self.fake_conf

85 )

86 self._driver.do_setup(self._context)

87 m_auth_token.assert_called_once()

89 def test_do_setup(self):

90 session = self._driver.rest.session

91 self.assertEqual(self._driver._backend_name, "vast")

92 self.assertEqual(self._driver._vippool_name, "vippool")

93 self.assertEqual(self._driver._root_export, "/fake")

94 self.assertFalse(session.ssl_verify)

95 self.assertEqual(session.base_url, "https://test:443/api")

97 @ddt.data("vast_mgmt_user", "vast_vippool_name", "vast_mgmt_host")

98 def test_do_setup_missing_required_fields(self, missing_field):

99 self.fake_conf.set_default(missing_field, None)

100 _driver = driver.VASTShareDriver(

101 execute=mock.MagicMock(), configuration=self.fake_conf

102 )

103 with self.assertRaises(exception.VastDriverException):

104 _driver.do_setup(self._context)

105

106 def test_do_setup_with_api_token(self):

107 self.fake_conf.set_default("vast_mgmt_user", None)

108 self.fake_conf.set_default("vast_mgmt_password", None)

109 self.fake_conf.set_default("vast_api_token", "test_token")

110 _driver = driver.VASTShareDriver(

111 execute=mock.MagicMock(), configuration=self.fake_conf

112 )

113 _driver.do_setup(self._context)

114 self.assertEqual(_driver.rest.session.token, "test_token")

115

116 @mock.patch(

117 "manila.share.drivers.vastdata.rest.Session.get",

118 mock.MagicMock(return_value=fake_metrics),

119 )

120 def test_update_share_stats(self):

121 self._driver._update_share_stats()

122 result = self._driver._stats

123 self.assertEqual(result["share_backend_name"], "vast")

124 self.assertEqual(result["driver_handles_share_servers"], False)

125 self.assertEqual(result["vendor_name"], "VAST STORAGE")

126 self.assertEqual(result["driver_version"], "1.0")

127 self.assertEqual(result["storage_protocol"], "NFS")

128 self.assertEqual(result["total_capacity_gb"], 471.1061706542969)

129 self.assertEqual(result["free_capacity_gb"], 450.2256333641708)

130 self.assertEqual(result["reserved_percentage"], 0)

131 self.assertEqual(result["reserved_snapshot_percentage"], 0)

132 self.assertEqual(result["reserved_share_extend_percentage"], 0)

133 self.assertIs(result["qos"], False)

134 self.assertIsNone(result["pools"])

135 self.assertIs(result["snapshot_support"], True)

136 self.assertIs(result["create_share_from_snapshot_support"], False)

137 self.assertIs(result["revert_to_snapshot_support"], False)

138 self.assertIs(result["mount_snapshot_support"], False)

139 self.assertIsNone(result["replication_domain"])

140 self.assertIsNone(result["filter_function"])

141 self.assertIsNone(result["goodness_function"])

142 self.assertIs(result["security_service_update_support"], False)

143 self.assertIs(result["network_allocation_update_support"], False)

144 self.assertIs(result["share_server_multiple_subnet_support"], False)

145 self.assertIs(result["mount_point_name_support"], False)

146 self.assertEqual(result["data_reduction"], 1.2)

147 self.assertEqual(result["provisioned_capacity_gb"], 20.880537290126085)

148 self.assertEqual(

149 result["share_group_stats"],

150 {"consistent_snapshot_support": None}

151 )

152 self.assertIs(result["ipv4_support"], True)

153 self.assertIs(result["ipv6_support"], False)

154

155 @ddt.idata(

156 itertools.product(

157 [1073741824, 1], ["NFS", "SMB"], ["fakeid", None]

158 )

159 )

160 @ddt.unpack

161 def test_create_shares(self, capacity, proto, policy):

162 share = fake_share.fake_share(share_proto=proto)

163 mock_rest = self._create_mocked_rest_api()

164 mock_rest.view_policies.ensure.return_value = driver_util.Bunch(id=1)

165 mock_rest.quotas.ensure.return_value = driver_util.Bunch(

166 id=2, hard_limit=capacity

167 )

168 mock_rest.views.ensure.return_value = driver_util.Bunch(

169 id=3, policy=policy

170 )

171 mock_rest.vip_pools.vips.return_value = ["1.1.1.0", "1.1.1.1"]

172 with mock.patch.object(self._driver, "rest", mock_rest):

173 if proto != "NFS":

174 with self.assertRaises(exception.InvalidShare) as exc:

175 self._driver.create_share(self._context, share)

176 self.assertIn(

177 "Invalid NAS protocol supplied",

178 str(exc.exception)

179 )

180 elif capacity == 1:

181 with self.assertRaises(exception.ManilaException) as exc:

182 self._driver.create_share(self._context, share)

183 self.assertIn(

184 "Share already exists with different capacity",

185 str(exc.exception)

186 )

187 else:

188

189 location = self._driver.create_share(self._context, share)

190 mock_rest.vip_pools.vips.assert_called_once_with(

191 pool_name="vippool"

192 )

193 mock_rest.view_policies.ensure.assert_called_once_with(

194 name="fakeid"

195 )

196 mock_rest.quotas.ensure.assert_called_once_with(

197 name="fakeid",

198 path="/fake/manila-fakeid",

199 create_dir=True,

200 hard_limit=capacity,

201 )

202 mock_rest.views.ensure.assert_called_once_with(

203 name="fakeid", path="/fake/manila-fakeid", policy_id=1

204 )

205 self.assertListEqual(

206 location,

207 [

208 {

209 'path': '1.1.1.0:/fake/manila-fakeid',

210 'is_admin_only': False

211 },

212 {

213 'path': '1.1.1.1:/fake/manila-fakeid',

214 'is_admin_only': False

215 },

216 ]

217 )

218 if not policy:

219 mock_rest.views.update.assert_called_once_with(

220 3, policy_id=1

221 )

222 else:

223 mock_rest.views.update.assert_not_called()

224

225 def test_delete_share(self):

226 share = fake_share.fake_share(share_proto="NFS")

227 mock_rest = self._create_mocked_rest_api()

228 with mock.patch.object(self._driver, "rest", mock_rest):

229 self._driver.delete_share(self._context, share)

230 mock_rest.folders.delete.assert_called_once_with(

231 path="/fake/manila-fakeid"

232 )

233 mock_rest.views.delete.assert_called_once_with(name="fakeid")

234 mock_rest.quotas.delete.assert_called_once_with(name="fakeid")

235 mock_rest.view_policies.delete.assert_called_once_with(name="fakeid")

236

237 def test_update_access_rules_wrong_proto(self):

238 share = fake_share.fake_share(share_proto="SMB")

239 access_rules = [

240 {

241 "access_level": "rw",

242 "access_to": "127.0.0.1",

243 "access_type": "ip"

244 }

245 ]

246 res = self._driver.update_access(

247 self._context,

248 share,

249 access_rules,

250 None,

251 None,

252 None

253 )

254 self.assertIsNone(res)

255

256 def test_update_access_add_rules_no_policy(self):

257 share = fake_share.fake_share(share_proto="NFS")

258 mock_rest = self._create_mocked_rest_api()

259 mock_rest.view_policies.one.return_value = None

260 access_rules = [

261 {

262 "access_level": "rw",

263 "access_to": "127.0.0.1",

264 "access_type": "ip"

265 }

266 ]

267 with mock.patch.object(self._driver, "rest", mock_rest):

268 with self.assertRaises(exception.ManilaException) as exc:

269 self._driver.update_access(

270 self._context, share, access_rules, None, None, None

271 )

272 self.assertIn("Policy not found", str(exc.exception))

273

274 @ddt.data(

275 (["*"], ["10.10.10.1", "10.10.10.2"]),

276 (["10.10.10.1", "10.10.10.2"], []),

277 (["*"], []),

278 )

279 @ddt.unpack

280 def test_update_access_add_rules(self, rw, ro):

281 share = fake_share.fake_share(share_proto="NFS")

282 mock_rest = self._create_mocked_rest_api()

283 mock_rest.view_policies.one.return_value = driver_util.Bunch(

284 id=1, nfs_read_write=rw, nfs_read_only=ro

285 )

286 access_rules = [

287 {

288 "access_level": "rw",

289 "access_to": "127.0.0.1",

290 "access_type": "ip"

291 }

292 ]

293 with mock.patch.object(self._driver, "rest", mock_rest):

294 failed_rules = self._driver.update_access(

295 self._context,

296 share,

297 access_rules,

298 None,

299 None,

300 None

301 )

302

303 expected_ro = set(ro)

304 if rw == ["*"]:

305 expected_rw = {"127.0.0.1"}

306 else:

307 expected_rw = set(["127.0.0.1"] + rw)

308 kw = mock_rest.view_policies.update.call_args.kwargs

309 self.assertEqual(kw["name"], "fakeid")

310 self.assertSetEqual(set(kw["nfs_read_write"]), expected_rw)

311 self.assertSetEqual(set(kw["nfs_read_only"]), expected_ro)

312 self.assertEqual(kw["nfs_no_squash"], ["*"])

313 self.assertEqual(kw["nfs_root_squash"], ["*"])

314 self.assertFalse(failed_rules)

315

316 # and the same for ro

317 mock_rest = self._create_mocked_rest_api()

318 mock_rest.view_policies.one.return_value = driver_util.Bunch(

319 id=1, nfs_read_write=rw, nfs_read_only=ro

320 )

321 access_rules = [

322 {

323 "access_level": "ro",

324 "access_to": "127.0.0.1",

325 "access_type": "ip"

326 }

327 ]

328 with mock.patch.object(self._driver, "rest", mock_rest):

329 failed_rules = self._driver.update_access(

330 self._context,

331 share,

332 access_rules,

333 None,

334 None,

335 None

336 )

337

338 expected_rw = set(rw)

339 if ro == ["*"]: 339 ↛ 340line 339 didn't jump to line 340 because the condition on line 339 was never true

340 expected_ro = {"127.0.0.1"}

341 else:

342 expected_ro = set(["127.0.0.1"] + ro)

343 kw = mock_rest.view_policies.update.call_args.kwargs

344 self.assertEqual(kw["name"], "fakeid")

345 self.assertSetEqual(set(kw["nfs_read_write"]), expected_rw)

346 self.assertSetEqual(set(kw["nfs_read_only"]), expected_ro)

347 self.assertEqual(kw["nfs_no_squash"], ["*"])

348 self.assertEqual(kw["nfs_root_squash"], ["*"])

349 self.assertFalse(failed_rules)

350

351 @ddt.data(

352 (["*"], ["10.10.10.1", "10.10.10.2"]),

353 (["10.10.10.1", "10.10.10.2"], []),

354 (["*"], []),

355 )

356 @ddt.unpack

357 def test_update_access_delete_rules(self, rw, ro):

358 share = fake_share.fake_share(share_proto="NFS")

359 mock_rest = self._create_mocked_rest_api()

360 mock_rest.view_policies.one.return_value = driver_util.Bunch(

361 id=1, nfs_read_write=rw, nfs_read_only=ro

362 )

363 delete_rules = [

364 {

365 "access_level": "rw",

366 "access_to": "10.10.10.1",

367 "access_type": "ip"

368 }

369 ]

370 with mock.patch.object(self._driver, "rest", mock_rest):

371 failed_rules = self._driver.update_access(

372 self._context, share,

373 None,

374 None,

375 delete_rules,

376 None,

377 )

378

379 expected_ro = set(ro)

380 if rw == ["*"]:

381 expected_rw = set(rw)

382 else:

383 expected_rw = set([r for r in rw if r != "10.10.10.1"])

384 kw = mock_rest.view_policies.update.call_args.kwargs

385 self.assertEqual(kw["name"], "fakeid")

386 self.assertSetEqual(set(kw["nfs_read_write"]), expected_rw)

387 self.assertSetEqual(set(kw["nfs_read_only"]), expected_ro)

388 self.assertEqual(kw["nfs_no_squash"], ["*"])

389 self.assertEqual(kw["nfs_root_squash"], ["*"])

390 self.assertFalse(failed_rules)

391

392 # and the same for ro

393 mock_rest = self._create_mocked_rest_api()

394 mock_rest.view_policies.one.return_value = driver_util.Bunch(

395 id=1, nfs_read_write=rw, nfs_read_only=ro

396 )

397 delete_rules = [

398 {

399 "access_level": "ro",

400 "access_to": "10.10.10.1",

401 "access_type": "ip"

402 }

403 ]

404 with mock.patch.object(self._driver, "rest", mock_rest):

405 failed_rules = self._driver.update_access(

406 self._context, share, None, None, delete_rules, None

407 )

408

409 expected_rw = set(rw)

410 if ro == ["*"]: 410 ↛ 411line 410 didn't jump to line 411 because the condition on line 410 was never true

411 expected_ro = set(ro)

412 else:

413 expected_ro = set([r for r in ro if r != "10.10.10.1"])

414 kw = mock_rest.view_policies.update.call_args.kwargs

415 self.assertEqual(kw["name"], "fakeid")

416 self.assertSetEqual(set(kw["nfs_read_write"]), expected_rw)

417 self.assertSetEqual(set(kw["nfs_read_only"]), expected_ro)

418 self.assertEqual(kw["nfs_no_squash"], ["*"])

419 self.assertEqual(kw["nfs_root_squash"], ["*"])

420 self.assertFalse(failed_rules)

421

422 def test_update_access_for_cidr(self):

423 share = fake_share.fake_share(share_proto="NFS")

424 mock_rest = self._create_mocked_rest_api()

425 mock_rest.view_policies.one.return_value = driver_util.Bunch(

426 id=1, nfs_read_write=["10.0.0.1"], nfs_read_only=["*"]

427 )

428 access_rules = [

429 {

430 "access_level": "ro",

431 "access_to": "10.0.0.1/29",

432 "access_type": "ip",

433 "access_id": 12345,

434 }

435 ]

436 with mock.patch.object(self._driver, "rest", mock_rest):

437 failed_rules = self._driver.update_access(

438 self._context,

439 share,

440 access_rules,

441 None,

442 None,

443 None

444 )

445 kw = mock_rest.view_policies.update.call_args.kwargs

446 self.assertEqual(kw["name"], "fakeid")

447 self.assertSetEqual(set(kw["nfs_read_write"]), {"10.0.0.1"})

448 self.assertSetEqual(

449 set(kw["nfs_read_only"]),

450 {

451 '10.0.0.1',

452 '10.0.0.3',

453 '10.0.0.2',

454 '10.0.0.6',

455 '10.0.0.5',

456 '10.0.0.4'

457 }

458 )

459 self.assertFalse(failed_rules)

460

461 delete_rules = [

462 {

463 "access_level": "ro",

464 "access_to": "10.0.0.1/30",

465 "access_type": "ip",

466 "access_id": 12345,

467 }

468 ]

469 mock_rest.view_policies.one.return_value = driver_util.Bunch(

470 id=1, nfs_read_write=["10.0.0.1"],

471 nfs_read_only=[

472 '10.0.0.1',

473 '10.0.0.3',

474 '10.0.0.2',

475 '10.0.0.6',

476 '10.0.0.5',

477 '10.0.0.4',

478 ]

479 )

480 with mock.patch.object(self._driver, "rest", mock_rest):

481 failed_rules = self._driver.update_access(

482 self._context,

483 share,

484 None,

485 None,

486 delete_rules,

487 None

488 )

489 kw = mock_rest.view_policies.update.call_args.kwargs

490 self.assertEqual(kw["name"], "fakeid")

491 self.assertSetEqual(set(kw["nfs_read_write"]), {"10.0.0.1"})

492 self.assertSetEqual(

493 set(kw["nfs_read_only"]),

494 {'10.0.0.6', '10.0.0.3', '10.0.0.4', '10.0.0.5'}

495 )

496 self.assertFalse(failed_rules)

497

498 def test_update_access_for_invalid_rules(self):

499 share = fake_share.fake_share(share_proto="NFS")

500 mock_rest = self._create_mocked_rest_api()

501 mock_rest.view_policies.one.return_value = driver_util.Bunch(

502 id=1, nfs_read_write=["10.0.0.1"], nfs_read_only=["*"]

503 )

504 access_rules = [

505 {

506 "access_level": "ry",

507 "access_to": "10.0.0.1",

508 "access_type": "ip",

509 "access_id": 12345,

510 "id": 12345,

511 },

512 {

513 "access_level": "ro",

514 "access_to": "10.0.0.2",

515 "access_type": "ip",

516 "access_id": 12346,

517 "id": 12346,

518 },

519 {

520 "access_level": "ro",

521 "access_to": "10.0.0.2/33",

522 "access_type": "ip",

523 "access_id": 12347,

524 "id": 12347,

525 },

526 {

527 "access_level": "rw",

528 "access_to": "10.0.0.2.4",

529 "access_type": "ip",

530 "access_id": 12348,

531 "id": 12348,

532 }

533 ]

534 with mock.patch.object(self._driver, "rest", mock_rest):

535 failed_rules = self._driver.update_access(

536 self._context,

537 share,

538 access_rules,

539 None,

540 None,

541 None

542 )

543 kw = mock_rest.view_policies.update.call_args.kwargs

544 self.assertEqual(kw["name"], "fakeid")

545 self.assertSetEqual(set(kw["nfs_read_write"]), {"10.0.0.1"})

546 self.assertSetEqual(set(kw["nfs_read_only"]), {'10.0.0.2'})

547 self.assertDictEqual(

548 failed_rules,

549 {

550 12345: {'state': 'error'},

551 12347: {'state': 'error'},

552 12348: {'state': 'error'}

553 }

554 )

555

556 def test_resize_share_quota_not_found(self):

557 share = fake_share.fake_share(share_proto="NFS")

558 mock_rest = self._create_mocked_rest_api()

559 mock_rest.quotas.one.return_value = None

560 with mock.patch.object(self._driver, "rest", mock_rest):

561 with self.assertRaises(exception.ShareNotFound) as exc:

562 self._driver.extend_share(share, 10000)

563 self.assertIn("could not be found", str(exc.exception))

564

565 def test_resize_share_ok(self):

566 share = fake_share.fake_share(share_proto="NFS")

567 mock_rest = self._create_mocked_rest_api()

568 mock_rest.quotas.one.return_value = driver_util.Bunch(

569 id=1, used_effective_capacity=1073741824

570 )

571 with mock.patch.object(self._driver, "rest", mock_rest):

572 self._driver.extend_share(share, 50)

573 mock_rest.quotas.update.assert_called_with(

574 1, hard_limit=53687091200

575 )

576 mock_rest.quotas.update.reset()

577 self._driver.shrink_share(share, 20)

578 mock_rest.quotas.update.assert_called_with(

579 1, hard_limit=21474836480

580 )

581

582 def test_resize_share_exceeded_hard_limit(self):

583 share = fake_share.fake_share(

584 share_proto="NFS"

585 )

586 mock_rest = self._create_mocked_rest_api()

587 mock_rest.quotas.one.return_value = driver_util.Bunch(

588 id=1, used_effective_capacity=10737418240

589 ) # 10GB

590 with mock.patch.object(self._driver, "rest", mock_rest):

591 with self.assertRaises(exception.ShareShrinkingPossibleDataLoss):

592 self._driver.shrink_share(share, 9.7)

593 self._driver.shrink_share(share, 10)

594

595 def test_create_snapshot(self):

596 snapshot = driver_util.Bunch(

597 name="fakesnap", share_instance_id="fakeid"

598 )

599 mock_rest = self._create_mocked_rest_api()

600 with mock.patch.object(self._driver, "rest", mock_rest):

601 self._driver.create_snapshot(self._context, snapshot, None)

602 mock_rest.snapshots.create.assert_called_once_with(

603 path="/fake/manila-fakeid", name="fakesnap"

604 )

605

606 def test_delete_snapshot(self):

607 snapshot = driver_util.Bunch(

608 name="fakesnap", share_instance_id="fakeid"

609 )

610 mock_rest = self._create_mocked_rest_api()

611 with mock.patch.object(self._driver, "rest", mock_rest):

612 self._driver.delete_snapshot(self._context, snapshot, None)

613 mock_rest.snapshots.delete.assert_called_once_with(name="fakesnap")

614

615 def test_network_allocation_number(self):

616 self.assertEqual(self._driver.get_network_allocations_number(), 0)

617

618 @ddt.data([], ['fake/path/1', 'fake/path'])

619 def test_ensure_shares(self, fake_export_locations):

620 mock_rest = self._create_mocked_rest_api()

621 mock_rest.view_policies.ensure.return_value = driver_util.Bunch(id=1)

622 mock_rest.quotas.ensure.return_value = driver_util.Bunch(

623 id=2, hard_limit=1073741824

624 )

625 mock_rest.views.ensure.return_value = driver_util.Bunch(

626 id=3, policy="test_policy"

627 )

628 shares = [

629 fake_share.fake_share(

630 id=_id,

631 share_id=share_id,

632 share_proto="NFS",

633 export_locations=fake_export_locations,

634 )

635 for _id, share_id in enumerate(["123", "456", "789"], 1)

636 ]

637 mock_rest.vip_pools.vips.return_value = ["1.1.1.0", "1.1.1.1"]

638 with mock.patch.object(self._driver, "rest", mock_rest):

639 locations = self._driver.ensure_shares(self._context, shares)

640

641 common = {"is_admin_only": False}

642 self.assertDictEqual(

643 locations,

644 {

645 1: {

646 "export_locations": [

647 {"path": "1.1.1.0:/fake/manila-1", **common},

648 {"path": "1.1.1.1:/fake/manila-1", **common},

649 ]

650 },

651 2: {

652 "export_locations": [

653 {"path": "1.1.1.0:/fake/manila-2", **common},

654 {"path": "1.1.1.1:/fake/manila-2", **common},

655 ]

656 },

657 3: {

658 "export_locations": [

659 {"path": "1.1.1.0:/fake/manila-3", **common},

660 {"path": "1.1.1.1:/fake/manila-3", **common},

661 ]

662 },

663 },

664 )

665

666 def test_backend_info(self):

667 backend_info = self._driver.get_backend_info(self._context)

668 self.assertDictEqual(

669 backend_info,

670 {'vast_vippool_name': 'vippool', 'vast_mgmt_host': 'test'}

671 )

672

673

674class TestPolicyPayloadFromRules(unittest.TestCase):

675 def test_policy_payload_from_rules_update(self):

676 rules = [{"access_level": "rw", "access_to": "127.0.0.1"}]

677 policy = mock.MagicMock()

678 policy.nfs_read_write = ["127.0.0.1"]

679 policy.nfs_read_only = []

680 result = driver.policy_payload_from_rules(rules, policy, "update")

681 self.assertEqual(

682 result, {"nfs_read_write": ["127.0.0.1"], "nfs_read_only": []}

683 )

684

685 def test_policy_payload_from_rules_deny(self):

686 rules = [{"access_level": "rw", "access_to": "127.0.0.1"}]

687 policy = mock.MagicMock()

688 policy.nfs_read_write = ["127.0.0.1"]

689 policy.nfs_read_only = []

690 result = driver.policy_payload_from_rules(rules, policy, "deny")

691 self.assertEqual(result, {"nfs_read_write": [], "nfs_read_only": []})

692

693 def test_policy_payload_from_rules_invalid_action(self):

694 rules = [{"access_level": "rw", "access_to": "127.0.0.1"}]

695 with self.assertRaises(ValueError):

696 driver.policy_payload_from_rules(rules, None, "invalid")

697

698 def test_policy_payload_from_rules_invalid_ip(self):

699 rules = [{"access_level": "rw", "access_to": "1.0.0.257"}]

700 with self.assertRaises(netaddr.core.AddrFormatError):

701 driver.policy_payload_from_rules(rules, None, "deny")

702

703

704class TestValidateAccessRules(unittest.TestCase):

705 def test_validate_access_rules_invalid_type(self):

706 rule = {"access_type": "INVALID", "access_level": "rw"}

707 with self.assertRaises(exception.InvalidShareAccess):

708 driver.validate_access_rule(rule)

709

710 def test_validate_access_rules_invalid_level(self):

711 rule = {"access_type": "ip", "access_level": "INVALID"}

712 with self.assertRaises(exception.InvalidShareAccessLevel):

713 driver.validate_access_rule(rule)

Coverage for manila/tests/share/drivers/vastdata/test_driver.py: 99%

336 statements