diff options
Diffstat (limited to 'src/ceph-volume/ceph_volume/tests/util')
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py | 89 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_device.py | 577 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_disk.py | 540 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_encryption.py | 53 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_prepare.py | 422 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_system.py | 279 | ||||
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_util.py | 116 |
7 files changed, 2076 insertions, 0 deletions
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py new file mode 100644 index 00000000..d4565ef4 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py @@ -0,0 +1,89 @@ +import argparse +import pytest +import os +from ceph_volume import exceptions +from ceph_volume.util import arg_validators + + +class TestOSDPath(object): + + def setup(self): + self.validator = arg_validators.OSDPath() + + def test_is_not_root(self, monkeypatch): + monkeypatch.setattr(os, 'getuid', lambda: 100) + with pytest.raises(exceptions.SuperUserError): + self.validator('') + + def test_path_is_not_a_directory(self, is_root, tmpfile, monkeypatch): + monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False) + validator = arg_validators.OSDPath() + with pytest.raises(argparse.ArgumentError): + validator(tmpfile()) + + def test_files_are_missing(self, is_root, tmpdir, monkeypatch): + tmppath = str(tmpdir) + monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False) + validator = arg_validators.OSDPath() + with pytest.raises(argparse.ArgumentError) as error: + validator(tmppath) + assert 'Required file (ceph_fsid) was not found in OSD' in str(error.value) + + +class TestExcludeGroupOptions(object): + + def setup(self): + self.parser = argparse.ArgumentParser() + + def test_flags_in_one_group(self): + argv = ['<prog>', '--filestore', '--bar'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + result = arg_validators.exclude_group_options( + self.parser, + ['filestore', 'bluestore'], + argv=argv + ) + assert result is None + + def test_flags_in_no_group(self): + argv = ['<prog>', '--foo', '--bar'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + result = arg_validators.exclude_group_options( + self.parser, + ['filestore', 'bluestore'], + argv=argv + ) + assert result is None + + def test_flags_conflict(self, capsys): + argv = ['<prog>', '--filestore', '--bluestore'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + + arg_validators.exclude_group_options( + self.parser, ['filestore', 'bluestore'], argv=argv + ) + stdout, stderr = capsys.readouterr() + assert 'Cannot use --filestore (filestore) with --bluestore (bluestore)' in stderr + + +class TestValidDevice(object): + + def setup(self): + self.validator = arg_validators.ValidDevice() + + def test_path_is_valid(self, fake_call): + result = self.validator('/') + assert result.abspath == '/' + + def test_path_is_invalid(self, fake_call): + with pytest.raises(argparse.ArgumentError): + self.validator('/device/does/not/exist') diff --git a/src/ceph-volume/ceph_volume/tests/util/test_device.py b/src/ceph-volume/ceph_volume/tests/util/test_device.py new file mode 100644 index 00000000..7dd8982f --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py @@ -0,0 +1,577 @@ +import pytest +from copy import deepcopy +from ceph_volume.util import device +from ceph_volume.api import lvm as api + + +class TestDevice(object): + + def test_sys_api(self, monkeypatch, device_info): + volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg', + lv_tags={}, lv_path='/dev/VolGroup/lv') + volumes = [] + volumes.append(volume) + monkeypatch.setattr(api, 'get_lvs', lambda **kwargs: + deepcopy(volumes)) + + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.sys_api + assert "foo" in disk.sys_api + + def test_lvm_size(self, monkeypatch, device_info): + volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg', + lv_tags={}, lv_path='/dev/VolGroup/lv') + volumes = [] + volumes.append(volume) + monkeypatch.setattr(api, 'get_lvs', lambda **kwargs: + deepcopy(volumes)) + + # 5GB in size + data = {"/dev/sda": {"size": "5368709120"}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.lvm_size.gb == 4 + + def test_lvm_size_rounds_down(self, device_info): + # 5.5GB in size + data = {"/dev/sda": {"size": "5905580032"}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.lvm_size.gb == 4 + + def test_is_lv(self, device_info): + data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"} + lsblk = {"TYPE": "lvm"} + device_info(lv=data,lsblk=lsblk) + disk = device.Device("vg/lv") + assert disk.is_lv + + def test_vgs_is_empty(self, device_info, monkeypatch): + BarPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", + pv_tags={}) + pvolumes = [] + pvolumes.append(BarPVolume) + lsblk = {"TYPE": "disk"} + device_info(lsblk=lsblk) + monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: {}) + + disk = device.Device("/dev/nvme0n1") + assert disk.vgs == [] + + def test_vgs_is_not_empty(self, device_info, monkeypatch): + vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6, + vg_extent_size=1073741824) + monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg]) + lsblk = {"TYPE": "disk"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/nvme0n1") + assert len(disk.vgs) == 1 + + def test_device_is_device(self, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "device"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_device is True + + def test_device_is_rotational(self, device_info): + data = {"/dev/sda": {"rotational": "1"}} + lsblk = {"TYPE": "device"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + def test_device_is_not_rotational(self, device_info): + data = {"/dev/sda": {"rotational": "0"}} + lsblk = {"TYPE": "device"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.rotational + + def test_device_is_rotational_lsblk(self, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "device", "ROTA": "1"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + def test_device_is_not_rotational_lsblk(self, device_info): + data = {"/dev/sda": {"rotational": "0"}} + lsblk = {"TYPE": "device", "ROTA": "0"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.rotational + + def test_device_is_rotational_defaults_true(self, device_info): + # rotational will default true if no info from sys_api or lsblk is found + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "device", "foo": "bar"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + def test_disk_is_device(self, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "disk"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_device is True + + def test_is_partition(self, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert disk.is_partition + + def test_is_not_acceptable_device(self, device_info): + data = {"/dev/dm-0": {"foo": "bar"}} + lsblk = {"TYPE": "mpath"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/dm-0") + assert not disk.is_device + + def test_is_not_lvm_memeber(self, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert not disk.is_lvm_member + + def test_is_lvm_memeber(self, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert not disk.is_lvm_member + + def test_is_mapper_device(self, device_info): + lsblk = {"TYPE": "lvm"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/mapper/foo") + assert disk.is_mapper + + def test_dm_is_mapper_device(self, device_info): + lsblk = {"TYPE": "lvm"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/dm-4") + assert disk.is_mapper + + def test_is_not_mapper_device(self, device_info): + lsblk = {"TYPE": "disk"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.is_mapper + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "disable_kernel_queries") + def test_is_ceph_disk_lsblk(self, monkeypatch, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "disable_kernel_queries") + def test_is_ceph_disk_blkid(self, monkeypatch, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "disable_kernel_queries") + def test_is_ceph_disk_member_not_available_lsblk(self, monkeypatch, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member + assert not disk.available + assert "Used by ceph-disk" in disk.rejected_reasons + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "disable_kernel_queries") + def test_is_ceph_disk_member_not_available_blkid(self, monkeypatch, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member + assert not disk.available + assert "Used by ceph-disk" in disk.rejected_reasons + + def test_reject_removable_device(self, device_info): + data = {"/dev/sdb": {"removable": 1}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sdb") + assert not disk.available + + def test_accept_non_removable_device(self, device_info): + data = {"/dev/sdb": {"removable": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sdb") + assert disk.available + + def test_reject_not_acceptable_device(self, device_info): + data = {"/dev/dm-0": {"foo": "bar"}} + lsblk = {"TYPE": "mpath"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/dm-0") + assert not disk.available + + def test_reject_readonly_device(self, device_info): + data = {"/dev/cdrom": {"ro": 1}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/cdrom") + assert not disk.available + + def test_reject_smaller_than_5gb(self, device_info): + data = {"/dev/sda": {"size": 5368709119}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.available, 'too small device is available' + + def test_accept_non_readonly_device(self, device_info): + data = {"/dev/sda": {"ro": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.available + + def test_reject_bluestore_device(self, monkeypatch, patch_bluestore_label, device_info): + patch_bluestore_label.return_value = True + lsblk = {"TYPE": "disk"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.available + assert "Has BlueStore device label" in disk.rejected_reasons + + @pytest.mark.usefixtures("device_info_not_ceph_disk_member", + "disable_kernel_queries") + def test_is_not_ceph_disk_member_lsblk(self, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member is False + + def test_existing_vg_available(self, monkeypatch, device_info): + vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=1536, + vg_extent_size=4194304) + monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg]) + lsblk = {"TYPE": "disk"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/nvme0n1") + assert disk.available_lvm + assert not disk.available + assert not disk.available_raw + + def test_existing_vg_too_small(self, monkeypatch, device_info): + vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=4, + vg_extent_size=1073741824) + monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg]) + lsblk = {"TYPE": "disk"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/nvme0n1") + assert not disk.available_lvm + assert not disk.available + assert not disk.available_raw + + def test_multiple_existing_vgs(self, monkeypatch, device_info): + vg1 = api.VolumeGroup(vg_name='foo/bar', vg_free_count=1000, + vg_extent_size=4194304) + vg2 = api.VolumeGroup(vg_name='foo/bar', vg_free_count=536, + vg_extent_size=4194304) + monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg1, vg2]) + lsblk = {"TYPE": "disk"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/nvme0n1") + assert disk.available_lvm + assert not disk.available + assert not disk.available_raw + + @pytest.mark.parametrize("ceph_type", ["data", "block"]) + def test_used_by_ceph(self, device_info, + monkeypatch, ceph_type): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "part"} + FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", + lv_uuid="0000", pv_tags={}, vg_name="vg") + pvolumes = [] + pvolumes.append(FooPVolume) + lv_data = {"lv_name": "lv", "lv_path": "vg/lv", "vg_name": "vg", + "lv_uuid": "0000", "lv_tags": + "ceph.osd_id=0,ceph.type="+ceph_type} + volumes = [] + lv = api.Volume(**lv_data) + volumes.append(lv) + monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes) + monkeypatch.setattr(api, 'get_lvs', lambda **kwargs: + deepcopy(volumes)) + + device_info(devices=data, lsblk=lsblk, lv=lv_data) + vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6, + vg_extent_size=1073741824) + monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg]) + disk = device.Device("/dev/sda") + assert disk.used_by_ceph + + def test_not_used_by_ceph(self, device_info, monkeypatch): + FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg") + pvolumes = [] + pvolumes.append(FooPVolume) + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "part"} + lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": "journal"}} + monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes) + + device_info(devices=data, lsblk=lsblk, lv=lv_data) + disk = device.Device("/dev/sda") + assert not disk.used_by_ceph + + def test_get_device_id(self, device_info): + udev = {k:k for k in ['ID_VENDOR', 'ID_MODEL', 'ID_SCSI_SERIAL']} + lsblk = {"TYPE": "disk"} + device_info(udevadm=udev,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk._get_device_id() == 'ID_VENDOR_ID_MODEL_ID_SCSI_SERIAL' + + + +class TestDeviceEncryption(object): + + def test_partition_is_not_encrypted_lsblk(self, device_info): + lsblk = {'TYPE': 'part', 'FSTYPE': 'xfs'} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is False + + def test_partition_is_encrypted_lsblk(self, device_info): + lsblk = {'TYPE': 'part', 'FSTYPE': 'crypto_LUKS'} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is True + + def test_partition_is_not_encrypted_blkid(self, device_info): + lsblk = {'TYPE': 'part'} + blkid = {'TYPE': 'ceph data'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is False + + def test_partition_is_encrypted_blkid(self, device_info): + lsblk = {'TYPE': 'part'} + blkid = {'TYPE': 'crypto_LUKS'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is True + + def test_mapper_is_encrypted_luks1(self, device_info, monkeypatch): + status = {'type': 'LUKS1'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + def test_mapper_is_encrypted_luks2(self, device_info, monkeypatch): + status = {'type': 'LUKS2'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + def test_mapper_is_encrypted_plain(self, device_info, monkeypatch): + status = {'type': 'PLAIN'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + def test_mapper_is_not_encrypted_plain(self, device_info, monkeypatch): + monkeypatch.setattr(device, 'encryption_status', lambda x: {}) + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is False + + def test_lv_is_encrypted_blkid(self, device_info): + lsblk = {'TYPE': 'lvm'} + blkid = {'TYPE': 'crypto_LUKS'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = {} + assert disk.is_encrypted is True + + def test_lv_is_not_encrypted_blkid(self, factory, device_info): + lsblk = {'TYPE': 'lvm'} + blkid = {'TYPE': 'xfs'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = factory(encrypted=None) + assert disk.is_encrypted is False + + def test_lv_is_encrypted_lsblk(self, device_info): + lsblk = {'FSTYPE': 'crypto_LUKS', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = {} + assert disk.is_encrypted is True + + def test_lv_is_not_encrypted_lsblk(self, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = factory(encrypted=None) + assert disk.is_encrypted is False + + def test_lv_is_encrypted_lvm_api(self, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = factory(encrypted=True) + assert disk.is_encrypted is True + + def test_lv_is_not_encrypted_lvm_api(self, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = factory(encrypted=False) + assert disk.is_encrypted is False + + +class TestDeviceOrdering(object): + + def setup(self): + self.data = { + "/dev/sda": {"removable": 0}, + "/dev/sdb": {"removable": 1}, # invalid + "/dev/sdc": {"removable": 0}, + "/dev/sdd": {"removable": 1}, # invalid + } + + def test_valid_before_invalid(self, device_info): + lsblk = {"TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk) + sda = device.Device("/dev/sda") + sdb = device.Device("/dev/sdb") + + assert sda < sdb + assert sdb > sda + + def test_valid_alphabetical_ordering(self, device_info): + lsblk = {"TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk) + sda = device.Device("/dev/sda") + sdc = device.Device("/dev/sdc") + + assert sda < sdc + assert sdc > sda + + def test_invalid_alphabetical_ordering(self, device_info): + lsblk = {"TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk) + sdb = device.Device("/dev/sdb") + sdd = device.Device("/dev/sdd") + + assert sdb < sdd + assert sdd > sdb + + +class TestCephDiskDevice(object): + + def test_partlabel_lsblk(self, device_info): + lsblk = {"TYPE": "disk", "PARTLABEL": ""} + device_info(lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.partlabel == '' + + def test_partlabel_blkid(self, device_info): + blkid = {"TYPE": "disk", "PARTLABEL": "ceph data"} + device_info(blkid=blkid) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.partlabel == 'ceph data' + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "disable_kernel_queries") + def test_is_member_blkid(self, monkeypatch, patch_bluestore_label): + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.is_member is True + + def test_reject_removable_device(self, device_info): + data = {"/dev/sdb": {"removable": 1}} + device_info(devices=data) + disk = device.Device("/dev/sdb") + assert not disk.available + + def test_accept_non_removable_device(self, device_info): + data = {"/dev/sdb": {"removable": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sdb") + assert disk.available + + def test_reject_readonly_device(self, device_info): + data = {"/dev/cdrom": {"ro": 1}} + device_info(devices=data) + disk = device.Device("/dev/cdrom") + assert not disk.available + + def test_reject_smaller_than_5gb(self, device_info): + data = {"/dev/sda": {"size": 5368709119}} + device_info(devices=data) + disk = device.Device("/dev/sda") + assert not disk.available, 'too small device is available' + + def test_accept_non_readonly_device(self, device_info): + data = {"/dev/sda": {"ro": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.available + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "disable_kernel_queries") + def test_is_member_lsblk(self, patch_bluestore_label, device_info): + lsblk = {"TYPE": "disk", "PARTLABEL": "ceph"} + device_info(lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.is_member is True + + def test_unknown_type(self, device_info): + lsblk = {"TYPE": "disk", "PARTLABEL": "gluster"} + device_info(lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.type == 'unknown' + + ceph_types = ['data', 'wal', 'db', 'lockbox', 'journal', 'block'] + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "disable_kernel_queries") + def test_type_blkid(self, monkeypatch, device_info, ceph_partlabel): + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.type in self.ceph_types + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "lsblk_ceph_disk_member", + "disable_kernel_queries") + def test_type_lsblk(self, device_info, ceph_partlabel): + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.type in self.ceph_types diff --git a/src/ceph-volume/ceph_volume/tests/util/test_disk.py b/src/ceph-volume/ceph_volume/tests/util/test_disk.py new file mode 100644 index 00000000..5f4d5734 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py @@ -0,0 +1,540 @@ +import os +import pytest +from mock.mock import patch +from ceph_volume.util import disk + + +class TestLsblkParser(object): + + def test_parses_whitespace_values(self): + output = 'NAME="sdaa5" PARTLABEL="ceph data" RM="0" SIZE="10M" RO="0" TYPE="part"' + result = disk._lsblk_parser(output) + assert result['PARTLABEL'] == 'ceph data' + + def test_ignores_bogus_pairs(self): + output = 'NAME="sdaa5" PARTLABEL RM="0" SIZE="10M" RO="0" TYPE="part" MOUNTPOINT=""' + result = disk._lsblk_parser(output) + assert result['SIZE'] == '10M' + + +class TestBlkidParser(object): + + def test_parses_whitespace_values(self): + output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"''' # noqa + result = disk._blkid_parser(output) + assert result['PARTLABEL'] == 'ceph data' + + def test_ignores_unmapped(self): + output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"''' # noqa + result = disk._blkid_parser(output) + assert len(result.keys()) == 4 + + def test_translates_to_partuuid(self): + output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"''' # noqa + result = disk._blkid_parser(output) + assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f' + + +class TestBlkid(object): + + def test_parses_translated(self, stub_call): + output = '''/dev/sdb1: UUID="62416664-cbaf-40bd-9689-10bd337379c3" TYPE="xfs" PART_ENTRY_SCHEME="gpt" PART_ENTRY_NAME="ceph data" PART_ENTRY_UUID="b89c03bc-bf58-4338-a8f8-a2f484852b4f"''' # noqa + stub_call((output.split(), [], 0)) + result = disk.blkid('/dev/sdb1') + assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f' + assert result['PARTLABEL'] == 'ceph data' + assert result['UUID'] == '62416664-cbaf-40bd-9689-10bd337379c3' + assert result['TYPE'] == 'xfs' + +class TestUdevadmProperty(object): + + def test_good_output(self, stub_call): + output = """ID_MODEL=SK_hynix_SC311_SATA_512GB +ID_PART_TABLE_TYPE=gpt +ID_SERIAL_SHORT=MS83N71801150416A""".split() + stub_call((output, [], 0)) + result = disk.udevadm_property('dev/sda') + assert result['ID_MODEL'] == 'SK_hynix_SC311_SATA_512GB' + assert result['ID_PART_TABLE_TYPE'] == 'gpt' + assert result['ID_SERIAL_SHORT'] == 'MS83N71801150416A' + + def test_property_filter(self, stub_call): + output = """ID_MODEL=SK_hynix_SC311_SATA_512GB +ID_PART_TABLE_TYPE=gpt +ID_SERIAL_SHORT=MS83N71801150416A""".split() + stub_call((output, [], 0)) + result = disk.udevadm_property('dev/sda', ['ID_MODEL', + 'ID_SERIAL_SHORT']) + assert result['ID_MODEL'] == 'SK_hynix_SC311_SATA_512GB' + assert 'ID_PART_TABLE_TYPE' not in result + + def test_fail_on_broken_output(self, stub_call): + output = ["ID_MODEL:SK_hynix_SC311_SATA_512GB"] + stub_call((output, [], 0)) + with pytest.raises(ValueError): + disk.udevadm_property('dev/sda') + + +class TestDeviceFamily(object): + + def test_groups_multiple_devices(self, stub_call): + out = [ + 'NAME="sdaa5" PARLABEL="ceph lockbox"', + 'NAME="sdaa" RO="0"', + 'NAME="sdaa1" PARLABEL="ceph data"', + 'NAME="sdaa2" PARLABEL="ceph journal"', + ] + stub_call((out, '', 0)) + result = disk.device_family('sdaa5') + assert len(result) == 4 + + def test_parses_output_correctly(self, stub_call): + names = ['sdaa', 'sdaa5', 'sdaa1', 'sdaa2'] + out = [ + 'NAME="sdaa5" PARLABEL="ceph lockbox"', + 'NAME="sdaa" RO="0"', + 'NAME="sdaa1" PARLABEL="ceph data"', + 'NAME="sdaa2" PARLABEL="ceph journal"', + ] + stub_call((out, '', 0)) + result = disk.device_family('sdaa5') + for parsed in result: + assert parsed['NAME'] in names + + +class TestHumanReadableSize(object): + + def test_bytes(self): + result = disk.human_readable_size(800) + assert result == '800.00 B' + + def test_kilobytes(self): + result = disk.human_readable_size(800*1024) + assert result == '800.00 KB' + + def test_megabytes(self): + result = disk.human_readable_size(800*1024*1024) + assert result == '800.00 MB' + + def test_gigabytes(self): + result = disk.human_readable_size(8.19*1024*1024*1024) + assert result == '8.19 GB' + + def test_terabytes(self): + result = disk.human_readable_size(81.2*1024*1024*1024*1024) + assert result == '81.20 TB' + + +class TestSizeFromHumanReadable(object): + + def test_bytes(self): + result = disk.size_from_human_readable('2') + assert result == disk.Size(b=2) + + def test_kilobytes(self): + result = disk.size_from_human_readable('2 K') + assert result == disk.Size(kb=2) + + def test_megabytes(self): + result = disk.size_from_human_readable('2 M') + assert result == disk.Size(mb=2) + + def test_gigabytes(self): + result = disk.size_from_human_readable('2 G') + assert result == disk.Size(gb=2) + + def test_terrabytes(self): + result = disk.size_from_human_readable('2 T') + assert result == disk.Size(tb=2) + + def test_case(self): + result = disk.size_from_human_readable('2 t') + assert result == disk.Size(tb=2) + + def test_space(self): + result = disk.size_from_human_readable('2T') + assert result == disk.Size(tb=2) + + def test_float(self): + result = disk.size_from_human_readable('2.0') + assert result == disk.Size(b=2) + result = disk.size_from_human_readable('2.0T') + assert result == disk.Size(tb=2) + result = disk.size_from_human_readable('1.8T') + assert result == disk.Size(tb=1.8) + + +class TestSizeParse(object): + + def test_bytes(self): + result = disk.Size.parse('2') + assert result == disk.Size(b=2) + + def test_kilobytes(self): + result = disk.Size.parse('2K') + assert result == disk.Size(kb=2) + + def test_megabytes(self): + result = disk.Size.parse('2M') + assert result == disk.Size(mb=2) + + def test_gigabytes(self): + result = disk.Size.parse('2G') + assert result == disk.Size(gb=2) + + def test_terrabytes(self): + result = disk.Size.parse('2T') + assert result == disk.Size(tb=2) + + def test_tb(self): + result = disk.Size.parse('2Tb') + assert result == disk.Size(tb=2) + + def test_case(self): + result = disk.Size.parse('2t') + assert result == disk.Size(tb=2) + + def test_space(self): + result = disk.Size.parse('2T') + assert result == disk.Size(tb=2) + + def test_float(self): + result = disk.Size.parse('2.0') + assert result == disk.Size(b=2) + result = disk.Size.parse('2.0T') + assert result == disk.Size(tb=2) + result = disk.Size.parse('1.8T') + 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_path(self, tmpdir): + path = os.path.join(str(tmpdir), 'block') + os.makedirs(path) + return path + + 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, 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')) + result = disk.get_devices(_sys_block_path=block_path) + assert len(result.keys()) == 1 + 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') + os.makedirs(block_sda_path) + tmpfile('size', '1024', directory=block_sda_path) + 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 + 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') + os.makedirs(block_sda_path) + os.makedirs(sda_queue_path) + tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path) + 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') + os.makedirs(block_sda_path) + os.makedirs(sda_queue_path) + tmpfile('logical_block_size', contents='99', directory=sda_queue_path) + 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') + os.makedirs(block_sda_path) + os.makedirs(sda_queue_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) + 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') + os.makedirs(block_sda_path) + os.makedirs(sda_queue_path) + tmpfile('rotational', contents='1', directory=sda_queue_path) + result = disk.get_devices(_sys_block_path=block_path) + assert result[sda_path]['rotational'] == '1' + + +class TestSizeCalculations(object): + + @pytest.mark.parametrize('aliases', [ + ('b', 'bytes'), + ('kb', 'kilobytes'), + ('mb', 'megabytes'), + ('gb', 'gigabytes'), + ('tb', 'terabytes'), + ]) + def test_aliases(self, aliases): + short_alias, long_alias = aliases + s = disk.Size(b=1) + short_alias = getattr(s, short_alias) + long_alias = getattr(s, long_alias) + assert short_alias == long_alias + + @pytest.mark.parametrize('values', [ + ('b', 857619069665.28), + ('kb', 837518622.72), + ('mb', 817889.28), + ('gb', 798.72), + ('tb', 0.78), + ]) + def test_terabytes(self, values): + # regardless of the input value, all the other values correlate to each + # other the same, every time + unit, value = values + s = disk.Size(**{unit: value}) + assert s.b == 857619069665.28 + assert s.kb == 837518622.72 + assert s.mb == 817889.28 + assert s.gb == 798.72 + assert s.tb == 0.78 + + +class TestSizeOperators(object): + + @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001]) + def test_gigabytes_is_smaller(self, larger): + assert disk.Size(gb=1) < disk.Size(mb=larger) + + @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001]) + def test_gigabytes_is_larger(self, smaller): + assert disk.Size(gb=1) > disk.Size(mb=smaller) + + @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001, 1024]) + def test_gigabytes_is_smaller_or_equal(self, larger): + assert disk.Size(gb=1) <= disk.Size(mb=larger) + + @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001, 1024]) + def test_gigabytes_is_larger_or_equal(self, smaller): + assert disk.Size(gb=1) >= disk.Size(mb=smaller) + + @pytest.mark.parametrize('values', [ + ('b', 857619069665.28), + ('kb', 837518622.72), + ('mb', 817889.28), + ('gb', 798.72), + ('tb', 0.78), + ]) + def test_equality(self, values): + unit, value = values + s = disk.Size(**{unit: value}) + # both tb and b, since b is always calculated regardless, and is useful + # when testing tb + assert disk.Size(tb=0.78) == s + assert disk.Size(b=857619069665.28) == s + + @pytest.mark.parametrize('values', [ + ('b', 857619069665.28), + ('kb', 837518622.72), + ('mb', 817889.28), + ('gb', 798.72), + ('tb', 0.78), + ]) + def test_inequality(self, values): + unit, value = values + s = disk.Size(**{unit: value}) + # both tb and b, since b is always calculated regardless, and is useful + # when testing tb + assert disk.Size(tb=1) != s + assert disk.Size(b=100) != s + + +class TestSizeOperations(object): + + def test_assignment_addition_with_size_objects(self): + result = disk.Size(mb=256) + disk.Size(gb=1) + assert result.gb == 1.25 + assert result.gb.as_int() == 1 + assert result.gb.as_float() == 1.25 + + def test_self_addition_with_size_objects(self): + base = disk.Size(mb=256) + base += disk.Size(gb=1) + assert base.gb == 1.25 + + def test_self_addition_does_not_alter_state(self): + base = disk.Size(mb=256) + base + disk.Size(gb=1) + assert base.mb == 256 + + def test_addition_with_non_size_objects(self): + with pytest.raises(TypeError): + disk.Size(mb=100) + 4 + + def test_assignment_subtraction_with_size_objects(self): + base = disk.Size(gb=1) + base -= disk.Size(mb=256) + assert base.mb == 768 + + def test_self_subtraction_does_not_alter_state(self): + base = disk.Size(gb=1) + base - disk.Size(mb=256) + assert base.gb == 1 + + def test_subtraction_with_size_objects(self): + result = disk.Size(gb=1) - disk.Size(mb=256) + assert result.mb == 768 + + def test_subtraction_with_non_size_objects(self): + with pytest.raises(TypeError): + disk.Size(mb=100) - 4 + + def test_multiplication_with_size_objects(self): + with pytest.raises(TypeError): + disk.Size(mb=100) * disk.Size(mb=1) + + def test_multiplication_with_non_size_objects(self): + base = disk.Size(gb=1) + result = base * 2 + assert result.gb == 2 + assert result.gb.as_int() == 2 + + def test_division_with_size_objects(self): + result = disk.Size(gb=1) / disk.Size(mb=1) + assert int(result) == 1024 + + def test_division_with_non_size_objects(self): + base = disk.Size(gb=1) + result = base / 2 + assert result.mb == 512 + assert result.mb.as_int() == 512 + + def test_division_with_non_size_objects_without_state(self): + base = disk.Size(gb=1) + base / 2 + assert base.gb == 1 + assert base.gb.as_int() == 1 + + +class TestSizeAttributes(object): + + def test_attribute_does_not_exist(self): + with pytest.raises(AttributeError): + disk.Size(mb=1).exabytes + + +class TestSizeFormatting(object): + + def test_default_formatting_tb_to_b(self): + size = disk.Size(tb=0.0000000001) + result = "%s" % size + assert result == "109.95 B" + + def test_default_formatting_tb_to_kb(self): + size = disk.Size(tb=0.00000001) + result = "%s" % size + assert result == "10.74 KB" + + def test_default_formatting_tb_to_mb(self): + size = disk.Size(tb=0.000001) + result = "%s" % size + assert result == "1.05 MB" + + def test_default_formatting_tb_to_gb(self): + size = disk.Size(tb=0.001) + result = "%s" % size + assert result == "1.02 GB" + + def test_default_formatting_tb_to_tb(self): + size = disk.Size(tb=10) + result = "%s" % size + assert result == "10.00 TB" + + +class TestSizeSpecificFormatting(object): + + def test_formatting_b(self): + size = disk.Size(b=2048) + result = "%s" % size.b + assert "%s" % size.b == "%s" % size.bytes + assert result == "2048.00 B" + + def test_formatting_kb(self): + size = disk.Size(kb=5700) + result = "%s" % size.kb + assert "%s" % size.kb == "%s" % size.kilobytes + assert result == "5700.00 KB" + + def test_formatting_mb(self): + size = disk.Size(mb=4000) + result = "%s" % size.mb + assert "%s" % size.mb == "%s" % size.megabytes + assert result == "4000.00 MB" + + def test_formatting_gb(self): + size = disk.Size(gb=77777) + result = "%s" % size.gb + assert "%s" % size.gb == "%s" % size.gigabytes + assert result == "77777.00 GB" + + def test_formatting_tb(self): + size = disk.Size(tb=1027) + result = "%s" % size.tb + assert "%s" % size.tb == "%s" % size.terabytes + assert result == "1027.00 TB" diff --git a/src/ceph-volume/ceph_volume/tests/util/test_encryption.py b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py new file mode 100644 index 00000000..e1420b44 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py @@ -0,0 +1,53 @@ +from ceph_volume.util import encryption + + +class TestStatus(object): + + def test_skips_unuseful_lines(self, stub_call): + out = ['some line here', ' device: /dev/sdc1'] + stub_call((out, '', 0)) + assert encryption.status('/dev/sdc1') == {'device': '/dev/sdc1'} + + def test_removes_extra_quotes(self, stub_call): + out = ['some line here', ' device: "/dev/sdc1"'] + stub_call((out, '', 0)) + assert encryption.status('/dev/sdc1') == {'device': '/dev/sdc1'} + + def test_ignores_bogus_lines(self, stub_call): + out = ['some line here', ' '] + stub_call((out, '', 0)) + assert encryption.status('/dev/sdc1') == {} + + +class TestDmcryptClose(object): + + def test_mapper_exists(self, fake_run, tmpfile): + file_name = tmpfile(name='mapper-device') + encryption.dmcrypt_close(file_name) + arguments = fake_run.calls[0]['args'][0] + assert arguments[0] == 'cryptsetup' + assert arguments[1] == 'remove' + assert arguments[2].startswith('/') + + def test_mapper_does_not_exist(self, fake_run): + file_name = '/path/does/not/exist' + encryption.dmcrypt_close(file_name) + assert fake_run.calls == [] + + +class TestDmcryptKey(object): + + def test_dmcrypt_with_default_size(self, conf_ceph_stub): + conf_ceph_stub('[global]\nfsid=asdf-lkjh') + result = encryption.create_dmcrypt_key() + assert len(result) == 172 + + def test_dmcrypt_with_custom_size(self, conf_ceph_stub): + conf_ceph_stub(''' + [global] + fsid=asdf + [osd] + osd_dmcrypt_size=8 + ''') + result = encryption.create_dmcrypt_key() + assert len(result) == 172 diff --git a/src/ceph-volume/ceph_volume/tests/util/test_prepare.py b/src/ceph-volume/ceph_volume/tests/util/test_prepare.py new file mode 100644 index 00000000..ced5d49e --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_prepare.py @@ -0,0 +1,422 @@ +import pytest +from textwrap import dedent +import json +from ceph_volume.util import prepare +from ceph_volume.util.prepare import system +from ceph_volume import conf +from ceph_volume.tests.conftest import Factory + + +class TestOSDIDAvailable(object): + + def test_false_if_id_is_none(self): + assert not prepare.osd_id_available(None) + + def test_returncode_is_not_zero(self, monkeypatch): + monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: ('', '', 1)) + with pytest.raises(RuntimeError): + prepare.osd_id_available(1) + + def test_id_does_exist_but_not_available(self, monkeypatch): + stdout = dict(nodes=[ + dict(id=0, status="up"), + ]) + stdout = ['', json.dumps(stdout)] + monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0)) + result = prepare.osd_id_available(0) + assert not result + + def test_id_does_not_exist(self, monkeypatch): + stdout = dict(nodes=[ + dict(id=0), + ]) + stdout = ['', json.dumps(stdout)] + monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0)) + result = prepare.osd_id_available(1) + assert not result + + def test_invalid_osd_id(self, monkeypatch): + stdout = dict(nodes=[ + dict(id=0), + ]) + stdout = ['', json.dumps(stdout)] + monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0)) + result = prepare.osd_id_available("foo") + assert not result + + def test_returns_true_when_id_is_destroyed(self, monkeypatch): + stdout = dict(nodes=[ + dict(id=0, status="destroyed"), + ]) + stdout = ['', json.dumps(stdout)] + monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0)) + result = prepare.osd_id_available(0) + assert result + + +class TestFormatDevice(object): + + def test_include_force(self, fake_run, monkeypatch): + monkeypatch.setattr(conf, 'ceph', Factory(get_list=lambda *a, **kw: [])) + prepare.format_device('/dev/sxx') + flags = fake_run.calls[0]['args'][0] + assert '-f' in flags + + def test_device_is_always_appended(self, fake_run, conf_ceph): + conf_ceph(get_list=lambda *a, **kw: []) + prepare.format_device('/dev/sxx') + flags = fake_run.calls[0]['args'][0] + assert flags[-1] == '/dev/sxx' + + def test_extra_flags_are_added(self, fake_run, conf_ceph): + conf_ceph(get_list=lambda *a, **kw: ['--why-yes']) + prepare.format_device('/dev/sxx') + flags = fake_run.calls[0]['args'][0] + assert '--why-yes' in flags + + def test_default_options(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234""")) + conf.cluster = 'ceph' + prepare.format_device('/dev/sda1') + expected = [ + 'mkfs', '-t', 'xfs', + '-f', '-i', 'size=2048', # default flags + '/dev/sda1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_multiple_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mkfs options xfs = -f -i size=1024""")) + conf.cluster = 'ceph' + prepare.format_device('/dev/sda1') + expected = [ + 'mkfs', '-t', 'xfs', + '-f', '-i', 'size=1024', + '/dev/sda1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_multiple_options_will_get_the_force_flag(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mkfs options xfs = -i size=1024""")) + conf.cluster = 'ceph' + prepare.format_device('/dev/sda1') + expected = [ + 'mkfs', '-t', 'xfs', + '-f', '-i', 'size=1024', + '/dev/sda1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_underscore_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd_mkfs_options_xfs = -i size=128""")) + conf.cluster = 'ceph' + prepare.format_device('/dev/sda1') + expected = [ + 'mkfs', '-t', 'xfs', + '-f', '-i', 'size=128', + '/dev/sda1'] + assert expected == fake_run.calls[0]['args'][0] + + +mkfs_filestore_flags = [ + 'ceph-osd', + '--cluster', + '--osd-objectstore', 'filestore', + '--mkfs', + '-i', + '--monmap', + '--keyfile', '-', # goes through stdin + '--osd-data', + '--osd-journal', + '--osd-uuid', + '--setuser', 'ceph', + '--setgroup', 'ceph' +] + + +class TestOsdMkfsFilestore(object): + + @pytest.mark.parametrize('flag', mkfs_filestore_flags) + def test_keyring_is_used(self, fake_call, monkeypatch, flag): + monkeypatch.setattr(prepare, '__release__', 'mimic') + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_filestore(1, 'asdf', keyring='secret') + assert flag in fake_call.calls[0]['args'][0] + + def test_keyring_is_used_luminous(self, fake_call, monkeypatch): + monkeypatch.setattr(prepare, '__release__', 'luminous') + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_filestore(1, 'asdf', keyring='secret') + assert '--keyfile' not in fake_call.calls[0]['args'][0] + + +class TestOsdMkfsBluestore(object): + + def test_keyring_is_added(self, fake_call, monkeypatch): + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_bluestore(1, 'asdf', keyring='secret') + assert '--keyfile' in fake_call.calls[0]['args'][0] + + def test_keyring_is_not_added(self, fake_call, monkeypatch): + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_bluestore(1, 'asdf') + assert '--keyfile' not in fake_call.calls[0]['args'][0] + + def test_keyring_is_not_added_luminous(self, fake_call, monkeypatch): + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_bluestore(1, 'asdf') + monkeypatch.setattr(prepare, '__release__', 'luminous') + assert '--keyfile' not in fake_call.calls[0]['args'][0] + + def test_wal_is_added(self, fake_call, monkeypatch): + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_bluestore(1, 'asdf', wal='/dev/smm1') + assert '--bluestore-block-wal-path' in fake_call.calls[0]['args'][0] + assert '/dev/smm1' in fake_call.calls[0]['args'][0] + + def test_db_is_added(self, fake_call, monkeypatch): + monkeypatch.setattr(system, 'chown', lambda path: True) + prepare.osd_mkfs_bluestore(1, 'asdf', db='/dev/smm2') + assert '--bluestore-block-db-path' in fake_call.calls[0]['args'][0] + assert '/dev/smm2' in fake_call.calls[0]['args'][0] + + +class TestMountOSD(object): + + def test_default_options(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234""")) + conf.cluster = 'ceph' + prepare.mount_osd('/dev/sda1', 1) + expected = [ + 'mount', '-t', 'xfs', '-o', + 'rw,noatime,inode64', # default flags + '/dev/sda1', '/var/lib/ceph/osd/ceph-1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_mount_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mount options xfs = rw""")) + conf.cluster = 'ceph' + prepare.mount_osd('/dev/sda1', 1) + expected = [ + 'mount', '-t', 'xfs', '-o', + 'rw', + '/dev/sda1', '/var/lib/ceph/osd/ceph-1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_multiple_whitespace_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mount options xfs = rw auto exec""")) + conf.cluster = 'ceph' + prepare.mount_osd('/dev/sda1', 1) + expected = [ + 'mount', '-t', 'xfs', '-o', + 'rw,auto,exec', + '/dev/sda1', '/var/lib/ceph/osd/ceph-1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_multiple_comma_whitespace_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mount options xfs = rw, auto, exec""")) + conf.cluster = 'ceph' + prepare.mount_osd('/dev/sda1', 1) + expected = [ + 'mount', '-t', 'xfs', '-o', + 'rw,auto,exec', + '/dev/sda1', '/var/lib/ceph/osd/ceph-1'] + assert expected == fake_run.calls[0]['args'][0] + + def test_underscore_mount_options_are_used(self, conf_ceph_stub, fake_run): + conf_ceph_stub(dedent("""[global] + fsid = 1234lkjh1234 + [osd] + osd mount options xfs = rw""")) + conf.cluster = 'ceph' + prepare.mount_osd('/dev/sda1', 1) + expected = [ + 'mount', '-t', 'xfs', '-o', + 'rw', + '/dev/sda1', '/var/lib/ceph/osd/ceph-1'] + assert expected == fake_run.calls[0]['args'][0] + + +ceph_conf_mount_values = [ + ['rw,', 'auto,' 'exec'], + ['rw', 'auto', 'exec'], + [' rw ', ' auto ', ' exec '], + ['rw,', 'auto,', 'exec,'], + [',rw ', ',auto ', ',exec,'], + [',rw,', ',auto,', ',exec,'], +] + +string_mount_values = [ + 'rw, auto exec ', + 'rw auto exec', + ',rw, auto, exec,', + ' rw auto exec ', + ' rw,auto,exec ', + 'rw,auto,exec', + ',rw,auto,exec,', + 'rw,auto,exec ', + 'rw, auto, exec ', +] + + +class TestNormalizeFlags(object): + # a bit overkill since most of this is already tested in prepare.mount_osd + # tests + + @pytest.mark.parametrize("flags", ceph_conf_mount_values) + def test_normalize_lists(self, flags): + result = sorted(prepare._normalize_mount_flags(flags).split(',')) + assert ','.join(result) == 'auto,exec,rw' + + @pytest.mark.parametrize("flags", string_mount_values) + def test_normalize_strings(self, flags): + result = sorted(prepare._normalize_mount_flags(flags).split(',')) + assert ','.join(result) == 'auto,exec,rw' + + @pytest.mark.parametrize("flags", ceph_conf_mount_values) + def test_normalize_extra_flags(self, flags): + result = prepare._normalize_mount_flags(flags, extras=['discard']) + assert sorted(result.split(',')) == ['auto', 'discard', 'exec', 'rw'] + + @pytest.mark.parametrize("flags", ceph_conf_mount_values) + def test_normalize_duplicate_extra_flags(self, flags): + result = prepare._normalize_mount_flags(flags, extras=['rw', 'discard']) + assert sorted(result.split(',')) == ['auto', 'discard', 'exec', 'rw'] + + @pytest.mark.parametrize("flags", string_mount_values) + def test_normalize_strings_flags(self, flags): + result = sorted(prepare._normalize_mount_flags(flags, extras=['discard']).split(',')) + assert ','.join(result) == 'auto,discard,exec,rw' + + @pytest.mark.parametrize("flags", string_mount_values) + def test_normalize_strings_duplicate_flags(self, flags): + result = sorted(prepare._normalize_mount_flags(flags, extras=['discard','rw']).split(',')) + assert ','.join(result) == 'auto,discard,exec,rw' + + +class TestMkfsFilestore(object): + + def test_non_zero_exit_status(self, stub_call, monkeypatch): + conf.cluster = 'ceph' + monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True) + stub_call(([], [], 1)) + with pytest.raises(RuntimeError) as error: + prepare.osd_mkfs_filestore('1', 'asdf-1234', 'keyring') + assert "Command failed with exit code 1" in str(error.value) + + def test_non_zero_exit_formats_command_correctly(self, stub_call, monkeypatch): + conf.cluster = 'ceph' + monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True) + stub_call(([], [], 1)) + with pytest.raises(RuntimeError) as error: + prepare.osd_mkfs_filestore('1', 'asdf-1234', 'keyring') + expected = ' '.join([ + 'ceph-osd', + '--cluster', + 'ceph', + '--osd-objectstore', 'filestore', '--mkfs', + '-i', '1', '--monmap', '/var/lib/ceph/osd/ceph-1/activate.monmap', + '--keyfile', '-', '--osd-data', '/var/lib/ceph/osd/ceph-1/', + '--osd-journal', '/var/lib/ceph/osd/ceph-1/journal', + '--osd-uuid', 'asdf-1234', + '--setuser', 'ceph', '--setgroup', 'ceph']) + assert expected in str(error.value) + + +class TestMkfsBluestore(object): + + def test_non_zero_exit_status(self, stub_call, monkeypatch): + conf.cluster = 'ceph' + monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True) + stub_call(([], [], 1)) + with pytest.raises(RuntimeError) as error: + prepare.osd_mkfs_bluestore('1', 'asdf-1234', keyring='keyring') + assert "Command failed with exit code 1" in str(error.value) + + def test_non_zero_exit_formats_command_correctly(self, stub_call, monkeypatch): + conf.cluster = 'ceph' + monkeypatch.setattr('ceph_volume.util.prepare.system.chown', lambda x: True) + stub_call(([], [], 1)) + with pytest.raises(RuntimeError) as error: + prepare.osd_mkfs_bluestore('1', 'asdf-1234', keyring='keyring') + expected = ' '.join([ + 'ceph-osd', + '--cluster', + 'ceph', + '--osd-objectstore', 'bluestore', '--mkfs', + '-i', '1', '--monmap', '/var/lib/ceph/osd/ceph-1/activate.monmap', + '--keyfile', '-', '--osd-data', '/var/lib/ceph/osd/ceph-1/', + '--osd-uuid', 'asdf-1234', + '--setuser', 'ceph', '--setgroup', 'ceph']) + assert expected in str(error.value) + + +class TestGetJournalSize(object): + + def test_undefined_size_fallbacks_formatted(self, conf_ceph_stub): + conf_ceph_stub(dedent(""" + [global] + fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f + """)) + result = prepare.get_journal_size() + assert result == '5G' + + def test_undefined_size_fallbacks_unformatted(self, conf_ceph_stub): + conf_ceph_stub(dedent(""" + [global] + fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f + """)) + result = prepare.get_journal_size(lv_format=False) + assert result.gb.as_int() == 5 + + def test_defined_size_unformatted(self, conf_ceph_stub): + conf_ceph_stub(dedent(""" + [global] + fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f + + [osd] + osd journal size = 10240 + """)) + result = prepare.get_journal_size(lv_format=False) + assert result.gb.as_int() == 10 + + def test_defined_size_formatted(self, conf_ceph_stub): + conf_ceph_stub(dedent(""" + [global] + fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f + + [osd] + osd journal size = 10240 + """)) + result = prepare.get_journal_size() + assert result == '10G' + + def test_refuse_tiny_journals(self, conf_ceph_stub): + conf_ceph_stub(dedent(""" + [global] + fsid = a25d19a6-7d57-4eda-b006-78e35d2c4d9f + + [osd] + osd journal size = 1024 + """)) + with pytest.raises(RuntimeError) as error: + prepare.get_journal_size() + assert 'journal sizes must be larger' in str(error.value) + assert 'detected: 1024.00 MB' in str(error.value) diff --git a/src/ceph-volume/ceph_volume/tests/util/test_system.py b/src/ceph-volume/ceph_volume/tests/util/test_system.py new file mode 100644 index 00000000..74999949 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_system.py @@ -0,0 +1,279 @@ +import os +import pwd +import getpass +import pytest +from textwrap import dedent +from ceph_volume.util import system +from mock.mock import patch + + +class TestMkdirP(object): + + def test_existing_dir_does_not_raise_w_chown(self, monkeypatch, tmpdir): + user = pwd.getpwnam(getpass.getuser()) + uid, gid = user[2], user[3] + monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,)) + path = str(tmpdir) + system.mkdir_p(path) + assert os.path.isdir(path) + + def test_new_dir_w_chown(self, monkeypatch, tmpdir): + user = pwd.getpwnam(getpass.getuser()) + uid, gid = user[2], user[3] + monkeypatch.setattr(system, 'get_ceph_user_ids', lambda: (uid, gid,)) + path = os.path.join(str(tmpdir), 'new') + system.mkdir_p(path) + assert os.path.isdir(path) + + def test_existing_dir_does_not_raise_no_chown(self, tmpdir): + path = str(tmpdir) + system.mkdir_p(path, chown=False) + assert os.path.isdir(path) + + def test_new_dir_no_chown(self, tmpdir): + path = os.path.join(str(tmpdir), 'new') + system.mkdir_p(path, chown=False) + assert os.path.isdir(path) + + +@pytest.fixture +def fake_proc(tmpdir, monkeypatch): + PROCDIR = str(tmpdir) + proc_path = os.path.join(PROCDIR, 'mounts') + with open(proc_path, 'w') as f: + f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0 + rootfs / rootfs rw 0 0 + sysfs /sys sysfs rw,seclabel,nosuid,nodev,noexec,relatime 0 0 + proc /proc proc rw,nosuid,nodev,noexec,relatime 0 0 + devtmpfs /dev devtmpfs rw,seclabel,nosuid,size=238292k,nr_inodes=59573,mode=755 0 0 + securityfs /sys/kernel/security securityfs rw,nosuid,nodev,noexec,relatime 0 0 + tmpfs /dev/shm tmpfs rw,seclabel,nosuid,nodev 0 0 + devpts /dev/pts devpts rw,seclabel,nosuid,noexec,relatime,gid=5,mode=620,ptmxmode=000 0 0 + tmpfs /run tmpfs rw,seclabel,nosuid,nodev,mode=755 0 0 + tmpfs /sys/fs/cgroup tmpfs ro,seclabel,nosuid,nodev,noexec,mode=755 0 0 + cgroup /sys/fs/cgroup/systemd cgroup rw,nosuid,nodev,noexec,relatime,xattr,release_agent=/usr/lib/systemd/systemd-cgroups-agent,name=systemd 0 0 + cgroup /sys/fs/cgroup/freezer cgroup rw,nosuid,nodev,noexec,relatime,freezer 0 0 + configfs /sys/kernel/config configfs rw,relatime 0 0 + /dev/mapper/VolGroup00-LogVol00 / xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0 + selinuxfs /sys/fs/selinux selinuxfs rw,relatime 0 0 + debugfs /sys/kernel/debug debugfs rw,relatime 0 0 + hugetlbfs /dev/hugepages hugetlbfs rw,seclabel,relatime 0 0 + mqueue /dev/mqueue mqueue rw,seclabel,relatime 0 0 + sunrpc /far/lib/nfs/rpc_pipefs rpc_pipefs rw,relatime 0 0 + /dev/sde4 /two/field/path + nfsd /proc/fs/nfsd nfsd rw,relatime 0 0 + /dev/sde2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0 + tmpfs /far/lib/ceph/osd/ceph-5 tmpfs rw,seclabel,relatime 0 0 + tmpfs /far/lib/ceph/osd/ceph-7 tmpfs rw,seclabel,relatime 0 0 + /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,seclabel,noatime,attr2,inode64,noquota 0 0 + tmpfs /run/user/1000 tmpfs rw,seclabel,nosuid,nodev,relatime,size=50040k,mode=700,uid=1000,gid=1000 0 0 + /dev/sdc2 /boot xfs rw,seclabel,relatime,attr2,inode64,noquota 0 0 + tmpfs /run/user/1000 tmpfs rw,seclabel,mode=700,uid=1000,gid=1000 0 0""")) + monkeypatch.setattr(system, 'PROCDIR', PROCDIR) + monkeypatch.setattr(os.path, 'exists', lambda x: True) + + +class TestPathIsMounted(object): + + def test_is_mounted(self, fake_proc): + assert system.path_is_mounted('/boot') is True + + def test_is_not_mounted(self, fake_proc): + assert system.path_is_mounted('/far/fib/feph') is False + + def test_is_not_mounted_at_destination(self, fake_proc): + assert system.path_is_mounted('/boot', destination='/dev/sda1') is False + + def test_is_mounted_at_destination(self, fake_proc): + assert system.path_is_mounted('/boot', destination='/dev/sdc2') is True + + +class TestDeviceIsMounted(object): + + def test_is_mounted(self, fake_proc): + assert system.device_is_mounted('/dev/sda1') is True + + def test_path_is_not_device(self, fake_proc): + assert system.device_is_mounted('/far/lib/ceph/osd/ceph-7') is False + + def test_is_not_mounted_at_destination(self, fake_proc): + assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/test-1') is False + + def test_is_mounted_at_destination(self, fake_proc): + assert system.device_is_mounted('/dev/sda1', destination='/far/lib/ceph/osd/ceph-7') is False + + def test_is_realpath_dev_mounted_at_destination(self, fake_proc, monkeypatch): + monkeypatch.setattr(system.os.path, 'realpath', lambda x: '/dev/sda1' if 'foo' in x else x) + result = system.device_is_mounted('/dev/maper/foo', destination='/far/lib/ceph/osd/ceph-0') + assert result is True + + def test_is_realpath_path_mounted_at_destination(self, fake_proc, monkeypatch): + monkeypatch.setattr( + system.os.path, 'realpath', + lambda x: '/far/lib/ceph/osd/ceph-0' if 'symlink' in x else x) + result = system.device_is_mounted('/dev/sda1', destination='/symlink/lib/ceph/osd/ceph-0') + assert result is True + + +class TestGetMounts(object): + + def test_not_mounted(self, tmpdir, monkeypatch): + PROCDIR = str(tmpdir) + proc_path = os.path.join(PROCDIR, 'mounts') + with open(proc_path, 'w') as f: + f.write('') + monkeypatch.setattr(system, 'PROCDIR', PROCDIR) + assert system.get_mounts() == {} + + def test_is_mounted_(self, fake_proc): + result = system.get_mounts() + assert result['/dev/sdc2'] == ['/boot'] + + def test_ignores_two_fields(self, fake_proc): + result = system.get_mounts() + assert result.get('/dev/sde4') is None + + def test_tmpfs_is_reported(self, fake_proc): + result = system.get_mounts() + assert result['tmpfs'][0] == '/dev/shm' + + def test_non_skip_devs_arent_reported(self, fake_proc): + result = system.get_mounts() + assert result.get('cgroup') is None + + def test_multiple_mounts_are_appended(self, fake_proc): + result = system.get_mounts() + assert len(result['tmpfs']) == 7 + + def test_nonexistent_devices_are_skipped(self, tmpdir, monkeypatch): + PROCDIR = str(tmpdir) + proc_path = os.path.join(PROCDIR, 'mounts') + with open(proc_path, 'w') as f: + f.write(dedent("""nfsd /proc/fs/nfsd nfsd rw,relatime 0 0 + /dev/sda1 /far/lib/ceph/osd/ceph-0 xfs rw,attr2,inode64,noquota 0 0 + /dev/sda2 /far/lib/ceph/osd/ceph-1 xfs rw,attr2,inode64,noquota 0 0""")) + monkeypatch.setattr(system, 'PROCDIR', PROCDIR) + monkeypatch.setattr(os.path, 'exists', lambda x: False if x == '/dev/sda1' else True) + result = system.get_mounts() + assert result.get('/dev/sda1') is None + + +class TestIsBinary(object): + + def test_is_binary(self, tmpfile): + binary_path = tmpfile(contents='asd\n\nlkjh\x00') + assert system.is_binary(binary_path) + + def test_is_not_binary(self, tmpfile): + binary_path = tmpfile(contents='asd\n\nlkjh0') + assert system.is_binary(binary_path) is False + + +class TestGetFileContents(object): + + def test_path_does_not_exist(self, tmpdir): + filepath = os.path.join(str(tmpdir), 'doesnotexist') + assert system.get_file_contents(filepath, 'default') == 'default' + + def test_path_has_contents(self, tmpfile): + interesting_file = tmpfile(contents="1") + result = system.get_file_contents(interesting_file) + assert result == "1" + + def test_path_has_multiline_contents(self, tmpfile): + interesting_file = tmpfile(contents="0\n1") + result = system.get_file_contents(interesting_file) + assert result == "0\n1" + + def test_exception_returns_default(self, tmpfile): + interesting_file = tmpfile(contents="0") + # remove read, causes IOError + os.chmod(interesting_file, 0o000) + result = system.get_file_contents(interesting_file) + assert result == '' + + +class TestWhich(object): + + def test_executable_exists_but_is_not_file(self, monkeypatch): + monkeypatch.setattr(system.os.path, 'isfile', lambda x: False) + monkeypatch.setattr(system.os.path, 'exists', lambda x: True) + assert system.which('exedir') == 'exedir' + + def test_executable_does_not_exist(self, monkeypatch): + monkeypatch.setattr(system.os.path, 'isfile', lambda x: False) + monkeypatch.setattr(system.os.path, 'exists', lambda x: False) + assert system.which('exedir') == 'exedir' + + def test_executable_exists_as_file(self, monkeypatch): + monkeypatch.setattr(system.os, 'getenv', lambda x, y: '') + monkeypatch.setattr(system.os.path, 'isfile', lambda x: x != 'ceph') + monkeypatch.setattr(system.os.path, 'exists', lambda x: x != 'ceph') + assert system.which('ceph') == '/usr/local/bin/ceph' + + def test_warnings_when_executable_isnt_matched(self, monkeypatch, capsys): + monkeypatch.setattr(system.os.path, 'isfile', lambda x: True) + monkeypatch.setattr(system.os.path, 'exists', lambda x: False) + system.which('exedir') + cap = capsys.readouterr() + assert 'Executable exedir not in PATH' in cap.err + +@pytest.fixture +def stub_which(monkeypatch): + def apply(value='/bin/restorecon'): + monkeypatch.setattr(system, 'which', lambda x: value) + return apply + + +# python2 has no FileNotFoundError +try: + FileNotFoundError +except NameError: + FileNotFoundError = OSError + + +class TestSetContext(object): + + def setup(self): + try: + os.environ.pop('CEPH_VOLUME_SKIP_RESTORECON') + except KeyError: + pass + + @pytest.mark.parametrize('value', ['1', 'True', 'true', 'TRUE', 'yes']) + def test_set_context_skips(self, stub_call, fake_run, value): + stub_call(('', '', 0)) + os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value + system.set_context('/tmp/foo') + assert fake_run.calls == [] + + @pytest.mark.parametrize('value', ['0', 'False', 'false', 'FALSE', 'no']) + def test_set_context_doesnt_skip_with_env(self, stub_call, stub_which, fake_run, value): + stub_call(('', '', 0)) + stub_which() + os.environ['CEPH_VOLUME_SKIP_RESTORECON'] = value + system.set_context('/tmp/foo') + assert len(fake_run.calls) + + def test_set_context_skips_on_executable(self, stub_call, stub_which, fake_run): + stub_call(('', '', 0)) + stub_which('restorecon') + system.set_context('/tmp/foo') + assert fake_run.calls == [] + + def test_set_context_no_skip_on_executable(self, stub_call, stub_which, fake_run): + stub_call(('', '', 0)) + stub_which('/bin/restorecon') + system.set_context('/tmp/foo') + assert len(fake_run.calls) + + @patch('ceph_volume.process.call') + def test_selinuxenabled_doesnt_exist(self, mocked_call, fake_run): + mocked_call.side_effect = FileNotFoundError() + system.set_context('/tmp/foo') + assert fake_run.calls == [] + + def test_selinuxenabled_is_not_enabled(self, stub_call, fake_run): + stub_call(('', '', 1)) + system.set_context('/tmp/foo') + assert fake_run.calls == [] diff --git a/src/ceph-volume/ceph_volume/tests/util/test_util.py b/src/ceph-volume/ceph_volume/tests/util/test_util.py new file mode 100644 index 00000000..1a094d33 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_util.py @@ -0,0 +1,116 @@ +import pytest +from ceph_volume import util + + +class TestAsBytes(object): + + def test_bytes_just_gets_returned(self): + bytes_string = "contents".encode('utf-8') + assert util.as_bytes(bytes_string) == bytes_string + + def test_string_gets_converted_to_bytes(self): + result = util.as_bytes('contents') + assert isinstance(result, bytes) + + +class TestStrToInt(object): + + def test_passing_a_float_str_comma(self): + result = util.str_to_int("1,99") + assert result == 1 + + def test_passing_a_float_does_not_round_comma(self): + result = util.str_to_int("1,99", round_down=False) + assert result == 2 + + @pytest.mark.parametrize("value", ['2', 2]) + def test_passing_an_int(self, value): + result = util.str_to_int(value) + assert result == 2 + + @pytest.mark.parametrize("value", ['1.99', 1.99]) + def test_passing_a_float(self, value): + result = util.str_to_int(value) + assert result == 1 + + @pytest.mark.parametrize("value", ['1.99', 1.99]) + def test_passing_a_float_does_not_round(self, value): + result = util.str_to_int(value, round_down=False) + assert result == 2 + + def test_text_is_not_an_integer_like(self): + with pytest.raises(RuntimeError) as error: + util.str_to_int("1.4GB") + assert str(error.value) == "Unable to convert to integer: '1.4GB'" + + def test_input_is_not_string(self): + with pytest.raises(RuntimeError) as error: + util.str_to_int(None) + assert str(error.value) == "Unable to convert to integer: 'None'" + + +def true_responses(upper_casing=False): + if upper_casing: + return ['Y', 'YES', ''] + return ['y', 'yes', ''] + + +def false_responses(upper_casing=False): + if upper_casing: + return ['N', 'NO'] + return ['n', 'no'] + + +def invalid_responses(): + return [9, 0.1, 'h', [], {}, None] + + +class TestStrToBool(object): + + @pytest.mark.parametrize('response', true_responses()) + def test_trueish(self, response): + assert util.str_to_bool(response) is True + + @pytest.mark.parametrize('response', false_responses()) + def test_falseish(self, response): + assert util.str_to_bool(response) is False + + @pytest.mark.parametrize('response', true_responses(True)) + def test_trueish_upper(self, response): + assert util.str_to_bool(response) is True + + @pytest.mark.parametrize('response', false_responses(True)) + def test_falseish_upper(self, response): + assert util.str_to_bool(response) is False + + @pytest.mark.parametrize('response', invalid_responses()) + def test_invalid(self, response): + with pytest.raises(ValueError): + util.str_to_bool(response) + + +class TestPromptBool(object): + + @pytest.mark.parametrize('response', true_responses()) + def test_trueish(self, response): + fake_input = lambda x: response + qx = 'what the what?' + assert util.prompt_bool(qx, input_=fake_input) is True + + @pytest.mark.parametrize('response', false_responses()) + def test_falseish(self, response): + fake_input = lambda x: response + qx = 'what the what?' + assert util.prompt_bool(qx, input_=fake_input) is False + + def test_try_again_true(self): + responses = ['g', 'h', 'y'] + fake_input = lambda x: responses.pop(0) + qx = 'what the what?' + assert util.prompt_bool(qx, input_=fake_input) is True + + def test_try_again_false(self): + responses = ['g', 'h', 'n'] + fake_input = lambda x: responses.pop(0) + qx = 'what the what?' + assert util.prompt_bool(qx, input_=fake_input) is False |