From 8deb088f101894763d27a6c31a3e8a2e96fc8320 Mon Sep 17 00:00:00 2001 From: Alfredo Deza Date: Thu, 3 May 2018 14:33:08 -0400 Subject: [PATCH] ceph-volume tests verify get_devices captures device metadata Signed-off-by: Alfredo Deza --- .../ceph_volume/tests/util/test_disk.py | 144 +++++++++++++++++- 1 file changed, 143 insertions(+), 1 deletion(-) 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 3339fe4483eb6..b5f07021b3748 100644 --- 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' -- 2.39.5