ceph/src/ceph-volume/ceph_volume/tests/util/test_arg_validators.py

   1 import argparse
   2 import pytest
   3 import os
   4 from ceph_volume import exceptions, process
   5 from ceph_volume.util import arg_validators
   6 from mock.mock import patch, MagicMock
   7
   8
   9 class TestOSDPath(object):
  10
  11     def setup(self):
  12         self.validator = arg_validators.OSDPath()
  13
  14     def test_is_not_root(self, monkeypatch):
  15         monkeypatch.setattr(os, 'getuid', lambda: 100)
  16         with pytest.raises(exceptions.SuperUserError):
  17             self.validator('')
  18
  19     def test_path_is_not_a_directory(self, is_root, tmpfile, monkeypatch):
  20         monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False)
  21         validator = arg_validators.OSDPath()
  22         with pytest.raises(argparse.ArgumentError):
  23             validator(tmpfile())
  24
  25     def test_files_are_missing(self, is_root, tmpdir, monkeypatch):
  26         tmppath = str(tmpdir)
  27         monkeypatch.setattr(arg_validators.disk, 'is_partition', lambda x: False)
  28         validator = arg_validators.OSDPath()
  29         with pytest.raises(argparse.ArgumentError) as error:
  30             validator(tmppath)
  31         assert 'Required file (ceph_fsid) was not found in OSD' in str(error.value)
  32
  33
  34 class TestExcludeGroupOptions(object):
  35
  36     def setup(self):
  37         self.parser = argparse.ArgumentParser()
  38
  39     def test_flags_in_one_group(self):
  40         argv = ['<prog>', '--filestore', '--bar']
  41         filestore_group = self.parser.add_argument_group('filestore')
  42         bluestore_group = self.parser.add_argument_group('bluestore')
  43         filestore_group.add_argument('--filestore')
  44         bluestore_group.add_argument('--bluestore')
  45         result = arg_validators.exclude_group_options(
  46             self.parser,
  47             ['filestore', 'bluestore'],
  48             argv=argv
  49         )
  50         assert result is None
  51
  52     def test_flags_in_no_group(self):
  53         argv = ['<prog>', '--foo', '--bar']
  54         filestore_group = self.parser.add_argument_group('filestore')
  55         bluestore_group = self.parser.add_argument_group('bluestore')
  56         filestore_group.add_argument('--filestore')
  57         bluestore_group.add_argument('--bluestore')
  58         result = arg_validators.exclude_group_options(
  59             self.parser,
  60             ['filestore', 'bluestore'],
  61             argv=argv
  62         )
  63         assert result is None
  64
  65     def test_flags_conflict(self, capsys):
  66         argv = ['<prog>', '--filestore', '--bluestore']
  67         filestore_group = self.parser.add_argument_group('filestore')
  68         bluestore_group = self.parser.add_argument_group('bluestore')
  69         filestore_group.add_argument('--filestore')
  70         bluestore_group.add_argument('--bluestore')
  71
  72         arg_validators.exclude_group_options(
  73             self.parser, ['filestore', 'bluestore'], argv=argv
  74         )
  75         stdout, stderr = capsys.readouterr()
  76         assert 'Cannot use --filestore (filestore) with --bluestore (bluestore)' in stderr
  77
  78
  79 class TestValidDevice(object):
  80
  81     def setup(self):
  82         self.validator = arg_validators.ValidDevice()
  83
  84     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
  85     def test_path_is_valid(self, m_has_bs_label, fake_call, patch_bluestore_label):
  86         result = self.validator('/')
  87         assert result.abspath == '/'
  88
  89     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
  90     def test_path_is_invalid(self, m_has_bs_label, fake_call, patch_bluestore_label):
  91         with pytest.raises(argparse.ArgumentError):
  92             self.validator('/device/does/not/exist')
  93
  94     @patch('ceph_volume.util.arg_validators.Device')
  95     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
  96     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
  97     def test_dev_has_partitions(self, m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
  98         mocked_device.return_value = MagicMock(
  99             exists=True,
 100             has_partitions=True,
 101         )
 102         with pytest.raises(RuntimeError):
 103             self.validator('/dev/foo')
 104
 105 class TestValidZapDevice(object):
 106     def setup(self):
 107         self.validator = arg_validators.ValidZapDevice()
 108
 109     @patch('ceph_volume.util.arg_validators.Device')
 110     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 111     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 112     def test_device_has_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device):
 113         mocked_device.return_value = MagicMock(
 114             used_by_ceph=False,
 115             exists=True,
 116             has_partitions=True,
 117             has_gpt_headers=False,
 118             has_fs=False
 119         )
 120         self.validator.zap = False
 121         with pytest.raises(RuntimeError):
 122             assert self.validator('/dev/foo')
 123
 124     @patch('ceph_volume.util.arg_validators.Device')
 125     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 126     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 127     def test_device_has_no_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device):
 128         mocked_device.return_value = MagicMock(
 129             used_by_ceph=False,
 130             exists=True,
 131             has_partitions=False,
 132             has_gpt_headers=False,
 133             has_fs=False
 134         )
 135         self.validator.zap = False
 136         assert self.validator('/dev/foo')
 137
 138 class TestValidDataDevice(object):
 139     def setup(self):
 140         self.validator = arg_validators.ValidDataDevice()
 141
 142     @patch('ceph_volume.util.arg_validators.Device')
 143     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 144     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 145     def test_device_used_by_ceph(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 146         mocked_device.return_value = MagicMock(
 147             used_by_ceph=True,
 148             exists=True,
 149             has_partitions=False,
 150             has_gpt_headers=False
 151         )
 152         with pytest.raises(SystemExit):
 153             self.validator.zap = False
 154             self.validator('/dev/foo')
 155
 156     @patch('ceph_volume.util.arg_validators.Device')
 157     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 158     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 159     def test_device_has_fs(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 160         mocked_device.return_value = MagicMock(
 161             used_by_ceph=False,
 162             exists=True,
 163             has_partitions=False,
 164             has_gpt_headers=False,
 165             has_fs=True
 166         )
 167         with pytest.raises(RuntimeError):
 168             self.validator.zap = False
 169             self.validator('/dev/foo')
 170
 171     @patch('ceph_volume.util.arg_validators.Device')
 172     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=True)
 173     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 174     def test_device_has_bs_signature(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 175         mocked_device.return_value = MagicMock(
 176             used_by_ceph=False,
 177             exists=True,
 178             has_partitions=False,
 179             has_gpt_headers=False,
 180             has_fs=False
 181         )
 182         with pytest.raises(RuntimeError):
 183             self.validator.zap = False
 184             self.validator('/dev/foo')
 185
 186 class TestValidRawDevice(object):
 187     def setup(self):
 188         self.validator = arg_validators.ValidRawDevice()
 189
 190     @patch('ceph_volume.util.arg_validators.Device')
 191     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 192     @patch('ceph_volume.util.arg_validators.disk.blkid')
 193     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 194     def test_dmcrypt_device_already_prepared(self,  m_get_single_lv, m_blkid, m_has_bs_label, mocked_device, fake_call, monkeypatch):
 195         def mock_call(cmd, **kw):
 196             return ('', '', 1)
 197         monkeypatch.setattr(process, 'call', mock_call)
 198         m_blkid.return_value = {'UUID': '8fd92779-ad78-437c-a06f-275f7170fa74', 'TYPE': 'crypto_LUKS'}
 199         mocked_device.return_value = MagicMock(
 200             used_by_ceph=False,
 201             exists=True,
 202             has_partitions=False,
 203             has_gpt_headers=False,
 204             has_fs=False
 205         )
 206         with pytest.raises(SystemExit):
 207             self.validator.zap = False
 208             self.validator('/dev/foo')
 209
 210     @patch('ceph_volume.util.arg_validators.Device')
 211     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 212     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 213     def test_device_already_prepared(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 214         mocked_device.return_value = MagicMock(
 215             used_by_ceph=False,
 216             exists=True,
 217             has_partitions=False,
 218             has_gpt_headers=False,
 219             has_fs=False
 220         )
 221         with pytest.raises(SystemExit):
 222             self.validator.zap = False
 223             self.validator('/dev/foo')
 224
 225     @patch('ceph_volume.util.arg_validators.Device')
 226     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 227     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 228     def test_device_not_prepared(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch):
 229         def mock_call(cmd, **kw):
 230             return ('', '', 1)
 231         monkeypatch.setattr(process, 'call', mock_call)
 232         mocked_device.return_value = MagicMock(
 233             used_by_ceph=False,
 234             exists=True,
 235             has_partitions=False,
 236             has_gpt_headers=False,
 237             has_fs=False
 238         )
 239         self.validator.zap = False
 240         assert self.validator('/dev/foo')
 241
 242     @patch('ceph_volume.util.arg_validators.Device')
 243     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 244     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 245     def test_device_has_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call, monkeypatch):
 246         def mock_call(cmd, **kw):
 247             return ('', '', 1)
 248         monkeypatch.setattr(process, 'call', mock_call)
 249         mocked_device.return_value = MagicMock(
 250             used_by_ceph=False,
 251             exists=True,
 252             has_partitions=True,
 253             has_gpt_headers=False,
 254             has_fs=False
 255         )
 256         self.validator.zap = False
 257         with pytest.raises(RuntimeError):
 258             assert self.validator('/dev/foo')
 259
 260 class TestValidBatchDevice(object):
 261     def setup(self):
 262         self.validator = arg_validators.ValidBatchDevice()
 263
 264     @patch('ceph_volume.util.arg_validators.Device')
 265     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 266     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 267     def test_device_is_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 268         mocked_device.return_value = MagicMock(
 269             used_by_ceph=False,
 270             exists=True,
 271             has_partitions=False,
 272             has_gpt_headers=False,
 273             has_fs=False,
 274             is_partition=True
 275         )
 276         with pytest.raises(argparse.ArgumentError):
 277             self.validator.zap = False
 278             self.validator('/dev/foo')
 279
 280     @patch('ceph_volume.util.arg_validators.Device')
 281     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 282     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 283     def test_device_is_not_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 284         mocked_device.return_value = MagicMock(
 285             used_by_ceph=False,
 286             exists=True,
 287             has_partitions=False,
 288             has_gpt_headers=False,
 289             has_fs=False,
 290             is_partition=False
 291         )
 292         self.validator.zap = False
 293         assert self.validator('/dev/foo')
 294
 295 class TestValidBatchDataDevice(object):
 296     def setup(self):
 297         self.validator = arg_validators.ValidBatchDataDevice()
 298
 299     @patch('ceph_volume.util.arg_validators.Device')
 300     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 301     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 302     def test_device_is_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 303         mocked_device.return_value = MagicMock(
 304             used_by_ceph=False,
 305             exists=True,
 306             has_partitions=False,
 307             has_gpt_headers=False,
 308             has_fs=False,
 309             is_partition=True
 310         )
 311         with pytest.raises(argparse.ArgumentError):
 312             self.validator.zap = False
 313             assert self.validator('/dev/foo')
 314
 315     @patch('ceph_volume.util.arg_validators.Device')
 316     @patch('ceph_volume.util.arg_validators.disk.has_bluestore_label', return_value=False)
 317     @patch('ceph_volume.api.lvm.get_single_lv', return_value=None)
 318     def test_device_is_not_partition(self,  m_get_single_lv, m_has_bs_label, mocked_device, fake_call):
 319         mocked_device.return_value = MagicMock(
 320             used_by_ceph=False,
 321             exists=True,
 322             has_partitions=False,
 323             has_gpt_headers=False,
 324             has_fs=False,
 325             is_partition=False
 326         )
 327         self.validator.zap = False
 328         assert self.validator('/dev/foo')
 329
 330
 331 class TestValidFraction(object):
 332
 333     def setup(self):
 334         self.validator = arg_validators.ValidFraction()
 335
 336     def test_fraction_is_valid(self, fake_call):
 337         result = self.validator('0.8')
 338         assert result == 0.8
 339
 340     def test_fraction_not_float(self, fake_call):
 341         with pytest.raises(ValueError):
 342             self.validator('xyz')
 343
 344     def test_fraction_is_nan(self, fake_call):
 345         with pytest.raises(argparse.ArgumentError):
 346             self.validator('NaN')
 347
 348     def test_fraction_is_negative(self, fake_call):
 349         with pytest.raises(argparse.ArgumentError):
 350             self.validator('-1.0')
 351
 352     def test_fraction_is_zero(self, fake_call):
 353         with pytest.raises(argparse.ArgumentError):
 354             self.validator('0.0')
 355
 356     def test_fraction_is_greater_one(self, fake_call):
 357         with pytest.raises(argparse.ArgumentError):
 358             self.validator('1.1')