ceph-volume: update unit tests

This commit consolidates all necessary changes to address unit tests
related to the introduction of Python type annotations.

Signed-off-by: Guillaume Abrioux <[email protected]>

Author: guits

src/ceph-volume/ceph_volume/tests/api/test_lvm.py

@@ -109,8 +109,8 @@ class TestVolumeGroupSizing(object):
 
     def setup_method(self):
         self.vg = api.VolumeGroup(vg_name='ceph',
-                                         vg_extent_size=1073741824,
-                                         vg_free_count=1024)
+                                  vg_extent_size=1073741824,
+                                  vg_free_count=1024)
 
     def test_parts_and_size_errors(self):
         with pytest.raises(ValueError) as error:
@@ -194,8 +194,8 @@ def setup_method(self):
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_uses_size(self, m_get_single_lv, m_call, m_run, monkeypatch):
         m_get_single_lv.return_value = self.foo_volume
-        api.create_lv('foo', 0, vg=self.foo_group, size=419430400, tags={'ceph.type': 'data'})
-        expected = (['lvcreate', '--yes', '-l', '100', '-n', 'foo-0', 'foo_group'])
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group, size=419430400, tags={'ceph.type': 'data'})
+        expected = (['lvcreate', '--yes', '-l', '100', '-n', 'foo-1234-abcd', 'foo_group'])
         m_run.assert_called_with(expected, run_on_host=True)
 
     @patch('ceph_volume.api.lvm.process.run')
@@ -209,8 +209,8 @@ def test_uses_size_adjust_if_1percent_over(self, m_get_single_lv, m_call, m_run,
                                     vg_free_count="1000")
         m_get_single_lv.return_value = foo_volume
         # 423624704 should be just under 1% off of the available size 419430400
-        api.create_lv('foo', 0, vg=foo_group, size=4232052736, tags={'ceph.type': 'data'})
-        expected = ['lvcreate', '--yes', '-l', '1000', '-n', 'foo-0', 'foo_group']
+        api.create_lv('foo', '1234-abcd', vg=foo_group, size=4232052736, tags={'ceph.type': 'data'})
+        expected = ['lvcreate', '--yes', '-l', '1000', '-n', 'foo-1234-abcd', 'foo_group']
         m_run.assert_called_with(expected, run_on_host=True)
 
     @patch('ceph_volume.api.lvm.process.run')
@@ -219,15 +219,15 @@ def test_uses_size_adjust_if_1percent_over(self, m_get_single_lv, m_call, m_run,
     def test_uses_size_too_large(self, m_get_single_lv, m_call, m_run, monkeypatch):
         m_get_single_lv.return_value = self.foo_volume
         with pytest.raises(RuntimeError):
-            api.create_lv('foo', 0, vg=self.foo_group, size=5368709120, tags={'ceph.type': 'data'})
+            api.create_lv('foo', '1234-abcd', vg=self.foo_group, size=5368709120, tags={'ceph.type': 'data'})
 
     @patch('ceph_volume.api.lvm.process.run')
     @patch('ceph_volume.api.lvm.process.call')
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_uses_extents(self, m_get_single_lv, m_call, m_run, monkeypatch):
         m_get_single_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']
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group, extents='50', tags={'ceph.type': 'data'})
+        expected = ['lvcreate', '--yes', '-l', '50', '-n', 'foo-1234-abcd', 'foo_group']
         m_run.assert_called_with(expected, run_on_host=True)
 
     @pytest.mark.parametrize("test_input,expected",
@@ -238,17 +238,17 @@ def test_uses_extents(self, m_get_single_lv, m_call, m_run, monkeypatch):
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_uses_slots(self, m_get_single_lv, m_call, m_run, monkeypatch, test_input, expected):
         m_get_single_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']
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group, slots=test_input, tags={'ceph.type': 'data'})
+        expected = ['lvcreate', '--yes', '-l', str(expected), '-n', 'foo-1234-abcd', 'foo_group']
         m_run.assert_called_with(expected, run_on_host=True)
 
     @patch('ceph_volume.api.lvm.process.run')
     @patch('ceph_volume.api.lvm.process.call')
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_uses_all(self, m_get_single_lv, m_call, m_run, monkeypatch):
         m_get_single_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']
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group, tags={'ceph.type': 'data'})
+        expected = ['lvcreate', '--yes', '-l', '100%FREE', '-n', 'foo-1234-abcd', 'foo_group']
         m_run.assert_called_with(expected, run_on_host=True)
 
     @patch('ceph_volume.api.lvm.process.run')
@@ -257,7 +257,7 @@ def test_uses_all(self, m_get_single_lv, m_call, m_run, monkeypatch):
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_calls_to_set_tags_default(self, m_get_single_lv, m_set_tags, m_call, m_run, monkeypatch):
         m_get_single_lv.return_value = self.foo_volume
-        api.create_lv('foo', 0, vg=self.foo_group)
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group)
         tags = {
             "ceph.osd_id": "null",
             "ceph.type": "null",
@@ -272,7 +272,7 @@ def test_calls_to_set_tags_default(self, m_get_single_lv, m_set_tags, m_call, m_
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_calls_to_set_tags_arg(self, m_get_single_lv, m_set_tags, m_call, m_run, monkeypatch):
         m_get_single_lv.return_value = self.foo_volume
-        api.create_lv('foo', 0, vg=self.foo_group, tags={'ceph.type': 'data'})
+        api.create_lv('foo', '1234-abcd', vg=self.foo_group, tags={'ceph.type': 'data'})
         tags = {
             "ceph.type": "data",
             "ceph.data_device": "/path"
@@ -288,7 +288,7 @@ def test_create_vg(self, m_get_single_lv, m_create_vg, m_get_device_vgs, m_call,
                        m_run, monkeypatch):
         m_get_single_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'})
+        api.create_lv('foo', '1234-abcd', device='dev/foo', size='5G', tags={'ceph.type': 'data'})
         m_create_vg.assert_called_with('dev/foo', name_prefix='ceph')
 
 

src/ceph-volume/ceph_volume/tests/conftest.py

@@ -291,7 +291,7 @@ def disable_kernel_queries(monkeypatch):
 
 
 @pytest.fixture(params=[
-    '', 'ceph data', 'ceph journal', 'ceph block',
+    'ceph data', 'ceph journal', 'ceph block',
     'ceph block.wal', 'ceph block.db', 'ceph lockbox'])
 def ceph_partlabel(request):
     return request.param

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

@@ -13,7 +13,8 @@ class TestBatch(object):
 
     def test_batch_instance(self, is_root):
         b = batch.Batch([])
-        b.main()
+        with pytest.raises(SystemExit):
+            b.main()
 
     def test_invalid_osd_ids_passed(self):
         with pytest.raises(SystemExit):

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

@@ -262,7 +262,7 @@ def test_report_by_osd_id_for_just_block_dev(self, monkeypatch):
         monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs)
 
         listing = lvm.listing.List([])
-        result = listing.single_report(0)
+        result = listing.single_report('0')
         assert result['0'][0]['name'] == 'lv1'
         assert result['0'][0]['lv_tags'] == tags
         assert result['0'][0]['lv_path'] == '/dev/vg/lv1'
@@ -277,7 +277,7 @@ def test_report_by_osd_id_for_just_data_dev(self, monkeypatch):
         monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs)
 
         listing = lvm.listing.List([])
-        result = listing.single_report(0)
+        result = listing.single_report('0')
         assert result['0'][0]['name'] == 'lv1'
         assert result['0'][0]['lv_tags'] == tags
         assert result['0'][0]['lv_path'] == '/dev/vg/lv1'
@@ -298,7 +298,7 @@ def test_report_by_osd_id_for_just_block_wal_and_db_dev(self, monkeypatch):
         monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs)
 
         listing = lvm.listing.List([])
-        result = listing.single_report(0)
+        result = listing.single_report('0')
         assert result['0'][0]['name'] == 'lv1'
         assert result['0'][0]['lv_tags'] == tags1
         assert result['0'][0]['lv_path'] == '/dev/vg/lv1'
@@ -324,7 +324,7 @@ def test_report_by_osd_id_for_data_and_journal_dev(self, monkeypatch):
         monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs)
 
         listing = lvm.listing.List([])
-        result = listing.single_report(0)
+        result = listing.single_report('0')
         assert result['0'][0]['name'] == 'lv1'
         assert result['0'][0]['lv_tags'] == tags1
         assert result['0'][0]['lv_path'] == '/dev/vg/lv1'

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

@@ -1,7 +1,7 @@
 import pytest
 from ceph_volume.devices import lvm
 from ceph_volume.api import lvm as api
-from mock.mock import patch, Mock
+from mock.mock import patch
 from ceph_volume import objectstore
 
 
@@ -72,19 +72,6 @@ def test_main_shows_full_help(self, m_create_key, capsys):
         assert 'Use the bluestore objectstore' in stdout
         assert 'A physical device or logical' in stdout
 
-    @patch('ceph_volume.api.lvm.is_ceph_device')
-    def test_safe_prepare_osd_already_created(self, m_create_key, m_is_ceph_device):
-        m_is_ceph_device.return_value = True
-        with pytest.raises(RuntimeError) as error:
-            self.p.args = Mock()
-            self.p.args.data = '/dev/sdfoo'
-            self.p.args.with_tpm = '0'
-            self.p.get_lv = Mock()
-            self.p.objectstore = objectstore.lvmbluestore.LvmBlueStore(args=self.p.args)
-            self.p.objectstore.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, m_create_key):
         self.p.objectstore = objectstore.lvmbluestore.LvmBlueStore(args=[])
         result = self.p.objectstore.setup_device(device_type='data',

src/ceph-volume/ceph_volume/tests/objectstore/test_lvmbluestore.py

@@ -142,21 +142,6 @@ def test_prepare_data_device_fails(self, factory):
         assert ('Cannot use device (/dev/foo). '
         'A vg/lv path or an existing device is needed') == str(error.value)
 
-    @patch('ceph_volume.api.lvm.is_ceph_device', Mock(return_value=True))
-    @patch('ceph_volume.api.lvm.get_single_lv')
-    def test_safe_prepare_is_ceph_device(self, m_get_single_lv, factory):
-        args = factory(data='/dev/foo')
-        self.lvm_bs.args = args
-        m_get_single_lv.return_value = Volume(lv_name='lv_foo',
-                                              lv_path='/fake-path',
-                                              vg_name='vg_foo',
-                                              lv_tags='',
-                                              lv_uuid='fake-uuid')
-        self.lvm_bs.prepare = MagicMock()
-        with pytest.raises(RuntimeError) as error:
-            self.lvm_bs.safe_prepare(args)
-        assert str(error.value) == 'skipping /dev/foo, it is already prepared'
-
     @patch('ceph_volume.api.lvm.is_ceph_device', Mock(return_value=False))
     @patch('ceph_volume.api.lvm.get_single_lv')
     def test_safe_prepare(self, m_get_single_lv, factory):
@@ -287,15 +272,17 @@ def test_setup_device_is_device(self, m_create_lv, m_set_tags):
                                           {},
                                           1,
                                           1)
-        assert m_create_lv.mock_calls == [call('osd-block',
-                                               'd83fa1ca-bd68-4c75-bdc2-464da58e8abd',
+        assert m_create_lv.mock_calls == [call(name_prefix='osd-block',
+                                               uuid='d83fa1ca-bd68-4c75-bdc2-464da58e8abd',
+                                               vg=None,
                                                device='/dev/foo',
+                                               slots=1,
+                                               extents=None,
+                                               size=1,
                                                tags={'ceph.type': 'block',
                                                      'ceph.vdo': '0',
                                                      'ceph.block_device': '/fake-path',
-                                                     'ceph.block_uuid': 'fake-uuid'},
-                                               slots=1,
-                                               size=1)]
+                                                     'ceph.block_uuid': 'fake-uuid'})]
         assert result == ('/fake-path',
                          'fake-uuid',
                          {'ceph.type': 'block',

src/ceph-volume/ceph_volume/tests/util/test_device.py

@@ -1,3 +1,4 @@
+#  type: ignore
 import os
 import pytest
 from copy import deepcopy
@@ -591,7 +592,7 @@ def test_lv_is_encrypted_lvm_api(self, fake_call, factory, device_info, monkeypa
         blkid = {'TYPE': 'mapper'}
         device_info(lsblk=lsblk, blkid=blkid)
         disk = device.Device("/dev/sda")
-        disk.lv_api = factory(encrypted=True)
+        disk.lv_api = api.Volume(**{'lv_name': 'lv1', 'lv_tags': 'ceph.encrypted=1'})
         assert disk.is_encrypted is True
 
     @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False)