Coverage for manila/tests/share/drivers/dell_emc/plugins/powerflex/test_connection.py: 100%
206 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright (c) 2023 EMC Corporation.
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.
16from unittest import mock
18import ddt
19from oslo_log import log
20from oslo_utils import units
22from manila.common import constants as const
23from manila import exception
24from manila.share.drivers.dell_emc.plugins.powerflex import connection
25from manila import test
27LOG = log.getLogger(__name__)
30@ddt.ddt
31class PowerFlexTest(test.TestCase):
32 """Integration test for the PowerFlex Manila driver."""
34 POWERFLEX_ADDR = "192.168.0.110"
35 SHARE_NAME = "Manila-UT-filesystem"
36 STORAGE_POOL_ID = "28515fee00000000"
37 FILESYSTEM_ID = "6432b79e-1cc3-0414-3ffd-2a50fb1ccff3"
38 NFS_EXPORT_ID = "6433a2b2-6d60-f737-9f3b-2a50fb1ccff3"
39 NFS_EXPORT_NAME = "Manila-UT-filesystem"
40 SNAPSHOT_NAME = "Manila-UT-filesystem-snap"
41 SNAPSHOT_PATH = "Manila-UT-filesystem"
42 SNAPSHOT_ID = "75758d63-2946-4c07-9118-9a6c6027d5e7"
43 NAS_SERVER_IP = "192.168.11.23"
45 class MockConfig(object):
46 def safe_get(self, value):
47 if value == "dell_nas_backend_host":
48 return "192.168.0.110"
49 elif value == "dell_nas_backend_port":
50 return 443
51 elif value == "dell_nas_login":
52 return "admin"
53 elif value == "dell_nas_password":
54 return "pwd"
55 elif value == "powerflex_storage_pool":
56 return "Env8-SP-SW_SSD-1"
57 elif value == "powerflex_protection_domain":
58 return "Env8-PD-1"
59 elif value == "dell_nas_server":
60 return "env8nasserver"
61 else:
62 return None
64 @mock.patch(
65 "manila.share.drivers.dell_emc.plugins.powerflex.object_manager."
66 "StorageObjectManager",
67 autospec=True,
68 )
69 def setUp(self, mock_powerflex_manager):
70 super(PowerFlexTest, self).setUp()
72 self._mock_powerflex_manager = mock_powerflex_manager.return_value
73 self.storage_connection = connection.PowerFlexStorageConnection(LOG)
75 self.mock_context = mock.Mock("Context")
76 self.mock_emc_driver = mock.Mock("EmcDriver")
78 self._mock_config = self.MockConfig()
79 self.mock_emc_driver.attach_mock(self._mock_config, "configuration")
80 self.storage_connection.connect(
81 self.mock_emc_driver, self.mock_context
82 )
84 @mock.patch(
85 "manila.share.drivers.dell_emc.plugins.powerflex.object_manager."
86 "StorageObjectManager",
87 autospec=True,
88 )
89 def test_connect(self, mock_powerflex_manager):
90 storage_connection = connection.PowerFlexStorageConnection(LOG)
92 # execute method under test
93 storage_connection.connect(self.mock_emc_driver, self.mock_context)
95 # verify connect sets driver params appropriately
96 mock_config = self.MockConfig()
97 server_addr = mock_config.safe_get("dell_nas_backend_host")
98 self.assertEqual(server_addr, storage_connection.rest_ip)
99 expected_port = int(mock_config.safe_get("dell_nas_backend_port"))
100 self.assertEqual(expected_port, storage_connection.rest_port)
101 self.assertEqual(
102 "https://{0}:{1}".format(server_addr, expected_port),
103 storage_connection.host_url,
104 )
105 expected_username = mock_config.safe_get("dell_nas_login")
106 self.assertEqual(expected_username, storage_connection.rest_username)
107 expected_password = mock_config.safe_get("dell_nas_password")
108 self.assertEqual(expected_password, storage_connection.rest_password)
109 expected_erify_certificates = mock_config.safe_get(
110 "dell_ssl_cert_verify"
111 )
112 self.assertEqual(
113 expected_erify_certificates, storage_connection.verify_certificate
114 )
116 def test_create_share_nfs(self):
117 self._mock_powerflex_manager.get_storage_pool_id.return_value = (
118 self.STORAGE_POOL_ID
119 )
120 self._mock_powerflex_manager.create_filesystem.return_value = (
121 self.FILESYSTEM_ID
122 )
123 self._mock_powerflex_manager.create_nfs_export.return_value = (
124 self.NFS_EXPORT_ID
125 )
126 self._mock_powerflex_manager.get_nfs_export_name.return_value = (
127 self.NFS_EXPORT_NAME
128 )
129 self._mock_powerflex_manager.get_nas_server_interfaces.return_value = (
130 [self.NAS_SERVER_IP]
131 )
133 self.assertFalse(
134 self._mock_powerflex_manager.get_storage_pool_id.called
135 )
136 self.assertFalse(self._mock_powerflex_manager.create_filesystem.called)
137 self.assertFalse(self._mock_powerflex_manager.create_nfs_export.called)
138 self.assertFalse(
139 self._mock_powerflex_manager.get_nfs_export_name.called
140 )
142 # create the share
143 share = {"name": self.SHARE_NAME, "share_proto": "NFS", "size": 8}
144 locations = self.storage_connection.create_share(
145 self.mock_context, share, None
146 )
148 # verify location and API call made
149 expected_locations = [{"path": "%s:/%s" % (
150 self.NAS_SERVER_IP,
151 self.SHARE_NAME,
152 )}]
153 self.assertEqual(expected_locations, locations)
154 self._mock_powerflex_manager.get_storage_pool_id.assert_called_with(
155 self._mock_config.safe_get("powerflex_protection_domain"),
156 self._mock_config.safe_get("powerflex_storage_pool"),
157 )
158 self._mock_powerflex_manager.create_filesystem.assert_called_with(
159 self.STORAGE_POOL_ID,
160 self._mock_config.safe_get("dell_nas_server"),
161 self.SHARE_NAME,
162 8 * units.Gi,
163 )
164 self._mock_powerflex_manager.create_nfs_export.assert_called_with(
165 self.FILESYSTEM_ID, self.SHARE_NAME
166 )
167 self._mock_powerflex_manager.get_nfs_export_name.assert_called_with(
168 self.NFS_EXPORT_ID
169 )
171 def test_create_share_nfs_filesystem_id_not_found(self):
172 share = {"name": self.SHARE_NAME, "share_proto": "NFS", "size": 8}
173 self._mock_powerflex_manager.create_filesystem.return_value = None
175 self.assertRaises(
176 exception.ShareBackendException,
177 self.storage_connection.create_share,
178 self.mock_context,
179 share,
180 share_server=None,
181 )
183 def test_create_share_nfs_backend_failure(self):
184 share = {"name": self.SHARE_NAME, "share_proto": "NFS", "size": 8}
185 self._mock_powerflex_manager.create_nfs_export.return_value = False
187 self.assertRaises(
188 exception.ShareBackendException,
189 self.storage_connection.create_share,
190 self.mock_context,
191 share,
192 share_server=None,
193 )
195 def test_create_snapshot(self):
196 self._mock_powerflex_manager.get_fsid_from_export_name.return_value = (
197 self.FILESYSTEM_ID
198 )
199 self._mock_powerflex_manager.create_snapshot.return_value = True
201 snapshot = {
202 "name": self.SNAPSHOT_NAME,
203 "share_name": self.SNAPSHOT_PATH,
204 "id": self.SNAPSHOT_ID,
205 }
206 self.storage_connection.create_snapshot(
207 self.mock_context, snapshot, None
208 )
210 # verify the create snapshot API call is executed
211 self._mock_powerflex_manager.get_fsid_from_export_name. \
212 assert_called_with(
213 self.SNAPSHOT_PATH
214 )
215 self._mock_powerflex_manager.create_snapshot.assert_called_with(
216 self.SNAPSHOT_NAME, self.FILESYSTEM_ID
217 )
219 def test_create_snapshot_failure(self):
220 self._mock_powerflex_manager.get_fsid_from_export_name.return_value = (
221 self.FILESYSTEM_ID
222 )
223 self._mock_powerflex_manager.create_snapshot.return_value = False
225 snapshot = {
226 "name": self.SNAPSHOT_NAME,
227 "share_name": self.SNAPSHOT_PATH,
228 "id": self.SNAPSHOT_ID,
229 }
230 self.storage_connection.create_snapshot(
231 self.mock_context, snapshot, None
232 )
234 def test_delete_share_nfs(self):
235 share = {"name": self.SHARE_NAME, "share_proto": "NFS"}
237 self._mock_powerflex_manager.get_filesystem_id.return_value = (
238 self.FILESYSTEM_ID
239 )
241 self.assertFalse(self._mock_powerflex_manager.get_filesystem_id.called)
242 self.assertFalse(self._mock_powerflex_manager.delete_filesystem.called)
244 # delete the share
245 self.storage_connection.delete_share(self.mock_context, share, None)
247 # verify share delete
248 self._mock_powerflex_manager.get_filesystem_id.assert_called_with(
249 self.SHARE_NAME
250 )
251 self._mock_powerflex_manager.delete_filesystem.assert_called_with(
252 self.FILESYSTEM_ID
253 )
255 def test_delete_nfs_share_backend_failure(self):
256 share = {"name": self.SHARE_NAME, "share_proto": "NFS"}
258 self._mock_powerflex_manager.delete_filesystem.return_value = False
259 self.assertRaises(
260 exception.ShareBackendException,
261 self.storage_connection.delete_share,
262 self.mock_context,
263 share,
264 None,
265 )
267 def test_delete_nfs_share_share_does_not_exist(self):
268 self._mock_powerflex_manager.get_filesystem_id.return_value = None
269 share = {"name": self.SHARE_NAME, "share_proto": "NFS"}
271 # verify the calling delete on a non-existent share returns and does
272 # not throw exception
273 self.storage_connection.delete_share(self.mock_context, share, None)
274 self.assertTrue(self._mock_powerflex_manager.get_filesystem_id.called)
275 self.assertFalse(self._mock_powerflex_manager.delete_filesystem.called)
277 def test_delete_snapshot(self):
278 self._mock_powerflex_manager.get_fsid_from_snapshot_name. \
279 return_value = (
280 self.FILESYSTEM_ID
281 )
283 self.assertFalse(
284 self._mock_powerflex_manager.get_fsid_from_snapshot_name.called
285 )
286 self.assertFalse(self._mock_powerflex_manager.delete_filesystem.called)
288 # delete the created snapshot
289 snapshot = {
290 "name": self.SNAPSHOT_NAME,
291 "share_name": self.SNAPSHOT_PATH,
292 "id": self.SNAPSHOT_ID,
293 }
294 self.storage_connection.delete_snapshot(
295 self.mock_context, snapshot, None
296 )
298 # verify the API call was made to delete the snapshot
299 self._mock_powerflex_manager.get_fsid_from_snapshot_name. \
300 assert_called_with(
301 self.SNAPSHOT_NAME
302 )
303 self._mock_powerflex_manager.delete_filesystem.assert_called_with(
304 self.FILESYSTEM_ID
305 )
307 def test_delete_snapshot_backend_failure(self):
308 self._mock_powerflex_manager.get_fsid_from_snapshot_name. \
309 return_value = (
310 self.FILESYSTEM_ID
311 )
312 self._mock_powerflex_manager.delete_filesystem.return_value = False
314 self.assertFalse(
315 self._mock_powerflex_manager.get_fsid_from_snapshot_name.called
316 )
317 self.assertFalse(self._mock_powerflex_manager.delete_filesystem.called)
319 snapshot = {
320 "name": self.SNAPSHOT_NAME,
321 "share_name": self.SNAPSHOT_PATH,
322 "id": self.SNAPSHOT_ID,
323 }
324 # verify the API call was made to delete the snapshot
325 self.assertRaises(
326 exception.ShareBackendException,
327 self.storage_connection.delete_snapshot,
328 self.mock_context,
329 snapshot,
330 None,
331 )
332 self._mock_powerflex_manager.get_fsid_from_snapshot_name. \
333 assert_called_with(
334 self.SNAPSHOT_NAME
335 )
336 self._mock_powerflex_manager.delete_filesystem.assert_called_with(
337 self.FILESYSTEM_ID
338 )
340 def test_extend_share(self):
341 new_share_size = 20
342 share = {
343 "name": self.SHARE_NAME,
344 "share_proto": "NFS",
345 "size": new_share_size,
346 }
347 self._mock_powerflex_manager.get_filesystem_id.return_value = (
348 self.FILESYSTEM_ID
349 )
350 self.assertFalse(self._mock_powerflex_manager.get_filesystem_id.called)
352 self.storage_connection.extend_share(share, new_share_size)
354 self._mock_powerflex_manager.get_filesystem_id.assert_called_with(
355 self.SHARE_NAME
356 )
357 expected_quota_size = new_share_size * units.Gi
358 self._mock_powerflex_manager.extend_export.assert_called_once_with(
359 self.FILESYSTEM_ID, expected_quota_size
360 )
362 def test_update_access_add_nfs(self):
363 share = {"name": self.SHARE_NAME, "share_proto": "NFS"}
365 self._mock_powerflex_manager.get_nfs_export_id.return_value = (
366 self.NFS_EXPORT_ID
367 )
368 self._mock_powerflex_manager.set_export_access.return_value = True
370 self.assertFalse(self._mock_powerflex_manager.get_nfs_export_id.called)
371 self.assertFalse(self._mock_powerflex_manager.set_export_access.called)
373 nfs_rw_ip = "192.168.0.10"
374 nfs_ro_ip = "192.168.0.11"
375 nfs_access_rw = {
376 "access_type": "ip",
377 "access_to": nfs_rw_ip,
378 "access_level": const.ACCESS_LEVEL_RW,
379 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd08",
380 }
381 nfs_access_ro = {
382 "access_type": "ip",
383 "access_to": nfs_ro_ip,
384 "access_level": const.ACCESS_LEVEL_RO,
385 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd08",
386 }
387 access_rules = [nfs_access_rw, nfs_access_ro]
389 self.storage_connection.update_access(
390 self.mock_context,
391 share,
392 access_rules,
393 add_rules=None,
394 delete_rules=None,
395 share_server=None,
396 )
398 self._mock_powerflex_manager.get_nfs_export_id.assert_called_once_with(
399 self.SHARE_NAME
400 )
401 self._mock_powerflex_manager.set_export_access.assert_called_once_with(
402 self.NFS_EXPORT_ID, {nfs_rw_ip}, {nfs_ro_ip}
403 )
405 def test_update_access_add_nfs_invalid_acess_type(self):
406 share = {
407 "name": self.SHARE_NAME,
408 "share_proto": "NFS",
409 "display_name": "foo_display_name",
410 }
412 nfs_rw_ip = "192.168.0.10"
413 nfs_ro_ip = "192.168.0.11"
414 nfs_access_rw = {
415 "access_type": "invalid_type",
416 "access_to": nfs_rw_ip,
417 "access_level": const.ACCESS_LEVEL_RW,
418 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd08",
419 }
420 nfs_access_ro = {
421 "access_type": "invalid_type",
422 "access_to": nfs_ro_ip,
423 "access_level": const.ACCESS_LEVEL_RO,
424 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd09",
425 }
426 access_rules = [nfs_access_rw, nfs_access_ro]
428 self._mock_powerflex_manager.get_nfs_export_id.return_value = (
429 self.NFS_EXPORT_ID
430 )
432 access_updates = self.storage_connection.update_access(
433 self.mock_context,
434 share,
435 access_rules,
436 add_rules=None,
437 delete_rules=None,
438 share_server=None,
439 )
441 self._mock_powerflex_manager.set_export_access.assert_called_once_with(
442 self.NFS_EXPORT_ID, set(), set()
443 )
445 self.assertIsNotNone(access_updates)
447 def test_update_access_add_nfs_backend_failure(self):
448 share = {
449 "name": self.SHARE_NAME,
450 "share_proto": "NFS",
451 "display_name": "foo_display_name",
452 }
454 self._mock_powerflex_manager.get_nfs_export_id.return_value = (
455 self.NFS_EXPORT_ID
456 )
457 self._mock_powerflex_manager.set_export_access.return_value = False
459 self.assertFalse(self._mock_powerflex_manager.get_nfs_export_id.called)
460 self.assertFalse(self._mock_powerflex_manager.set_export_access.called)
462 nfs_rw_ip = "192.168.0.10"
463 nfs_ro_ip = "192.168.0.11"
464 nfs_access_rw = {
465 "access_type": "ip",
466 "access_to": nfs_rw_ip,
467 "access_level": const.ACCESS_LEVEL_RW,
468 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd08",
469 }
470 nfs_access_ro = {
471 "access_type": "ip",
472 "access_to": nfs_ro_ip,
473 "access_level": const.ACCESS_LEVEL_RO,
474 "access_id": "09960614-8574-4e03-89cf-7cf267b0bd08",
475 }
476 access_rules = [nfs_access_rw, nfs_access_ro]
478 self.assertRaises(
479 exception.ShareBackendException,
480 self.storage_connection.update_access,
481 self.mock_context,
482 share,
483 access_rules,
484 add_rules=None,
485 delete_rules=None,
486 share_server=None,
487 )
489 def test_update_share_stats(self):
490 data = dict(
491 share_backend_name='powerflex',
492 vendor_name='Dell EMC',
493 storage_protocol='NFS_CIFS',
494 snapshot_support=True,
495 create_share_from_snapshot_support=True)
496 stats = dict(
497 maxCapacityInKb=4826330112,
498 capacityInUseInKb=53217280,
499 netUnusedCapacityInKb=1566080512,
500 primaryVacInKb=184549376)
502 self._mock_powerflex_manager.get_storage_pool_id.return_value = (
503 self.STORAGE_POOL_ID
504 )
505 self._mock_powerflex_manager.get_storage_pool_statistic. \
506 return_value = stats
507 self.storage_connection.update_share_stats(data)
508 self.assertEqual(data['storage_protocol'], 'NFS')
509 self.assertEqual(data['create_share_from_snapshot_support'], False)
510 self.assertEqual(data['driver_version'], connection.VERSION)
511 self.assertIsNotNone(data['pools'])
513 def test_get_default_filter_function(self):
514 filter = self.storage_connection.get_default_filter_function()
515 self.assertEqual(filter, "share.size >= 3")
517 def test_create_share_from_snapshot(self):
518 self.assertRaises(
519 NotImplementedError,
520 self.storage_connection.create_share_from_snapshot,
521 self.mock_context,
522 share=None,
523 snapshot=None,
524 )
526 def test_allow_access(self):
527 self.assertRaises(
528 NotImplementedError,
529 self.storage_connection.allow_access,
530 self.mock_context,
531 share=None,
532 access=None,
533 share_server=None,
534 )
536 def test_deny_access(self):
537 self.assertRaises(
538 NotImplementedError,
539 self.storage_connection.deny_access,
540 self.mock_context,
541 share=None,
542 access=None,
543 share_server=None,
544 )
546 def test_setup_server(self):
547 self.assertRaises(
548 NotImplementedError,
549 self.storage_connection.setup_server,
550 network_info=None,
551 )
553 def test_teardown_server(self):
554 self.assertRaises(
555 NotImplementedError,
556 self.storage_connection.teardown_server,
557 server_details=None,
558 )