From 77952b93f8299675aa47a37289ef5cf7815d0147 Mon Sep 17 00:00:00 2001 From: Jan Fajerski Date: Mon, 11 Nov 2019 14:35:43 +0100 Subject: [PATCH] ceph-volume: refactor tests for refactored get_devices Signed-off-by: Jan Fajerski (cherry picked from commit 4749f4c59dc4edb1fd689e18e8d9f358f95cb945) --- src/ceph-volume/ceph_volume/tests/conftest.py | 5 + .../tests/functional/batch/tox.ini | 1 + .../ceph_volume/tests/functional/lvm/tox.ini | 1 + .../tests/functional/simple/tox.ini | 1 + .../ceph_volume/tests/util/test_disk.py | 268 ++++++------------ src/ceph-volume/tox.ini | 1 + 6 files changed, 90 insertions(+), 187 deletions(-) diff --git a/src/ceph-volume/ceph_volume/tests/conftest.py b/src/ceph-volume/ceph_volume/tests/conftest.py index b108ce2f1256f..b7f18ac0ea8eb 100644 --- a/src/ceph-volume/ceph_volume/tests/conftest.py +++ b/src/ceph-volume/ceph_volume/tests/conftest.py @@ -1,5 +1,6 @@ import os import pytest +from mock.mock import patch from ceph_volume.util import disk from ceph_volume.util.constants import ceph_disk_guids from ceph_volume.api import lvm as lvm_api @@ -263,6 +264,10 @@ def device_info_not_ceph_disk_member(monkeypatch, request): monkeypatch.setattr("ceph_volume.util.device.disk.blkid", lambda path: {'PARTLABEL': request.param[1]}) +@pytest.fixture +def patched_get_block_devs_lsblk(): + with patch('ceph_volume.util.disk.get_block_devs_lsblk') as p: + yield p @pytest.fixture def device_info(monkeypatch): diff --git a/src/ceph-volume/ceph_volume/tests/functional/batch/tox.ini b/src/ceph-volume/ceph_volume/tests/functional/batch/tox.ini index d7c6a6c5e48d3..ffe0503d7348f 100644 --- a/src/ceph-volume/ceph_volume/tests/functional/batch/tox.ini +++ b/src/ceph-volume/ceph_volume/tests/functional/batch/tox.ini @@ -3,6 +3,7 @@ envlist = {centos7,xenial}-{bluestore,filestore}-{single_type,single_type_dmcryp skipsdist = True [testenv] +deps = mock whitelist_externals = vagrant bash diff --git a/src/ceph-volume/ceph_volume/tests/functional/lvm/tox.ini b/src/ceph-volume/ceph_volume/tests/functional/lvm/tox.ini index d9cbe6b0bdcf6..07292afccdf6f 100644 --- a/src/ceph-volume/ceph_volume/tests/functional/lvm/tox.ini +++ b/src/ceph-volume/ceph_volume/tests/functional/lvm/tox.ini @@ -3,6 +3,7 @@ envlist = {centos7,xenial}-{filestore,bluestore}-{create,prepare_activate,dmcryp skipsdist = True [testenv] +deps = mock whitelist_externals = vagrant bash diff --git a/src/ceph-volume/ceph_volume/tests/functional/simple/tox.ini b/src/ceph-volume/ceph_volume/tests/functional/simple/tox.ini index 9ab6fee513fd1..0e3677d7744f8 100644 --- a/src/ceph-volume/ceph_volume/tests/functional/simple/tox.ini +++ b/src/ceph-volume/ceph_volume/tests/functional/simple/tox.ini @@ -3,6 +3,7 @@ envlist = {centos7,xenial}-{filestore,bluestore}-{activate,dmcrypt_plain,dmcrypt skipsdist = True [testenv] +deps = mock whitelist_externals = vagrant bash 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 68068a809ef98..4f776567d336c 100644 --- a/src/ceph-volume/ceph_volume/tests/util/test_disk.py +++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py @@ -1,5 +1,6 @@ import os import pytest +from mock.mock import patch from ceph_volume.util import disk @@ -101,90 +102,6 @@ class TestDeviceFamily(object): assert parsed['NAME'] in names -class TestMapDevPaths(object): - - def test_errors_return_empty_mapping(self, tmpdir): - bad_dir = os.path.join(str(tmpdir), 'nonexisting') - assert disk._map_dev_paths(bad_dir) == {} - - def test_base_name_and_abspath(self, tmpfile): - sda_path = tmpfile(name='sda', contents='') - directory = os.path.dirname(sda_path) - result = disk._map_dev_paths(directory) - assert len(result.keys()) == 1 - assert result['sda'] == sda_path - - def test_abspath_included(self, tmpfile): - sda_path = tmpfile(name='sda', contents='') - directory = os.path.dirname(sda_path) - result = disk._map_dev_paths(directory, include_abspath=True) - assert sorted(result.keys()) == sorted(['sda', sda_path]) - assert result['sda'] == sda_path - assert result[sda_path] == 'sda' - - def test_realpath_included(self, tmpfile): - sda_path = tmpfile(name='sda', contents='') - directory = os.path.dirname(sda_path) - dm_path = os.path.join(directory, 'dm-0') - os.symlink(sda_path, os.path.join(directory, 'dm-0')) - result = disk._map_dev_paths(directory, include_realpath=True) - assert sorted(result.keys()) == sorted(['sda', 'dm-0']) - assert result['sda'] == dm_path - assert result['dm-0'] == dm_path - - def test_absolute_and_realpath_included(self, tmpfile): - dm_path = tmpfile(name='dm-0', contents='') - directory = os.path.dirname(dm_path) - sda_path = os.path.join(directory, 'sda') - os.symlink(sda_path, os.path.join(directory, 'sda')) - result = disk._map_dev_paths(directory, include_realpath=True, include_abspath=True) - assert sorted(result.keys()) == sorted([dm_path, sda_path, 'sda', 'dm-0']) - assert result['sda'] == sda_path - assert result['dm-0'] == dm_path - assert result[sda_path] == sda_path - assert result[dm_path] == 'dm-0' - - -class TestGetBlockDevs(object): - - def test_loop_devices_are_missing(self, tmpfile): - path = os.path.dirname(tmpfile(name='loop0', contents='')) - result = disk.get_block_devs(sys_block_path=path) - assert result == [] - - 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'] - - -class TestGetDevDevs(object): - - def test_abspaths_are_included(self, tmpfile): - sda_path = tmpfile(name='sda', contents='') - directory = os.path.dirname(sda_path) - result = disk.get_dev_devs(directory) - assert sorted(result.keys()) == sorted(['sda', sda_path]) - assert result['sda'] == sda_path - assert result[sda_path] == 'sda' - - -class TestGetMapperDevs(object): - - def test_abspaths_and_realpaths_are_included(self, tmpfile): - dm_path = tmpfile(name='dm-0', contents='') - directory = os.path.dirname(dm_path) - sda_path = os.path.join(directory, 'sda') - os.symlink(sda_path, os.path.join(directory, 'sda')) - result = disk.get_mapper_devs(directory) - assert sorted(result.keys()) == sorted([dm_path, sda_path, 'sda', 'dm-0']) - assert result['sda'] == sda_path - assert result['dm-0'] == dm_path - assert result[sda_path] == sda_path - assert result[dm_path] == 'dm-0' - - class TestHumanReadableSize(object): def test_bytes(self): @@ -247,141 +164,118 @@ class TestSizeFromHumanReadable(object): assert result == disk.Size(tb=1.8) +class TestGetBlockDevsLsblk(object): + + @patch('ceph_volume.process.call') + def test_return_structure(self, patched_call): + lsblk_stdout = [ + '/dev/dm-0 /dev/mapper/ceph--8b2684eb--56ff--49e4--8f28--522e04cbd6ab-osd--data--9fc29fbf--3b5b--4066--be10--61042569b5a7 lvm', + '/dev/vda /dev/vda disk', + '/dev/vda1 /dev/vda1 part', + '/dev/vdb /dev/vdb disk',] + patched_call.return_value = (lsblk_stdout, '', 0) + disks = disk.get_block_devs_lsblk() + assert len(disks) == len(lsblk_stdout) + assert len(disks[0]) == 3 + + @patch('ceph_volume.process.call') + def test_empty_lsblk(self, patched_call): + patched_call.return_value = ([], '', 0) + disks = disk.get_block_devs_lsblk() + assert len(disks) == 0 + + @patch('ceph_volume.process.call') + def test_raise_on_failure(self, patched_call): + patched_call.return_value = ([], 'error', 1) + with pytest.raises(OSError): + disk.get_block_devs_lsblk() + + 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 setup_path(self, tmpdir): + path = os.path.join(str(tmpdir), 'block') + os.makedirs(path) + return path - def test_no_devices_are_found_errors(self, tmpdir): - block_path, dev_path, mapper_path = self.setup_paths(tmpdir) - os.makedirs(os.path.join(block_path, 'sda')) - result = disk.get_devices( - _sys_block_path=block_path, # has 1 device - _dev_path=str(tmpdir), # exists but no devices - _mapper_path='/does/not/exist/path') # does not exist + def test_no_devices_are_found(self, tmpdir, patched_get_block_devs_lsblk): + patched_get_block_devs_lsblk.return_value = [] + result = disk.get_devices(_sys_block_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') + def test_sda_block_is_found(self, tmpdir, patched_get_block_devs_lsblk): + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(tmpdir) 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) + result = disk.get_devices(_sys_block_path=block_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_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) + assert result[sda_path]['human_readable_size'] == '0.00 B' + assert result[sda_path]['model'] == '' + assert result[sda_path]['partitions'] == {} + + + def test_sda_size(self, tmpfile, tmpdir, patched_get_block_devs_lsblk): + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(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): + result = disk.get_devices(_sys_block_path=block_path) + assert list(result.keys()) == [sda_path] + assert result[sda_path]['human_readable_size'] == '512.00 KB' + + def test_sda_sectorsize_fallsback(self, tmpfile, tmpdir, patched_get_block_devs_lsblk): # if no sectorsize, it will use queue/hw_sector_size - block_path, dev_path, mapper_path = self.setup_paths(tmpdir) + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(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) + result = disk.get_devices(_sys_block_path=block_path) + assert list(result.keys()) == [sda_path] + assert result[sda_path]['sectorsize'] == '1024' + + def test_sda_sectorsize_from_logical_block(self, tmpfile, tmpdir, patched_get_block_devs_lsblk): + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(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) + result = disk.get_devices(_sys_block_path=block_path) + assert result[sda_path]['sectorsize'] == '99' + + def test_sda_sectorsize_does_not_fallback(self, tmpfile, tmpdir, patched_get_block_devs_lsblk): + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(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) + result = disk.get_devices(_sys_block_path=block_path) + assert result[sda_path]['sectorsize'] == '99' + + def test_is_rotational(self, tmpfile, tmpdir, patched_get_block_devs_lsblk): + sda_path = '/dev/sda' + patched_get_block_devs_lsblk.return_value = [[sda_path, sda_path, 'disk']] + block_path = self.setup_path(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' + result = disk.get_devices(_sys_block_path=block_path) + assert result[sda_path]['rotational'] == '1' class TestSizeCalculations(object): diff --git a/src/ceph-volume/tox.ini b/src/ceph-volume/tox.ini index fce465def1881..6db5556b66fa2 100644 --- a/src/ceph-volume/tox.ini +++ b/src/ceph-volume/tox.ini @@ -4,6 +4,7 @@ envlist = py27, py35, py36, flake8 [testenv] deps= pytest + mock commands=py.test -v {posargs:ceph_volume/tests} --ignore=ceph_volume/tests/functional [testenv:flake8] -- 2.39.5