Coverage for manila/tests/share/drivers/container/test_storage_helper.py: 99%
151 statements
« prev ^ index » next coverage.py v7.11.0, created at 2026-02-18 22:19 +0000
1# Copyright 2016 Mirantis, Inc.
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.
15"""Unit tests for the Storage helper module."""
17import functools
18from unittest import mock
20import ddt
22from manila import exception
23from manila.share import configuration
24from manila.share.drivers.container import storage_helper
25from manila import test
26from manila.tests import fake_share as base_fake_share
27from manila.tests.share.drivers.container.fakes import fake_share
30@ddt.ddt
31class LVMHelperTestCase(test.TestCase):
32 """Tests ContainerShareDriver"""
34 def setUp(self):
35 super(LVMHelperTestCase, self).setUp()
36 self.share = fake_share()
37 self.fake_conf = configuration.Configuration(None)
38 self.fake_conf.container_volume_mount_path = "/tmp/shares"
39 self.LVMHelper = storage_helper.LVMHelper(configuration=self.fake_conf)
40 self.context = mock.Mock()
42 def fake_exec_sync(self, *args, **kwargs):
43 kwargs['execute_arguments'].append(args)
44 try:
45 ret_val = kwargs['ret_val']
46 except KeyError:
47 ret_val = None
48 return ret_val
50 def test_lvmhelper_setup_explodes_in_gore_on_no_config_supplied(self):
51 self.assertRaises(exception.ManilaException,
52 storage_helper.LVMHelper,
53 None)
55 @ddt.data("62.50g 72.50g", " 72.50g 62.50g\n", " <62.50g <72.50g\n")
56 def test_get_share_server_pools(self, ret_vgs):
57 expected_result = [{'reserved_percentage': 0,
58 'reserved_snapshot_percentage': 0,
59 'reserved_share_extend_percentage': 0,
60 'pool_name': 'manila_docker_volumes',
61 'total_capacity_gb': 72.5,
62 'free_capacity_gb': 62.5}]
63 self.mock_object(self.LVMHelper, "_execute",
64 mock.Mock(return_value=(ret_vgs, 0)))
66 result = self.LVMHelper.get_share_server_pools()
68 self.assertEqual(expected_result, result)
70 def test__get_lv_device(self):
71 fake_share_name = 'fakeshareid'
72 self.assertEqual("/dev/manila_docker_volumes/%s" % fake_share_name,
73 self.LVMHelper._get_lv_device(fake_share_name))
75 def test__get_lv_folder(self):
76 fake_share_name = 'fakeshareid'
77 self.assertEqual("/tmp/shares/%s" % fake_share_name,
78 self.LVMHelper._get_lv_folder(fake_share_name))
80 def test_provide_storage(self):
81 actual_arguments = []
82 fake_share_name = 'fakeshareid'
83 expected_arguments = [
84 ('lvcreate', '-p', 'rw', '-L', '1G', '-n', 'fakeshareid',
85 'manila_docker_volumes'),
86 ('mkfs.ext4', '/dev/manila_docker_volumes/fakeshareid'),
87 ]
88 self.LVMHelper._execute = functools.partial(
89 self.fake_exec_sync, execute_arguments=actual_arguments,
90 ret_val='')
92 self.LVMHelper.provide_storage(fake_share_name, 1)
94 self.assertEqual(expected_arguments, actual_arguments)
96 @ddt.data(None, exception.ProcessExecutionError)
97 def test__try_to_unmount_device(self, side_effect):
98 device = {}
99 mock_warning = self.mock_object(storage_helper.LOG, 'warning')
100 mock_execute = self.mock_object(self.LVMHelper, '_execute',
101 mock.Mock(side_effect=side_effect))
102 self.LVMHelper._try_to_unmount_device(device)
104 mock_execute.assert_called_once_with(
105 "umount", device, run_as_root=True
106 )
107 if side_effect is not None:
108 mock_warning.assert_called_once()
110 def test_remove_storage(self):
111 fake_share_name = 'fakeshareid'
112 fake_device = {}
114 mock_get_lv_device = self.mock_object(
115 self.LVMHelper, '_get_lv_device',
116 mock.Mock(return_value=fake_device))
117 mock_try_to_umount = self.mock_object(self.LVMHelper,
118 '_try_to_unmount_device')
119 mock_execute = self.mock_object(self.LVMHelper, '_execute')
121 self.LVMHelper.remove_storage(fake_share_name)
123 mock_get_lv_device.assert_called_once_with(
124 fake_share_name
125 )
126 mock_try_to_umount.assert_called_once_with(fake_device)
127 mock_execute.assert_called_once_with(
128 'lvremove', '-f', '--autobackup', 'n', fake_device,
129 run_as_root=True
130 )
132 def test_remove_storage_lvremove_failed(self):
133 fake_share_name = 'fakeshareid'
135 def fake_execute(*args, **kwargs):
136 if 'lvremove' in args:
137 raise exception.ProcessExecutionError()
139 self.mock_object(storage_helper.LOG, "warning")
140 self.mock_object(self.LVMHelper, "_execute", fake_execute)
142 self.LVMHelper.remove_storage(fake_share_name)
144 self.assertTrue(storage_helper.LOG.warning.called)
146 @ddt.data(None, exception.ProcessExecutionError)
147 def test_rename_storage(self, side_effect):
148 fake_old_share_name = 'fake_old_name'
149 fake_new_share_name = 'fake_new_name'
150 fake_new_device = "/dev/new_device"
151 fake_old_device = "/dev/old_device"
153 mock_get_lv_device = self.mock_object(
154 self.LVMHelper, '_get_lv_device',
155 mock.Mock(side_effect=[fake_old_device, fake_new_device]))
156 mock_try_to_umount = self.mock_object(self.LVMHelper,
157 '_try_to_unmount_device')
159 mock_execute = self.mock_object(self.LVMHelper, '_execute',
160 mock.Mock(side_effect=side_effect))
162 if side_effect is None:
163 self.LVMHelper.rename_storage(fake_old_share_name,
164 fake_new_share_name)
165 else:
166 self.assertRaises(exception.ProcessExecutionError,
167 self.LVMHelper.rename_storage,
168 fake_old_share_name, fake_new_share_name)
169 mock_try_to_umount.assert_called_once_with(fake_old_device)
170 mock_execute.assert_called_once_with(
171 "lvrename", "--autobackup", "n", fake_old_device, fake_new_device,
172 run_as_root=True
173 )
174 mock_get_lv_device.assert_has_calls([
175 mock.call(fake_old_share_name),
176 mock.call(fake_new_share_name)
177 ])
179 def test_extend_share(self):
180 actual_arguments = []
181 expected_arguments = [
182 ('lvextend', '-L', 'shareG', '-n',
183 '/dev/manila_docker_volumes/fakeshareid'),
184 ('e2fsck', '-f', '-y', '/dev/manila_docker_volumes/fakeshareid'),
185 ('resize2fs', '/dev/manila_docker_volumes/fakeshareid'),
186 ]
187 fake_share_name = 'fakeshareid'
188 self.LVMHelper._execute = functools.partial(
189 self.fake_exec_sync, execute_arguments=actual_arguments,
190 ret_val='')
192 self.LVMHelper.extend_share(fake_share_name, 'share', 3)
194 self.assertEqual(expected_arguments, actual_arguments)
196 def test_get_size(self):
197 share_name = 'fakeshareid'
198 fake_old_device = {}
200 mock_get_lv_device = self.mock_object(
201 self.LVMHelper, '_get_lv_device',
202 mock.Mock(return_value=fake_old_device))
203 mock_execute = self.mock_object(self.LVMHelper, '_execute',
204 mock.Mock(return_value=[1, "args"]))
206 result = self.LVMHelper.get_size(share_name)
208 mock_execute.assert_called_once_with(
209 "lvs", "-o", "lv_size", "--noheadings", "--nosuffix", "--units",
210 "g", fake_old_device, run_as_root=True
211 )
212 mock_get_lv_device.assert_called_once_with(share_name)
213 self.assertEqual(result, 1)
215 @ddt.data({'source_host': 'host@back1#vg1', 'dest_host': 'host@back2#vg2',
216 'compatible': False},
217 {'source_host': 'host@back1#vg1', 'dest_host': 'host@back2#vg1',
218 'compatible': True},
219 {'source_host': 'host@back1#vg1', 'dest_host': 'host@back1#vg1',
220 'compatible': True})
221 @ddt.unpack
222 def test_migration_check_compatibility(
223 self, source_host, dest_host, compatible):
224 mock_exception_log = self.mock_object(storage_helper.LOG, 'exception')
226 source_share = base_fake_share.fake_share_instance(host=source_host)
227 dest_share = base_fake_share.fake_share_instance(host=dest_host)
229 migration_compatibility = self.LVMHelper.migration_check_compatibility(
230 self.context, source_share, dest_share, share_server=None,
231 destination_share_server=None)
233 expected_compatibility = {
234 'compatible': compatible,
235 'writable': True,
236 'nondisruptive': False,
237 'preserve_metadata': True,
238 'preserve_snapshots': False,
239 }
240 self.assertDictEqual(expected_compatibility, migration_compatibility)
241 if not compatible:
242 mock_exception_log.assert_called_once()
244 def test_migration_continue(self):
245 end1Phase = self.LVMHelper.migration_continue(
246 self.context, None, None, None, None, share_server=None,
247 destination_share_server=None)
248 self.assertTrue(end1Phase)
250 def test_migration_get_progress(self):
251 progress = self.LVMHelper.migration_get_progress(
252 self.context, None, None, None, None, share_server=None,
253 destination_share_server=None)
254 expected_progress = {
255 'total_progress': 100,
256 }
257 self.assertDictEqual(expected_progress, progress)
259 @ddt.data({'source_host': 'host@back1', 'dest_host': 'host@back1',
260 'shares_specs': {}},
261 {'source_host': 'host@back1', 'dest_host': 'host@back2#vg1',
262 'shares_specs': {'shares_req_spec': [
263 {'share_instance_properties': {'host': 'host@back1#vg2'}}
264 ]}})
265 @ddt.unpack
266 def test_share_server_migration_check_compatibility_false(
267 self, source_host, dest_host, shares_specs):
268 not_compatible = {
269 'compatible': False,
270 'writable': None,
271 'nondisruptive': None,
272 'preserve_snapshots': None,
273 'migration_cancel': None,
274 'migration_get_progress': None,
275 }
276 mock_error_log = self.mock_object(storage_helper.LOG, 'error')
278 source_server = {'id': 'fake_id', 'host': source_host}
279 migration_compatibility = (
280 self.LVMHelper.share_server_migration_check_compatibility(
281 self.context, source_server, dest_host, None, None,
282 shares_specs))
284 self.assertDictEqual(not_compatible, migration_compatibility)
285 mock_error_log.assert_called_once()
287 @ddt.data({'source_host': 'host@back1', 'dest_host': 'host@back2#vg1',
288 'shares_specs': {'shares_req_spec': [
289 {'share_instance_properties': {'host': 'host@back1#vg1'}}
290 ]}},
291 {'source_host': 'host@back1', 'dest_host': 'host@back2',
292 'shares_specs': {'shares_req_spec': [
293 {'share_instance_properties': {'host': 'host@back1#vg1'}}
294 ]}})
295 @ddt.unpack
296 def test_share_server_migration_check_compatibility_true(
297 self, source_host, dest_host, shares_specs):
298 compatible = {
299 'compatible': True,
300 'writable': True,
301 'nondisruptive': False,
302 'preserve_snapshots': False,
303 'migration_cancel': True,
304 'migration_get_progress': True,
305 }
307 source_server = {'id': 'fake_id', 'host': source_host}
308 migration_compatibility = (
309 self.LVMHelper.share_server_migration_check_compatibility(
310 self.context, source_server, dest_host, None, None,
311 shares_specs))
313 self.assertDictEqual(compatible, migration_compatibility)
315 def test_share_server_migration_continue(self):
316 end1Phase = self.LVMHelper.share_server_migration_continue(
317 self.context, None, None, None, None)
318 self.assertTrue(end1Phase)
320 def test_share_server_migration_get_progess(self):
321 progress = self.LVMHelper.share_server_migration_get_progress(
322 self.context, None, None, None, None)
323 expected_progress = {
324 'total_progress': 100,
325 }
326 self.assertDictEqual(expected_progress, progress)
328 def test_get_share_pool_name(self):
329 fake_vg_name = 'fake_vg'
330 self.LVMHelper.configuration.container_volume_group = fake_vg_name
332 vg_name = self.LVMHelper.get_share_pool_name('fake_share_id')
333 self.assertEqual(vg_name, fake_vg_name)