ceph-volume: refactor device path handling for LVM lookups

This consolidates the conditional checks for device paths to
reduce redundancy and improve readability and adds logic to
handle both '/dev/mapper' and '/dev/dm-' paths uniformly by
introducing a utility function `get_lvm_mapper_path_from_dm()`.

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

Author: guits

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

@@ -1,6 +1,6 @@
 import pytest
 from ceph_volume.util import disk
-from mock.mock import patch, MagicMock
+from mock.mock import patch, Mock, MagicMock, mock_open
 
 
 class TestFunctions:
@@ -38,6 +38,11 @@ def test_is_partition(self):
         assert disk.is_partition('sda1')
 
 
+    @patch('os.path.exists', Mock(return_value=True))
+    def test_get_lvm_mapper_path_from_dm(self):
+        with patch('builtins.open', mock_open(read_data='test--foo--vg-test--foo--lv')):
+            assert disk.get_lvm_mapper_path_from_dm('/dev/dm-123') == '/dev/mapper/test--foo--vg-test--foo--lv'
+
 class TestLsblkParser(object):
 
     def test_parses_whitespace_values(self):

src/ceph-volume/ceph_volume/util/arg_validators.py

@@ -92,6 +92,9 @@ def __call__(self, dev_path):
         super().get_device(dev_path)
         return self._format_device(self._is_valid_device())
 
+    def _format_device(self, device: Device) -> str:
+        return device.path
+
     def _is_valid_device(self, raise_sys_exit=True):
         out, err, rc = process.call([
 	    'ceph-bluestore-tool', 'show-label',

src/ceph-volume/ceph_volume/util/device.py

@@ -211,12 +211,21 @@ def _parse(self):
                         lv = _lv
                         break
         else:
+            filters = {}
             if self.path[0] == '/':
-                lv = lvm.get_single_lv(filters={'lv_path': self.path})
+                lv_mapper_path: str = self.path
+                field: str = 'lv_path'
+
+                if self.path.startswith('/dev/mapper') or self.path.startswith('/dev/dm-'):
+                    path = os.path.realpath(self.path) if self.path.startswith('/dev/mapper') else self.path
+                    lv_mapper_path = disk.get_lvm_mapper_path_from_dm(path)
+                    field = 'lv_dm_path'
+
+                filters = {field: lv_mapper_path}
             else:
                 vgname, lvname = self.path.split('/')
-                lv = lvm.get_single_lv(filters={'lv_name': lvname,
-                                                'vg_name': vgname})
+                filters = {'lv_name': lvname, 'vg_name': vgname}
+            lv = lvm.get_single_lv(filters=filters)
 
         if lv:
             self.lv_api = lv

src/ceph-volume/ceph_volume/util/disk.py

@@ -1104,3 +1104,28 @@ def get_parent_device_from_mapper(mapper: str, abspath: bool = True) -> str:
         except KeyError:
             pass
     return result
+
+
+def get_lvm_mapper_path_from_dm(path: str, sys_block: str = '/sys/block') -> str:
+    """_summary_
+    Retrieve the logical volume path for a given device.
+
+    This function takes the path of a device and returns the corresponding
+    logical volume path by reading the 'dm/name' file within the sysfs
+    directory.
+
+    Args:
+        path (str): The device path for which to retrieve the logical volume path.
+        sys_block (str, optional): The base sysfs block directory. Defaults to '/sys/block'.
+
+    Returns:
+        str: The device mapper path in the form of '/dev/dm-X'.
+    """
+    result: str = ''
+    dev: str = os.path.basename(path)
+    sys_block_path: str = os.path.join(sys_block, dev, 'dm/name')
+    if os.path.exists(sys_block_path):
+        with open(sys_block_path, 'r') as f:
+            content: str = f.read()
+            result = f'/dev/mapper/{content}'
+    return result