Coverage for manila/tests/share/test_share_utils.py: 100%
113 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2011 OpenStack Foundation
2# Copyright (c) 2015 Rushil Chugh
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.
17"""Tests For miscellaneous util methods used with share."""
19from unittest import mock
21import ddt
23from manila.common import constants
24from manila.share import utils as share_utils
25from manila import test
28@ddt.ddt
29class ShareUtilsTestCase(test.TestCase):
30 def test_extract_host_without_pool(self):
31 host = 'Host@Backend'
32 self.assertEqual(
33 'Host@Backend', share_utils.extract_host(host))
35 def test_extract_host_only_return_host(self):
36 host = 'Host@Backend'
37 self.assertEqual(
38 'Host', share_utils.extract_host(host, 'host'))
40 def test_extract_host_only_return_pool(self):
41 host = 'Host@Backend'
42 self.assertIsNone(
43 share_utils.extract_host(host, 'pool'))
45 def test_extract_host_only_return_backend(self):
46 host = 'Host@Backend'
47 self.assertEqual(
48 'Host@Backend', share_utils.extract_host(host, 'backend'))
50 def test_extract_host_missing_backend_and_pool(self):
51 host = 'Host'
52 # Default level is 'backend'
53 self.assertEqual(
54 'Host', share_utils.extract_host(host))
56 def test_extract_host_only_return_backend_name(self):
57 host = 'Host@Backend#Pool'
58 self.assertEqual(
59 'Backend', share_utils.extract_host(host, 'backend_name'))
61 def test_extract_host_only_return_backend_name_index_error(self):
62 host = 'Host#Pool'
64 self.assertRaises(IndexError,
65 share_utils.extract_host,
66 host, 'backend_name')
68 def test_extract_host_missing_backend(self):
69 host = 'Host#Pool'
70 self.assertEqual(
71 'Host', share_utils.extract_host(host))
72 self.assertEqual(
73 'Host', share_utils.extract_host(host, 'host'))
75 def test_extract_host_missing_backend_only_return_backend(self):
76 host = 'Host#Pool'
77 self.assertEqual(
78 'Host', share_utils.extract_host(host, 'backend'))
80 def test_extract_host_missing_backend_only_return_pool(self):
81 host = 'Host#Pool'
82 self.assertEqual(
83 'Pool', share_utils.extract_host(host, 'pool'))
84 self.assertEqual(
85 'Pool', share_utils.extract_host(host, 'pool', True))
87 def test_extract_host_missing_pool(self):
88 host = 'Host@Backend'
89 self.assertIsNone(
90 share_utils.extract_host(host, 'pool'))
92 def test_extract_host_missing_pool_use_default_pool(self):
93 host = 'Host@Backend'
94 self.assertEqual(
95 '_pool0', share_utils.extract_host(host, 'pool', True))
97 def test_extract_host_with_default_pool(self):
98 host = 'Host'
99 # Default_pool_name doesn't work for level other than 'pool'
100 self.assertEqual(
101 'Host', share_utils.extract_host(host, 'host', True))
102 self.assertEqual(
103 'Host', share_utils.extract_host(host, 'host', False))
104 self.assertEqual(
105 'Host', share_utils.extract_host(host, 'backend', True))
106 self.assertEqual(
107 'Host', share_utils.extract_host(host, 'backend', False))
109 def test_extract_host_with_pool(self):
110 host = 'Host@Backend#Pool'
111 self.assertEqual(
112 'Host@Backend', share_utils.extract_host(host))
113 self.assertEqual(
114 'Host', share_utils.extract_host(host, 'host'))
115 self.assertEqual(
116 'Host@Backend', share_utils.extract_host(host, 'backend'),)
117 self.assertEqual(
118 'Pool', share_utils.extract_host(host, 'pool'))
119 self.assertEqual(
120 'Pool', share_utils.extract_host(host, 'pool', True))
122 def test_append_host_with_host_and_pool(self):
123 host = 'Host'
124 pool = 'Pool'
125 expected = 'Host#Pool'
126 self.assertEqual(expected,
127 share_utils.append_host(host, pool))
129 def test_append_host_with_host(self):
130 host = 'Host'
131 pool = None
132 expected = 'Host'
133 self.assertEqual(expected,
134 share_utils.append_host(host, pool))
136 def test_append_host_with_pool(self):
137 host = None
138 pool = 'pool'
139 expected = None
140 self.assertEqual(expected,
141 share_utils.append_host(host, pool))
143 def test_append_host_with_no_values(self):
144 host = None
145 pool = None
146 expected = None
147 self.assertEqual(expected,
148 share_utils.append_host(host, pool))
150 def test_get_active_replica_success(self):
151 replica_list = [{'id': '123456',
152 'replica_state': constants.REPLICA_STATE_IN_SYNC},
153 {'id': '654321',
154 'replica_state': constants.REPLICA_STATE_ACTIVE},
155 ]
156 replica = share_utils.get_active_replica(replica_list)
157 self.assertEqual('654321', replica['id'])
159 def test_get_active_replica_not_exist(self):
160 replica_list = [{'id': '123456',
161 'replica_state': constants.REPLICA_STATE_IN_SYNC},
162 {'id': '654321',
163 'replica_state': constants.REPLICA_STATE_OUT_OF_SYNC},
164 ]
165 replica = share_utils.get_active_replica(replica_list)
166 self.assertIsNone(replica)
168 @ddt.data(
169 {'fake_subnet': [{'neutron_net_id': 'fake_nn_id',
170 'neutron_subnet_id': 'fake_nsb_id'}],
171 'fake_new_subnet': [{'neutron_net_id': 'fake_nn_id',
172 'neutron_subnet_id': 'fake_nsb_id'}],
173 'is_compatible': True},
174 {'fake_subnet': [{'neutron_net_id': 'fake_nn_id',
175 'neutron_subnet_id': 'fake_nsb_id'}],
176 'fake_new_subnet': [{'neutron_net_id': 'fake_nn_id',
177 'neutron_subnet_id': 'fake_nsb_id2'}],
178 'is_compatible': False},
179 {'fake_subnet': [{'neutron_net_id': 'fake_nn_id',
180 'neutron_subnet_id': 'fake_nsb_id'},
181 {'neutron_net_id': 'fake_nn_id2',
182 'neutron_subnet_id': 'fake_nsb_id2'}],
183 'fake_new_subnet': [{'neutron_net_id': 'fake_nn_id',
184 'neutron_subnet_id': 'fake_nsb_id'}],
185 'is_compatible': False}
186 )
187 @ddt.unpack
188 def test_is_az_subnets_compatible(self, fake_subnet, fake_new_subnet,
189 is_compatible):
190 expected_result = is_compatible
191 result = share_utils.is_az_subnets_compatible(fake_subnet,
192 fake_new_subnet)
193 self.assertEqual(expected_result, result)
196class NotifyUsageTestCase(test.TestCase):
197 @mock.patch('manila.share.utils._usage_from_share')
198 @mock.patch('manila.share.utils.CONF')
199 @mock.patch('manila.share.utils.rpc')
200 def test_notify_about_share_usage(self, mock_rpc, mock_conf, mock_usage):
201 mock_conf.host = 'host1'
202 output = share_utils.notify_about_share_usage(mock.sentinel.context,
203 mock.sentinel.share,
204 mock.sentinel.
205 share_instance,
206 'test_suffix')
207 self.assertIsNone(output)
208 mock_usage.assert_called_once_with(mock.sentinel.share,
209 mock.sentinel.share_instance)
210 mock_rpc.get_notifier.assert_called_once_with('share',
211 'host1')
212 mock_rpc.get_notifier.return_value.info.assert_called_once_with(
213 mock.sentinel.context,
214 'share.test_suffix',
215 mock_usage.return_value)
217 @mock.patch('manila.share.utils._usage_from_share')
218 @mock.patch('manila.share.utils.CONF')
219 @mock.patch('manila.share.utils.rpc')
220 def test_notify_about_share_usage_with_kwargs(self, mock_rpc, mock_conf,
221 mock_usage):
222 mock_conf.host = 'host1'
223 output = share_utils.notify_about_share_usage(mock.sentinel.context,
224 mock.sentinel.share,
225 mock.sentinel.
226 share_instance,
227 'test_suffix',
228 extra_usage_info={
229 'a': 'b', 'c': 'd'},
230 host='host2')
231 self.assertIsNone(output)
232 mock_usage.assert_called_once_with(mock.sentinel.share,
233 mock.sentinel.share_instance,
234 a='b', c='d')
235 mock_rpc.get_notifier.assert_called_once_with('share',
236 'host2')
237 mock_rpc.get_notifier.return_value.info.assert_called_once_with(
238 mock.sentinel.context,
239 'share.test_suffix',
240 mock_usage.return_value)