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 <gabrioux@ibm.com>

diff --git a/src/ceph-volume/ceph_volume/api/lvm.py b/src/ceph-volume/ceph_volume/api/lvm.py
index d689c314ba2be3a5edf8159d5b2d62914721f267..417742a401365f2443f7c2779132cf8238d34d64 100644 (file)
--- a/src/ceph-volume/ceph_volume/api/lvm.py
+++ b/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 @@ class Volume(Lvm):
         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

diff --git a/src/ceph-volume/ceph_volume/tests/api/test_lvm.py b/src/ceph-volume/ceph_volume/tests/api/test_lvm.py
index 6b62eccbbb6d55b0319f4f0e94cb46f43b3689d6..35f7dea4f10d3c110413eee581534da28a8e63b9 100644 (file)
--- a/src/ceph-volume/ceph_volume/tests/api/test_lvm.py
+++ b/src/ceph-volume/ceph_volume/tests/api/test_lvm.py
@@ -31,20 +31,20 @@ class TestParseTags(object):
         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)
 

diff --git a/src/ceph-volume/ceph_volume/tests/util/test_device.py b/src/ceph-volume/ceph_volume/tests/util/test_device.py
index cb145d10f2c0ccb24aecd949cbb2e2f80d6e32fa..965ca5bdf54cc5d6820aa2be5a6424068984792e 100644 (file)
--- a/src/ceph-volume/ceph_volume/tests/util/test_device.py
+++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py
@@ -50,7 +50,7 @@ class TestDevice(object):
 
     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 @@ class TestDevice(object):
                                          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")