]>
git.proxmox.com Git - ceph.git/blob - ceph/src/ceph-volume/ceph_volume/tests/api/test_lvm.py
3 from mock
.mock
import patch
4 from ceph_volume
import process
, exceptions
5 from ceph_volume
.api
import lvm
as api
8 class TestParseTags(object):
10 def test_no_tags_means_empty_dict(self
):
11 result
= api
.parse_tags('')
14 def test_single_tag_gets_parsed(self
):
15 result
= api
.parse_tags('ceph.osd_something=1')
16 assert result
== {'ceph.osd_something': '1'}
18 def test_non_ceph_tags_are_skipped(self
):
19 result
= api
.parse_tags('foo')
22 def test_mixed_non_ceph_tags(self
):
23 result
= api
.parse_tags('foo,ceph.bar=1')
24 assert result
== {'ceph.bar': '1'}
26 def test_multiple_csv_expands_in_dict(self
):
27 result
= api
.parse_tags('ceph.osd_something=1,ceph.foo=2,ceph.fsid=0000')
28 # assert them piecemeal to avoid the un-ordered dict nature
29 assert result
['ceph.osd_something'] == '1'
30 assert result
['ceph.foo'] == '2'
31 assert result
['ceph.fsid'] == '0000'
34 class TestVolume(object):
36 def test_is_ceph_device(self
):
37 lv_tags
= "ceph.type=data,ceph.osd_id=0"
38 osd
= api
.Volume(lv_name
='osd/volume', lv_tags
=lv_tags
)
39 assert api
.is_ceph_device(osd
)
41 @pytest.mark
.parametrize('dev',[
43 api
.VolumeGroup(vg_name
='foo'),
44 api
.Volume(lv_name
='vg/no_osd', lv_tags
='', lv_path
='lv/path'),
45 api
.Volume(lv_name
='vg/no_osd', lv_tags
='ceph.osd_id=null', lv_path
='lv/path'),
48 def test_is_not_ceph_device(self
, dev
):
49 assert not api
.is_ceph_device(dev
)
51 def test_no_empty_lv_name(self
):
52 with pytest
.raises(ValueError):
53 api
.Volume(lv_name
='', lv_tags
='')
56 class TestVolumeGroup(object):
58 def test_volume_group_no_empty_name(self
):
59 with pytest
.raises(ValueError):
60 api
.VolumeGroup(vg_name
='')
63 class TestVolumeGroupFree(object):
65 def test_integer_gets_produced(self
):
66 vg
= api
.VolumeGroup(vg_name
='nosize', vg_free_count
=100, vg_extent_size
=4194304)
67 assert vg
.free
== 100 * 4194304
70 class TestCreateLVs(object):
72 def setup_method(self
):
73 self
.vg
= api
.VolumeGroup(vg_name
='ceph',
74 vg_extent_size
=1073741824,
75 vg_extent_count
=99999999,
78 def test_creates_correct_lv_number_from_parts(self
, monkeypatch
):
79 monkeypatch
.setattr('ceph_volume.api.lvm.create_lv', lambda *a
, **kw
: (a
, kw
))
80 lvs
= api
.create_lvs(self
.vg
, parts
=4)
83 def test_suffixes_the_size_arg(self
, monkeypatch
):
84 monkeypatch
.setattr('ceph_volume.api.lvm.create_lv', lambda *a
, **kw
: (a
, kw
))
85 lvs
= api
.create_lvs(self
.vg
, parts
=4)
86 assert lvs
[0][1]['extents'] == 249
88 def test_only_uses_free_size(self
, monkeypatch
):
89 monkeypatch
.setattr('ceph_volume.api.lvm.create_lv', lambda *a
, **kw
: (a
, kw
))
90 vg
= api
.VolumeGroup(vg_name
='ceph',
91 vg_extent_size
=1073741824,
92 vg_extent_count
=99999999,
94 lvs
= api
.create_lvs(vg
, parts
=4)
95 assert lvs
[0][1]['extents'] == 250
97 def test_null_tags_are_set_by_default(self
, monkeypatch
):
98 monkeypatch
.setattr('ceph_volume.api.lvm.create_lv', lambda *a
, **kw
: (a
, kw
))
99 kwargs
= api
.create_lvs(self
.vg
, parts
=4)[0][1]
100 assert list(kwargs
['tags'].values()) == ['null', 'null', 'null', 'null']
102 def test_fallback_to_one_part(self
, monkeypatch
):
103 monkeypatch
.setattr('ceph_volume.api.lvm.create_lv', lambda *a
, **kw
: (a
, kw
))
104 lvs
= api
.create_lvs(self
.vg
)
108 class TestVolumeGroupSizing(object):
110 def setup_method(self
):
111 self
.vg
= api
.VolumeGroup(vg_name
='ceph',
112 vg_extent_size
=1073741824,
115 def test_parts_and_size_errors(self
):
116 with pytest
.raises(ValueError) as error
:
117 self
.vg
.sizing(parts
=4, size
=10)
118 assert "Cannot process sizing" in str(error
.value
)
120 def test_zero_parts_produces_100_percent(self
):
121 result
= self
.vg
.sizing(parts
=0)
122 assert result
['percentages'] == 100
124 def test_two_parts_produces_50_percent(self
):
125 result
= self
.vg
.sizing(parts
=2)
126 assert result
['percentages'] == 50
128 def test_two_parts_produces_half_size(self
):
129 result
= self
.vg
.sizing(parts
=2)
130 assert result
['sizes'] == 512
132 def test_half_size_produces_round_sizes(self
):
133 result
= self
.vg
.sizing(size
=512)
134 assert result
['sizes'] == 512
135 assert result
['percentages'] == 50
136 assert result
['parts'] == 2
138 def test_bit_more_than_half_size_allocates_full_size(self
):
139 # 513 can't allocate more than 1, so it just fallsback to using the
141 result
= self
.vg
.sizing(size
=513)
142 assert result
['sizes'] == 1024
143 assert result
['percentages'] == 100
144 assert result
['parts'] == 1
146 def test_extents_are_halfed_rounded_down(self
):
147 result
= self
.vg
.sizing(size
=512)
148 assert result
['extents'] == 512
150 def test_bit_less_size_rounds_down(self
):
151 result
= self
.vg
.sizing(size
=129)
152 assert result
['sizes'] == 146
153 assert result
['percentages'] == 14
154 assert result
['parts'] == 7
156 def test_unable_to_allocate_past_free_size(self
):
157 with pytest
.raises(exceptions
.SizeAllocationError
):
158 self
.vg
.sizing(size
=2048)
161 class TestRemoveLV(object):
163 def test_removes_lv(self
, monkeypatch
):
164 def mock_call(cmd
, **kw
):
166 monkeypatch
.setattr(process
, 'call', mock_call
)
167 assert api
.remove_lv("vg/lv")
169 def test_removes_lv_object(self
, fake_call
):
170 foo_volume
= api
.Volume(lv_name
='foo', lv_path
='/path', vg_name
='foo_group', lv_tags
='')
171 api
.remove_lv(foo_volume
)
172 # last argument from the list passed to process.call
173 assert fake_call
.calls
[0]['args'][0][-1] == '/path'
175 def test_fails_to_remove_lv(self
, monkeypatch
):
176 def mock_call(cmd
, **kw
):
178 monkeypatch
.setattr(process
, 'call', mock_call
)
179 with pytest
.raises(RuntimeError):
180 api
.remove_lv("vg/lv")
183 class TestCreateLV(object):
185 def setup_method(self
):
186 self
.foo_volume
= api
.Volume(lv_name
='foo', lv_path
='/path', vg_name
='foo_group', lv_tags
='')
187 self
.foo_group
= api
.VolumeGroup(vg_name
='foo_group',
188 vg_extent_size
="4194304",
189 vg_extent_count
="100",
192 @patch('ceph_volume.api.lvm.process.run')
193 @patch('ceph_volume.api.lvm.process.call')
194 @patch('ceph_volume.api.lvm.get_single_lv')
195 def test_uses_size(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
):
196 m_get_single_lv
.return_value
= self
.foo_volume
197 api
.create_lv('foo', 0, vg
=self
.foo_group
, size
=419430400, tags
={'ceph.type': 'data'})
198 expected
= (['lvcreate', '--yes', '-l', '100', '-n', 'foo-0', 'foo_group'])
199 m_run
.assert_called_with(expected
, run_on_host
=True)
201 @patch('ceph_volume.api.lvm.process.run')
202 @patch('ceph_volume.api.lvm.process.call')
203 @patch('ceph_volume.api.lvm.get_single_lv')
204 def test_uses_size_adjust_if_1percent_over(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
):
205 foo_volume
= api
.Volume(lv_name
='foo', lv_path
='/path', vg_name
='foo_group', lv_tags
='')
206 foo_group
= api
.VolumeGroup(vg_name
='foo_group',
207 vg_extent_size
="4194304",
208 vg_extent_count
="1000",
209 vg_free_count
="1000")
210 m_get_single_lv
.return_value
= foo_volume
211 # 423624704 should be just under 1% off of the available size 419430400
212 api
.create_lv('foo', 0, vg
=foo_group
, size
=4232052736, tags
={'ceph.type': 'data'})
213 expected
= ['lvcreate', '--yes', '-l', '1000', '-n', 'foo-0', 'foo_group']
214 m_run
.assert_called_with(expected
, run_on_host
=True)
216 @patch('ceph_volume.api.lvm.process.run')
217 @patch('ceph_volume.api.lvm.process.call')
218 @patch('ceph_volume.api.lvm.get_single_lv')
219 def test_uses_size_too_large(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
):
220 m_get_single_lv
.return_value
= self
.foo_volume
221 with pytest
.raises(RuntimeError):
222 api
.create_lv('foo', 0, vg
=self
.foo_group
, size
=5368709120, tags
={'ceph.type': 'data'})
224 @patch('ceph_volume.api.lvm.process.run')
225 @patch('ceph_volume.api.lvm.process.call')
226 @patch('ceph_volume.api.lvm.get_single_lv')
227 def test_uses_extents(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
):
228 m_get_single_lv
.return_value
= self
.foo_volume
229 api
.create_lv('foo', 0, vg
=self
.foo_group
, extents
='50', tags
={'ceph.type': 'data'})
230 expected
= ['lvcreate', '--yes', '-l', '50', '-n', 'foo-0', 'foo_group']
231 m_run
.assert_called_with(expected
, run_on_host
=True)
233 @pytest.mark
.parametrize("test_input,expected",
236 @patch('ceph_volume.api.lvm.process.run')
237 @patch('ceph_volume.api.lvm.process.call')
238 @patch('ceph_volume.api.lvm.get_single_lv')
239 def test_uses_slots(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
, test_input
, expected
):
240 m_get_single_lv
.return_value
= self
.foo_volume
241 api
.create_lv('foo', 0, vg
=self
.foo_group
, slots
=test_input
, tags
={'ceph.type': 'data'})
242 expected
= ['lvcreate', '--yes', '-l', str(expected
), '-n', 'foo-0', 'foo_group']
243 m_run
.assert_called_with(expected
, run_on_host
=True)
245 @patch('ceph_volume.api.lvm.process.run')
246 @patch('ceph_volume.api.lvm.process.call')
247 @patch('ceph_volume.api.lvm.get_single_lv')
248 def test_uses_all(self
, m_get_single_lv
, m_call
, m_run
, monkeypatch
):
249 m_get_single_lv
.return_value
= self
.foo_volume
250 api
.create_lv('foo', 0, vg
=self
.foo_group
, tags
={'ceph.type': 'data'})
251 expected
= ['lvcreate', '--yes', '-l', '100%FREE', '-n', 'foo-0', 'foo_group']
252 m_run
.assert_called_with(expected
, run_on_host
=True)
254 @patch('ceph_volume.api.lvm.process.run')
255 @patch('ceph_volume.api.lvm.process.call')
256 @patch('ceph_volume.api.lvm.Volume.set_tags')
257 @patch('ceph_volume.api.lvm.get_single_lv')
258 def test_calls_to_set_tags_default(self
, m_get_single_lv
, m_set_tags
, m_call
, m_run
, monkeypatch
):
259 m_get_single_lv
.return_value
= self
.foo_volume
260 api
.create_lv('foo', 0, vg
=self
.foo_group
)
262 "ceph.osd_id": "null",
264 "ceph.cluster_fsid": "null",
265 "ceph.osd_fsid": "null",
267 m_set_tags
.assert_called_with(tags
)
269 @patch('ceph_volume.api.lvm.process.run')
270 @patch('ceph_volume.api.lvm.process.call')
271 @patch('ceph_volume.api.lvm.Volume.set_tags')
272 @patch('ceph_volume.api.lvm.get_single_lv')
273 def test_calls_to_set_tags_arg(self
, m_get_single_lv
, m_set_tags
, m_call
, m_run
, monkeypatch
):
274 m_get_single_lv
.return_value
= self
.foo_volume
275 api
.create_lv('foo', 0, vg
=self
.foo_group
, tags
={'ceph.type': 'data'})
278 "ceph.data_device": "/path"
280 m_set_tags
.assert_called_with(tags
)
282 @patch('ceph_volume.api.lvm.process.run')
283 @patch('ceph_volume.api.lvm.process.call')
284 @patch('ceph_volume.api.lvm.get_device_vgs')
285 @patch('ceph_volume.api.lvm.create_vg')
286 @patch('ceph_volume.api.lvm.get_single_lv')
287 def test_create_vg(self
, m_get_single_lv
, m_create_vg
, m_get_device_vgs
, m_call
,
289 m_get_single_lv
.return_value
= self
.foo_volume
290 m_get_device_vgs
.return_value
= []
291 api
.create_lv('foo', 0, device
='dev/foo', size
='5G', tags
={'ceph.type': 'data'})
292 m_create_vg
.assert_called_with('dev/foo', name_prefix
='ceph')
295 class TestTags(object):
297 def setup_method(self
):
298 self
.foo_volume_clean
= api
.Volume(lv_name
='foo_clean', lv_path
='/pathclean',
301 self
.foo_volume
= api
.Volume(lv_name
='foo', lv_path
='/path',
303 lv_tags
='ceph.foo0=bar0,ceph.foo1=bar1,ceph.foo2=bar2')
305 def test_set_tag(self
, monkeypatch
, capture
):
306 monkeypatch
.setattr(process
, 'run', capture
)
307 monkeypatch
.setattr(process
, 'call', capture
)
308 self
.foo_volume_clean
.set_tag('foo', 'bar')
309 expected
= ['lvchange', '--addtag', 'foo=bar', '/pathclean']
310 assert capture
.calls
[0]['args'][0] == expected
311 assert self
.foo_volume_clean
.tags
== {'foo': 'bar'}
313 def test_set_clear_tag(self
, monkeypatch
, capture
):
314 monkeypatch
.setattr(process
, 'run', capture
)
315 monkeypatch
.setattr(process
, 'call', capture
)
316 self
.foo_volume_clean
.set_tag('foo', 'bar')
317 assert self
.foo_volume_clean
.tags
== {'foo': 'bar'}
318 self
.foo_volume_clean
.clear_tag('foo')
319 expected
= ['lvchange', '--deltag', 'foo=bar', '/pathclean']
320 assert self
.foo_volume_clean
.tags
== {}
321 assert capture
.calls
[1]['args'][0] == expected
323 def test_set_tags(self
, monkeypatch
, capture
):
324 monkeypatch
.setattr(process
, 'run', capture
)
325 monkeypatch
.setattr(process
, 'call', capture
)
326 tags
= {'ceph.foo0': 'bar0', 'ceph.foo1': 'bar1', 'ceph.foo2': 'bar2'}
327 assert self
.foo_volume
.tags
== tags
329 tags
= {'ceph.foo0': 'bar0', 'ceph.foo1': 'baz1', 'ceph.foo2': 'baz2'}
330 self
.foo_volume
.set_tags(tags
)
331 assert self
.foo_volume
.tags
== tags
333 self
.foo_volume
.set_tag('ceph.foo1', 'other1')
334 tags
['ceph.foo1'] = 'other1'
335 assert self
.foo_volume
.tags
== tags
338 sorted(['lvchange', '--deltag', 'ceph.foo0=bar0', '--deltag',
339 'ceph.foo1=bar1', '--deltag', 'ceph.foo2=bar2', '/path']),
340 sorted(['lvchange', '--deltag', 'ceph.foo1=baz1', '/path']),
341 sorted(['lvchange', '--addtag', 'ceph.foo0=bar0', '--addtag',
342 'ceph.foo1=baz1', '--addtag', 'ceph.foo2=baz2', '/path']),
343 sorted(['lvchange', '--addtag', 'ceph.foo1=other1', '/path']),
345 # The order isn't guaranted
346 for call
in capture
.calls
:
347 assert sorted(call
['args'][0]) in expected
348 assert len(capture
.calls
) == len(expected
)
350 def test_clear_tags(self
, monkeypatch
, capture
):
351 monkeypatch
.setattr(process
, 'run', capture
)
352 monkeypatch
.setattr(process
, 'call', capture
)
353 tags
= {'ceph.foo0': 'bar0', 'ceph.foo1': 'bar1', 'ceph.foo2': 'bar2'}
355 self
.foo_volume_clean
.set_tags(tags
)
356 assert self
.foo_volume_clean
.tags
== tags
357 self
.foo_volume_clean
.clear_tags()
358 assert self
.foo_volume_clean
.tags
== {}
361 sorted(['lvchange', '--addtag', 'ceph.foo0=bar0', '--addtag',
362 'ceph.foo1=bar1', '--addtag', 'ceph.foo2=bar2',
364 sorted(['lvchange', '--deltag', 'ceph.foo0=bar0', '--deltag',
365 'ceph.foo1=bar1', '--deltag', 'ceph.foo2=bar2',
368 # The order isn't guaranted
369 for call
in capture
.calls
:
370 assert sorted(call
['args'][0]) in expected
371 assert len(capture
.calls
) == len(expected
)
374 class TestExtendVG(object):
376 def setup_method(self
):
377 self
.foo_volume
= api
.VolumeGroup(vg_name
='foo', lv_tags
='')
379 def test_uses_single_device_in_list(self
, monkeypatch
, fake_run
):
380 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
381 api
.extend_vg(self
.foo_volume
, ['/dev/sda'])
382 expected
= ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
383 assert fake_run
.calls
[0]['args'][0] == expected
385 def test_uses_single_device(self
, monkeypatch
, fake_run
):
386 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
387 api
.extend_vg(self
.foo_volume
, '/dev/sda')
388 expected
= ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
389 assert fake_run
.calls
[0]['args'][0] == expected
391 def test_uses_multiple_devices(self
, monkeypatch
, fake_run
):
392 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
393 api
.extend_vg(self
.foo_volume
, ['/dev/sda', '/dev/sdb'])
394 expected
= ['vgextend', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
395 assert fake_run
.calls
[0]['args'][0] == expected
398 class TestReduceVG(object):
400 def setup_method(self
):
401 self
.foo_volume
= api
.VolumeGroup(vg_name
='foo', lv_tags
='')
403 def test_uses_single_device_in_list(self
, monkeypatch
, fake_run
):
404 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
405 api
.reduce_vg(self
.foo_volume
, ['/dev/sda'])
406 expected
= ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
407 assert fake_run
.calls
[0]['args'][0] == expected
409 def test_uses_single_device(self
, monkeypatch
, fake_run
):
410 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
411 api
.reduce_vg(self
.foo_volume
, '/dev/sda')
412 expected
= ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
413 assert fake_run
.calls
[0]['args'][0] == expected
415 def test_uses_multiple_devices(self
, monkeypatch
, fake_run
):
416 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
417 api
.reduce_vg(self
.foo_volume
, ['/dev/sda', '/dev/sdb'])
418 expected
= ['vgreduce', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
419 assert fake_run
.calls
[0]['args'][0] == expected
422 class TestCreateVG(object):
424 def setup_method(self
):
425 self
.foo_volume
= api
.VolumeGroup(vg_name
='foo', lv_tags
='')
427 def test_no_name(self
, monkeypatch
, fake_run
):
428 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
429 api
.create_vg('/dev/sda')
430 result
= fake_run
.calls
[0]['args'][0]
431 assert '/dev/sda' in result
432 assert result
[-2].startswith('ceph-')
434 def test_devices_list(self
, monkeypatch
, fake_run
):
435 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
436 api
.create_vg(['/dev/sda', '/dev/sdb'], name
='ceph')
437 result
= fake_run
.calls
[0]['args'][0]
438 expected
= ['vgcreate', '--force', '--yes', 'ceph', '/dev/sda', '/dev/sdb']
439 assert result
== expected
441 def test_name_prefix(self
, monkeypatch
, fake_run
):
442 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
443 api
.create_vg('/dev/sda', name_prefix
='master')
444 result
= fake_run
.calls
[0]['args'][0]
445 assert '/dev/sda' in result
446 assert result
[-2].startswith('master-')
448 def test_specific_name(self
, monkeypatch
, fake_run
):
449 monkeypatch
.setattr(api
, 'get_single_vg', lambda **kw
: True)
450 api
.create_vg('/dev/sda', name
='master')
451 result
= fake_run
.calls
[0]['args'][0]
452 assert '/dev/sda' in result
453 assert result
[-2] == 'master'
456 # The following tests are pretty gnarly. VDO detection is very convoluted and
457 # involves correlating information from device mappers, realpaths, slaves of
458 # those mappers, and parents or related mappers. This makes it very hard to
459 # patch nicely or keep tests short and readable. These tests are trying to
460 # ensure correctness, the better approach will be to do some functional testing
466 def disable_kvdo_path(monkeypatch
):
467 monkeypatch
.setattr('os.path.isdir', lambda x
, **kw
: False)
471 def enable_kvdo_path(monkeypatch
):
472 monkeypatch
.setattr('os.path.isdir', lambda x
, **kw
: True)
475 # Stub for os.listdir
478 class ListDir(object):
480 def __init__(self
, paths
):
482 self
._normalize
_paths
()
483 self
.listdir
= os
.listdir
485 def _normalize_paths(self
):
486 for k
, v
in self
.paths
.items():
487 self
.paths
[k
.rstrip('/')] = v
.rstrip('/')
489 def add(self
, original
, fake
):
490 self
.paths
[original
.rstrip('/')] = fake
.rstrip('/')
492 def __call__(self
, path
):
493 return self
.listdir(self
.paths
[path
.rstrip('/')])
496 @pytest.fixture(scope
='function')
497 def listdir(monkeypatch
):
498 def apply(paths
=None, stub
=None):
500 stub
= ListDir(paths
)
502 for original
, fake
in paths
.items():
503 stub
.add(original
, fake
)
505 monkeypatch
.setattr('os.listdir', stub
)
509 @pytest.fixture(scope
='function')
510 def makedirs(tmpdir
):
511 def create(directory
):
512 path
= os
.path
.join(str(tmpdir
), directory
)
515 create
.base
= str(tmpdir
)
519 class TestIsVdo(object):
521 def test_no_vdo_dir(self
, disable_kvdo_path
):
522 assert api
._is
_vdo
('/path') is False
524 def test_exceptions_return_false(self
, monkeypatch
):
527 monkeypatch
.setattr('ceph_volume.api.lvm._is_vdo', throw
)
528 assert api
.is_vdo('/path') == '0'
530 def test_is_vdo_returns_a_string(self
, monkeypatch
):
531 monkeypatch
.setattr('ceph_volume.api.lvm._is_vdo', lambda x
, **kw
: True)
532 assert api
.is_vdo('/path') == '1'
534 def test_kvdo_dir_no_devices(self
, makedirs
, enable_kvdo_path
, listdir
, monkeypatch
):
535 kvdo_path
= makedirs('sys/kvdo')
536 listdir(paths
={'/sys/kvdo': kvdo_path
})
537 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x
, **kw
: [])
538 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_parents', lambda x
, **kw
: [])
539 assert api
._is
_vdo
('/dev/mapper/vdo0') is False
541 def test_vdo_slaves_found_and_matched(self
, makedirs
, enable_kvdo_path
, listdir
, monkeypatch
):
542 kvdo_path
= makedirs('sys/kvdo')
543 listdir(paths
={'/sys/kvdo': kvdo_path
})
544 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x
, **kw
: ['/dev/dm-3'])
545 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_parents', lambda x
, **kw
: [])
546 assert api
._is
_vdo
('/dev/dm-3') is True
548 def test_vdo_parents_found_and_matched(self
, makedirs
, enable_kvdo_path
, listdir
, monkeypatch
):
549 kvdo_path
= makedirs('sys/kvdo')
550 listdir(paths
={'/sys/kvdo': kvdo_path
})
551 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x
, **kw
: [])
552 monkeypatch
.setattr('ceph_volume.api.lvm._vdo_parents', lambda x
, **kw
: ['/dev/dm-4'])
553 assert api
._is
_vdo
('/dev/dm-4') is True
556 class TestVdoSlaves(object):
558 def test_slaves_are_not_found(self
, makedirs
, listdir
, monkeypatch
):
559 slaves_path
= makedirs('sys/block/vdo0/slaves')
560 listdir(paths
={'/sys/block/vdo0/slaves': slaves_path
})
561 monkeypatch
.setattr('ceph_volume.api.lvm.os.path.exists', lambda x
, **kw
: True)
562 result
= sorted(api
._vdo
_slaves
(['vdo0']))
563 assert '/dev/mapper/vdo0' in result
564 assert 'vdo0' in result
566 def test_slaves_are_found(self
, makedirs
, listdir
, monkeypatch
):
567 slaves_path
= makedirs('sys/block/vdo0/slaves')
568 makedirs('sys/block/vdo0/slaves/dm-4')
569 makedirs('dev/mapper/vdo0')
570 listdir(paths
={'/sys/block/vdo0/slaves': slaves_path
})
571 monkeypatch
.setattr('ceph_volume.api.lvm.os.path.exists', lambda x
, **kw
: True)
572 result
= sorted(api
._vdo
_slaves
(['vdo0']))
573 assert '/dev/dm-4' in result
574 assert 'dm-4' in result
577 class TestVDOParents(object):
579 def test_parents_are_found(self
, makedirs
, listdir
):
580 block_path
= makedirs('sys/block')
581 slaves_path
= makedirs('sys/block/dm-4/slaves')
582 makedirs('sys/block/dm-4/slaves/dm-3')
584 '/sys/block/dm-4/slaves': slaves_path
,
585 '/sys/block': block_path
})
586 result
= api
._vdo
_parents
(['dm-3'])
587 assert '/dev/dm-4' in result
588 assert 'dm-4' in result
590 def test_parents_are_not_found(self
, makedirs
, listdir
):
591 block_path
= makedirs('sys/block')
592 slaves_path
= makedirs('sys/block/dm-4/slaves')
593 makedirs('sys/block/dm-4/slaves/dm-5')
595 '/sys/block/dm-4/slaves': slaves_path
,
596 '/sys/block': block_path
})
597 result
= api
._vdo
_parents
(['dm-3'])
601 class TestSplitNameParser(object):
603 def test_keys_are_parsed_without_prefix(self
):
604 line
= ["DM_VG_NAME='/dev/mapper/vg';DM_LV_NAME='lv';DM_LV_LAYER=''"]
605 result
= api
._splitname
_parser
(line
)
606 assert result
['VG_NAME'] == 'vg'
607 assert result
['LV_NAME'] == 'lv'
608 assert result
['LV_LAYER'] == ''
610 def test_vg_name_sans_mapper(self
):
611 line
= ["DM_VG_NAME='/dev/mapper/vg';DM_LV_NAME='lv';DM_LV_LAYER=''"]
612 result
= api
._splitname
_parser
(line
)
613 assert '/dev/mapper' not in result
['VG_NAME']
616 class TestGetDeviceVgs(object):
618 @patch('ceph_volume.process.call')
619 @patch('ceph_volume.api.lvm._output_parser')
620 def test_get_device_vgs_with_empty_pv(self
, patched_output_parser
, pcall
):
621 patched_output_parser
.return_value
= [{'vg_name': ''}]
622 pcall
.return_value
= ('', '', '')
623 vgs
= api
.get_device_vgs('/dev/foo')
626 class TestGetDeviceLvs(object):
628 @patch('ceph_volume.process.call')
629 @patch('ceph_volume.api.lvm._output_parser')
630 def test_get_device_lvs_with_empty_vg(self
, patched_output_parser
, pcall
):
631 patched_output_parser
.return_value
= [{'lv_name': ''}]
632 pcall
.return_value
= ('', '', '')
633 vgs
= api
.get_device_lvs('/dev/foo')
637 # NOTE: api.convert_filters_to_str() and api.convert_tags_to_str() should get
638 # tested automatically while testing api.make_filters_lvmcmd_ready()
639 class TestMakeFiltersLVMCMDReady(object):
641 def test_with_no_filters_and_no_tags(self
):
642 retval
= api
.make_filters_lvmcmd_ready(None, None)
644 assert isinstance(retval
, str)
647 def test_with_filters_and_no_tags(self
):
648 filters
= {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
650 retval
= api
.make_filters_lvmcmd_ready(filters
, None)
652 assert isinstance(retval
, str)
653 for k
, v
in filters
.items():
657 def test_with_no_filters_and_with_tags(self
):
658 tags
= {'ceph.type': 'data', 'ceph.osd_id': '0'}
660 retval
= api
.make_filters_lvmcmd_ready(None, tags
)
662 assert isinstance(retval
, str)
663 assert 'tags' in retval
664 for k
, v
in tags
.items():
667 assert retval
.find('tags') < retval
.find(k
) < retval
.find(v
)
669 def test_with_filters_and_tags(self
):
670 filters
= {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
671 tags
= {'ceph.type': 'data', 'ceph.osd_id': '0'}
673 retval
= api
.make_filters_lvmcmd_ready(filters
, tags
)
675 assert isinstance(retval
, str)
676 for f
, t
in zip(filters
.items(), tags
.items()):
677 assert f
[0] in retval
678 assert f
[1] in retval
679 assert t
[0] in retval
680 assert t
[1] in retval
681 assert retval
.find(f
[0]) < retval
.find(f
[1]) < \
682 retval
.find('tags') < retval
.find(t
[0]) < retval
.find(t
[1])
685 class TestGetPVs(object):
687 def test_get_pvs(self
, monkeypatch
):
688 pv1
= api
.PVolume(pv_name
='/dev/sda', pv_uuid
='0000', pv_tags
={},
690 pv2
= api
.PVolume(pv_name
='/dev/sdb', pv_uuid
='0001', pv_tags
={},
693 stdout
= ['{};{};{};{};;'.format(pv1
.pv_name
, pv1
.pv_tags
, pv1
.pv_uuid
, pv1
.vg_name
),
694 '{};{};{};{};;'.format(pv2
.pv_name
, pv2
.pv_tags
, pv2
.pv_uuid
, pv2
.vg_name
)]
695 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
698 assert len(pvs_
) == len(pvs
)
699 for pv
, pv_
in zip(pvs
, pvs_
):
700 assert pv_
.pv_name
== pv
.pv_name
702 def test_get_pvs_single_pv(self
, monkeypatch
):
703 pv1
= api
.PVolume(pv_name
='/dev/sda', pv_uuid
='0000', pv_tags
={},
706 stdout
= ['{};;;;;;'.format(pv1
.pv_name
)]
707 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
710 assert len(pvs_
) == 1
711 assert pvs_
[0].pv_name
== pvs
[0].pv_name
713 def test_get_pvs_empty(self
, monkeypatch
):
714 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: ('', '', 0))
715 assert api
.get_pvs() == []
718 class TestGetVGs(object):
720 def test_get_vgs(self
, monkeypatch
):
721 vg1
= api
.VolumeGroup(vg_name
='vg1')
722 vg2
= api
.VolumeGroup(vg_name
='vg2')
724 stdout
= ['{};;;;;;'.format(vg1
.vg_name
),
725 '{};;;;;;'.format(vg2
.vg_name
)]
726 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
729 assert len(vgs_
) == len(vgs
)
730 for vg
, vg_
in zip(vgs
, vgs_
):
731 assert vg_
.vg_name
== vg
.vg_name
733 def test_get_vgs_single_vg(self
, monkeypatch
):
734 vg1
= api
.VolumeGroup(vg_name
='vg'); vgs
= [vg1
]
735 stdout
= ['{};;;;;;'.format(vg1
.vg_name
)]
736 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
739 assert len(vgs_
) == 1
740 assert vgs_
[0].vg_name
== vgs
[0].vg_name
742 def test_get_vgs_empty(self
, monkeypatch
):
743 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: ('', '', 0))
744 assert api
.get_vgs() == []
747 class TestGetLVs(object):
749 def test_get_lvs(self
, monkeypatch
):
750 lv1
= api
.Volume(lv_tags
='ceph.type=data', lv_path
='/dev/vg1/lv1',
751 lv_name
='lv1', vg_name
='vg1')
752 lv2
= api
.Volume(lv_tags
='ceph.type=data', lv_path
='/dev/vg2/lv2',
753 lv_name
='lv2', vg_name
='vg2')
755 stdout
= ['{};{};{};{}'.format(lv1
.lv_tags
, lv1
.lv_path
, lv1
.lv_name
,
757 '{};{};{};{}'.format(lv2
.lv_tags
, lv2
.lv_path
, lv2
.lv_name
,
759 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
762 assert len(lvs_
) == len(lvs
)
763 for lv
, lv_
in zip(lvs
, lvs_
):
764 assert lv
.__dict
__ == lv_
.__dict
__
766 def test_get_lvs_single_lv(self
, monkeypatch
):
767 stdout
= ['ceph.type=data;/dev/vg/lv;lv;vg']
768 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: (stdout
, '', 0))
770 lvs
.append((api
.Volume(lv_tags
='ceph.type=data',
771 lv_path
='/dev/vg/lv',
772 lv_name
='lv', vg_name
='vg')))
775 assert len(lvs_
) == len(lvs
)
776 assert lvs
[0].__dict
__ == lvs_
[0].__dict
__
778 def test_get_lvs_empty(self
, monkeypatch
):
779 monkeypatch
.setattr(api
.process
, 'call', lambda x
,**kw
: ('', '', 0))
780 assert api
.get_lvs() == []
783 class TestGetSinglePV(object):
785 @patch('ceph_volume.devices.lvm.prepare.api.get_pvs')
786 def test_get_single_pv_multiple_matches_raises_runtimeerror(self
, m_get_pvs
):
788 fake_pvs
.append(api
.PVolume(pv_name
='/dev/sda', pv_tags
={}))
789 fake_pvs
.append(api
.PVolume(pv_name
='/dev/sdb', pv_tags
={}))
791 m_get_pvs
.return_value
= fake_pvs
793 with pytest
.raises(RuntimeError) as e
:
795 assert "matched more than 1 PV present on this host." in str(e
.value
)
797 @patch('ceph_volume.devices.lvm.prepare.api.get_pvs')
798 def test_get_single_pv_no_match_returns_none(self
, m_get_pvs
):
799 m_get_pvs
.return_value
= []
801 pv
= api
.get_single_pv()
804 @patch('ceph_volume.devices.lvm.prepare.api.get_pvs')
805 def test_get_single_pv_one_match(self
, m_get_pvs
):
807 fake_pvs
.append(api
.PVolume(pv_name
='/dev/sda', pv_tags
={}))
808 m_get_pvs
.return_value
= fake_pvs
810 pv
= api
.get_single_pv()
812 assert isinstance(pv
, api
.PVolume
)
813 assert pv
.name
== '/dev/sda'
816 class TestGetSingleVG(object):
818 @patch('ceph_volume.devices.lvm.prepare.api.get_vgs')
819 def test_get_single_vg_multiple_matches_raises_runtimeerror(self
, m_get_vgs
):
821 fake_vgs
.append(api
.VolumeGroup(vg_name
='vg1'))
822 fake_vgs
.append(api
.VolumeGroup(vg_name
='vg2'))
824 m_get_vgs
.return_value
= fake_vgs
826 with pytest
.raises(RuntimeError) as e
:
828 assert "matched more than 1 VG present on this host." in str(e
.value
)
830 @patch('ceph_volume.devices.lvm.prepare.api.get_vgs')
831 def test_get_single_vg_no_match_returns_none(self
, m_get_vgs
):
832 m_get_vgs
.return_value
= []
834 vg
= api
.get_single_vg()
837 @patch('ceph_volume.devices.lvm.prepare.api.get_vgs')
838 def test_get_single_vg_one_match(self
, m_get_vgs
):
840 fake_vgs
.append(api
.VolumeGroup(vg_name
='vg1'))
841 m_get_vgs
.return_value
= fake_vgs
843 vg
= api
.get_single_vg()
845 assert isinstance(vg
, api
.VolumeGroup
)
846 assert vg
.name
== 'vg1'
848 class TestGetSingleLV(object):
850 @patch('ceph_volume.devices.lvm.prepare.api.get_lvs')
851 def test_get_single_lv_multiple_matches_raises_runtimeerror(self
, m_get_lvs
):
853 fake_lvs
.append(api
.Volume(lv_name
='lv1',
854 lv_path
='/dev/vg1/lv1',
857 lv_uuid
='fake-uuid'))
858 fake_lvs
.append(api
.Volume(lv_name
='lv1',
859 lv_path
='/dev/vg2/lv1',
862 lv_uuid
='fake-uuid'))
863 m_get_lvs
.return_value
= fake_lvs
865 with pytest
.raises(RuntimeError) as e
:
867 assert "matched more than 1 LV present on this host" in str(e
.value
)
869 @patch('ceph_volume.devices.lvm.prepare.api.get_lvs')
870 def test_get_single_lv_no_match_returns_none(self
, m_get_lvs
):
871 m_get_lvs
.return_value
= []
873 lv
= api
.get_single_lv()
876 @patch('ceph_volume.devices.lvm.prepare.api.get_lvs')
877 def test_get_single_lv_one_match(self
, m_get_lvs
):
879 fake_lvs
.append(api
.Volume(lv_name
='lv1', lv_path
='/dev/vg1/lv1', vg_name
='vg1', lv_tags
='', lv_uuid
='fake-uuid'))
880 m_get_lvs
.return_value
= fake_lvs
882 lv_
= api
.get_single_lv()
884 assert isinstance(lv_
, api
.Volume
)
885 assert lv_
.name
== 'lv1'