diff options
Diffstat (limited to 'src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py')
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py | 368 |
1 files changed, 368 insertions, 0 deletions
diff --git a/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py new file mode 100644 index 000000000..c6349308e --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py @@ -0,0 +1,368 @@ +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_method(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_method(self): + self.parser = argparse.ArgumentParser() + + def test_flags_in_one_group(self): + argv = ['<prog>', '--filestore', '--bar'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + result = arg_validators.exclude_group_options( + self.parser, + ['filestore', 'bluestore'], + argv=argv + ) + assert result is None + + def test_flags_in_no_group(self): + argv = ['<prog>', '--foo', '--bar'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + result = arg_validators.exclude_group_options( + self.parser, + ['filestore', 'bluestore'], + argv=argv + ) + assert result is None + + def test_flags_conflict(self, capsys): + argv = ['<prog>', '--filestore', '--bluestore'] + filestore_group = self.parser.add_argument_group('filestore') + bluestore_group = self.parser.add_argument_group('bluestore') + filestore_group.add_argument('--filestore') + bluestore_group.add_argument('--bluestore') + + arg_validators.exclude_group_options( + self.parser, ['filestore', 'bluestore'], argv=argv + ) + stdout, stderr = capsys.readouterr() + assert 'Cannot use --filestore (filestore) with --bluestore (bluestore)' in stderr + + +class TestValidDevice(object): + + def setup_method(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_method(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_method(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_method(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_method(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_method(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') + + +class TestValidFraction(object): + + def setup_method(self): + self.validator = arg_validators.ValidFraction() + + def test_fraction_is_valid(self, fake_call): + result = self.validator('0.8') + assert result == 0.8 + + def test_fraction_not_float(self, fake_call): + with pytest.raises(ValueError): + self.validator('xyz') + + def test_fraction_is_nan(self, fake_call): + with pytest.raises(argparse.ArgumentError): + self.validator('NaN') + + def test_fraction_is_negative(self, fake_call): + with pytest.raises(argparse.ArgumentError): + self.validator('-1.0') + + def test_fraction_is_zero(self, fake_call): + with pytest.raises(argparse.ArgumentError): + self.validator('0.0') + + def test_fraction_is_greater_one(self, fake_call): + with pytest.raises(argparse.ArgumentError): + self.validator('1.1') |