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