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
   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     def test_excludes_filestore_bluestore_flags(self, capsys, device_info):
  70         device_info()
  71         with pytest.raises(SystemExit):
  72             lvm.prepare.Prepare(argv=['--data', '/dev/sdfoo', '--filestore', '--bluestore']).main()
  73         stdout, stderr = capsys.readouterr()
  74         expected = 'Cannot use --filestore (filestore) with --bluestore (bluestore)'
  75         assert expected in stderr
  76
  77     def test_excludes_other_filestore_bluestore_flags(self, capsys, device_info):
  78         device_info()
  79         with pytest.raises(SystemExit):
  80             lvm.prepare.Prepare(argv=[
  81                 '--bluestore', '--data', '/dev/sdfoo',
  82                 '--journal', '/dev/sf14',
  83             ]).main()
  84         stdout, stderr = capsys.readouterr()
  85         expected = 'Cannot use --bluestore (bluestore) with --journal (filestore)'
  86         assert expected in stderr
  87
  88     def test_excludes_block_and_journal_flags(self, capsys, device_info):
  89         device_info()
  90         with pytest.raises(SystemExit):
  91             lvm.prepare.Prepare(argv=[
  92                 '--bluestore', '--data', '/dev/sdfoo', '--block.db', 'vg/ceph1',
  93                 '--journal', '/dev/sf14',
  94             ]).main()
  95         stdout, stderr = capsys.readouterr()
  96         expected = 'Cannot use --block.db (bluestore) with --journal (filestore)'
  97         assert expected in stderr
  98
  99     def test_journal_is_required_with_filestore(self, is_root, monkeypatch, device_info):
 100         monkeypatch.setattr("os.path.exists", lambda path: True)
 101         device_info()
 102         with pytest.raises(SystemExit) as error:
 103             lvm.prepare.Prepare(argv=['--filestore', '--data', '/dev/sdfoo']).main()
 104         expected = '--journal is required when using --filestore'
 105         assert expected in str(error.value)
 106
 107     @patch('ceph_volume.devices.lvm.prepare.api.is_ceph_device')
 108     def test_safe_prepare_osd_already_created(self, m_is_ceph_device):
 109         m_is_ceph_device.return_value = True
 110         with pytest.raises(RuntimeError) as error:
 111             prepare = lvm.prepare.Prepare(argv=[])
 112             prepare.args = Mock()
 113             prepare.args.data = '/dev/sdfoo'
 114             prepare.get_lv = Mock()
 115             prepare.safe_prepare()
 116             expected = 'skipping {}, it is already prepared'.format('/dev/sdfoo')
 117             assert expected in str(error.value)
 118
 119     def test_setup_device_device_name_is_none(self):
 120         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name=None, tags={'ceph.type': 'data'}, size=0, slots=None)
 121         assert result == ('', '', {'ceph.type': 'data'})
 122
 123     @patch('ceph_volume.api.lvm.Volume.set_tags')
 124     @patch('ceph_volume.devices.lvm.prepare.api.get_first_lv')
 125     def test_setup_device_lv_passed(self, m_get_first_lv, m_set_tags):
 126         fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
 127         m_get_first_lv.return_value = fake_volume
 128         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='vg_foo/lv_foo', tags={'ceph.type': 'data'}, size=0, slots=None)
 129
 130         assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
 131                                                     'ceph.vdo': '0',
 132                                                     'ceph.data_uuid': 'fake-uuid',
 133                                                     'ceph.data_device': '/fake-path'})
 134
 135     @patch('ceph_volume.devices.lvm.prepare.api.create_lv')
 136     @patch('ceph_volume.api.lvm.Volume.set_tags')
 137     @patch('ceph_volume.util.disk.is_device')
 138     def test_setup_device_device_passed(self, m_is_device, m_set_tags, m_create_lv):
 139         fake_volume = api.Volume(lv_name='lv_foo', lv_path='/fake-path', vg_name='vg_foo', lv_tags='', lv_uuid='fake-uuid')
 140         m_is_device.return_value = True
 141         m_create_lv.return_value = fake_volume
 142         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
 143
 144         assert result == ('/fake-path', 'fake-uuid', {'ceph.type': 'data',
 145                                                     'ceph.vdo': '0',
 146                                                     'ceph.data_uuid': 'fake-uuid',
 147                                                     'ceph.data_device': '/fake-path'})
 148
 149     @patch('ceph_volume.devices.lvm.prepare.Prepare.get_ptuuid')
 150     @patch('ceph_volume.devices.lvm.prepare.api.get_first_lv')
 151     def test_setup_device_partition_passed(self, m_get_first_lv, m_get_ptuuid):
 152         m_get_first_lv.side_effect = ValueError()
 153         m_get_ptuuid.return_value = 'fake-uuid'
 154         result = lvm.prepare.Prepare([]).setup_device(device_type='data', device_name='/dev/sdx', tags={'ceph.type': 'data'}, size=0, slots=None)
 155
 156         assert result == ('/dev/sdx', 'fake-uuid', {'ceph.type': 'data',
 157                                                     'ceph.vdo': '0',
 158                                                     'ceph.data_uuid': 'fake-uuid',
 159                                                     'ceph.data_device': '/dev/sdx'})
 160
 161
 162 class TestActivate(object):
 163
 164     def test_main_spits_help_with_no_arguments(self, capsys):
 165         lvm.activate.Activate([]).main()
 166         stdout, stderr = capsys.readouterr()
 167         assert 'Activate OSDs by discovering them with' in stdout
 168
 169     def test_main_shows_full_help(self, capsys):
 170         with pytest.raises(SystemExit):
 171             lvm.activate.Activate(argv=['--help']).main()
 172         stdout, stderr = capsys.readouterr()
 173         assert 'optional arguments' in stdout
 174         assert 'positional arguments' in stdout