update ceph source to reef 18.2.0

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

diff --git a/ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py b/ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py
index c69394fd69479223b43128b7e9c4dc073aa6f6b0..0a356988eebc5151d9c685d243810e48e3aba597 100644 (file)
--- a/ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py
+++ b/ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_prepare.py
@@ -1,5 +1,7 @@
 import pytest
 from ceph_volume.devices import lvm
 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):
 
 
 class TestLVM(object):


@@ -7,7 +9,7 @@ class TestLVM(object):
     def test_main_spits_help_with_no_arguments(self, capsys):
         lvm.main.LVM([]).main()
         stdout, stderr = capsys.readouterr()
     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()
 
     def test_main_shows_activate_subcommands(self, capsys):
         lvm.main.LVM([]).main()


@@ -22,6 +24,33 @@ class TestLVM(object):
         assert 'Format an LVM device' in stdout
 
 
         assert 'Format an LVM device' in stdout
 
 

+class TestPrepareDevice(object):
+
+    def test_cannot_use_device(self, factory):
+        args = factory(data='/dev/var/foo')
+        with pytest.raises(RuntimeError) as error:
+            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)
+
+
+class TestGetClusterFsid(object):
+
+    def test_fsid_is_passed_in(self, factory):
+        args = factory(cluster_fsid='aaaa-1111')
+        prepare_obj = lvm.prepare.Prepare([])
+        prepare_obj.args = args
+        assert prepare_obj.get_cluster_fsid() == 'aaaa-1111'
+
+    def test_fsid_is_read_from_ceph_conf(self, factory, conf_ceph_stub):
+        conf_ceph_stub('[global]\nfsid = bbbb-2222')
+        prepare_obj = lvm.prepare.Prepare([])
+        prepare_obj.args = factory(cluster_fsid=None)
+        assert prepare_obj.get_cluster_fsid() == 'bbbb-2222'
+
+

 class TestPrepare(object):
 
     def test_main_spits_help_with_no_arguments(self, capsys):
 class TestPrepare(object):
 
     def test_main_spits_help_with_no_arguments(self, capsys):


@@ -33,24 +62,66 @@ class TestPrepare(object):
         with pytest.raises(SystemExit):
             lvm.prepare.Prepare(argv=['--help']).main()
         stdout, stderr = capsys.readouterr()
         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
 
         assert 'Use the bluestore objectstore' in stdout
         assert 'A physical device or logical' in stdout
 

-
-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
+    @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=['--osd-id', 'foo']).main()

 
 
 class TestActivate(object):
 
 
 class TestActivate(object):


@@ -66,4 +137,3 @@ class TestActivate(object):
         stdout, stderr = capsys.readouterr()
         assert 'optional arguments' in stdout
         assert 'positional arguments' in stdout
         stdout, stderr = capsys.readouterr()
         assert 'optional arguments' in stdout
         assert 'positional arguments' in stdout

-