assert result['ceph.fsid'] == '0000'
-class TestGetAPIVgs(object):
-
- def test_report_is_emtpy(self, monkeypatch):
- monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('\n\n', '', 0))
- assert api.get_api_vgs() == []
-
- def test_report_has_stuff(self, monkeypatch):
- report = [' VolGroup00']
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: (report, '', 0))
- assert api.get_api_vgs() == [{'vg_name': 'VolGroup00'}]
-
- def test_report_has_stuff_with_empty_attrs(self, monkeypatch):
- report = [' VolGroup00 ;;;;;;4194304']
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: (report, '', 0))
- result = api.get_api_vgs()[0]
- assert len(result.keys()) == 7
- assert result['vg_name'] == 'VolGroup00'
- assert result['vg_extent_size'] == '4194304'
-
- def test_report_has_multiple_items(self, monkeypatch):
- report = [' VolGroup00;;;;;;;', ' ceph_vg;;;;;;;']
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: (report, '', 0))
- result = api.get_api_vgs()
- assert result[0]['vg_name'] == 'VolGroup00'
- assert result[1]['vg_name'] == 'ceph_vg'
-
-
-class TestGetAPILvs(object):
-
- def test_report_is_emtpy(self, monkeypatch):
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: ('', '', 0))
- assert api.get_api_lvs() == []
-
- def test_report_has_stuff(self, monkeypatch):
- report = [' ;/path;VolGroup00;root']
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: (report, '', 0))
- result = api.get_api_lvs()
- assert result[0]['lv_name'] == 'VolGroup00'
-
- def test_report_has_multiple_items(self, monkeypatch):
- report = [' ;/path;VolName;root', ';/dev/path;ceph_lv;ceph_vg']
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: (report, '', 0))
- result = api.get_api_lvs()
- assert result[0]['lv_name'] == 'VolName'
- assert result[1]['lv_name'] == 'ceph_lv'
-
-
-@pytest.fixture
-def volumes(monkeypatch):
- monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
- volumes = api.Volumes()
- volumes._purge()
- # also patch api.Volumes so that when it is called, it will use the newly
- # created fixture, with whatever the test method wants to append to it
- monkeypatch.setattr(api, 'Volumes', lambda: volumes)
- return volumes
-
-
-@pytest.fixture
-def volume_groups(monkeypatch):
- monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
- vgs = api.VolumeGroups()
- vgs._purge()
- return vgs
-
-
-class TestGetLV(object):
-
- def test_nothing_is_passed_in(self):
- # so we return a None
- assert api.get_lv() is None
-
- def test_single_lv_is_matched(self, volumes, monkeypatch):
- FooVolume = api.Volume(lv_name='foo', lv_path='/dev/vg/foo', lv_tags="ceph.type=data")
- volumes.append(FooVolume)
- monkeypatch.setattr(api, 'Volumes', lambda: volumes)
- assert api.get_lv(lv_name='foo') == FooVolume
-
- def test_single_lv_is_matched_by_uuid(self, volumes, monkeypatch):
- FooVolume = api.Volume(
- lv_name='foo', lv_path='/dev/vg/foo',
- lv_uuid='1111', lv_tags="ceph.type=data")
- volumes.append(FooVolume)
- monkeypatch.setattr(api, 'Volumes', lambda: volumes)
- assert api.get_lv(lv_uuid='1111') == FooVolume
-
-
-class TestGetPV(object):
-
- def test_nothing_is_passed_in(self):
- # so we return a None
- assert api.get_pv() is None
-
- def test_single_pv_is_not_matched(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", pv_tags={}, vg_name="vg")
- pvolumes.append(FooPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- assert api.get_pv(pv_uuid='foo') is None
-
- def test_single_pv_is_matched(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={})
- pvolumes.append(FooPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- assert api.get_pv(pv_uuid='0000') == FooPVolume
-
- def test_multiple_pvs_is_matched_by_uuid(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={}, lv_uuid="0000000")
- BarPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={})
- pvolumes.append(FooPVolume)
- pvolumes.append(BarPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- assert api.get_pv(pv_uuid='0000') == FooPVolume
-
- def test_multiple_pvs_is_matched_by_name(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={}, lv_uuid="0000000")
- BarPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={})
- pvolumes.append(FooPVolume)
- pvolumes.append(BarPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- assert api.get_pv(pv_name='/dev/sda') == FooPVolume
-
- def test_multiple_pvs_is_matched_by_tags(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(vg_name="vg1", pv_name='/dev/sdc', pv_uuid="1000", pv_tags="ceph.foo=bar", lv_uuid="0000000")
- BarPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags="ceph.foo=bar")
- pvolumes.append(FooPVolume)
- pvolumes.append(BarPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- with pytest.raises(exceptions.MultiplePVsError):
- api.get_pv(pv_tags={"ceph.foo": "bar"})
-
- def test_single_pv_is_matched_by_uuid(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(
- pv_name='/dev/vg/foo',
- pv_uuid='1111', pv_tags="ceph.type=data", vg_name="vg")
- pvolumes.append(FooPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- assert api.get_pv(pv_uuid='1111') == FooPVolume
-
- def test_vg_name_is_set(self, pvolumes, monkeypatch):
- FooPVolume = api.PVolume(
- pv_name='/dev/vg/foo',
- pv_uuid='1111', pv_tags="ceph.type=data", vg_name="vg")
- pvolumes.append(FooPVolume)
- monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
- pv = api.get_pv(pv_name="/dev/vg/foo")
- assert pv.vg_name == "vg"
-
-
-class TestPVolumes(object):
-
- def test_filter_by_tag_does_not_match_one(self, pvolumes, monkeypatch):
- pv_tags = "ceph.type=journal,ceph.osd_id=1,ceph.fsid=000-aaa"
- FooPVolume = api.PVolume(
- pv_name='/dev/vg/foo',
- pv_uuid='1111', pv_tags=pv_tags, vg_name='vg')
- pvolumes.append(FooPVolume)
- assert pvolumes.filter(pv_tags={'ceph.type': 'journal',
- 'ceph.osd_id': '2'}) == []
-
- def test_filter_by_tags_matches(self, pvolumes, monkeypatch):
- pv_tags = "ceph.type=journal,ceph.osd_id=1"
- FooPVolume = api.PVolume(
- pv_name='/dev/vg/foo',
- pv_uuid='1111', pv_tags=pv_tags, vg_name="vg")
- pvolumes.append(FooPVolume)
- assert pvolumes.filter(pv_tags={'ceph.type': 'journal',
- 'ceph.osd_id': '1'}) == [FooPVolume]
-
-
-class TestGetVG(object):
-
- def test_nothing_is_passed_in(self):
- # so we return a None
- assert api.get_vg() is None
-
- def test_single_vg_is_matched(self, volume_groups, monkeypatch):
- FooVG = api.VolumeGroup(vg_name='foo')
- volume_groups.append(FooVG)
- monkeypatch.setattr(api, 'VolumeGroups', lambda: volume_groups)
- assert api.get_vg(vg_name='foo') == FooVG
-
-
class TestVolume(object):
def test_is_ceph_device(self):
@pytest.mark.parametrize('dev',[
'/dev/sdb',
api.VolumeGroup(vg_name='foo'),
- api.Volume(lv_name='vg/no_osd', lv_tags=''),
- api.Volume(lv_name='vg/no_osd', lv_tags='ceph.osd_id=null'),
+ api.Volume(lv_name='vg/no_osd', lv_tags='', lv_path='lv/path'),
+ api.Volume(lv_name='vg/no_osd', lv_tags='ceph.osd_id=null', lv_path='lv/path'),
None,
])
def test_is_not_ceph_device(self, dev):
api.Volume(lv_name='', lv_tags='')
-class TestVolumes(object):
-
- def test_volume_get_has_no_volumes(self, volumes):
- assert volumes.get() is None
-
- def test_volume_get_filtered_has_no_volumes(self, volumes):
- assert volumes.get(lv_name='ceph') is None
-
- def test_volume_has_multiple_matches(self, volumes):
- volume1 = volume2 = api.Volume(lv_name='foo', lv_path='/dev/vg/lv', lv_tags='')
- volumes.append(volume1)
- volumes.append(volume2)
- with pytest.raises(exceptions.MultipleLVsError):
- volumes.get(lv_name='foo')
-
- def test_as_dict_infers_type_from_tags(self, volumes):
- lv_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags=lv_tags)
- volumes.append(osd)
- result = volumes.get(lv_tags={'ceph.type': 'data'}).as_dict()
- assert result['type'] == 'data'
-
- def test_as_dict_populates_path_from_lv_api(self, volumes):
- lv_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags=lv_tags)
- volumes.append(osd)
- result = volumes.get(lv_tags={'ceph.type': 'data'}).as_dict()
- assert result['path'] == '/dev/vg/lv'
-
- def test_find_the_correct_one(self, volumes):
- volume1 = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags='')
- volume2 = api.Volume(lv_name='volume2', lv_path='/dev/vg/lv', lv_tags='')
- volumes.append(volume1)
- volumes.append(volume2)
- assert volumes.get(lv_name='volume1') == volume1
-
- def test_filter_by_tag(self, volumes):
- lv_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags=lv_tags)
- journal = api.Volume(lv_name='volume2', lv_path='/dev/vg/lv', lv_tags='ceph.type=journal')
- volumes.append(osd)
- volumes.append(journal)
- volumes.filter(lv_tags={'ceph.type': 'data'})
- assert len(volumes) == 1
- assert volumes[0].lv_name == 'volume1'
-
- def test_filter_by_tag_does_not_match_one(self, volumes):
- lv_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags=lv_tags)
- journal = api.Volume(lv_name='volume2', lv_path='/dev/vg/lv', lv_tags='ceph.osd_id=1,ceph.type=journal')
- volumes.append(osd)
- volumes.append(journal)
- # note the different osd_id!
- volumes.filter(lv_tags={'ceph.type': 'data', 'ceph.osd_id': '2'})
- assert volumes == []
-
- def test_filter_by_vg_name(self, volumes):
- lv_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.Volume(lv_name='volume1', vg_name='ceph_vg', lv_tags=lv_tags)
- journal = api.Volume(lv_name='volume2', vg_name='system_vg', lv_tags='ceph.type=journal')
- volumes.append(osd)
- volumes.append(journal)
- volumes.filter(vg_name='ceph_vg')
- assert len(volumes) == 1
- assert volumes[0].lv_name == 'volume1'
-
- def test_filter_by_lv_path(self, volumes):
- osd = api.Volume(lv_name='volume1', lv_path='/dev/volume1', lv_tags='')
- journal = api.Volume(lv_name='volume2', lv_path='/dev/volume2', lv_tags='')
- volumes.append(osd)
- volumes.append(journal)
- volumes.filter(lv_path='/dev/volume1')
- assert len(volumes) == 1
- assert volumes[0].lv_name == 'volume1'
-
- def test_filter_by_lv_uuid(self, volumes):
- osd = api.Volume(lv_name='volume1', lv_path='/dev/volume1', lv_uuid='1111', lv_tags='')
- journal = api.Volume(lv_name='volume2', lv_path='/dev/volume2', lv_uuid='', lv_tags='')
- volumes.append(osd)
- volumes.append(journal)
- volumes.filter(lv_uuid='1111')
- assert len(volumes) == 1
- assert volumes[0].lv_name == 'volume1'
-
- def test_filter_by_lv_uuid_nothing_found(self, volumes):
- osd = api.Volume(lv_name='volume1', lv_path='/dev/volume1', lv_uuid='1111', lv_tags='')
- journal = api.Volume(lv_name='volume2', lv_path='/dev/volume2', lv_uuid='', lv_tags='')
- volumes.append(osd)
- volumes.append(journal)
- volumes.filter(lv_uuid='22222')
- assert volumes == []
-
- def test_filter_requires_params(self, volumes):
- with pytest.raises(TypeError):
- volumes.filter()
-
-
class TestVolumeGroup(object):
def test_volume_group_no_empty_name(self):
api.VolumeGroup(vg_name='')
-class TestVolumeGroups(object):
-
- def test_volume_get_has_no_volume_groups(self, volume_groups):
- assert volume_groups.get() is None
-
- def test_volume_get_filtered_has_no_volumes(self, volume_groups):
- assert volume_groups.get(vg_name='ceph') is None
-
- def test_volume_has_multiple_matches(self, volume_groups):
- volume1 = volume2 = api.VolumeGroup(vg_name='foo', lv_path='/dev/vg/lv', lv_tags='')
- volume_groups.append(volume1)
- volume_groups.append(volume2)
- with pytest.raises(exceptions.MultipleVGsError):
- volume_groups.get(vg_name='foo')
-
- def test_find_the_correct_one(self, volume_groups):
- volume1 = api.VolumeGroup(vg_name='volume1', lv_tags='')
- volume2 = api.VolumeGroup(vg_name='volume2', lv_tags='')
- volume_groups.append(volume1)
- volume_groups.append(volume2)
- assert volume_groups.get(vg_name='volume1') == volume1
-
- def test_filter_by_tag(self, volume_groups):
- vg_tags = "ceph.group=dmcache"
- osd = api.VolumeGroup(vg_name='volume1', vg_tags=vg_tags)
- journal = api.VolumeGroup(vg_name='volume2', vg_tags='ceph.group=plain')
- volume_groups.append(osd)
- volume_groups.append(journal)
- volume_groups = volume_groups.filter(vg_tags={'ceph.group': 'dmcache'})
- assert len(volume_groups) == 1
- assert volume_groups[0].vg_name == 'volume1'
-
- def test_filter_by_tag_does_not_match_one(self, volume_groups):
- vg_tags = "ceph.group=dmcache,ceph.disk_type=ssd"
- osd = api.VolumeGroup(vg_name='volume1', vg_path='/dev/vg/lv', vg_tags=vg_tags)
- volume_groups.append(osd)
- volume_groups = volume_groups.filter(vg_tags={'ceph.group': 'data', 'ceph.disk_type': 'ssd'})
- assert volume_groups == []
-
- def test_filter_by_vg_name(self, volume_groups):
- vg_tags = "ceph.type=data,ceph.fsid=000-aaa"
- osd = api.VolumeGroup(vg_name='ceph_vg', vg_tags=vg_tags)
- journal = api.VolumeGroup(vg_name='volume2', vg_tags='ceph.type=journal')
- volume_groups.append(osd)
- volume_groups.append(journal)
- volume_groups = volume_groups.filter(vg_name='ceph_vg')
- assert len(volume_groups) == 1
- assert volume_groups[0].vg_name == 'ceph_vg'
-
- def test_filter_requires_params(self, volume_groups):
- with pytest.raises(TypeError):
- volume_groups = volume_groups.filter()
-
-
class TestVolumeGroupFree(object):
def test_integer_gets_produced(self):
self.vg.sizing(size=2048)
-class TestGetLVFromArgument(object):
-
- def setup(self):
- self.foo_volume = api.Volume(
- lv_name='foo', lv_path='/path/to/lv',
- vg_name='foo_group', lv_tags=''
- )
-
- def test_non_absolute_path_is_not_valid(self, volumes):
- volumes.append(self.foo_volume)
- assert api.get_lv_from_argument('foo') is None
-
- def test_too_many_slashes_is_invalid(self, volumes):
- volumes.append(self.foo_volume)
- assert api.get_lv_from_argument('path/to/lv') is None
-
- def test_absolute_path_is_not_lv(self, volumes):
- volumes.append(self.foo_volume)
- assert api.get_lv_from_argument('/path') is None
-
- def test_absolute_path_is_lv(self, volumes):
- volumes.append(self.foo_volume)
- assert api.get_lv_from_argument('/path/to/lv') == self.foo_volume
-
-
class TestRemoveLV(object):
def test_removes_lv(self, monkeypatch):
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_uses_size(self, m_get_lv, m_call, m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_uses_size(self, m_get_first_lv, m_call, m_run, monkeypatch):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group, size=5368709120, tags={'ceph.type': 'data'})
expected = ['lvcreate', '--yes', '-l', '1280', '-n', 'foo-0', 'foo_group']
m_run.assert_called_with(expected)
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_uses_extents(self, m_get_lv, m_call, m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_uses_extents(self, m_get_first_lv, m_call, m_run, monkeypatch):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group, extents='50', tags={'ceph.type': 'data'})
expected = ['lvcreate', '--yes', '-l', '50', '-n', 'foo-0', 'foo_group']
m_run.assert_called_with(expected)
(3, 33),])
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_uses_slots(self, m_get_lv, m_call, m_run, monkeypatch, test_input, expected):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_uses_slots(self, m_get_first_lv, m_call, m_run, monkeypatch, test_input, expected):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group, slots=test_input, tags={'ceph.type': 'data'})
expected = ['lvcreate', '--yes', '-l', str(expected), '-n', 'foo-0', 'foo_group']
m_run.assert_called_with(expected)
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_uses_all(self, m_get_lv, m_call, m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_uses_all(self, m_get_first_lv, m_call, m_run, monkeypatch):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group, tags={'ceph.type': 'data'})
expected = ['lvcreate', '--yes', '-l', '100%FREE', '-n', 'foo-0', 'foo_group']
m_run.assert_called_with(expected)
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
@patch('ceph_volume.api.lvm.Volume.set_tags')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_calls_to_set_tags_default(self, m_get_lv, m_set_tags, m_call, m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_calls_to_set_tags_default(self, m_get_first_lv, m_set_tags, m_call, m_run, monkeypatch):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group)
tags = {
"ceph.osd_id": "null",
@patch('ceph_volume.api.lvm.process.run')
@patch('ceph_volume.api.lvm.process.call')
@patch('ceph_volume.api.lvm.Volume.set_tags')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_calls_to_set_tags_arg(self, m_get_lv, m_set_tags, m_call, m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_calls_to_set_tags_arg(self, m_get_first_lv, m_set_tags, m_call, m_run, monkeypatch):
+ m_get_first_lv.return_value = self.foo_volume
api.create_lv('foo', 0, vg=self.foo_group, tags={'ceph.type': 'data'})
tags = {
"ceph.type": "data",
@patch('ceph_volume.api.lvm.process.call')
@patch('ceph_volume.api.lvm.get_device_vgs')
@patch('ceph_volume.api.lvm.create_vg')
- @patch('ceph_volume.api.lvm.get_lv')
- def test_create_vg(self, m_get_lv, m_create_vg, m_get_device_vgs, m_call,
+ @patch('ceph_volume.api.lvm.get_first_lv')
+ def test_create_vg(self, m_get_first_lv, m_create_vg, m_get_device_vgs, m_call,
m_run, monkeypatch):
- m_get_lv.return_value = self.foo_volume
+ m_get_first_lv.return_value = self.foo_volume
m_get_device_vgs.return_value = []
api.create_lv('foo', 0, device='dev/foo', size='5G', tags={'ceph.type': 'data'})
m_create_vg.assert_called_with('dev/foo', name_prefix='ceph')
self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
def test_uses_single_device_in_list(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.extend_vg(self.foo_volume, ['/dev/sda'])
expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
assert fake_run.calls[0]['args'][0] == expected
def test_uses_single_device(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.extend_vg(self.foo_volume, '/dev/sda')
expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda']
assert fake_run.calls[0]['args'][0] == expected
def test_uses_multiple_devices(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.extend_vg(self.foo_volume, ['/dev/sda', '/dev/sdb'])
expected = ['vgextend', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
assert fake_run.calls[0]['args'][0] == expected
self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
def test_uses_single_device_in_list(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.reduce_vg(self.foo_volume, ['/dev/sda'])
expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
assert fake_run.calls[0]['args'][0] == expected
def test_uses_single_device(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.reduce_vg(self.foo_volume, '/dev/sda')
expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda']
assert fake_run.calls[0]['args'][0] == expected
def test_uses_multiple_devices(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.reduce_vg(self.foo_volume, ['/dev/sda', '/dev/sdb'])
expected = ['vgreduce', '--force', '--yes', 'foo', '/dev/sda', '/dev/sdb']
assert fake_run.calls[0]['args'][0] == expected
self.foo_volume = api.VolumeGroup(vg_name='foo', lv_tags='')
def test_no_name(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.create_vg('/dev/sda')
result = fake_run.calls[0]['args'][0]
assert '/dev/sda' in result
assert result[-2].startswith('ceph-')
def test_devices_list(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.create_vg(['/dev/sda', '/dev/sdb'], name='ceph')
result = fake_run.calls[0]['args'][0]
expected = ['vgcreate', '--force', '--yes', 'ceph', '/dev/sda', '/dev/sdb']
assert result == expected
def test_name_prefix(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.create_vg('/dev/sda', name_prefix='master')
result = fake_run.calls[0]['args'][0]
assert '/dev/sda' in result
assert result[-2].startswith('master-')
def test_specific_name(self, monkeypatch, fake_run):
- monkeypatch.setattr(api, 'get_vg', lambda **kw: True)
+ monkeypatch.setattr(api, 'get_first_vg', lambda **kw: True)
api.create_vg('/dev/sda', name='master')
result = fake_run.calls[0]['args'][0]
assert '/dev/sda' in result
assert '/dev/mapper' not in result['VG_NAME']
-class TestIsLV(object):
-
- def test_is_not_an_lv(self, monkeypatch):
- monkeypatch.setattr(api.process, 'call', lambda x, **kw: ('', '', 0))
- monkeypatch.setattr(api, 'dmsetup_splitname', lambda x, **kw: {})
- assert api.is_lv('/dev/sda1', lvs=[]) is False
-
- def test_lvs_not_found(self, monkeypatch, volumes):
- CephVolume = api.Volume(lv_name='foo', lv_path='/dev/vg/foo', lv_tags="ceph.type=data")
- volumes.append(CephVolume)
- splitname = {'LV_NAME': 'data', 'VG_NAME': 'ceph'}
- monkeypatch.setattr(api, 'dmsetup_splitname', lambda x, **kw: splitname)
- assert api.is_lv('/dev/sda1', lvs=volumes) is False
-
- def test_is_lv(self, monkeypatch, volumes):
- CephVolume = api.Volume(
- vg_name='ceph', lv_name='data',
- lv_path='/dev/vg/foo', lv_tags="ceph.type=data"
- )
- volumes.append(CephVolume)
- splitname = {'LV_NAME': 'data', 'VG_NAME': 'ceph'}
- monkeypatch.setattr(api, 'dmsetup_splitname', lambda x, **kw: splitname)
- assert api.is_lv('/dev/sda1', lvs=volumes) is True
-
class TestGetDeviceVgs(object):
@patch('ceph_volume.process.call')
pcall.return_value = ('', '', '')
vgs = api.get_device_lvs('/dev/foo')
assert vgs == []
+
+
+# NOTE: api.convert_filters_to_str() and api.convert_tags_to_str() should get
+# tested automatically while testing api.make_filters_lvmcmd_ready()
+class TestMakeFiltersLVMCMDReady(object):
+
+ def test_with_no_filters_and_no_tags(self):
+ retval = api.make_filters_lvmcmd_ready(None, None)
+
+ assert isinstance(retval, str)
+ assert retval == ''
+
+ def test_with_filters_and_no_tags(self):
+ filters = {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
+
+ retval = api.make_filters_lvmcmd_ready(filters, None)
+
+ assert isinstance(retval, str)
+ for k, v in filters.items():
+ assert k in retval
+ assert v in retval
+
+ def test_with_no_filters_and_with_tags(self):
+ tags = {'ceph.type': 'data', 'ceph.osd_id': '0'}
+
+ retval = api.make_filters_lvmcmd_ready(None, tags)
+
+ assert isinstance(retval, str)
+ assert 'tags' in retval
+ for k, v in tags.items():
+ assert k in retval
+ assert v in retval
+ assert retval.find('tags') < retval.find(k) < retval.find(v)
+
+ def test_with_filters_and_tags(self):
+ filters = {'lv_name': 'lv1', 'lv_path': '/dev/sda'}
+ tags = {'ceph.type': 'data', 'ceph.osd_id': '0'}
+
+ retval = api.make_filters_lvmcmd_ready(filters, tags)
+
+ assert isinstance(retval, str)
+ for f, t in zip(filters.items(), tags.items()):
+ assert f[0] in retval
+ assert f[1] in retval
+ assert t[0] in retval
+ assert t[1] in retval
+ assert retval.find(f[0]) < retval.find(f[1]) < \
+ retval.find('tags') < retval.find(t[0]) < retval.find(t[1])
+
+
+class TestGetPVs(object):
+
+ def test_get_pvs(self, monkeypatch):
+ pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
+ vg_name='vg1')
+ pv2 = api.PVolume(pv_name='/dev/sdb', pv_uuid='0001', pv_tags={},
+ vg_name='vg2')
+ pvs = [pv1, pv2]
+ stdout = ['{};{};{};{};;'.format(pv1.pv_name, pv1.pv_tags, pv1.pv_uuid, pv1.vg_name),
+ '{};{};{};{};;'.format(pv2.pv_name, pv2.pv_tags, pv2.pv_uuid, pv2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ pvs_ = api.get_pvs()
+ assert len(pvs_) == len(pvs)
+ for pv, pv_ in zip(pvs, pvs_):
+ assert pv_.pv_name == pv.pv_name
+
+ def test_get_pvs_single_pv(self, monkeypatch):
+ pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
+ vg_name='vg1')
+ pvs = [pv1]
+ stdout = ['{};;;;;;'.format(pv1.pv_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ pvs_ = api.get_pvs()
+ assert len(pvs_) == 1
+ assert pvs_[0].pv_name == pvs[0].pv_name
+
+ def test_get_pvs_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ assert api.get_pvs() == []
+
+
+class TestGetVGs(object):
+
+ def test_get_vgs(self, monkeypatch):
+ vg1 = api.VolumeGroup(vg_name='vg1')
+ vg2 = api.VolumeGroup(vg_name='vg2')
+ vgs = [vg1, vg2]
+ stdout = ['{};;;;;;'.format(vg1.vg_name),
+ '{};;;;;;'.format(vg2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ vgs_ = api.get_vgs()
+ assert len(vgs_) == len(vgs)
+ for vg, vg_ in zip(vgs, vgs_):
+ assert vg_.vg_name == vg.vg_name
+
+ def test_get_vgs_single_vg(self, monkeypatch):
+ vg1 = api.VolumeGroup(vg_name='vg'); vgs = [vg1]
+ stdout = ['{};;;;;;'.format(vg1.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ vgs_ = api.get_vgs()
+ assert len(vgs_) == 1
+ assert vgs_[0].vg_name == vgs[0].vg_name
+
+ def test_get_vgs_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ assert api.get_vgs() == []
+
+
+class TestGetLVs(object):
+
+ def test_get_lvs(self, monkeypatch):
+ lv1 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg1/lv1',
+ lv_name='lv1', vg_name='vg1')
+ lv2 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg2/lv2',
+ lv_name='lv2', vg_name='vg2')
+ lvs = [lv1, lv2]
+ stdout = ['{};{};{};{}'.format(lv1.lv_tags, lv1.lv_path, lv1.lv_name,
+ lv1.vg_name),
+ '{};{};{};{}'.format(lv2.lv_tags, lv2.lv_path, lv2.lv_name,
+ lv2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ lvs_ = api.get_lvs()
+ assert len(lvs_) == len(lvs)
+ for lv, lv_ in zip(lvs, lvs_):
+ assert lv.__dict__ == lv_.__dict__
+
+ def test_get_lvs_single_lv(self, monkeypatch):
+ stdout = ['ceph.type=data;/dev/vg/lv;lv;vg']
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+ lvs = []
+ lvs.append((api.Volume(lv_tags='ceph.type=data',
+ lv_path='/dev/vg/lv',
+ lv_name='lv', vg_name='vg')))
+
+ lvs_ = api.get_lvs()
+ assert len(lvs_) == len(lvs)
+ assert lvs[0].__dict__ == lvs_[0].__dict__
+
+ def test_get_lvs_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ assert api.get_lvs() == []
+
+
+class TestGetFirstPV(object):
+
+ def test_get_first_pv(self, monkeypatch):
+ pv1 = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
+ vg_name='vg1')
+ pv2 = api.PVolume(pv_name='/dev/sdb', pv_uuid='0001', pv_tags={},
+ vg_name='vg2')
+ stdout = ['{};{};{};{};;'.format(pv1.pv_name, pv1.pv_tags, pv1.pv_uuid, pv1.vg_name),
+ '{};{};{};{};;'.format(pv2.pv_name, pv2.pv_tags, pv2.pv_uuid, pv2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ pv_ = api.get_first_pv()
+ assert isinstance(pv_, api.PVolume)
+ assert pv_.pv_name == pv1.pv_name
+
+ def test_get_first_pv_single_pv(self, monkeypatch):
+ pv = api.PVolume(pv_name='/dev/sda', pv_uuid='0000', pv_tags={},
+ vg_name='vg1')
+ stdout = ['{};;;;;;'.format(pv.pv_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ pv_ = api.get_first_pv()
+ assert isinstance(pv_, api.PVolume)
+ assert pv_.pv_name == pv.pv_name
+
+ def test_get_first_pv_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ assert api.get_first_pv() == []
+
+
+class TestGetFirstVG(object):
+
+ def test_get_first_vg(self, monkeypatch):
+ vg1 = api.VolumeGroup(vg_name='vg1')
+ vg2 = api.VolumeGroup(vg_name='vg2')
+ stdout = ['{};;;;;;'.format(vg1.vg_name), '{};;;;;;'.format(vg2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ vg_ = api.get_first_vg()
+ assert isinstance(vg_, api.VolumeGroup)
+ assert vg_.vg_name == vg1.vg_name
+
+ def test_get_first_vg_single_vg(self, monkeypatch):
+ vg = api.VolumeGroup(vg_name='vg')
+ stdout = ['{};;;;;;'.format(vg.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ vg_ = api.get_first_vg()
+ assert isinstance(vg_, api.VolumeGroup)
+ assert vg_.vg_name == vg.vg_name
+
+ def test_get_first_vg_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ vg_ = api.get_first_vg()
+ assert vg_ == []
+
+
+class TestGetFirstLV(object):
+
+ def test_get_first_lv(self, monkeypatch):
+ lv1 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg1/lv1',
+ lv_name='lv1', vg_name='vg1')
+ lv2 = api.Volume(lv_tags='ceph.type=data', lv_path='/dev/vg2/lv2',
+ lv_name='lv2', vg_name='vg2')
+ stdout = ['{};{};{};{}'.format(lv1.lv_tags, lv1.lv_path, lv1.lv_name,
+ lv1.vg_name),
+ '{};{};{};{}'.format(lv2.lv_tags, lv2.lv_path, lv2.lv_name,
+ lv2.vg_name)]
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+
+ lv_ = api.get_first_lv()
+ assert isinstance(lv_, api.Volume)
+ assert lv_.lv_name == lv1.lv_name
+
+ def test_get_first_lv_single_lv(self, monkeypatch):
+ stdout = ['ceph.type=data;/dev/vg/lv;lv;vg']
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: (stdout, '', 0))
+ lv = api.Volume(lv_tags='ceph.type=data',
+ lv_path='/dev/vg/lv',
+ lv_name='lv', vg_name='vg')
+
+ lv_ = api.get_first_lv()
+ assert isinstance(lv_, api.Volume)
+ assert lv_.lv_name == lv.lv_name
+
+ def test_get_first_lv_empty(self, monkeypatch):
+ monkeypatch.setattr(api.process, 'call', lambda x,**kw: ('', '', 0))
+ assert api.get_lvs() == []
projects / ceph.git / blobdiff