ceph-volume: add type annotations to devices.lvm.common

This commit adds the Python type annotations to `devices.lvm.common`.

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

Author: guits

src/ceph-volume/ceph_volume/devices/lvm/common.py

@@ -3,8 +3,9 @@
 from ceph_volume import terminal
 from ceph_volume.devices.lvm.zap import Zap
 import argparse
+from typing import Any, Dict, Optional
 
-def rollback_osd(args, osd_id=None):
+def rollback_osd(osd_id: Optional[str] = None) -> None:
     """
     When the process of creating or preparing fails, the OSD needs to be
     destroyed so that the ID can be reused.  This prevents from leaving the ID
@@ -35,7 +36,7 @@ def rollback_osd(args, osd_id=None):
     Zap(['--destroy', '--osd-id', osd_id]).main()
 
 
-common_args = {
+common_args: Dict[str, Any] = {
     '--objectstore': {
         'dest': 'objectstore',
         'help': 'The OSD objectstore.',
@@ -95,7 +96,7 @@ def rollback_osd(args, osd_id=None):
     },
 }
 
-bluestore_args = {
+bluestore_args: Dict[str, Any] = {
     '--bluestore': {
         'action': 'store_true',
         'help': 'Use the bluestore objectstore. (DEPRECATED: use --objectstore instead)',
@@ -139,16 +140,16 @@ def rollback_osd(args, osd_id=None):
 }
 
 
-def get_default_args():
+def get_default_args() -> Dict[str, Any]:
     defaults = {}
-    def format_name(name):
+    def format_name(name: str) -> str:
         return name.strip('-').replace('-', '_').replace('.', '_')
     for argset in (common_args, bluestore_args):
         defaults.update({format_name(name): val.get('default', None) for name, val in argset.items()})
     return defaults
 
 
-def common_parser(prog, description):
+def common_parser(prog: str, description: str) -> argparse.ArgumentParser:
     """
     Both prepare and create share the same parser, those are defined here to
     avoid duplication

src/ceph-volume/ceph_volume/devices/lvm/create.py

@@ -38,7 +38,7 @@ def create(self) -> None:
                 except Exception:
                     logger.exception('lvm activate was unable to complete, while creating the OSD')
                     logger.info('will rollback OSD ID creation')
-                    rollback_osd(self.args, osd_id)
+                    rollback_osd(osd_id)
                     raise
                 terminal.success("ceph-volume lvm create successful for: %s" % self.args.data)
 

src/ceph-volume/ceph_volume/objectstore/lvmbluestore.py

@@ -131,7 +131,7 @@ def safe_prepare(self,
         except Exception:
             logger.exception('lvm prepare was unable to complete')
             logger.info('will rollback OSD ID creation')
-            rollback_osd(self.args, self.osd_id)
+            rollback_osd(self.osd_id)
             raise
         terminal.success("ceph-volume lvm prepare successful for: %s" %
                          self.args.data)

src/ceph-volume/ceph_volume/objectstore/rawbluestore.py

@@ -75,7 +75,7 @@ def safe_prepare(self,
         except Exception:
             logger.exception('raw prepare was unable to complete')
             logger.info('will rollback OSD ID creation')
-            rollback_osd(self.args, self.osd_id)
+            rollback_osd(self.osd_id)
             raise
         dmcrypt_log = 'dmcrypt' if hasattr(args, 'dmcrypt') else 'clear'
         terminal.success("ceph-volume raw {} prepare "

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

@@ -188,7 +188,7 @@ def test_safe_prepare_raises_exception(self, m_get_single_lv, m_rollback_osd, fa
         m_rollback_osd.return_value = MagicMock()
         with pytest.raises(Exception):
             self.lvm_bs.safe_prepare()
-        assert m_rollback_osd.mock_calls == [call(self.lvm_bs.args, '111')]
+        assert m_rollback_osd.mock_calls == [call('111')]
 
     @patch('ceph_volume.objectstore.baseobjectstore.BaseObjectStore.get_ptuuid', Mock(return_value='c6798f59-01'))
     @patch('ceph_volume.api.lvm.Volume.set_tags', MagicMock())

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

@@ -60,7 +60,7 @@ def test_safe_prepare_raises_exception(self,
         self.raw_bs.osd_id = self.raw_bs.args.osd_id
         with pytest.raises(Exception):
             self.raw_bs.safe_prepare()
-        assert m_rollback_osd.mock_calls == [call(self.raw_bs.args, '1')]
+        assert m_rollback_osd.mock_calls == [call('1')]
 
     @patch('ceph_volume.objectstore.rawbluestore.RawBlueStore.prepare', MagicMock())
     def test_safe_prepare(self,