ceph-volume: Refactor is_ceph_device to simplify error handling

Replace the try-except block with a direct get() call on lv.tags to
check for ceph.osd_id. This avoids catching AttributeError unnecessarily
and makes the logic more concise. Additionally, the warning message is
now logged when osd_id is None or 'null', ensuring consistency in
error handling.
Also, this fixes the unit test `api/test_lvm.py::TestVolume::test_is_not_ceph_device()`
as `api.lvm.is_ceph_device()` expects a `Volume` object.

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

Author: guits

src/ceph-volume/ceph_volume/api/lvm.py

@@ -322,16 +322,13 @@ def dmsetup_splitname(dev: str) -> Dict[str, Any]:
 
 
 def is_ceph_device(lv: "Volume") -> bool:
-    try:
-        lv.tags['ceph.osd_id']
-    except (KeyError, AttributeError):
-        logger.warning('device is not part of ceph: %s', lv)
-        return False
+    osd_id = lv.tags.get('ceph.osd_id', 'null')
 
-    if lv.tags['ceph.osd_id'] == 'null':
-        return False
-    else:
-        return True
+    if osd_id == 'null':
+         logger.warning('device is not part of ceph: %s', lv)
+         return False
+
+    return True
 
 class Lvm:
     def __init__(self, name_key: str, tags_key: str, **kw: Any) -> None:
@@ -907,6 +904,7 @@ def __init__(self, **kw: Any) -> None:
         self.lv_uuid: str = ''
         self.vg_name: str = ''
         self.lv_size: str = ''
+        self.tags: Dict[str, Any] = {}
         self.lv_tags: Dict[str, Any] = {}
         super().__init__('lv_name', 'lv_tags', **kw)
         self.lv_api = kw

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

@@ -31,20 +31,20 @@ def test_multiple_csv_expands_in_dict(self):
         assert result['ceph.fsid'] == '0000'
 
 
-class TestVolume(object):
-
+class TestVolume:
     def test_is_ceph_device(self):
         lv_tags = "ceph.type=data,ceph.osd_id=0"
         osd = api.Volume(lv_name='osd/volume', lv_tags=lv_tags)
         assert api.is_ceph_device(osd)
 
-    @pytest.mark.parametrize('dev',[
-        '/dev/sdb',
-        api.VolumeGroup(vg_name='foo'),
-        api.Volume(lv_name='vg/no_osd', lv_tags='', lv_path='lv/path'),
-        api.Volume(lv_name='vg/no_osd', lv_tags='ceph.osd_id=null', lv_path='lv/path'),
-        None,
-    ])
+    @pytest.mark.parametrize('dev',
+                             [api.VolumeGroup(vg_name='foo'),
+                              api.Volume(lv_name='vg/no_osd',
+                                         lv_tags='',
+                                         lv_path='lv/path'),
+                              api.Volume(lv_name='vg/no_osd',
+                                         lv_tags='ceph.osd_id=null',
+                                         lv_path='lv/path')])
     def test_is_not_ceph_device(self, dev):
         assert not api.is_ceph_device(dev)
 

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

@@ -50,7 +50,7 @@ def test_lvm_size_rounds_down(self, fake_call, device_info):
 
     def test_is_lv(self, fake_call, device_info, monkeypatch):
         monkeypatch.setattr('ceph_volume.util.device.Device.is_lv', lambda: True)
-        data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"}
+        data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv", "tags": {}}
         lsblk = {"TYPE": "lvm", "NAME": "vg-lv"}
         device_info(lv=data,lsblk=lsblk)
         disk = device.Device("vg/lv")
@@ -322,7 +322,7 @@ def test_reject_lv_symlink_to_device(self,
                                          fake_call):
         m_os_path_islink.return_value = True
         m_os_path_realpath.return_value = '/dev/mapper/vg-lv'
-        lv = {"lv_path": "/dev/vg/lv", "vg_name": "vg", "name": "lv"}
+        lv = {"lv_path": "/dev/vg/lv", "vg_name": "vg", "name": "lv", "tags": {}}
         lsblk = {"TYPE": "lvm", "NAME": "vg-lv"}
         device_info(lv=lv,lsblk=lsblk)
         disk = device.Device("/dev/vg/lv")