Coverage for manila/tests/share/drivers/dell_emc/plugins/unity/test_utils.py: 95%
93 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright (c) 2016 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.
16import ddt
17from oslo_utils import units
19from manila.share.drivers.dell_emc.plugins.unity import utils
20from manila import test
23class MockSP(object):
24 def __init__(self, sp_id):
25 self.sp_id = sp_id
27 def get_id(self):
28 return self.sp_id
31SPA = MockSP('spa')
32SPB = MockSP('spb')
35class MockPort(object):
36 def __init__(self, sp, port_id, mtu):
37 self._sp = sp
38 self.port_id = port_id
39 self.mtu = mtu
41 def get_id(self):
42 return self.port_id
44 @property
45 def parent_storage_processor(self):
46 return self._sp
49SPA_ETH0 = MockPort(SPA, 'spa_eth0', 1500)
50SPA_ETH1 = MockPort(SPA, 'spa_eth1', 9000)
51SPB_ETH0 = MockPort(SPB, 'spb_eth0', 1500)
52SPB_ETH1 = MockPort(SPB, 'spb_eth1', 9000)
53SPA_LA1 = MockPort(SPA, 'spa_la_1', 1500)
54SPB_LA1 = MockPort(SPB, 'spb_la_1', 1500)
57class MockPortV5(object):
58 def __init__(self, sp, port_id, mtu):
59 self._sp = sp
60 self.port_id = port_id
61 self.mtu = mtu
63 def get_id(self):
64 return self.port_id
66 @property
67 def storage_processor(self):
68 return self._sp
71SPA_LA2 = MockPort(SPA, 'spa_la_2', 1500)
72SPB_LA2 = MockPort(SPB, 'spb_la_2', 1500)
75@ddt.ddt
76class TestUtils(test.TestCase):
77 @ddt.data({'matcher': None,
78 'matched': {'pool_1', 'pool_2', 'nas_server_pool'},
79 'not_matched': set()},
80 {'matcher': ['*'],
81 'matched': {'pool_1', 'pool_2', 'nas_server_pool'},
82 'not_matched': set()},
83 {'matcher': ['pool_*'],
84 'matched': {'pool_1', 'pool_2'},
85 'not_matched': {'nas_server_pool'}},
86 {'matcher': ['*pool'],
87 'matched': {'nas_server_pool'},
88 'not_matched': {'pool_1', 'pool_2'}},
89 {'matcher': ['nas_server_pool'],
90 'matched': {'nas_server_pool'},
91 'not_matched': {'pool_1', 'pool_2'}},
92 {'matcher': ['nas_*', 'pool_*'],
93 'matched': {'pool_1', 'pool_2', 'nas_server_pool'},
94 'not_matched': set()})
95 def test_do_match(self, data):
96 full = ['pool_1 ', ' pool_2', ' nas_server_pool ']
97 matcher = data['matcher']
98 expected_matched = data['matched']
99 expected_not_matched = data['not_matched']
101 matched, not_matched = utils.do_match(full, matcher)
102 self.assertEqual(expected_matched, matched)
103 self.assertEqual(expected_not_matched, not_matched)
105 @ddt.data({'ports': [SPA_ETH0, SPB_ETH0],
106 'ids_conf': None,
107 'port_map': {'spa': {'spa_eth0'}, 'spb': {'spb_eth0'}},
108 'unmanaged': set()},
109 {'ports': [SPA_ETH0, SPB_ETH0],
110 'ids_conf': [' '],
111 'port_map': {'spa': {'spa_eth0'}, 'spb': {'spb_eth0'}},
112 'unmanaged': set()},
113 {'ports': [SPA_ETH0, SPB_ETH0, SPA_ETH1],
114 'ids_conf': ['spa*'],
115 'port_map': {'spa': {'spa_eth0', 'spa_eth1'}},
116 'unmanaged': {'spb_eth0'}},
117 {'ports': [SPA_LA2, SPB_LA2],
118 'ids_conf': None,
119 'port_map': {'spa': {'spa_la_2'}, 'spb': {'spb_la_2'}},
120 'unmanaged': set()},
121 )
122 @ddt.unpack
123 def test_match_ports(self, ports, ids_conf, port_map, unmanaged):
124 sp_ports_map, unmanaged_port_ids = utils.match_ports(ports,
125 ids_conf)
126 self.assertEqual(port_map, sp_ports_map)
127 self.assertEqual(unmanaged, unmanaged_port_ids)
129 def test_find_ports_by_mtu(self):
130 all_ports = [SPA_ETH0, SPB_ETH0, SPA_ETH1, SPB_ETH1, SPA_LA1,
131 SPB_LA1]
132 port_ids_conf = '*'
133 port_map = utils.find_ports_by_mtu(all_ports, port_ids_conf, 1500)
134 self.assertEqual({'spa': {'spa_eth0', 'spa_la_1'},
135 'spb': {'spb_eth0', 'spb_la_1'}},
136 port_map)
138 def test_gb_to_byte(self):
139 self.assertEqual(3 * units.Gi, utils.gib_to_byte(3))
141 def test_get_snapshot_id(self):
142 snapshot = {'provider_location': '23047-ef2344-4563cvw-r4323cwed',
143 'id': 'test_id'}
144 result = utils.get_snapshot_id(snapshot)
145 expected = '23047-ef2344-4563cvw-r4323cwed'
146 self.assertEqual(expected, result)
148 def test_get_snapshot_id_without_pl(self):
149 snapshot = {'provider_location': '', 'id': 'test_id'}
150 result = utils.get_snapshot_id(snapshot)
151 expected = 'test_id'
152 self.assertEqual(expected, result)
154 def test_get_nfs_share_id(self):
155 nfs_share = {'export_locations':
156 [{'path': '10.10.1.12:/addf-97e-46c-8ac6-55922f',
157 'share_instance_id': 'e24-457e-47-12c6-gf345'}],
158 'share_proto': 'NFS', 'id': 'test_nfs_id'}
159 result = utils.get_share_backend_id(nfs_share)
160 expected = 'addf-97e-46c-8ac6-55922f'
161 self.assertEqual(expected, result)
163 def test_get_nfs_share_id_without_path(self):
164 nfs_share = {'export_locations':
165 [{'path': '',
166 'share_instance_id': 'ev24-7e-4-12c6-g45245'}],
167 'share_proto': 'NFS', 'id': 'test_nfs_id'}
168 result = utils.get_share_backend_id(nfs_share)
169 expected = 'test_nfs_id'
170 self.assertEqual(expected, result)
172 def test_get_cifs_share_id(self):
173 cifs_share = {'export_locations':
174 [{'path': '\\\\17.66.5.3\\bdf-h4e-42c-122c5-b212',
175 'share_instance_id': 'ev4-47e-48-126-gfbh452'}],
176 'share_proto': 'CIFS', 'id': 'test_cifs_id'}
177 result = utils.get_share_backend_id(cifs_share)
178 expected = 'bdf-h4e-42c-122c5-b212'
179 self.assertEqual(expected, result)
181 def test_get_cifs_share_id_without_path(self):
182 cifs_share = {'export_locations':
183 [{'path': '',
184 'share_instance_id': 'ef4-47e-48-12c6-gf452'}],
185 'share_proto': 'CIFS', 'id': 'test_cifs_id'}
186 result = utils.get_share_backend_id(cifs_share)
187 expected = 'test_cifs_id'
188 self.assertEqual(expected, result)