From: Guillaume Abrioux <gabrioux@ibm.com>
Date: Wed, 24 Jan 2024 15:28:41 +0000 (+0100)
Subject: Merge pull request #54629 from guits/wip-63599-reef
X-Git-Tag: testing/wip-pdonnell-testing-20240430.123648-reef-debug~308
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=8903e60701fdb68a7f2ab35500ff7d8e138d7d6b;p=ceph-ci.git

Merge pull request #54629 from guits/wip-63599-reef

reef: ceph-volume: fixes fallback to stat in is_device and is_partition
---

8903e60701fdb68a7f2ab35500ff7d8e138d7d6b
diff --cc src/ceph-volume/ceph_volume/tests/util/test_disk.py
index ce1f9466fd5,94325b475ef..b287530dc17
--- a/src/ceph-volume/ceph_volume/tests/util/test_disk.py
+++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py
@@@ -1,38 -1,13 +1,43 @@@
  import os
  import pytest
  from ceph_volume.util import disk
 -from mock.mock import patch
 +from mock.mock import patch, MagicMock
 +
 +
 +class TestFunctions:
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=False))
 +    def test_is_device_path_does_not_exist(self):
 +        assert not disk.is_device('/dev/foo')
 +
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
 +    def test_is_device_dev_doesnt_startswith_dev(self):
 +        assert not disk.is_device('/foo')
 +
 +    @patch('ceph_volume.util.disk.allow_loop_devices', MagicMock(return_value=False))
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
 +    def test_is_device_loop_not_allowed(self):
 +        assert not disk.is_device('/dev/loop123')
 +
 +    @patch('ceph_volume.util.disk.lsblk', MagicMock(return_value={'NAME': 'foo', 'TYPE': 'disk'}))
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
 +    def test_is_device_type_disk(self):
 +        assert disk.is_device('/dev/foo')
 +
 +    @patch('ceph_volume.util.disk.lsblk', MagicMock(return_value={'NAME': 'foo', 'TYPE': 'mpath'}))
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
 +    def test_is_device_type_mpath(self):
 +        assert disk.is_device('/dev/foo')
 +
 +    @patch('ceph_volume.util.disk.lsblk', MagicMock(return_value={'NAME': 'foo1', 'TYPE': 'part'}))
 +    @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
 +    def test_is_device_type_part(self):
 +        assert not disk.is_device('/dev/foo1')
  
+     @patch('ceph_volume.util.disk.os.path.exists', MagicMock(return_value=True))
+     @patch('ceph_volume.util.disk.get_partitions', MagicMock(return_value={"sda1": "sda"}))
+     def test_is_partition(self):
+         assert disk.is_partition('sda1')
+ 
  
  class TestLsblkParser(object):
  
diff --cc src/ceph-volume/ceph_volume/util/disk.py
index 05db875fee4,c1d1360765f..7148612f329
--- a/src/ceph-volume/ceph_volume/util/disk.py
+++ b/src/ceph-volume/ceph_volume/util/disk.py
@@@ -359,15 -359,11 +359,15 @@@ def is_device(dev)
          if not allow_loop_devices():
              return False
  
 +    TYPE = lsblk(dev).get('TYPE')
 +    if TYPE:
 +        return TYPE in ['disk', 'mpath']
 +
      # fallback to stat
-     return _stat_is_device(os.lstat(dev).st_mode)
+     return _stat_is_device(os.lstat(dev).st_mode) and not is_partition(dev)
  
  
- def is_partition(dev):
+ def is_partition(dev: str) -> bool:
      """
      Boolean to determine if a given device is a partition, like /dev/sda1
      """