ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py

   1 import pytest
   2 from ceph_volume.devices import lvm
   3 from ceph_volume.api import lvm as api
   4 from mock.mock import patch, Mock, MagicMock
   5
   6
   7 class TestLVM(object):
   8
   9     def test_main_spits_help_with_no_arguments(self, capsys):
  10         lvm.main.LVM([]).main()
  11         stdout, stderr = capsys.readouterr()
  12         assert 'Use LVM and LVM-based technologies to deploy' in stdout
  13
  14     def test_main_shows_activate_subcommands(self, capsys):
  15         lvm.main.LVM([]).main()
  16         stdout, stderr = capsys.readouterr()
  17         assert 'activate ' in stdout
  18         assert 'Discover and mount' in stdout
  19
  20     def test_main_shows_prepare_subcommands(self, capsys):
  21         lvm.main.LVM([]).main()
  22         stdout, stderr = capsys.readouterr()
  23         assert 'prepare ' in stdout
  24         assert 'Format an LVM device' in stdout
  25
  26
  27 class TestPrepareDevice(object):
  28
  29     def test_cannot_use_device(self, factory):
  30         args = factory(data='/dev/var/foo')
  31         with pytest.raises(RuntimeError) as error:
  32             p = lvm.prepare.Prepare([])
  33             p.args = args
  34             p.prepare_data_device( 'data', '0')
  35         assert 'Cannot use device (/dev/var/foo)' in str(error.value)
  36         assert 'A vg/lv path or an existing device is needed' in str(error.value)
  37
  38
  39 class TestGetClusterFsid(object):
  40
  41     def test_fsid_is_passed_in(self, factory):
  42         args = factory(cluster_fsid='aaaa-1111')
  43         prepare_obj = lvm.prepare.Prepare([])
  44         prepare_obj.args = args
  45         assert prepare_obj.get_cluster_fsid() == 'aaaa-1111'
  46
  47     def test_fsid_is_read_from_ceph_conf(self, factory, conf_ceph_stub):
  48         conf_ceph_stub('[global]\nfsid = bbbb-2222')
  49         prepare_obj = lvm.prepare.Prepare([])
  50         prepare_obj.args = factory(cluster_fsid=None)
  51         assert prepare_obj.get_cluster_fsid() == 'bbbb-2222'
  52
  53
  54 class TestPrepare(object):
  55
  56     def test_main_spits_help_with_no_arguments(self, capsys):
  57         lvm.prepare.Prepare([]).main()
  58         stdout, stderr = capsys.readouterr()
  59         assert 'Prepare an OSD by assigning an ID and FSID' in stdout
  60
  61     def test_main_shows_full_help(self, capsys):
  62         with pytest.raises(SystemExit):
  63             lvm.prepare.Prepare(argv=['--help']).main()
  64         stdout, stderr = capsys.readouterr()
  65         assert 'Use the filestore objectstore' in stdout
  66         assert 'Use the bluestore objectstore' in stdout
  67         assert 'A physical device or logical' in stdout
  68
  69
  70     @patch('ceph_volume.util.disk.has_bluestore_label', return_value=False)
  71     def test_excludes_filestore_bluestore_flags(self, m_has_bs_label, fake_call, capsys, device_info):
  72         device_info()
  73         with pytest.raises(SystemExit):
  74             lvm.prepare.Prepare(argv=['--data', '/dev/sdfoo', '--filestore', '--bluestore']).main()
  75         stdout, stderr = capsys.readouterr()
  76         expected = 'Cannot use --filestore (filestore) with --bluestore (bluestore)'
  77         assert expected in stderr
  78
  79
  80     @patch('ceph_volume.util.disk.has_bluestore_label', return_value=False)
  81     def test_excludes_other_filestore_bluestore_flags(self, m_has_bs_label, fake_call, capsys, device_info):
  82         device_info()
  83         with pytest.raises(SystemExit):
  84             lvm.prepare.Prepare(argv=[
  85                 '--bluestore', '--data', '/dev/sdfoo',
  86                 '--journal', '/dev/sf14',
  87             ]).main()
  88         stdout, stderr = capsys.readouterr()
  89         expected = 'Cannot use --bluestore (bluestore) with --journal (filestore)'
  90         assert expected in stderr
  91
  92     @patch('ceph_volume.util.disk.has_bluestore_label', return_value=False)
  93     def test_excludes_block_and_journal_flags(self, m_has_bs_label, fake_call, capsys, device_info):
  94         device_info()
  95         with pytest.raises(SystemExit):
  96             lvm.prepare.Prepare(argv=[
  97                 '--bluestore', '--data', '/dev/sdfoo', '--block.db', 'vg/ceph1',
  98                 '--journal', '/dev/sf14',
  99             ]).main()
 100         stdout, stderr = capsys.readouterr()
 101         expected = 'Cannot use --block.db (bluestore) with --journal (filestore)'
 102         assert expected in stderr
 103
 104     @patch('ceph_volume.util.arg_validators.Device')
 105     @patch('ceph_volume.util.disk.has_bluestore_label', return_value=False)
 106     def test_journal_is_required_with_filestore(self, m_has_bs_label, m_device, is_root, monkeypatch, device_info):
 107         m_device.return_value = MagicMock(exists=True,
 108                                           has_fs=False,
 109                                           used_by_ceph=False,
 110                                           has_partitions=False,
 111                                           has_gpt_headers=False)
 112         monkeypatch.setattr("os.path.exists", lambda path: True)
 113         with pytest.raises(SystemExit) as error:
 114             lvm.prepare.Prepare(argv=['--filestore', '--data', '/dev/sdfoo']).main()
 115         expected = '--journal is required when using --filestore'
 116         assert expected in str(error.value)
 117
 118     @patch('ceph_volume.devices.lvm.prepare.api.is_ceph_device')
 119     def test_safe_prepare_osd_already_created(self, m_is_ceph_device):
 120         m_is_ceph_device.return_value = True
 121         with pytest.raises(RuntimeError) as error:
 122             prepare = lvm.prepare.Prepare(argv=[])
 123             prepare.args = Mock()
 124             prepare.args.data = '/dev/sdfoo'
 125             prepare.get_lv = Mock()
 126             prepare.safe_prepare()
 127             expected = 'skipping {}, it is already prepared'.format('/dev/sdfoo')
 128             assert expected in str(error.value)
 129
 130     def test_setup_device_device_name_is_none(self):
 131         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name=None, tags={'ceph.type': 'data'}, size=0, slots=None)
 132         assert result == ('', '', {'ceph.type': 'data'})
 133
 134     @patch('ceph_volume.api.lvm.Volume.set_tags')
 135     @patch('ceph_volume.devices.lvm.prepare.api.get_single_lv')
 136     def test_setup_device_lv_passed(self, m_get_single_lv, m_set_tags):
 137         fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
 138         m_get_single_lv.return_value = fake_volume
 139         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='vg_foo/lv_foo', tags={'ceph.type': 'data'}, size=0, slots=None)
 140
 141         assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
 142                                                     'ceph.vdo': '0',
 143                                                     'ceph.data_uuid': 'fake-uuid',
 144                                                     'ceph.data_device': '/fake-path'})
 145
 146     @patch('ceph_volume.devices.lvm.prepare.api.create_lv')
 147     @patch('ceph_volume.api.lvm.Volume.set_tags')
 148     @patch('ceph_volume.util.disk.is_device')
 149     def test_setup_device_device_passed(self, m_is_device, m_set_tags, m_create_lv):
 150         fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
 151         m_is_device.return_value = True
 152         m_create_lv.return_value = fake_volume
 153         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
 154
 155         assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
 156                                                     'ceph.vdo': '0',
 157                                                     'ceph.data_uuid': 'fake-uuid',
 158                                                     'ceph.data_device': '/fake-path'})
 159
 160     @patch('ceph_volume.devices.lvm.prepare.Prepare.get_ptuuid')
 161     @patch('ceph_volume.devices.lvm.prepare.api.get_single_lv')
 162     def test_setup_device_partition_passed(self, m_get_single_lv, m_get_ptuuid):
 163         m_get_single_lv.side_effect = ValueError()
 164         m_get_ptuuid.return_value = 'fake-uuid'
 165         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
 166
 167         assert result == ('/dev/sdx', 'fake-uuid', {'ceph.type': 'data',
 168                                                     'ceph.vdo': '0',
 169                                                     'ceph.data_uuid': 'fake-uuid',
 170                                                     'ceph.data_device': '/dev/sdx'})
 171
 172     def test_invalid_osd_id_passed(self):
 173         with pytest.raises(SystemExit):
 174             lvm.prepare.Prepare(argv=['--osd-id', 'foo']).main()
 175
 176
 177 class TestActivate(object):
 178
 179     def test_main_spits_help_with_no_arguments(self, capsys):
 180         lvm.activate.Activate([]).main()
 181         stdout, stderr = capsys.readouterr()
 182         assert 'Activate OSDs by discovering them with' in stdout
 183
 184     def test_main_shows_full_help(self, capsys):
 185         with pytest.raises(SystemExit):
 186             lvm.activate.Activate(argv=['--help']).main()
 187         stdout, stderr = capsys.readouterr()
 188         assert 'optional arguments' in stdout
 189         assert 'positional arguments' in stdout