From 19fcec84d8d7d21e796c7624e521b60d28ee21ed Mon Sep 17 00:00:00 2001 From: Daniel Baumann Date: Sun, 7 Apr 2024 20:45:59 +0200 Subject: Adding upstream version 16.2.11+ds. Signed-off-by: Daniel Baumann --- .../ceph_volume/tests/util/test_arg_validators.py | 338 ++++++++++ .../ceph_volume/tests/util/test_device.py | 704 +++++++++++++++++++++ .../ceph_volume/tests/util/test_disk.py | 524 +++++++++++++++ .../ceph_volume/tests/util/test_encryption.py | 138 ++++ .../ceph_volume/tests/util/test_prepare.py | 413 ++++++++++++ .../ceph_volume/tests/util/test_system.py | 309 +++++++++ .../ceph_volume/tests/util/test_util.py | 116 ++++ 7 files changed, 2542 insertions(+) create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_device.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_disk.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_encryption.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_prepare.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_system.py create mode 100644 src/ceph-volume/ceph_volume/tests/util/test_util.py (limited to 'src/ceph-volume/ceph_volume/tests/util') 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 000000000..59ca12619 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py @@ -0,0 +1,338 @@ +import argparse +import pytest +import os +from ceph_volume import exceptions, process +from ceph_volume.util import arg_validators +from mock.mock import patch, MagicMock + + +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, monkeypatch, fake_filesystem): + fake_file = fake_filesystem.create_file('/tmp/foo') + monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False) + validator = arg_validators.OSDPath() + with pytest.raises(argparse.ArgumentError): + validator(fake_file.path) + + 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 = ['', '--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 = ['', '--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 = ['', '--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, fake_filesystem): + self.validator = arg_validators.ValidDevice() + + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + def test_path_is_valid(self, m_has_bs_label, + fake_call, patch_bluestore_label, + device_info, monkeypatch): + monkeypatch.setattr('ceph_volume.util.device.Device.exists', lambda: True) + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + result = self.validator('/dev/sda') + assert result.path == '/dev/sda' + + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + def test_path_is_invalid(self, m_has_bs_label, + fake_call, patch_bluestore_label, + device_info): + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + with pytest.raises(argparse.ArgumentError): + self.validator('/device/does/not/exist') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_dev_has_partitions(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + exists=True, + has_partitions=True, + ) + with pytest.raises(RuntimeError): + self.validator('/dev/foo') + +class TestValidZapDevice(object): + def setup(self): + self.validator = arg_validators.ValidZapDevice() + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_has_partition(self, m_get_single_lv, m_has_bs_label, mocked_device): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=True, + has_gpt_headers=False, + has_fs=False + ) + self.validator.zap = False + with pytest.raises(RuntimeError): + assert self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_has_no_partition(self, m_get_single_lv, m_has_bs_label, mocked_device): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False + ) + self.validator.zap = False + assert self.validator('/dev/foo') + +class TestValidDataDevice(object): + def setup(self): + self.validator = arg_validators.ValidDataDevice() + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_used_by_ceph(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=True, + exists=True, + has_partitions=False, + has_gpt_headers=False + ) + with pytest.raises(SystemExit): + self.validator.zap = False + self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_has_fs(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=True + ) + with pytest.raises(RuntimeError): + self.validator.zap = False + self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=True) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_has_bs_signature(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False + ) + with pytest.raises(RuntimeError): + self.validator.zap = False + self.validator('/dev/foo') + +class TestValidRawDevice(object): + def setup(self): + self.validator = arg_validators.ValidRawDevice() + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.util.arg_validators.disk.blkid') + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_dmcrypt_device_already_prepared(self, m_get_single_lv, m_blkid, m_has_bs_label, mocked_device, fake_call, monkeypatch): + def mock_call(cmd, **kw): + return ('', '', 1) + monkeypatch.setattr(process, 'call', mock_call) + m_blkid.return_value = {'UUID': '8fd92779-ad78-437c-a06f-275f7170fa74', 'TYPE': 'crypto_LUKS'} + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False + ) + with pytest.raises(SystemExit): + self.validator.zap = False + self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_already_prepared(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False + ) + with pytest.raises(SystemExit): + self.validator.zap = False + self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_not_prepared(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch): + def mock_call(cmd, **kw): + return ('', '', 1) + monkeypatch.setattr(process, 'call', mock_call) + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False + ) + self.validator.zap = False + assert self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_has_partition(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch): + def mock_call(cmd, **kw): + return ('', '', 1) + monkeypatch.setattr(process, 'call', mock_call) + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=True, + has_gpt_headers=False, + has_fs=False + ) + self.validator.zap = False + with pytest.raises(RuntimeError): + assert self.validator('/dev/foo') + +class TestValidBatchDevice(object): + def setup(self): + self.validator = arg_validators.ValidBatchDevice() + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_is_partition(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False, + is_partition=True + ) + with pytest.raises(argparse.ArgumentError): + self.validator.zap = False + self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_is_not_partition(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False, + is_partition=False + ) + self.validator.zap = False + assert self.validator('/dev/foo') + +class TestValidBatchDataDevice(object): + def setup(self): + self.validator = arg_validators.ValidBatchDataDevice() + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_is_partition(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False, + is_partition=True + ) + with pytest.raises(argparse.ArgumentError): + self.validator.zap = False + assert self.validator('/dev/foo') + + @patch('ceph_volume.util.arg_validators.Device') + @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False) + @patch('ceph_volume.api.lvm.get_single_lv', return_value=None) + def test_device_is_not_partition(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call): + mocked_device.return_value = MagicMock( + used_by_ceph=False, + exists=True, + has_partitions=False, + has_gpt_headers=False, + has_fs=False, + is_partition=False + ) + self.validator.zap = False + assert self.validator('/dev/foo') 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 000000000..8eef3ff00 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py @@ -0,0 +1,704 @@ +import os +import pytest +from copy import deepcopy +from ceph_volume.util import device +from ceph_volume.api import lvm as api +from mock.mock import patch, mock_open + + +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", "NAME": "sda"} + 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", "NAME": "sda"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.lvm_size.gb == 4 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lvm_size_rounds_down(self, fake_call, device_info): + # 5.5GB in size + data = {"/dev/sda": {"size": "5905580032"}} + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.lvm_size.gb == 4 + + def test_is_lv(self, fake_call, device_info): + data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"} + lsblk = {"TYPE": "lvm", "NAME": "vg-lv"} + device_info(lv=data,lsblk=lsblk) + disk = device.Device("vg/lv") + assert disk.is_lv + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_vgs_is_empty(self, fake_call, device_info, monkeypatch): + BarPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", + pv_tags={}) + pvolumes = [] + pvolumes.append(BarPVolume) + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: {}) + + disk = device.Device("/dev/nvme0n1") + assert disk.vgs == [] + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_vgs_is_not_empty(self, fake_call, device_info, monkeypatch): + vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=6, + vg_extent_size=1073741824) + monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg]) + lsblk = {"TYPE": "disk", "NAME": "nvme0n1"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/nvme0n1") + assert len(disk.vgs) == 1 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_device(self, fake_call, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "device", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_device is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_loop_device_is_not_device(self, fake_call, device_info): + data = {"/dev/loop0": {"foo": "bar"}} + lsblk = {"TYPE": "loop"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/loop0") + assert disk.is_device is False + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_loop_device_is_device(self, fake_call, device_info): + data = {"/dev/loop0": {"foo": "bar"}} + lsblk = {"TYPE": "loop"} + os.environ["CEPH_VOLUME_ALLOW_LOOP_DEVICES"] = "1" + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/loop0") + assert disk.is_device is True + del os.environ["CEPH_VOLUME_ALLOW_LOOP_DEVICES"] + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_rotational(self, fake_call, device_info): + data = {"/dev/sda": {"rotational": "1"}} + lsblk = {"TYPE": "device", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_not_rotational(self, fake_call, device_info): + data = {"/dev/sda": {"rotational": "0"}} + lsblk = {"TYPE": "device", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.rotational + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_rotational_lsblk(self, fake_call, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "device", "ROTA": "1", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_not_rotational_lsblk(self, fake_call, device_info): + data = {"/dev/sda": {"rotational": "0"}} + lsblk = {"TYPE": "device", "ROTA": "0", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.rotational + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_device_is_rotational_defaults_true(self, fake_call, 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", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.rotational + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_disk_is_device(self, fake_call, device_info): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_device is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_partition(self, fake_call, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert disk.is_partition + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_mpath_device_is_device(self, fake_call, device_info): + data = {"/dev/foo": {"foo": "bar"}} + lsblk = {"TYPE": "mpath", "NAME": "foo"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/foo") + assert disk.is_device is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_not_lvm_member(self, fake_call, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert not disk.is_lvm_member + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_lvm_member(self, fake_call, device_info): + data = {"/dev/sda1": {"foo": "bar"}} + lsblk = {"TYPE": "part", "NAME": "sda1", "PKNAME": "sda"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/sda1") + assert not disk.is_lvm_member + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_mapper_device(self, fake_call, device_info): + lsblk = {"TYPE": "lvm", "NAME": "foo"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/mapper/foo") + assert disk.is_mapper + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_dm_is_mapper_device(self, fake_call, device_info): + lsblk = {"TYPE": "lvm", "NAME": "dm-4"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/dm-4") + assert disk.is_mapper + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_not_mapper_device(self, fake_call, device_info): + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.is_mapper + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_ceph_disk_lsblk(self, fake_call, monkeypatch, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member + + @pytest.mark.usefixtures("blkid_ceph_disk_member", + "lsblk_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_ceph_disk_blkid(self, fake_call, 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") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_ceph_disk_member_not_available_lsblk(self, fake_call, 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", + "lsblk_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_ceph_disk_member_not_available_blkid(self, fake_call, 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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_removable_device(self, fake_call, device_info): + data = {"/dev/sdb": {"removable": 1}} + lsblk = {"TYPE": "disk", "NAME": "sdb"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sdb") + assert not disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_device_with_gpt_headers(self, fake_call, device_info): + data = {"/dev/sdb": {"removable": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk", "NAME": "sdb"} + blkid= {"PTTYPE": "gpt"} + device_info( + devices=data, + blkid=blkid, + lsblk=lsblk, + ) + disk = device.Device("/dev/sdb") + assert not disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_accept_non_removable_device(self, fake_call, device_info): + data = {"/dev/sdb": {"removable": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk", "NAME": "sdb"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sdb") + assert disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_not_acceptable_device(self, fake_call, device_info): + data = {"/dev/dm-0": {"foo": "bar"}} + lsblk = {"TYPE": "mpath", "NAME": "dm-0"} + device_info(devices=data, lsblk=lsblk) + disk = device.Device("/dev/dm-0") + assert not disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + @patch('ceph_volume.util.device.os.path.realpath') + @patch('ceph_volume.util.device.os.path.islink') + def test_accept_symlink_to_device(self, + m_os_path_islink, + m_os_path_realpath, + device_info, + fake_call): + m_os_path_islink.return_value = True + m_os_path_realpath.return_value = '/dev/sdb' + data = {"/dev/sdb": {"ro": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/test_symlink") + print(disk) + print(disk.sys_api) + assert disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + @patch('ceph_volume.util.device.os.readlink') + @patch('ceph_volume.util.device.os.path.islink') + def test_reject_symlink_to_device_mapper(self, + m_os_path_islink, + m_os_readlink, + device_info, + fake_call): + m_os_path_islink.return_value = True + m_os_readlink.return_value = '/dev/dm-0' + data = {"/dev/mapper/mpatha": {"ro": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/mapper/mpatha") + assert disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_readonly_device(self, fake_call, device_info): + data = {"/dev/cdrom": {"ro": 1}} + lsblk = {"TYPE": "disk", "NAME": "cdrom"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/cdrom") + assert not disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_smaller_than_5gb(self, fake_call, device_info): + data = {"/dev/sda": {"size": 5368709119}} + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.available, 'too small device is available' + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_accept_non_readonly_device(self, fake_call, device_info): + data = {"/dev/sda": {"ro": 0, "size": 5368709120}} + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(devices=data,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.available + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_bluestore_device(self, fake_call, monkeypatch, patch_bluestore_label, device_info): + patch_bluestore_label.return_value = True + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.available + assert "Has BlueStore device label" in disk.rejected_reasons + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_reject_device_with_oserror(self, fake_call, monkeypatch, patch_bluestore_label, device_info): + patch_bluestore_label.side_effect = OSError('test failure') + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert not disk.available + assert "Failed to determine if device is BlueStore" in disk.rejected_reasons + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "device_info_not_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_not_ceph_disk_member_lsblk(self, fake_call, patch_bluestore_label): + disk = device.Device("/dev/sda") + assert disk.is_ceph_disk_member is False + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_existing_vg_available(self, fake_call, monkeypatch, device_info): + vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=1536, + vg_extent_size=4194304) + monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg]) + lsblk = {"TYPE": "disk", "NAME": "nvme0n1"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + lv = {"tags": {"ceph.osd_id": "1"}} + device_info(devices=data, lsblk=lsblk, lv=lv) + disk = device.Device("/dev/nvme0n1") + assert disk.available_lvm + assert not disk.available + assert not disk.available_raw + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_existing_vg_too_small(self, fake_call, monkeypatch, device_info): + vg = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=4, + vg_extent_size=1073741824) + monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg]) + lsblk = {"TYPE": "disk", "NAME": "nvme0n1"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + lv = {"tags": {"ceph.osd_id": "1"}} + device_info(devices=data, lsblk=lsblk, lv=lv) + disk = device.Device("/dev/nvme0n1") + assert not disk.available_lvm + assert not disk.available + assert not disk.available_raw + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_multiple_existing_vgs(self, fake_call, monkeypatch, device_info): + vg1 = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=1000, + vg_extent_size=4194304) + vg2 = api.VolumeGroup(pv_name='/dev/nvme0n1', vg_name='foo/bar', vg_free_count=536, + vg_extent_size=4194304) + monkeypatch.setattr(api, 'get_all_devices_vgs', lambda : [vg1, vg2]) + lsblk = {"TYPE": "disk", "NAME": "nvme0n1"} + data = {"/dev/nvme0n1": {"size": "6442450944"}} + lv = {"tags": {"ceph.osd_id": "1"}} + device_info(devices=data, lsblk=lsblk, lv=lv) + 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, fake_call, device_info, + monkeypatch, ceph_type): + data = {"/dev/sda": {"foo": "bar"}} + lsblk = {"TYPE": "part", "NAME": "sda", "PKNAME": "sda"} + 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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_not_used_by_ceph(self, fake_call, 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", "NAME": "sda", "PKNAME": "sda"} + 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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_get_device_id(self, fake_call, device_info): + udev = {k:k for k in ['ID_VENDOR', 'ID_MODEL', 'ID_SCSI_SERIAL']} + lsblk = {"TYPE": "disk", "NAME": "sda"} + device_info(udevadm=udev,lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk._get_device_id() == 'ID_VENDOR_ID_MODEL_ID_SCSI_SERIAL' + + def test_has_bluestore_label(self): + # patch device.Device __init__ function to do nothing since we want to only test the + # low-level behavior of has_bluestore_label + with patch.object(device.Device, "__init__", lambda self, path, with_lsm=False: None): + disk = device.Device("/dev/sda") + disk.path = "/dev/sda" + with patch('builtins.open', mock_open(read_data=b'bluestore block device\n')): + assert disk.has_bluestore_label + with patch('builtins.open', mock_open(read_data=b'not a bluestore block device\n')): + assert not disk.has_bluestore_label + + +class TestDeviceEncryption(object): + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partition_is_not_encrypted_lsblk(self, fake_call, device_info): + lsblk = {'TYPE': 'part', 'FSTYPE': 'xfs', 'NAME': 'sda', 'PKNAME': 'sda'} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is False + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partition_is_encrypted_lsblk(self, fake_call, device_info): + lsblk = {'TYPE': 'part', 'FSTYPE': 'crypto_LUKS', 'NAME': 'sda', 'PKNAME': 'sda'} + device_info(lsblk=lsblk) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partition_is_not_encrypted_blkid(self, fake_call, device_info): + lsblk = {'TYPE': 'part', 'NAME': 'sda', 'PKNAME': 'sda'} + blkid = {'TYPE': 'ceph data'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is False + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partition_is_encrypted_blkid(self, fake_call, device_info): + lsblk = {'TYPE': 'part', 'NAME': 'sda' ,'PKNAME': 'sda'} + blkid = {'TYPE': 'crypto_LUKS'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_mapper_is_encrypted_luks1(self, fake_call, device_info, monkeypatch): + status = {'type': 'LUKS1'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid','TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_mapper_is_encrypted_luks2(self, fake_call, device_info, monkeypatch): + status = {'type': 'LUKS2'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_mapper_is_encrypted_plain(self, fake_call, device_info, monkeypatch): + status = {'type': 'PLAIN'} + monkeypatch.setattr(device, 'encryption_status', lambda x: status) + lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_mapper_is_not_encrypted_plain(self, fake_call, device_info, monkeypatch): + monkeypatch.setattr(device, 'encryption_status', lambda x: {}) + lsblk = {'FSTYPE': 'xfs', 'NAME': 'uuid', 'TYPE': 'lvm'} + blkid = {'TYPE': 'mapper'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/mapper/uuid") + assert disk.is_encrypted is False + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_encrypted_blkid(self, fake_call, device_info): + lsblk = {'TYPE': 'lvm', 'NAME': 'sda'} + blkid = {'TYPE': 'crypto_LUKS'} + device_info(lsblk=lsblk, blkid=blkid) + disk = device.Device("/dev/sda") + disk.lv_api = {} + assert disk.is_encrypted is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_not_encrypted_blkid(self, fake_call, factory, device_info): + lsblk = {'TYPE': 'lvm', 'NAME': 'sda'} + 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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_encrypted_lsblk(self, fake_call, device_info): + lsblk = {'FSTYPE': 'crypto_LUKS', 'NAME': 'sda', '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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_not_encrypted_lsblk(self, fake_call, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', '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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_encrypted_lvm_api(self, fake_call, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', '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 + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_lv_is_not_encrypted_lvm_api(self, fake_call, factory, device_info): + lsblk = {'FSTYPE': 'xfs', 'NAME': 'sda', '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 + } + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_valid_before_invalid(self, fake_call, device_info): + lsblk_sda = {"NAME": "sda", "TYPE": "disk"} + lsblk_sdb = {"NAME": "sdb", "TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk_sda) + sda = device.Device("/dev/sda") + device_info(devices=self.data,lsblk=lsblk_sdb) + sdb = device.Device("/dev/sdb") + + assert sda < sdb + assert sdb > sda + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_valid_alphabetical_ordering(self, fake_call, device_info): + lsblk_sda = {"NAME": "sda", "TYPE": "disk"} + lsblk_sdc = {"NAME": "sdc", "TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk_sda) + sda = device.Device("/dev/sda") + device_info(devices=self.data,lsblk=lsblk_sdc) + sdc = device.Device("/dev/sdc") + + assert sda < sdc + assert sdc > sda + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_invalid_alphabetical_ordering(self, fake_call, device_info): + lsblk_sdb = {"NAME": "sdb", "TYPE": "disk"} + lsblk_sdd = {"NAME": "sdd", "TYPE": "disk"} + device_info(devices=self.data,lsblk=lsblk_sdb) + sdb = device.Device("/dev/sdb") + device_info(devices=self.data,lsblk=lsblk_sdd) + sdd = device.Device("/dev/sdd") + + assert sdb < sdd + assert sdd > sdb + + +class TestCephDiskDevice(object): + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partlabel_lsblk(self, fake_call, device_info): + lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": ""} + device_info(lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.partlabel == '' + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_partlabel_blkid(self, fake_call, device_info): + lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": "ceph data"} + blkid = {"TYPE": "disk", "PARTLABEL": "ceph data"} + device_info(blkid=blkid, lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.partlabel == 'ceph data' + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "blkid_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_member_blkid(self, fake_call, monkeypatch): + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.is_member is True + + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "disable_kernel_queries") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_is_member_lsblk(self, fake_call, patch_bluestore_label, device_info): + lsblk = {"TYPE": "disk", "NAME": "sda", "PARTLABEL": "ceph"} + device_info(lsblk=lsblk) + disk = device.CephDiskDevice(device.Device("/dev/sda")) + + assert disk.is_member is True + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_unknown_type(self, fake_call, device_info): + lsblk = {"TYPE": "disk", "NAME": "sda", "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'] + + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + @pytest.mark.usefixtures("lsblk_ceph_disk_member", + "blkid_ceph_disk_member", + "disable_kernel_queries") + def test_type_blkid(self, monkeypatch, fake_call, 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") + @patch("ceph_volume.util.disk.has_bluestore_label", lambda x: False) + def test_type_lsblk(self, fake_call, 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 000000000..fcd644a86 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_disk.py @@ -0,0 +1,524 @@ +import os +import pytest +from ceph_volume.util import disk +from mock.mock import patch + + +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' + + def test_petabytes(self): + result = disk.human_readable_size(9.23*1024*1024*1024*1024*1024) + assert result == '9.23 PB' + +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_terabytes(self): + result = disk.size_from_human_readable('2 T') + assert result == disk.Size(tb=2) + + def test_petabytes(self): + result = disk.size_from_human_readable('2 P') + assert result == disk.Size(pb=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_terabytes(self): + result = disk.Size.parse('2T') + assert result == disk.Size(tb=2) + + def test_petabytes(self): + result = disk.Size.parse('2P') + assert result == disk.Size(pb=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 TestGetDevices(object): + + def test_no_devices_are_found(self, tmpdir, patched_get_block_devs_sysfs): + patched_get_block_devs_sysfs.return_value = [] + result = disk.get_devices(_sys_block_path=str(tmpdir)) + assert result == {} + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_sda_block_is_found(self, patched_get_block_devs_sysfs, fake_filesystem): + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + result = disk.get_devices() + 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'] == {} + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_sda_size(self, patched_get_block_devs_sysfs, fake_filesystem): + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + fake_filesystem.create_file('/sys/block/sda/size', contents = '1024') + result = disk.get_devices() + assert list(result.keys()) == [sda_path] + assert result[sda_path]['human_readable_size'] == '512.00 KB' + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_sda_sectorsize_fallsback(self, patched_get_block_devs_sysfs, fake_filesystem): + # if no sectorsize, it will use queue/hw_sector_size + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + fake_filesystem.create_file('/sys/block/sda/queue/hw_sector_size', contents = '1024') + result = disk.get_devices() + assert list(result.keys()) == [sda_path] + assert result[sda_path]['sectorsize'] == '1024' + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_sda_sectorsize_from_logical_block(self, patched_get_block_devs_sysfs, fake_filesystem): + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + fake_filesystem.create_file('/sys/block/sda/queue/logical_block_size', contents = '99') + result = disk.get_devices() + assert result[sda_path]['sectorsize'] == '99' + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_sda_sectorsize_does_not_fallback(self, patched_get_block_devs_sysfs, fake_filesystem): + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + fake_filesystem.create_file('/sys/block/sda/queue/logical_block_size', contents = '99') + fake_filesystem.create_file('/sys/block/sda/queue/hw_sector_size', contents = '1024') + result = disk.get_devices() + assert result[sda_path]['sectorsize'] == '99' + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_is_rotational(self, patched_get_block_devs_sysfs, fake_filesystem): + sda_path = '/dev/sda' + patched_get_block_devs_sysfs.return_value = [[sda_path, sda_path, 'disk']] + fake_filesystem.create_file('/sys/block/sda/queue/rotational', contents = '1') + result = disk.get_devices() + assert result[sda_path]['rotational'] == '1' + + @patch('ceph_volume.util.disk.is_locked_raw_device', lambda x: False) + def test_is_ceph_rbd(self, patched_get_block_devs_sysfs, fake_filesystem): + rbd_path = '/dev/rbd0' + patched_get_block_devs_sysfs.return_value = [[rbd_path, rbd_path, 'disk']] + result = disk.get_devices() + assert rbd_path not in result + + +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" + + +class TestAllowLoopDevsWarning(object): + def test_loop_dev_warning(self, fake_call, caplog): + assert disk.allow_loop_devices() is False + assert not caplog.records + os.environ['CEPH_VOLUME_ALLOW_LOOP_DEVICES'] = "y" + assert disk.allow_loop_devices() is True + log = caplog.records[0] + assert log.levelname == "WARNING" + assert "will never be supported in production" in log.message + + +class TestHasBlueStoreLabel(object): + def test_device_path_is_a_path(self, fake_filesystem): + device_path = '/var/lib/ceph/osd/ceph-0' + fake_filesystem.create_dir(device_path) + assert not disk.has_bluestore_label(device_path) \ No newline at end of file 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 000000000..cd2ea8f18 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_encryption.py @@ -0,0 +1,138 @@ +from ceph_volume.util import encryption +from mock.mock import patch +import base64 + +class TestGetKeySize(object): + def test_get_size_from_conf_default(self, conf_ceph_stub): + conf_ceph_stub(''' + [global] + fsid=asdf + ''') + assert encryption.get_key_size_from_conf() == '512' + + def test_get_size_from_conf_custom(self, conf_ceph_stub): + conf_ceph_stub(''' + [global] + fsid=asdf + [osd] + osd_dmcrypt_key_size=256 + ''') + assert encryption.get_key_size_from_conf() == '256' + + def test_get_size_from_conf_custom_invalid(self, conf_ceph_stub): + conf_ceph_stub(''' + [global] + fsid=asdf + [osd] + osd_dmcrypt_key_size=1024 + ''') + assert encryption.get_key_size_from_conf() == '512' + +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, fake_filesystem): + file_name = fake_filesystem.create_file('mapper-device') + encryption.dmcrypt_close(file_name.path) + 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(self): + result = encryption.create_dmcrypt_key() + assert len(base64.b64decode(result)) == 128 + +class TestLuksFormat(object): + @patch('ceph_volume.util.encryption.process.call') + def test_luks_format_command_with_default_size(self, m_call, conf_ceph_stub): + conf_ceph_stub('[global]\nfsid=abcd') + expected = [ + 'cryptsetup', + '--batch-mode', + '--key-size', + '512', + '--key-file', + '-', + 'luksFormat', + '/dev/foo' + ] + encryption.luks_format('abcd', '/dev/foo') + assert m_call.call_args[0][0] == expected + + @patch('ceph_volume.util.encryption.process.call') + def test_luks_format_command_with_custom_size(self, m_call, conf_ceph_stub): + conf_ceph_stub('[global]\nfsid=abcd\n[osd]\nosd_dmcrypt_key_size=256') + expected = [ + 'cryptsetup', + '--batch-mode', + '--key-size', + '256', + '--key-file', + '-', + 'luksFormat', + '/dev/foo' + ] + encryption.luks_format('abcd', '/dev/foo') + assert m_call.call_args[0][0] == expected + + +class TestLuksOpen(object): + @patch('ceph_volume.util.encryption.process.call') + def test_luks_open_command_with_default_size(self, m_call, conf_ceph_stub): + conf_ceph_stub('[global]\nfsid=abcd') + expected = [ + 'cryptsetup', + '--key-size', + '512', + '--key-file', + '-', + '--allow-discards', + 'luksOpen', + '/dev/foo', + '/dev/bar' + ] + encryption.luks_open('abcd', '/dev/foo', '/dev/bar') + assert m_call.call_args[0][0] == expected + + @patch('ceph_volume.util.encryption.process.call') + def test_luks_open_command_with_custom_size(self, m_call, conf_ceph_stub): + conf_ceph_stub('[global]\nfsid=abcd\n[osd]\nosd_dmcrypt_key_size=256') + expected = [ + 'cryptsetup', + '--key-size', + '256', + '--key-file', + '-', + '--allow-discards', + 'luksOpen', + '/dev/foo', + '/dev/bar' + ] + encryption.luks_open('abcd', '/dev/foo', '/dev/bar') + assert m_call.call_args[0][0] == expected 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 000000000..080823307 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_prepare.py @@ -0,0 +1,413 @@ +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 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 000000000..5746f7023 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_system.py @@ -0,0 +1,309 @@ +import os +import pwd +import getpass +import pytest +from textwrap import dedent +from ceph_volume.util import system +from mock.mock import patch +from ceph_volume.tests.conftest import Factory + + +@pytest.fixture +def mock_find_executable_on_host(monkeypatch): + """ + Monkeypatches util.system.find_executable_on_host, so that a caller can add behavior to the response + """ + def apply(stdout=None, stderr=None, returncode=0): + stdout_stream = Factory(read=lambda: stdout) + stderr_stream = Factory(read=lambda: stderr) + return_value = Factory( + stdout=stdout_stream, + stderr=stderr_stream, + wait=lambda: returncode, + communicate=lambda x: (stdout, stderr, returncode) + ) + + monkeypatch.setattr( + 'ceph_volume.util.system.subprocess.Popen', + lambda *a, **kw: return_value) + + return apply + +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) + m = system.Mounts() + assert m.get_mounts() == {} + + def test_is_mounted_(self, fake_proc): + m = system.Mounts() + assert m.get_mounts()['/dev/sdc2'] == ['/boot'] + + def test_ignores_two_fields(self, fake_proc): + m = system.Mounts() + assert m.get_mounts().get('/dev/sde4') is None + + def test_tmpfs_is_reported(self, fake_proc): + m = system.Mounts() + assert m.get_mounts()['tmpfs'][0] == '/dev/shm' + + def test_non_skip_devs_arent_reported(self, fake_proc): + m = system.Mounts() + assert m.get_mounts().get('cgroup') is None + + def test_multiple_mounts_are_appended(self, fake_proc): + m = system.Mounts() + assert len(m.get_mounts()['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) + m = system.Mounts() + assert m.get_mounts().get('/dev/sda1') is None + + +class TestIsBinary(object): + + def test_is_binary(self, fake_filesystem): + binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh\x00') + assert system.is_binary(binary_path.path) + + def test_is_not_binary(self, fake_filesystem): + binary_path = fake_filesystem.create_file('/tmp/fake-file', contents='asd\n\nlkjh0') + assert system.is_binary(binary_path.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, fake_filesystem): + interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="1") + result = system.get_file_contents(interesting_file.path) + assert result == "1" + + def test_path_has_multiline_contents(self, fake_filesystem): + interesting_file = fake_filesystem.create_file('/tmp/fake-file', contents="0\n1") + result = system.get_file_contents(interesting_file.path) + assert result == "0\n1" + + def test_exception_returns_default(self): + with patch('builtins.open') as mocked_open: + mocked_open.side_effect = Exception() + result = system.get_file_contents('/tmp/fake-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 + + def test_run_on_host_found(self, mock_find_executable_on_host): + mock_find_executable_on_host(stdout="/sbin/lvs\n", stderr="some stderr message\n") + assert system.which('lvs', run_on_host=True) == '/sbin/lvs' + + def test_run_on_host_not_found(self, mock_find_executable_on_host): + mock_find_executable_on_host(stdout="", stderr="some stderr message\n") + assert system.which('lvs', run_on_host=True) == 'lvs' + +@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 000000000..1a094d33f --- /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 -- cgit v1.2.3