ceph-volume tests verify get_devices captures device metadata

Signed-off-by: Alfredo Deza <adeza@redhat.com>

diff --git a/src/ceph-volume/ceph_volume/tests/util/test_disk.py b/src/ceph-volume/ceph_volume/tests/util/test_disk.py
index 3339fe4483eb6ef0c962f1b1be8a807e8201422c..b5f07021b3748fb19580910c75c41829edc44bb6 100644 (file)
--- a/src/ceph-volume/ceph_volume/tests/util/test_disk.py
+++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py
@@ -52,7 +52,7 @@ class TestMapDevPaths(object):
         sda_path = tmpfile(name='sda', contents='')
         directory = os.path.dirname(sda_path)
         result = disk._map_dev_paths(directory)
-        assert result.keys() == ['sda']
+        assert len(result.keys()) == 1
         assert result['sda'] == sda_path
 
     def test_abspath_included(self, tmpfile):
@@ -96,6 +96,7 @@ class TestGetBlockDevs(object):
     def test_loop_devices_are_included(self, tmpfile):
         path = os.path.dirname(tmpfile(name='loop0', contents=''))
         result = disk.get_block_devs(sys_block_path=path, skip_loop=False)
+        assert len(result) == 1
         assert result == ['loop0']
 
 
@@ -146,3 +147,144 @@ class TestHumanReadableSize(object):
     def test_terabytes(self):
         result = disk.human_readable_size(81.2*1024*1024*1024*1024)
         assert result == '81.20 TB'
+
+
+class TestGetDevices(object):
+
+    def setup_paths(self, tmpdir):
+        paths = []
+        for directory in ['block', 'dev', 'mapper']:
+            path = os.path.join(str(tmpdir), directory)
+            paths.append(path)
+            os.makedirs(path)
+        return paths
+
+    def test_no_devices_are_found(self, tmpdir):
+        result = disk.get_devices(
+            _sys_block_path=str(tmpdir),
+            _dev_path=str(tmpdir),
+            _mapper_path=str(tmpdir))
+        assert result == {}
+
+    def test_sda_block_is_found(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(os.path.join(block_path, 'sda'))
+        os.makedirs(dev_sda_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert len(result.keys()) == 1
+        assert result[dev_sda_path]['human_readable_size'] == '0.00 B'
+        assert result[dev_sda_path]['model'] == ''
+        assert result[dev_sda_path]['partitions'] == {}
+
+    def test_sda_is_removable_gets_skipped(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        block_sda_path = os.path.join(block_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(dev_sda_path)
+
+        tmpfile('removable', contents='1', directory=block_sda_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert result == {}
+
+    def test_dm_device_is_not_used(self, monkeypatch, tmpdir):
+        # the link to the mapper is used instead
+        monkeypatch.setattr(disk.lvm, 'is_lv', lambda: True)
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        dev_dm_path = os.path.join(dev_path, 'dm-0')
+        ceph_data_path = os.path.join(mapper_path, 'ceph-data')
+        os.symlink(dev_dm_path, ceph_data_path)
+        block_dm_path = os.path.join(block_path, 'dm-0')
+        os.makedirs(block_dm_path)
+
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        result = list(result.keys())
+        assert len(result) == 1
+        assert result == [ceph_data_path]
+
+    def test_sda_size(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        block_sda_path = os.path.join(block_path, 'sda')
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(dev_sda_path)
+        tmpfile('size', '1024', directory=block_sda_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert list(result.keys()) == [dev_sda_path]
+        assert result[dev_sda_path]['human_readable_size'] == '512.00 KB'
+
+    def test_sda_sectorsize_fallsback(self, tmpfile, tmpdir):
+        # if no sectorsize, it will use queue/hw_sector_size
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        block_sda_path = os.path.join(block_path, 'sda')
+        sda_queue_path = os.path.join(block_sda_path, 'queue')
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(sda_queue_path)
+        os.makedirs(dev_sda_path)
+        tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert list(result.keys()) == [dev_sda_path]
+        assert result[dev_sda_path]['sectorsize'] == '1024'
+
+    def test_sda_sectorsize_from_logical_block(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        block_sda_path = os.path.join(block_path, 'sda')
+        sda_queue_path = os.path.join(block_sda_path, 'queue')
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(sda_queue_path)
+        os.makedirs(dev_sda_path)
+        tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert result[dev_sda_path]['sectorsize'] == '99'
+
+    def test_sda_sectorsize_does_not_fallback(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        block_sda_path = os.path.join(block_path, 'sda')
+        sda_queue_path = os.path.join(block_sda_path, 'queue')
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(sda_queue_path)
+        os.makedirs(dev_sda_path)
+        tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
+        tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert result[dev_sda_path]['sectorsize'] == '99'
+
+    def test_is_rotational(self, tmpfile, tmpdir):
+        block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
+        block_sda_path = os.path.join(block_path, 'sda')
+        sda_queue_path = os.path.join(block_sda_path, 'queue')
+        dev_sda_path = os.path.join(dev_path, 'sda')
+        os.makedirs(block_sda_path)
+        os.makedirs(sda_queue_path)
+        os.makedirs(dev_sda_path)
+        tmpfile('rotational', contents='1', directory=sda_queue_path)
+        result = disk.get_devices(
+            _sys_block_path=block_path,
+            _dev_path=dev_path,
+            _mapper_path=mapper_path)
+        assert result[dev_sda_path]['rotational'] == '1'