Coverage for manila/tests/share/drivers/veritas/test_veritas_isa.py: 100%
319 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2017 Veritas Technologies LLC.
2#
3# Licensed under the Apache License, Version 2.0 (the "License"); you may
4# not use this file except in compliance with the License. You may obtain
5# a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12# License for the specific language governing permissions and limitations
13# under the License.
14"""
15Unit tests for Veritas Manila driver.
16"""
17import hashlib
18import json
19from unittest import mock
21from oslo_config import cfg
22import requests
24from manila import context
25from manila import exception
26from manila.share import configuration as conf
27from manila.share.drivers.veritas import veritas_isa
28from manila import test
31CONF = cfg.CONF
32FAKE_BACKEND = 'fake_backend'
35class MockResponse(object):
36 def __init__(self):
37 self.status_code = 200
39 def json(self):
40 data = {'fake_key': 'fake_val'}
41 return json.dumps(data)
44class ACCESSShareDriverTestCase(test.TestCase):
45 """Tests ACCESSShareDriver."""
47 share = {
48 'id': 'fakeid',
49 'name': 'fakename',
50 'size': 1,
51 'share_proto': 'NFS',
52 'export_locations': [{'path': '10.20.30.40:/vx/fake_location'}],
53 'snapshot_id': False
54 }
56 share2 = {
57 'id': 'fakeid2',
58 'name': 'fakename2',
59 'size': 4,
60 'share_proto': 'NFS',
61 }
63 share3 = {
64 'id': 'fakeid3',
65 'name': 'fakename3',
66 'size': 2,
67 'share_proto': 'NFS',
68 'export_location': '/vx/fake_location',
69 'snapshot_id': True
70 }
72 snapshot = {
73 'id': 'fakesnapshotid',
74 'share_name': 'fakename',
75 'share_id': 'fakeid',
76 'name': 'fakesnapshotname',
77 'share_size': 1,
78 'share_proto': 'NFS',
79 'snapshot_id': 'fake_snap_id',
80 }
82 access = {
83 'id': 'fakeaccid',
84 'access_type': 'ip',
85 'access_to': '10.0.0.2',
86 'access_level': 'rw',
87 'state': 'active',
88 }
90 access2 = {
91 'id': 'fakeaccid2',
92 'access_type': 'user',
93 'access_to': '10.0.0.3',
94 'access_level': 'rw',
95 'state': 'active',
96 }
98 access3 = {
99 'id': 'fakeaccid3',
100 'access_type': 'ip',
101 'access_to': '10.0.0.4',
102 'access_level': 'rw+',
103 'state': 'active',
104 }
106 access4 = {
107 'id': 'fakeaccid',
108 'access_type': 'ip',
109 'access_to': '10.0.0.2',
110 'access_level': 'ro',
111 'state': 'active',
112 }
114 def setUp(self):
115 super(ACCESSShareDriverTestCase, self).setUp()
116 self._create_fake_config()
117 lcfg = self.configuration
118 self._context = context.get_admin_context()
119 self._driver = veritas_isa.ACCESSShareDriver(False, configuration=lcfg)
120 self._driver.do_setup(self._context)
122 def _create_fake_config(self):
123 def _safe_get(opt):
124 return getattr(self.configuration, opt)
126 self.mock_object(veritas_isa.ACCESSShareDriver, '_authenticate_access')
127 self.configuration = mock.Mock(spec=conf.Configuration)
128 self.configuration.safe_get = mock.Mock(side_effect=_safe_get)
129 self.configuration.va_server_ip = '1.1.1.1'
130 self.configuration.va_pool = 'pool1'
131 self.configuration.va_user = 'user'
132 self.configuration.va_pwd = 'passwd'
133 self.configuration.va_port = 14161
134 self.configuration.va_ssl = 'False'
135 self.configuration.va_fstype = 'simple'
136 self.configuration.network_config_group = 'fake_network_config_group'
137 self.configuration.admin_network_config_group = (
138 'fake_admin_network_config_group')
139 self.configuration.driver_handles_share_servers = False
140 self.configuration.share_backend_name = FAKE_BACKEND
141 self.configuration.replication_domain = 'Disable'
142 self.configuration.filter_function = 'Disable'
143 self.configuration.goodness_function = 'Disable'
144 self.configuration.max_shares_per_share_server = -1
145 self.configuration.max_share_server_size = -1
147 def test_create_share(self):
148 self.mock_object(self._driver, '_get_va_share_name')
149 self.mock_object(self._driver, '_get_va_share_path')
150 self.mock_object(self._driver, '_get_vip')
151 self.mock_object(self._driver, '_access_api')
152 length = len(self.share['name'])
153 index = int(length / 2)
154 name1 = self.share['name'][:index]
155 name2 = self.share['name'][index:]
156 crc1 = hashlib.md5(name1.encode('utf-8')).hexdigest()[:8]
157 crc2 = hashlib.md5(name2.encode('utf-8')).hexdigest()[:8]
159 share_name_to_ret = crc1 + '-' + crc2
160 share_path_to_ret = '/vx/' + crc1 + '-' + crc2
162 self._driver._get_va_share_name.return_value = share_name_to_ret
163 self._driver._get_va_share_path.return_value = share_path_to_ret
165 self._driver._get_vip.return_value = '1.1.1.1'
166 self._driver.create_share(self._context, self.share)
168 self.assertEqual(1, self._driver._get_vip.call_count)
169 self.assertEqual(1, self._driver._get_va_share_name.call_count)
170 self.assertEqual(1, self._driver._get_va_share_path.call_count)
172 def test_create_share_negative(self):
173 self.mock_object(self._driver, '_access_api')
175 self._driver._access_api.return_value = False
177 self.assertRaises(exception.ShareBackendException,
178 self._driver.create_share,
179 self._context,
180 self.share)
182 def test_create_share_from_snapshot(self):
183 self.mock_object(self._driver, '_get_vip')
185 sharename = self._driver._get_va_share_name(
186 self.snapshot['share_name'])
187 snapname = self._driver._get_va_snap_name(self.snapshot['name'])
188 sharepath = self._driver._get_va_share_path(sharename)
189 self._driver._get_vip.return_value = '1.1.1.1'
190 vip = self._driver._get_vip()
191 location = (str(vip) + ':' +
192 str(sharepath) + ':' + str(snapname))
194 ret = self._driver.create_share_from_snapshot(self._context,
195 self.share,
196 self.snapshot)
197 self.assertEqual(location, ret)
199 def test_delete_share(self):
200 self.mock_object(self._driver, '_access_api')
201 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
202 self._driver._does_item_exist_at_va_backend.return_value = True
203 self._driver.delete_share(self._context, self.share)
204 self.assertEqual(2, self._driver._access_api.call_count)
206 def test_delete_share_negative(self):
207 self.mock_object(self._driver, '_access_api')
208 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
210 self._driver._does_item_exist_at_va_backend.return_value = True
211 self._driver._access_api.return_value = False
213 self.assertRaises(exception.ShareBackendException,
214 self._driver.delete_share,
215 self._context, self.share)
217 def test_delete_share_if_share_created_from_snap(self):
218 self.mock_object(self._driver, '_access_api')
219 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
221 self._driver.delete_share(self._context, self.share3)
222 self.assertEqual(0,
223 (self._driver.
224 _does_item_exist_at_va_backend.call_count))
225 self.assertEqual(0, self._driver._access_api.call_count)
227 def test_delete_share_if_not_present_at_backend(self):
228 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
229 self.mock_object(self._driver, '_access_api')
231 self._driver._does_item_exist_at_va_backend.return_value = False
232 self._driver.delete_share(self._context, self.share)
233 self.assertEqual(1,
234 (self._driver.
235 _does_item_exist_at_va_backend.call_count))
236 self.assertEqual(0, self._driver._access_api.call_count)
238 def test_create_snapshot(self):
239 self.mock_object(self._driver, '_access_api')
240 self._driver.create_snapshot(self._context, self.snapshot)
241 self.assertEqual(2, self._driver._access_api.call_count)
243 def test_create_snapshot_negative(self):
244 self.mock_object(self._driver, '_access_api')
246 self._driver._access_api.return_value = False
248 self.assertRaises(exception.ShareBackendException,
249 self._driver.create_snapshot,
250 self._context,
251 self.snapshot)
253 def test_delete_snapshot(self):
254 self.mock_object(self._driver, '_access_api')
255 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
257 self._driver._does_item_exist_at_va_backend.return_value = True
258 self._driver.delete_snapshot(self._context, self.snapshot)
259 self.assertEqual(2, self._driver._access_api.call_count)
261 def test_delete_snapshot_negative(self):
262 self.mock_object(self._driver, '_access_api')
263 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
265 self._driver._does_item_exist_at_va_backend.return_value = True
266 self._driver._access_api.return_value = False
268 self.assertRaises(exception.ShareBackendException,
269 self._driver.delete_snapshot,
270 self._context, self.snapshot)
272 def test_delete_snapshot_if_not_present_at_backend(self):
273 self.mock_object(self._driver, '_does_item_exist_at_va_backend')
274 self.mock_object(self._driver, '_access_api')
276 self._driver._does_item_exist_at_va_backend.return_value = False
277 self._driver.delete_snapshot(self._context, self.snapshot)
278 self.assertEqual(1,
279 (self._driver.
280 _does_item_exist_at_va_backend.call_count))
281 self.assertEqual(0, self._driver._access_api.call_count)
283 def test_update_access_for_allow(self):
284 self.mock_object(self._driver, '_access_api')
285 self._driver.update_access(self._context, self.share, [],
286 [self.access], [], [])
287 self.assertEqual(2, self._driver._access_api.call_count)
289 def test_update_access_for_allow_negative(self):
290 self.mock_object(self._driver, '_access_api')
291 self._driver._access_api.return_value = False
292 self.assertRaises(exception.ShareBackendException,
293 self._driver.update_access,
294 self._context,
295 self.share, [], [self.access], [], [])
297 self.assertRaises(exception.InvalidShareAccess,
298 self._driver.update_access,
299 self._context,
300 self.share, [], [self.access2], [], [])
302 self.assertRaises(exception.InvalidShareAccessLevel,
303 self._driver.update_access,
304 self._context,
305 self.share, [], [self.access3], [], [])
307 def test_update_access_for_deny(self):
308 self.mock_object(self._driver, '_access_api')
309 self._driver.update_access(self._context, self.share,
310 [], [], [self.access], [])
311 self.assertEqual(2, self._driver._access_api.call_count)
313 def test_update_access_for_deny_negative(self):
314 self.mock_object(self._driver, '_access_api')
315 self._driver._access_api.return_value = False
316 self.assertRaises(exception.ShareBackendException,
317 self._driver.update_access,
318 self._context,
319 self.share, [], [], [self.access], [])
321 def test_update_access_for_deny_for_invalid_access_type(self):
322 self.mock_object(self._driver, '_access_api')
323 self._driver.update_access(self._context, self.share,
324 [], [], [self.access2], [])
325 self.assertEqual(0, self._driver._access_api.call_count)
327 def test_update_access_for_empty_rule_list(self):
328 self.mock_object(self._driver, '_allow_access')
329 self.mock_object(self._driver, '_deny_access')
330 self._driver.update_access(self._context, self.share,
331 [], [], [], [])
332 self.assertEqual(0, self._driver._allow_access.call_count)
333 self.assertEqual(0, self._driver._deny_access.call_count)
335 def test_update_access_for_access_rules(self):
336 self.mock_object(self._driver, '_fetch_existing_rule')
337 self.mock_object(self._driver, '_allow_access')
338 self.mock_object(self._driver, '_deny_access')
340 existing_a_rules = [{'access_level': 'rw',
341 'access_type': 'ip',
342 'access_to': '10.0.0.2'},
343 {'access_level': 'rw',
344 'access_type': 'ip',
345 'access_to': '10.0.0.3'}]
347 self._driver._fetch_existing_rule.return_value = existing_a_rules
348 d_rule = self._driver._return_access_lists_difference(existing_a_rules,
349 [self.access4])
351 a_rule = self._driver._return_access_lists_difference([self.access4],
352 existing_a_rules)
353 self._driver.update_access(self._context, self.share,
354 [self.access4], [], [], [])
356 self.assertEqual(d_rule, existing_a_rules)
357 self.assertEqual(a_rule, [self.access4])
358 self.assertEqual(1, self._driver._allow_access.call_count)
359 self.assertEqual(2, self._driver._deny_access.call_count)
361 def test_extend_share(self):
362 self.mock_object(self._driver, '_access_api')
363 new_size = 3
364 self._driver.extend_share(self.share, new_size)
365 self.assertEqual(1, self._driver._access_api.call_count)
367 def test_extend_share_negative(self):
368 self.mock_object(self._driver, '_access_api')
370 new_size = 3
371 self._driver._access_api.return_value = False
372 self.assertRaises(exception.ShareBackendException,
373 self._driver.extend_share,
374 self.share, new_size)
376 def test_shrink_share(self):
377 self.mock_object(self._driver, '_access_api')
378 new_size = 3
379 self._driver.shrink_share(self.share2, new_size)
380 self.assertEqual(1, self._driver._access_api.call_count)
382 def test_shrink_share_negative(self):
383 self.mock_object(self._driver, '_access_api')
385 new_size = 3
386 self._driver._access_api.return_value = False
387 self.assertRaises(exception.ShareBackendException,
388 self._driver.shrink_share,
389 self.share2, new_size)
391 def test__get_access_pool_details(self):
392 self.mock_object(self._driver, '_access_api')
394 pool_details = []
395 pool_details_dict = {}
396 pool_details_dict['device_group_name'] = 'fake_pool'
397 pool_details_dict['capacity'] = 10737418240
398 pool_details_dict['used_size'] = 9663676416
399 pool_details.append(pool_details_dict)
401 pool_details_dict2 = {}
402 pool_details_dict2['device_group_name'] = self.configuration.va_pool
403 pool_details_dict2['capacity'] = 10737418240
404 pool_details_dict2['used_size'] = 9663676416
405 pool_details.append(pool_details_dict2)
407 self._driver._access_api.return_value = pool_details
408 total_space, free_space = self._driver._get_access_pool_details()
409 self.assertEqual(10, total_space)
410 self.assertEqual(1, free_space)
412 def test__get_access_pool_details_negative(self):
413 self.mock_object(self._driver, '_access_api')
415 pool_details = []
416 self._driver._access_api.return_value = pool_details
417 self.assertRaises(exception.ShareBackendException,
418 self._driver._get_access_pool_details)
420 def test__update_share_stats(self):
421 self.mock_object(self._driver, '_authenticate_access')
422 self.mock_object(self._driver, '_get_access_pool_details')
424 self._driver._get_access_pool_details.return_value = (10, 9)
425 self._driver._update_share_stats()
426 data = {
427 'share_backend_name': FAKE_BACKEND,
428 'vendor_name': 'Veritas',
429 'driver_version': '1.0',
430 'storage_protocol': 'NFS',
431 'total_capacity_gb': 10,
432 'free_capacity_gb': 9,
433 'reserved_percentage': 0,
434 'reserved_snapshot_percentage': 0,
435 'reserved_share_extend_percentage': 0,
436 'QoS_support': False,
437 'create_share_from_snapshot_support': True,
438 'driver_handles_share_servers': False,
439 'filter_function': 'Disable',
440 'goodness_function': 'Disable',
441 'mount_point_name_support': False,
442 'ipv4_support': True,
443 'ipv6_support': False,
444 'mount_snapshot_support': False,
445 'pools': None,
446 'qos': False,
447 'replication_domain': 'Disable',
448 'revert_to_snapshot_support': False,
449 'share_group_stats': {'consistent_snapshot_support': None},
450 'snapshot_support': True,
451 'security_service_update_support': False,
452 'share_server_multiple_subnet_support': False,
453 'network_allocation_update_support': False,
454 'share_replicas_migration_support': False,
455 'encryption_support': None,
456 }
458 self.assertEqual(data, self._driver._stats)
460 def test__get_vip(self):
461 self.mock_object(self._driver, '_get_access_ips')
463 pool_list = []
464 ip1 = {'isconsoleip': 1, 'type': 'Virtual',
465 'status': 'ONLINE', 'ip': '1.1.1.2'}
466 ip2 = {'isconsoleip': 0, 'type': 'Virtual',
467 'status': 'ONLINE', 'ip': '1.1.1.4'}
468 ip3 = {'isconsoleip': 0, 'type': 'Virtual',
469 'status': 'OFFLINE', 'ip': '1.1.1.5'}
470 ip4 = {'isconsoleip': 0, 'type': 'Physical',
471 'status': 'OFFLINE', 'ip': '1.1.1.6'}
473 pool_list = [ip1, ip2, ip3, ip4]
475 self._driver._get_access_ips.return_value = pool_list
476 vip = self._driver._get_vip()
477 self.assertEqual('1.1.1.4', vip)
479 def test__get_access_ips(self):
480 self.mock_object(self._driver, '_access_api')
481 ip_list = ['1.1.1.2', '1.1.2.3', '1.1.1.4']
482 self._driver._access_api.return_value = ip_list
483 ret_value = self._driver._get_access_ips(self._driver.session,
484 self._driver.host)
485 self.assertEqual(ret_value, ip_list)
487 def test__access_api(self):
488 self.mock_object(requests, 'session')
490 provider = '%s:%s' % (self._driver.host, self._driver._port)
491 path = '/fake/path'
492 input_data = {}
493 mock_response = MockResponse()
494 session = requests.session
496 data = {'fake_key': 'fake_val'}
497 json_data = json.dumps(data)
499 session.request.return_value = mock_response
500 ret_value = self._driver._access_api(session, provider, path,
501 json.dumps(input_data), 'GET')
503 self.assertEqual(json_data, ret_value)
505 def test__access_api_ret_for_update_object(self):
506 self.mock_object(requests, 'session')
508 provider = '%s:%s' % (self._driver.host, self._driver._port)
509 path = self._driver._update_object
510 input_data = None
511 mock_response = MockResponse()
512 session = requests.session
514 session.request.return_value = mock_response
515 ret = self._driver._access_api(session, provider, path,
516 input_data, 'GET')
518 self.assertTrue(ret)
520 def test__access_api_negative(self):
521 session = self._driver.session
522 provider = '%s:%s' % (self._driver.host, self._driver._port)
523 path = '/fake/path'
524 input_data = {}
525 ret_value = self._driver._access_api(session, provider, path,
526 json.dumps(input_data), 'GET')
527 self.assertEqual(False, ret_value)
529 def test__get_api(self):
530 provider = '%s:%s' % (self._driver.host, self._driver._port)
531 tail = '/fake/path'
532 ret = self._driver._get_api(provider, tail)
534 api_root = 'https://%s/api' % (provider)
535 to_be_ret = api_root + tail
536 self.assertEqual(to_be_ret, ret)
538 def test__does_item_exist_at_va_backend(self):
539 self.mock_object(self._driver, '_access_api')
540 item_name = 'fake_item'
541 path = '/fake/path'
542 fake_item_list = [{'name': item_name}]
543 self._driver._access_api.return_value = fake_item_list
544 ret_value = self._driver._does_item_exist_at_va_backend(item_name,
545 path)
546 self.assertTrue(ret_value)
548 def test__does_item_exist_at_va_backend_negative(self):
549 self.mock_object(self._driver, '_access_api')
550 item_name = 'fake_item'
551 path = '/fake/path'
552 fake_item_list = [{'name': 'item2'}]
553 self._driver._access_api.return_value = fake_item_list
554 ret_value = self._driver._does_item_exist_at_va_backend(item_name,
555 path)
556 self.assertEqual(False, ret_value)
558 def test__fetch_existing_rule(self):
559 self.mock_object(self._driver, '_access_api')
560 fake_share = 'fake-share'
561 fake_access_list = []
562 list1 = []
563 list1.append({
564 'status': 'online',
565 'name': '/vx/fake-share',
566 'host_name': '10.0.0.1',
567 'privilege': 'rw'
568 })
569 list1.append({
570 'status': 'online',
571 'name': '/vx/fake-share',
572 'host_name': '10.0.0.2',
573 'privilege': 'rw'
574 })
575 list1.append({
576 'status': 'online',
577 'name': '/vx/fake-share',
578 'host_name': '10.0.0.3',
579 'privilege': 'ro'
580 })
581 list1.append({
582 'status': 'online',
583 'name': '/vx/fake-share2',
584 'host_name': '10.0.0.4',
585 'privilege': 'rw'
586 })
588 fake_access_list.append({
589 'shareType': 'NFS',
590 'shares': list1
591 })
593 fake_access_list.append({
594 'shareType': 'CIFS',
595 'shares': []
596 })
598 ret_access_list = []
599 ret_access_list.append({
600 'access_to': '10.0.0.1',
601 'access_level': 'rw',
602 'access_type': 'ip'
603 })
605 ret_access_list.append({
606 'access_to': '10.0.0.2',
607 'access_level': 'rw',
608 'access_type': 'ip'
609 })
611 ret_access_list.append({
612 'access_to': '10.0.0.3',
613 'access_level': 'ro',
614 'access_type': 'ip'
615 })
617 self._driver._access_api.return_value = fake_access_list
618 ret_value = self._driver._fetch_existing_rule(fake_share)
619 self.assertEqual(ret_access_list, ret_value)