import pytest
from ceph_volume.devices import lvm
+from ceph_volume.api import lvm as api
+from mock.mock import patch, Mock
class TestLVM(object):
def test_main_spits_help_with_no_arguments(self, capsys):
lvm.main.LVM([]).main()
stdout, stderr = capsys.readouterr()
- assert 'Use LVM and LVM-based technologies like dmcache to deploy' in stdout
+ assert 'Use LVM and LVM-based technologies to deploy' in stdout
def test_main_shows_activate_subcommands(self, capsys):
lvm.main.LVM([]).main()
class TestPrepareDevice(object):
- def test_cannot_use_device(self):
+ def test_cannot_use_device(self, factory):
+ args = factory(data='/dev/var/foo')
with pytest.raises(RuntimeError) as error:
- lvm.prepare.Prepare([]).prepare_device(
- '/dev/var/foo', 'data', 'asdf', '0')
+ p = lvm.prepare.Prepare([])
+ p.args = args
+ p.prepare_data_device( 'data', '0')
assert 'Cannot use device (/dev/var/foo)' in str(error.value)
assert 'A vg/lv path or an existing device is needed' in str(error.value)
with pytest.raises(SystemExit):
lvm.prepare.Prepare(argv=['--help']).main()
stdout, stderr = capsys.readouterr()
- assert 'Use the filestore objectstore' in stdout
assert 'Use the bluestore objectstore' in stdout
assert 'A physical device or logical' in stdout
- def test_excludes_filestore_bluestore_flags(self, capsys, device_info):
- device_info()
- with pytest.raises(SystemExit):
- lvm.prepare.Prepare(argv=['--data', '/dev/sdfoo', '--filestore', '--bluestore']).main()
- stdout, stderr = capsys.readouterr()
- expected = 'Cannot use --filestore (filestore) with --bluestore (bluestore)'
- assert expected in stdout
-
- def test_excludes_other_filestore_bluestore_flags(self, capsys, device_info):
- device_info()
- with pytest.raises(SystemExit):
- lvm.prepare.Prepare(argv=[
- '--bluestore', '--data', '/dev/sdfoo',
- '--journal', '/dev/sf14',
- ]).main()
- stdout, stderr = capsys.readouterr()
- expected = 'Cannot use --bluestore (bluestore) with --journal (filestore)'
- assert expected in stdout
-
- def test_excludes_block_and_journal_flags(self, capsys, device_info):
- device_info()
+ @patch('ceph_volume.devices.lvm.prepare.api.is_ceph_device')
+ def test_safe_prepare_osd_already_created(self, m_is_ceph_device):
+ m_is_ceph_device.return_value = True
+ with pytest.raises(RuntimeError) as error:
+ prepare = lvm.prepare.Prepare(argv=[])
+ prepare.args = Mock()
+ prepare.args.data = '/dev/sdfoo'
+ prepare.get_lv = Mock()
+ prepare.safe_prepare()
+ expected = 'skipping {}, it is already prepared'.format('/dev/sdfoo')
+ assert expected in str(error.value)
+
+ def test_setup_device_device_name_is_none(self):
+ result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name=None, tags={'ceph.type': 'data'}, size=0, slots=None)
+ assert result == ('', '', {'ceph.type': 'data'})
+
+ @patch('ceph_volume.api.lvm.Volume.set_tags')
+ @patch('ceph_volume.devices.lvm.prepare.api.get_single_lv')
+ def test_setup_device_lv_passed(self, m_get_single_lv, m_set_tags):
+ fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
+ m_get_single_lv.return_value = fake_volume
+ result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='vg_foo/lv_foo', tags={'ceph.type': 'data'}, size=0, slots=None)
+
+ assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
+ 'ceph.vdo': '0',
+ 'ceph.data_uuid': 'fake-uuid',
+ 'ceph.data_device': '/fake-path'})
+
+ @patch('ceph_volume.devices.lvm.prepare.api.create_lv')
+ @patch('ceph_volume.api.lvm.Volume.set_tags')
+ @patch('ceph_volume.util.disk.is_device')
+ def test_setup_device_device_passed(self, m_is_device, m_set_tags, m_create_lv):
+ fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
+ m_is_device.return_value = True
+ m_create_lv.return_value = fake_volume
+ result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
+
+ assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
+ 'ceph.vdo': '0',
+ 'ceph.data_uuid': 'fake-uuid',
+ 'ceph.data_device': '/fake-path'})
+
+ @patch('ceph_volume.devices.lvm.prepare.Prepare.get_ptuuid')
+ @patch('ceph_volume.devices.lvm.prepare.api.get_single_lv')
+ def test_setup_device_partition_passed(self, m_get_single_lv, m_get_ptuuid):
+ m_get_single_lv.side_effect = ValueError()
+ m_get_ptuuid.return_value = 'fake-uuid'
+ result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
+
+ assert result == ('/dev/sdx', 'fake-uuid', {'ceph.type': 'data',
+ 'ceph.vdo': '0',
+ 'ceph.data_uuid': 'fake-uuid',
+ 'ceph.data_device': '/dev/sdx'})
+
+ def test_invalid_osd_id_passed(self):
with pytest.raises(SystemExit):
- lvm.prepare.Prepare(argv=[
- '--bluestore', '--data', '/dev/sdfoo', '--block.db', 'vg/ceph1',
- '--journal', '/dev/sf14',
- ]).main()
- stdout, stderr = capsys.readouterr()
- expected = 'Cannot use --block.db (bluestore) with --journal (filestore)'
- assert expected in stdout
-
- def test_journal_is_required_with_filestore(self, is_root, monkeypatch, device_info):
- monkeypatch.setattr("os.path.exists", lambda path: True)
- device_info()
- with pytest.raises(SystemExit) as error:
- lvm.prepare.Prepare(argv=['--filestore', '--data', '/dev/sdfoo']).main()
- expected = '--journal is required when using --filestore'
- assert expected in str(error.value)
-
-
-class TestGetJournalLV(object):
-
- @pytest.mark.parametrize('arg', ['', '///', None, '/dev/sda1'])
- def test_no_journal_on_invalid_path(self, monkeypatch, arg):
- monkeypatch.setattr(lvm.prepare.api, 'get_lv', lambda **kw: False)
- prepare = lvm.prepare.Prepare([])
- assert prepare.get_lv(arg) is None
-
- def test_no_journal_lv_found(self, monkeypatch):
- # patch it with 0 so we know we are getting to get_lv
- monkeypatch.setattr(lvm.prepare.api, 'get_lv', lambda **kw: 0)
- prepare = lvm.prepare.Prepare([])
- assert prepare.get_lv('vg/lv') == 0
+ lvm.prepare.Prepare(argv=['--osd-id', 'foo']).main()
class TestActivate(object):