ceph/src/ceph-volume/ceph_volume/tests/api/test_lvm.py

   1 import os
   2 import pytest
   3 from mock.mock import patch
   4 from ceph_volume import process, exceptions
   5 from ceph_volume.api import lvm as api
   6
   7
   8 class TestParseTags(object):
   9
  10     def test_no_tags_means_empty_dict(self):
  11         result = api.parse_tags('')
  12         assert result == {}
  13
  14     def test_single_tag_gets_parsed(self):
  15         result = api.parse_tags('ceph.osd_something=1')
  16         assert result == {'ceph.osd_something': '1'}
  17
  18     def test_non_ceph_tags_are_skipped(self):
  19         result = api.parse_tags('foo')
  20         assert result == {}
  21
  22     def test_mixed_non_ceph_tags(self):
  23         result = api.parse_tags('foo,ceph.bar=1')
  24         assert result == {'ceph.bar': '1'}
  25
  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'
  32
  33
  34 class TestVolume(object):
  35
  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)
  40
  41     @pytest.mark.parametrize('dev',[
  42         '/dev/sdb',
  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'),
  46         None,
  47     ])
  48     def test_is_not_ceph_device(self, dev):
  49         assert not api.is_ceph_device(dev)
  50
  51     def test_no_empty_lv_name(self):
  52         with pytest.raises(ValueError):
  53             api.Volume(lv_name='', lv_tags='')
  54
  55
  56 class TestVolumeGroup(object):
  57
  58     def test_volume_group_no_empty_name(self):
  59         with pytest.raises(ValueError):
  60             api.VolumeGroup(vg_name='')
  61
  62
  63 class TestVolumeGroupFree(object):
  64
  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
  68
  69
  70 class TestCreateLVs(object):
  71
  72     def setup(self):
  73         self.vg = api.VolumeGroup(vg_name='ceph',
  74                                          vg_extent_size=1073741824,
  75                                          vg_extent_count=99999999,
  76                                          vg_free_count=999)
  77
  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)
  81         assert len(lvs) == 4
  82
  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
  87
  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,
  93                              vg_free_count=1000)
  94         lvs = api.create_lvs(vg, parts=4)
  95         assert lvs[0][1]['extents'] == 250
  96
  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']
 101
 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)
 105         assert len(lvs) == 1
 106
 107
 108 class TestVolumeGroupSizing(object):
 109
 110     def setup(self):
 111         self.vg = api.VolumeGroup(vg_name='ceph',
 112                                          vg_extent_size=1073741824,
 113                                          vg_free_count=1024)
 114
 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)
 119
 120     def test_zero_parts_produces_100_percent(self):
 121         result = self.vg.sizing(parts=0)
 122         assert result['percentages'] == 100
 123
 124     def test_two_parts_produces_50_percent(self):
 125         result = self.vg.sizing(parts=2)
 126         assert result['percentages'] == 50
 127
 128     def test_two_parts_produces_half_size(self):
 129         result = self.vg.sizing(parts=2)
 130         assert result['sizes'] == 512
 131
 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
 137
 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
 140         # whole device
 141         result = self.vg.sizing(size=513)
 142         assert result['sizes'] == 1024
 143         assert result['percentages'] == 100
 144         assert result['parts'] == 1
 145
 146     def test_extents_are_halfed_rounded_down(self):
 147         result = self.vg.sizing(size=512)
 148         assert result['extents'] == 512
 149
 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
 155
 156     def test_unable_to_allocate_past_free_size(self):
 157         with pytest.raises(exceptions.SizeAllocationError):
 158             self.vg.sizing(size=2048)
 159
 160
 161 class TestRemoveLV(object):
 162
 163     def test_removes_lv(self, monkeypatch):
 164         def mock_call(cmd, **kw):
 165             return ('', '', 0)
 166         monkeypatch.setattr(process, 'call', mock_call)
 167         assert api.remove_lv("vg/lv")
 168
 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'
 174
 175     def test_fails_to_remove_lv(self, monkeypatch):
 176         def mock_call(cmd, **kw):
 177             return ('', '', 1)
 178         monkeypatch.setattr(process, 'call', mock_call)
 179         with pytest.raises(RuntimeError):
 180             api.remove_lv("vg/lv")
 181
 182
 183 class TestCreateLV(object):
 184
 185     def setup(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_free_count=100)
 190
 191     @patch('ceph_volume.api.lvm.process.run')
 192     @patch('ceph_volume.api.lvm.process.call')
 193     @patch('ceph_volume.api.lvm.get_first_lv')
 194     def test_uses_size(self, m_get_first_lv, m_call, m_run, monkeypatch):
 195         m_get_first_lv.return_value = self.foo_volume
 196         api.create_lv('foo', 0, vg=self.foo_group, size=5368709120, tags={'ceph.type': 'data'})
 197         expected = ['lvcreate', '--yes', '-l', '1280', '-n', 'foo-0', 'foo_group']
 198         m_run.assert_called_with(expected)
 199
 200     @patch('ceph_volume.api.lvm.process.run')
 201     @patch('ceph_volume.api.lvm.process.call')
 202     @patch('ceph_volume.api.lvm.get_first_lv')
 203     def test_uses_extents(self, m_get_first_lv, m_call, m_run, monkeypatch):
 204         m_get_first_lv.return_value = self.foo_volume
 205         api.create_lv('foo', 0, vg=self.foo_group, extents='50', tags={'ceph.type': 'data'})
 206         expected = ['lvcreate', '--yes', '-l', '50', '-n', 'foo-0', 'foo_group']
 207         m_run.assert_called_with(expected)
 208
 209     @pytest.mark.parametrize("test_input,expected",
 210                              [(2, 50),
 211                               (3, 33),])
 212     @patch('ceph_volume.api.lvm.process.run')
 213     @patch('ceph_volume.api.lvm.process.call')
 214     @patch('ceph_volume.api.lvm.get_first_lv')
 215     def test_uses_slots(self, m_get_first_lv, m_call, m_run, monkeypatch, test_input, expected):
 216         m_get_first_lv.return_value = self.foo_volume
 217         api.create_lv('foo', 0, vg=self.foo_group, slots=test_input, tags={'ceph.type': 'data'})
 218         expected = ['lvcreate', '--yes', '-l', str(expected), '-n', 'foo-0', 'foo_group']
 219         m_run.assert_called_with(expected)
 220
 221     @patch('ceph_volume.api.lvm.process.run')
 222     @patch('ceph_volume.api.lvm.process.call')
 223     @patch('ceph_volume.api.lvm.get_first_lv')
 224     def test_uses_all(self, m_get_first_lv, m_call, m_run, monkeypatch):
 225         m_get_first_lv.return_value = self.foo_volume
 226         api.create_lv('foo', 0, vg=self.foo_group, tags={'ceph.type': 'data'})
 227         expected = ['lvcreate', '--yes', '-l', '100%FREE', '-n', 'foo-0', 'foo_group']
 228         m_run.assert_called_with(expected)
 229
 230     @patch('ceph_volume.api.lvm.process.run')
 231     @patch('ceph_volume.api.lvm.process.call')
 232     @patch('ceph_volume.api.lvm.Volume.set_tags')
 233     @patch('ceph_volume.api.lvm.get_first_lv')
 234     def test_calls_to_set_tags_default(self, m_get_first_lv, m_set_tags, m_call, m_run, monkeypatch):
 235         m_get_first_lv.return_value = self.foo_volume
 236         api.create_lv('foo', 0, vg=self.foo_group)
 237         tags = {
 238             "ceph.osd_id": "null",
 239             "ceph.type": "null",
 240             "ceph.cluster_fsid": "null",
 241             "ceph.osd_fsid": "null",
 242         }
 243         m_set_tags.assert_called_with(tags)
 244
 245     @patch('ceph_volume.api.lvm.process.run')
 246     @patch('ceph_volume.api.lvm.process.call')
 247     @patch('ceph_volume.api.lvm.Volume.set_tags')
 248     @patch('ceph_volume.api.lvm.get_first_lv')
 249     def test_calls_to_set_tags_arg(self, m_get_first_lv, m_set_tags, m_call, m_run, monkeypatch):
 250         m_get_first_lv.return_value = self.foo_volume
 251         api.create_lv('foo', 0, vg=self.foo_group, tags={'ceph.type': 'data'})
 252         tags = {
 253             "ceph.type": "data",
 254             "ceph.data_device": "/path"
 255         }
 256         m_set_tags.assert_called_with(tags)
 257
 258     @patch('ceph_volume.api.lvm.process.run')
 259     @patch('ceph_volume.api.lvm.process.call')
 260     @patch('ceph_volume.api.lvm.get_device_vgs')
 261     @patch('ceph_volume.api.lvm.create_vg')
 262     @patch('ceph_volume.api.lvm.get_first_lv')
 263     def test_create_vg(self, m_get_first_lv, m_create_vg, m_get_device_vgs, m_call,
 264                        m_run, monkeypatch):
 265         m_get_first_lv.return_value = self.foo_volume
 266         m_get_device_vgs.return_value = []
 267         api.create_lv('foo', 0, device='dev/foo', size='5G', tags={'ceph.type': 'data'})
 268         m_create_vg.assert_called_with('dev/foo', name_prefix='ceph')
 269
 270
 271 class TestTags(object):
 272
 273     def setup(self):
 274         self.foo_volume_clean = api.Volume(lv_name='foo_clean', lv_path='/pathclean',
 275             vg_name='foo_group',
 276             lv_tags='')
 277         self.foo_volume = api.Volume(lv_name='foo', lv_path='/path',
 278             vg_name='foo_group',
 279             lv_tags='ceph.foo0=bar0,ceph.foo1=bar1,ceph.foo2=bar2')
 280
 281     def test_set_tag(self, monkeypatch, capture):
 282         monkeypatch.setattr(process, 'run', capture)
 283         monkeypatch.setattr(process, 'call', capture)
 284         self.foo_volume_clean.set_tag('foo', 'bar')
 285         expected = ['lvchange', '--addtag', 'foo=bar', '/pathclean']
 286         assert capture.calls[0]['args'][0] == expected
 287         assert self.foo_volume_clean.tags == {'foo': 'bar'}
 288
 289     def test_set_clear_tag(self, monkeypatch, capture):
 290         monkeypatch.setattr(process, 'run', capture)
 291         monkeypatch.setattr(process, 'call', capture)
 292         self.foo_volume_clean.set_tag('foo', 'bar')
 293         assert self.foo_volume_clean.tags == {'foo': 'bar'}
 294         self.foo_volume_clean.clear_tag('foo')
 295         expected = ['lvchange', '--deltag', 'foo=bar', '/pathclean']
 296         assert self.foo_volume_clean.tags == {}
 297         assert capture.calls[1]['args'][0] == expected
 298
 299     def test_set_tags(self, monkeypatch, capture):
 300         monkeypatch.setattr(process, 'run', capture)
 301         monkeypatch.setattr(process, 'call', capture)
 302         tags = {'ceph.foo0': 'bar0', 'ceph.foo1': 'bar1', 'ceph.foo2': 'bar2'}
 303         assert self.foo_volume.tags == tags
 304
 305         tags = {'ceph.foo0': 'bar0', 'ceph.foo1': 'baz1', 'ceph.foo2': 'baz2'}
 306         self.foo_volume.set_tags(tags)
 307         assert self.foo_volume.tags == tags
 308
 309         self.foo_volume.set_tag('ceph.foo1', 'other1')
 310         tags['ceph.foo1'] = 'other1'
 311         assert self.foo_volume.tags == tags
 312
 313         expected = [
 314             sorted(['lvchange', '--deltag', 'ceph.foo0=bar0', '--deltag',
 315                     'ceph.foo1=bar1', '--deltag', 'ceph.foo2=bar2', '/path']),
 316             sorted(['lvchange', '--deltag', 'ceph.foo1=baz1', '/path']),
 317             sorted(['lvchange', '--addtag', 'ceph.foo0=bar0', '--addtag',
 318                     'ceph.foo1=baz1', '--addtag', 'ceph.foo2=baz2', '/path']),
 319             sorted(['lvchange', '--addtag', 'ceph.foo1=other1', '/path']),
 320         ]
 321         # The order isn't guaranted
 322         for call in capture.calls:
 323             assert sorted(call['args'][0]) in expected
 324         assert len(capture.calls) == len(expected)
 325
 326     def test_clear_tags(self, monkeypatch, capture):
 327         monkeypatch.setattr(process, 'run', capture)
 328         monkeypatch.setattr(process, 'call', capture)
 329         tags = {'ceph.foo0': 'bar0', 'ceph.foo1': 'bar1', 'ceph.foo2': 'bar2'}
 330
 331         self.foo_volume_clean.set_tags(tags)
 332         assert self.foo_volume_clean.tags == tags
 333         self.foo_volume_clean.clear_tags()
 334         assert self.foo_volume_clean.tags == {}
 335
 336         expected = [
 337             sorted(['lvchange', '--addtag', 'ceph.foo0=bar0', '--addtag',
 338                     'ceph.foo1=bar1', '--addtag', 'ceph.foo2=bar2',
 339                     '/pathclean']),
 340             sorted(['lvchange', '--deltag', 'ceph.foo0=bar0', '--deltag',
 341                     'ceph.foo1=bar1', '--deltag', 'ceph.foo2=bar2',
 342                     '/pathclean']),
 343         ]
 344         # The order isn't guaranted
 345         for call in capture.calls:
 346             assert sorted(call['args'][0]) in expected
 347         assert len(capture.calls) == len(expected)
 348
 349
 350 class TestExtendVG(object):
 351
 352     def setup(self):
 353         self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
 354
 355     def test_uses_single_device_in_list(self, monkeypatch, fake_run):
 356         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 357         api.extend_vg(self.foo_volume, ['/dev/sda'])
 358         expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
 359         assert fake_run.calls[0]['args'][0] == expected
 360
 361     def test_uses_single_device(self, monkeypatch, fake_run):
 362         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 363         api.extend_vg(self.foo_volume, '/dev/sda')
 364         expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
 365         assert fake_run.calls[0]['args'][0] == expected
 366
 367     def test_uses_multiple_devices(self, monkeypatch, fake_run):
 368         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 369         api.extend_vg(self.foo_volume, ['/dev/sda', '/dev/sdb'])
 370         expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
 371         assert fake_run.calls[0]['args'][0] == expected
 372
 373
 374 class TestReduceVG(object):
 375
 376     def setup(self):
 377         self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
 378
 379     def test_uses_single_device_in_list(self, monkeypatch, fake_run):
 380         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 381         api.reduce_vg(self.foo_volume, ['/dev/sda'])
 382         expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
 383         assert fake_run.calls[0]['args'][0] == expected
 384
 385     def test_uses_single_device(self, monkeypatch, fake_run):
 386         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 387         api.reduce_vg(self.foo_volume, '/dev/sda')
 388         expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
 389         assert fake_run.calls[0]['args'][0] == expected
 390
 391     def test_uses_multiple_devices(self, monkeypatch, fake_run):
 392         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 393         api.reduce_vg(self.foo_volume, ['/dev/sda', '/dev/sdb'])
 394         expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
 395         assert fake_run.calls[0]['args'][0] == expected
 396
 397
 398 class TestCreateVG(object):
 399
 400     def setup(self):
 401         self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
 402
 403     def test_no_name(self, monkeypatch, fake_run):
 404         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 405         api.create_vg('/dev/sda')
 406         result = fake_run.calls[0]['args'][0]
 407         assert '/dev/sda' in result
 408         assert result[-2].startswith('ceph-')
 409
 410     def test_devices_list(self, monkeypatch, fake_run):
 411         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 412         api.create_vg(['/dev/sda', '/dev/sdb'], name='ceph')
 413         result = fake_run.calls[0]['args'][0]
 414         expected = ['vgcreate', '--force', '--yes', 'ceph', '/dev/sda', '/dev/sdb']
 415         assert result == expected
 416
 417     def test_name_prefix(self, monkeypatch, fake_run):
 418         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 419         api.create_vg('/dev/sda', name_prefix='master')
 420         result = fake_run.calls[0]['args'][0]
 421         assert '/dev/sda' in result
 422         assert result[-2].startswith('master-')
 423
 424     def test_specific_name(self, monkeypatch, fake_run):
 425         monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
 426         api.create_vg('/dev/sda', name='master')
 427         result = fake_run.calls[0]['args'][0]
 428         assert '/dev/sda' in result
 429         assert result[-2] == 'master'
 430
 431 #
 432 # The following tests are pretty gnarly. VDO detection is very convoluted and
 433 # involves correlating information from device mappers, realpaths, slaves of
 434 # those mappers, and parents or related mappers.  This makes it very hard to
 435 # patch nicely or keep tests short and readable. These tests are trying to
 436 # ensure correctness, the better approach will be to do some functional testing
 437 # with VDO.
 438 #
 439
 440
 441 @pytest.fixture
 442 def disable_kvdo_path(monkeypatch):
 443     monkeypatch.setattr('os.path.isdir', lambda x, **kw: False)
 444
 445
 446 @pytest.fixture
 447 def enable_kvdo_path(monkeypatch):
 448     monkeypatch.setattr('os.path.isdir', lambda x, **kw: True)
 449
 450
 451 # Stub for os.listdir
 452
 453
 454 class ListDir(object):
 455
 456     def __init__(self, paths):
 457         self.paths = paths
 458         self._normalize_paths()
 459         self.listdir = os.listdir
 460
 461     def _normalize_paths(self):
 462         for k, v in self.paths.items():
 463             self.paths[k.rstrip('/')] = v.rstrip('/')
 464
 465     def add(self, original, fake):
 466         self.paths[original.rstrip('/')] = fake.rstrip('/')
 467
 468     def __call__(self, path):
 469         return self.listdir(self.paths[path.rstrip('/')])
 470
 471
 472 @pytest.fixture(scope='function')
 473 def listdir(monkeypatch):
 474     def apply(paths=None, stub=None):
 475         if not stub:
 476             stub = ListDir(paths)
 477         if paths:
 478             for original, fake in paths.items():
 479                 stub.add(original, fake)
 480
 481         monkeypatch.setattr('os.listdir', stub)
 482     return apply
 483
 484
 485 @pytest.fixture(scope='function')
 486 def makedirs(tmpdir):
 487     def create(directory):
 488         path = os.path.join(str(tmpdir), directory)
 489         os.makedirs(path)
 490         return path
 491     create.base = str(tmpdir)
 492     return create
 493
 494
 495 class TestIsVdo(object):
 496
 497     def test_no_vdo_dir(self, disable_kvdo_path):
 498         assert api._is_vdo('/path') is False
 499
 500     def test_exceptions_return_false(self, monkeypatch):
 501         def throw():
 502             raise Exception()
 503         monkeypatch.setattr('ceph_volume.api.lvm._is_vdo', throw)
 504         assert api.is_vdo('/path') == '0'
 505
 506     def test_is_vdo_returns_a_string(self, monkeypatch):
 507         monkeypatch.setattr('ceph_volume.api.lvm._is_vdo', lambda x, **kw: True)
 508         assert api.is_vdo('/path') == '1'
 509
 510     def test_kvdo_dir_no_devices(self, makedirs, enable_kvdo_path, listdir, monkeypatch):
 511         kvdo_path = makedirs('sys/kvdo')
 512         listdir(paths={'/sys/kvdo': kvdo_path})
 513         monkeypatch.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x, **kw: [])
 514         monkeypatch.setattr('ceph_volume.api.lvm._vdo_parents', lambda x, **kw: [])
 515         assert api._is_vdo('/dev/mapper/vdo0') is False
 516
 517     def test_vdo_slaves_found_and_matched(self, makedirs, enable_kvdo_path, listdir, monkeypatch):
 518         kvdo_path = makedirs('sys/kvdo')
 519         listdir(paths={'/sys/kvdo': kvdo_path})
 520         monkeypatch.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x, **kw: ['/dev/dm-3'])
 521         monkeypatch.setattr('ceph_volume.api.lvm._vdo_parents', lambda x, **kw: [])
 522         assert api._is_vdo('/dev/dm-3') is True
 523
 524     def test_vdo_parents_found_and_matched(self, makedirs, enable_kvdo_path, listdir, monkeypatch):
 525         kvdo_path = makedirs('sys/kvdo')
 526         listdir(paths={'/sys/kvdo': kvdo_path})
 527         monkeypatch.setattr('ceph_volume.api.lvm._vdo_slaves', lambda x, **kw: [])
 528         monkeypatch.setattr('ceph_volume.api.lvm._vdo_parents', lambda x, **kw: ['/dev/dm-4'])
 529         assert api._is_vdo('/dev/dm-4') is True
 530
 531
 532 class TestVdoSlaves(object):
 533
 534     def test_slaves_are_not_found(self, makedirs, listdir, monkeypatch):
 535         slaves_path = makedirs('sys/block/vdo0/slaves')
 536         listdir(paths={'/sys/block/vdo0/slaves': slaves_path})
 537         monkeypatch.setattr('ceph_volume.api.lvm.os.path.exists', lambda x, **kw: True)
 538         result = sorted(api._vdo_slaves(['vdo0']))
 539         assert '/dev/mapper/vdo0' in result
 540         assert 'vdo0' in result
 541
 542     def test_slaves_are_found(self, makedirs, listdir, monkeypatch):
 543         slaves_path = makedirs('sys/block/vdo0/slaves')
 544         makedirs('sys/block/vdo0/slaves/dm-4')
 545         makedirs('dev/mapper/vdo0')
 546         listdir(paths={'/sys/block/vdo0/slaves': slaves_path})
 547         monkeypatch.setattr('ceph_volume.api.lvm.os.path.exists', lambda x, **kw: True)
 548         result = sorted(api._vdo_slaves(['vdo0']))
 549         assert '/dev/dm-4' in result
 550         assert 'dm-4' in result
 551
 552
 553 class TestVDOParents(object):
 554
 555     def test_parents_are_found(self, makedirs, listdir):
 556         block_path = makedirs('sys/block')
 557         slaves_path = makedirs('sys/block/dm-4/slaves')
 558         makedirs('sys/block/dm-4/slaves/dm-3')
 559         listdir(paths={
 560             '/sys/block/dm-4/slaves': slaves_path,
 561             '/sys/block': block_path})
 562         result = api._vdo_parents(['dm-3'])
 563         assert '/dev/dm-4' in result
 564         assert 'dm-4' in result
 565
 566     def test_parents_are_not_found(self, makedirs, listdir):
 567         block_path = makedirs('sys/block')
 568         slaves_path = makedirs('sys/block/dm-4/slaves')
 569         makedirs('sys/block/dm-4/slaves/dm-5')
 570         listdir(paths={
 571             '/sys/block/dm-4/slaves': slaves_path,
 572             '/sys/block': block_path})
 573         result = api._vdo_parents(['dm-3'])
 574         assert result == []
 575
 576
 577 class TestSplitNameParser(object):
 578
 579     def test_keys_are_parsed_without_prefix(self):
 580         line = ["DM_VG_NAME='/dev/mapper/vg';DM_LV_NAME='lv';DM_LV_LAYER=''"]
 581         result = api._splitname_parser(line)
 582         assert result['VG_NAME'] == 'vg'
 583         assert result['LV_NAME'] == 'lv'
 584         assert result['LV_LAYER'] == ''
 585
 586     def test_vg_name_sans_mapper(self):
 587         line = ["DM_VG_NAME='/dev/mapper/vg';DM_LV_NAME='lv';DM_LV_LAYER=''"]
 588         result = api._splitname_parser(line)
 589         assert '/dev/mapper' not in result['VG_NAME']
 590
 591
 592 class TestGetDeviceVgs(object):
 593
 594     @patch('ceph_volume.process.call')
 595     @patch('ceph_volume.api.lvm._output_parser')
 596     def test_get_device_vgs_with_empty_pv(self, patched_output_parser, pcall):
 597         patched_output_parser.return_value = [{'vg_name': ''}]
 598         pcall.return_value = ('', '', '')
 599         vgs = api.get_device_vgs('/dev/foo')
 600         assert vgs == []
 601
 602 class TestGetDeviceLvs(object):
 603
 604     @patch('ceph_volume.process.call')
 605     @patch('ceph_volume.api.lvm._output_parser')
 606     def test_get_device_lvs_with_empty_vg(self, patched_output_parser, pcall):
 607         patched_output_parser.return_value = [{'lv_name': ''}]
 608         pcall.return_value = ('', '', '')
 609         vgs = api.get_device_lvs('/dev/foo')
 610         assert vgs == []
 611
 612
 613 # NOTE: api.convert_filters_to_str() and api.convert_tags_to_str() should get
 614 # tested automatically while testing api.make_filters_lvmcmd_ready()
 615 class TestMakeFiltersLVMCMDReady(object):
 616
 617     def test_with_no_filters_and_no_tags(self):
 618         retval = api.make_filters_lvmcmd_ready(None, None)
 619
 620         assert isinstance(retval, str)
 621         assert retval == ''
 622
 623     def test_with_filters_and_no_tags(self):
 624         filters = {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
 625
 626         retval = api.make_filters_lvmcmd_ready(filters, None)
 627
 628         assert isinstance(retval, str)
 629         for k, v in filters.items():
 630             assert k in retval
 631             assert v in retval
 632
 633     def test_with_no_filters_and_with_tags(self):
 634         tags = {'ceph.type': 'data', 'ceph.osd_id': '0'}
 635
 636         retval = api.make_filters_lvmcmd_ready(None, tags)
 637
 638         assert isinstance(retval, str)
 639         assert 'tags' in retval
 640         for k, v in tags.items():
 641             assert k in retval
 642             assert v in retval
 643             assert retval.find('tags') < retval.find(k) < retval.find(v)
 644
 645     def test_with_filters_and_tags(self):
 646         filters = {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
 647         tags = {'ceph.type': 'data', 'ceph.osd_id': '0'}
 648
 649         retval = api.make_filters_lvmcmd_ready(filters, tags)
 650
 651         assert isinstance(retval, str)
 652         for f, t in zip(filters.items(), tags.items()):
 653             assert f[0] in retval
 654             assert f[1] in retval
 655             assert t[0] in retval
 656             assert t[1] in retval
 657             assert retval.find(f[0]) < retval.find(f[1]) < \
 658                     retval.find('tags') < retval.find(t[0]) < retval.find(t[1])
 659
 660
 661 class TestGetPVs(object):
 662
 663     def test_get_pvs(self, monkeypatch):
 664         pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
 665                           vg_name='vg1')
 666         pv2 = api.PVolume(pv_name='/dev/sdb', pv_uuid='0001', pv_tags={},
 667                           vg_name='vg2')
 668         pvs = [pv1, pv2]
 669         stdout = ['{};{};{};{};;'.format(pv1.pv_name, pv1.pv_tags, pv1.pv_uuid, pv1.vg_name),
 670                   '{};{};{};{};;'.format(pv2.pv_name, pv2.pv_tags, pv2.pv_uuid, pv2.vg_name)]
 671         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 672
 673         pvs_ = api.get_pvs()
 674         assert len(pvs_) == len(pvs)
 675         for pv, pv_ in zip(pvs, pvs_):
 676             assert pv_.pv_name == pv.pv_name
 677
 678     def test_get_pvs_single_pv(self, monkeypatch):
 679         pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
 680                           vg_name='vg1')
 681         pvs = [pv1]
 682         stdout = ['{};;;;;;'.format(pv1.pv_name)]
 683         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 684
 685         pvs_ = api.get_pvs()
 686         assert len(pvs_) == 1
 687         assert pvs_[0].pv_name == pvs[0].pv_name
 688
 689     def test_get_pvs_empty(self, monkeypatch):
 690         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 691         assert api.get_pvs() == []
 692
 693
 694 class TestGetVGs(object):
 695
 696     def test_get_vgs(self, monkeypatch):
 697         vg1 = api.VolumeGroup(vg_name='vg1')
 698         vg2 = api.VolumeGroup(vg_name='vg2')
 699         vgs = [vg1, vg2]
 700         stdout = ['{};;;;;;'.format(vg1.vg_name),
 701                   '{};;;;;;'.format(vg2.vg_name)]
 702         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 703
 704         vgs_ = api.get_vgs()
 705         assert len(vgs_) == len(vgs)
 706         for vg, vg_ in zip(vgs, vgs_):
 707             assert vg_.vg_name == vg.vg_name
 708
 709     def test_get_vgs_single_vg(self, monkeypatch):
 710         vg1 = api.VolumeGroup(vg_name='vg'); vgs = [vg1]
 711         stdout = ['{};;;;;;'.format(vg1.vg_name)]
 712         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 713
 714         vgs_ = api.get_vgs()
 715         assert len(vgs_) == 1
 716         assert vgs_[0].vg_name == vgs[0].vg_name
 717
 718     def test_get_vgs_empty(self, monkeypatch):
 719         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 720         assert api.get_vgs() == []
 721
 722
 723 class TestGetLVs(object):
 724
 725     def test_get_lvs(self, monkeypatch):
 726         lv1 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg1/lv1',
 727                          lv_name='lv1', vg_name='vg1')
 728         lv2 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg2/lv2',
 729                          lv_name='lv2', vg_name='vg2')
 730         lvs = [lv1, lv2]
 731         stdout = ['{};{};{};{}'.format(lv1.lv_tags, lv1.lv_path, lv1.lv_name,
 732                                        lv1.vg_name),
 733                   '{};{};{};{}'.format(lv2.lv_tags, lv2.lv_path, lv2.lv_name,
 734                                        lv2.vg_name)]
 735         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 736
 737         lvs_ = api.get_lvs()
 738         assert len(lvs_) == len(lvs)
 739         for lv, lv_ in zip(lvs, lvs_):
 740             assert lv.__dict__ == lv_.__dict__
 741
 742     def test_get_lvs_single_lv(self, monkeypatch):
 743         stdout = ['ceph.type=data;/dev/vg/lv;lv;vg']
 744         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 745         lvs = []
 746         lvs.append((api.Volume(lv_tags='ceph.type=data',
 747                            lv_path='/dev/vg/lv',
 748                            lv_name='lv', vg_name='vg')))
 749
 750         lvs_ = api.get_lvs()
 751         assert len(lvs_) == len(lvs)
 752         assert lvs[0].__dict__ == lvs_[0].__dict__
 753
 754     def test_get_lvs_empty(self, monkeypatch):
 755         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 756         assert api.get_lvs() == []
 757
 758
 759 class TestGetFirstPV(object):
 760
 761     def test_get_first_pv(self, monkeypatch):
 762         pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
 763                           vg_name='vg1')
 764         pv2 = api.PVolume(pv_name='/dev/sdb', pv_uuid='0001', pv_tags={},
 765                           vg_name='vg2')
 766         stdout = ['{};{};{};{};;'.format(pv1.pv_name, pv1.pv_tags, pv1.pv_uuid, pv1.vg_name),
 767                   '{};{};{};{};;'.format(pv2.pv_name, pv2.pv_tags, pv2.pv_uuid, pv2.vg_name)]
 768         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 769
 770         pv_ = api.get_first_pv()
 771         assert isinstance(pv_, api.PVolume)
 772         assert pv_.pv_name == pv1.pv_name
 773
 774     def test_get_first_pv_single_pv(self, monkeypatch):
 775         pv = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
 776                          vg_name='vg1')
 777         stdout = ['{};;;;;;'.format(pv.pv_name)]
 778         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 779
 780         pv_ = api.get_first_pv()
 781         assert isinstance(pv_, api.PVolume)
 782         assert pv_.pv_name == pv.pv_name
 783
 784     def test_get_first_pv_empty(self, monkeypatch):
 785         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 786         assert api.get_first_pv() == []
 787
 788
 789 class TestGetFirstVG(object):
 790
 791     def test_get_first_vg(self, monkeypatch):
 792         vg1 = api.VolumeGroup(vg_name='vg1')
 793         vg2 = api.VolumeGroup(vg_name='vg2')
 794         stdout = ['{};;;;;;'.format(vg1.vg_name), '{};;;;;;'.format(vg2.vg_name)]
 795         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 796
 797         vg_ = api.get_first_vg()
 798         assert isinstance(vg_, api.VolumeGroup)
 799         assert vg_.vg_name == vg1.vg_name
 800
 801     def test_get_first_vg_single_vg(self, monkeypatch):
 802         vg = api.VolumeGroup(vg_name='vg')
 803         stdout = ['{};;;;;;'.format(vg.vg_name)]
 804         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 805
 806         vg_ = api.get_first_vg()
 807         assert isinstance(vg_, api.VolumeGroup)
 808         assert vg_.vg_name == vg.vg_name
 809
 810     def test_get_first_vg_empty(self, monkeypatch):
 811         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 812         vg_ = api.get_first_vg()
 813         assert vg_ == []
 814
 815
 816 class TestGetFirstLV(object):
 817
 818     def test_get_first_lv(self, monkeypatch):
 819         lv1 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg1/lv1',
 820                          lv_name='lv1', vg_name='vg1')
 821         lv2 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg2/lv2',
 822                          lv_name='lv2', vg_name='vg2')
 823         stdout = ['{};{};{};{}'.format(lv1.lv_tags, lv1.lv_path, lv1.lv_name,
 824                                        lv1.vg_name),
 825                   '{};{};{};{}'.format(lv2.lv_tags, lv2.lv_path, lv2.lv_name,
 826                                        lv2.vg_name)]
 827         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 828
 829         lv_ = api.get_first_lv()
 830         assert isinstance(lv_, api.Volume)
 831         assert lv_.lv_name == lv1.lv_name
 832
 833     def test_get_first_lv_single_lv(self, monkeypatch):
 834         stdout = ['ceph.type=data;/dev/vg/lv;lv;vg']
 835         monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
 836         lv = api.Volume(lv_tags='ceph.type=data',
 837                            lv_path='/dev/vg/lv',
 838                            lv_name='lv', vg_name='vg')
 839
 840         lv_ = api.get_first_lv()
 841         assert isinstance(lv_, api.Volume)
 842         assert lv_.lv_name == lv.lv_name
 843
 844     def test_get_first_lv_empty(self, monkeypatch):
 845         monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
 846         assert api.get_lvs() == []