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

   1 import os
   2 import pytest
   3 from ceph_volume.util import disk
   4
   5
   6 class TestLsblkParser(object):
   7
   8     def test_parses_whitespace_values(self):
   9         output = 'NAME="sdaa5" PARTLABEL="ceph data" RM="0" SIZE="10M" RO="0" TYPE="part"'
  10         result = disk._lsblk_parser(output)
  11         assert result['PARTLABEL'] == 'ceph data'
  12
  13     def test_ignores_bogus_pairs(self):
  14         output = 'NAME="sdaa5" PARTLABEL RM="0" SIZE="10M" RO="0" TYPE="part" MOUNTPOINT=""'
  15         result = disk._lsblk_parser(output)
  16         assert result['SIZE'] == '10M'
  17
  18
  19 class TestBlkidParser(object):
  20
  21     def test_parses_whitespace_values(self):
  22         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
  23         result = disk._blkid_parser(output)
  24         assert result['PARTLABEL'] == 'ceph data'
  25
  26     def test_ignores_unmapped(self):
  27         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
  28         result = disk._blkid_parser(output)
  29         assert len(result.keys()) == 4
  30
  31     def test_translates_to_partuuid(self):
  32         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
  33         result = disk._blkid_parser(output)
  34         assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f'
  35
  36
  37 class TestBlkid(object):
  38
  39     def test_parses_translated(self, stub_call):
  40         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
  41         stub_call((output.split(), [], 0))
  42         result = disk.blkid('/dev/sdb1')
  43         assert result['PARTUUID'] == 'b89c03bc-bf58-4338-a8f8-a2f484852b4f'
  44         assert result['PARTLABEL'] == 'ceph data'
  45         assert result['UUID'] == '62416664-cbaf-40bd-9689-10bd337379c3'
  46         assert result['TYPE'] == 'xfs'
  47
  48
  49 class TestDeviceFamily(object):
  50
  51     def test_groups_multiple_devices(self, stub_call):
  52         out = [
  53             'NAME="sdaa5" PARLABEL="ceph lockbox"',
  54             'NAME="sdaa" RO="0"',
  55             'NAME="sdaa1" PARLABEL="ceph data"',
  56             'NAME="sdaa2" PARLABEL="ceph journal"',
  57         ]
  58         stub_call((out, '', 0))
  59         result = disk.device_family('sdaa5')
  60         assert len(result) == 4
  61
  62     def test_parses_output_correctly(self, stub_call):
  63         names = ['sdaa', 'sdaa5', 'sdaa1', 'sdaa2']
  64         out = [
  65             'NAME="sdaa5" PARLABEL="ceph lockbox"',
  66             'NAME="sdaa" RO="0"',
  67             'NAME="sdaa1" PARLABEL="ceph data"',
  68             'NAME="sdaa2" PARLABEL="ceph journal"',
  69         ]
  70         stub_call((out, '', 0))
  71         result = disk.device_family('sdaa5')
  72         for parsed in result:
  73             assert parsed['NAME'] in names
  74
  75
  76 class TestMapDevPaths(object):
  77
  78     def test_errors_return_empty_mapping(self, tmpdir):
  79         bad_dir = os.path.join(str(tmpdir), 'nonexisting')
  80         assert disk._map_dev_paths(bad_dir) == {}
  81
  82     def test_base_name_and_abspath(self, tmpfile):
  83         sda_path = tmpfile(name='sda', contents='')
  84         directory = os.path.dirname(sda_path)
  85         result = disk._map_dev_paths(directory)
  86         assert len(result.keys()) == 1
  87         assert result['sda'] == sda_path
  88
  89     def test_abspath_included(self, tmpfile):
  90         sda_path = tmpfile(name='sda', contents='')
  91         directory = os.path.dirname(sda_path)
  92         result = disk._map_dev_paths(directory, include_abspath=True)
  93         assert sorted(result.keys()) == sorted(['sda', sda_path])
  94         assert result['sda'] == sda_path
  95         assert result[sda_path] == 'sda'
  96
  97     def test_realpath_included(self, tmpfile):
  98         sda_path = tmpfile(name='sda', contents='')
  99         directory = os.path.dirname(sda_path)
 100         dm_path = os.path.join(directory, 'dm-0')
 101         os.symlink(sda_path, os.path.join(directory, 'dm-0'))
 102         result = disk._map_dev_paths(directory, include_realpath=True)
 103         assert sorted(result.keys()) == sorted(['sda', 'dm-0'])
 104         assert result['sda'] == dm_path
 105         assert result['dm-0'] == dm_path
 106
 107     def test_absolute_and_realpath_included(self, tmpfile):
 108         dm_path = tmpfile(name='dm-0', contents='')
 109         directory = os.path.dirname(dm_path)
 110         sda_path = os.path.join(directory, 'sda')
 111         os.symlink(sda_path, os.path.join(directory, 'sda'))
 112         result = disk._map_dev_paths(directory, include_realpath=True, include_abspath=True)
 113         assert sorted(result.keys()) == sorted([dm_path, sda_path, 'sda', 'dm-0'])
 114         assert result['sda'] == sda_path
 115         assert result['dm-0'] == dm_path
 116         assert result[sda_path] == sda_path
 117         assert result[dm_path] == 'dm-0'
 118
 119
 120 class TestGetBlockDevs(object):
 121
 122     def test_loop_devices_are_missing(self, tmpfile):
 123         path = os.path.dirname(tmpfile(name='loop0', contents=''))
 124         result = disk.get_block_devs(sys_block_path=path)
 125         assert result == []
 126
 127     def test_loop_devices_are_included(self, tmpfile):
 128         path = os.path.dirname(tmpfile(name='loop0', contents=''))
 129         result = disk.get_block_devs(sys_block_path=path, skip_loop=False)
 130         assert len(result) == 1
 131         assert result == ['loop0']
 132
 133
 134 class TestGetDevDevs(object):
 135
 136     def test_abspaths_are_included(self, tmpfile):
 137         sda_path = tmpfile(name='sda', contents='')
 138         directory = os.path.dirname(sda_path)
 139         result = disk.get_dev_devs(directory)
 140         assert sorted(result.keys()) == sorted(['sda', sda_path])
 141         assert result['sda'] == sda_path
 142         assert result[sda_path] == 'sda'
 143
 144
 145 class TestGetMapperDevs(object):
 146
 147     def test_abspaths_and_realpaths_are_included(self, tmpfile):
 148         dm_path = tmpfile(name='dm-0', contents='')
 149         directory = os.path.dirname(dm_path)
 150         sda_path = os.path.join(directory, 'sda')
 151         os.symlink(sda_path, os.path.join(directory, 'sda'))
 152         result = disk.get_mapper_devs(directory)
 153         assert sorted(result.keys()) == sorted([dm_path, sda_path, 'sda', 'dm-0'])
 154         assert result['sda'] == sda_path
 155         assert result['dm-0'] == dm_path
 156         assert result[sda_path] == sda_path
 157         assert result[dm_path] == 'dm-0'
 158
 159
 160 class TestHumanReadableSize(object):
 161
 162     def test_bytes(self):
 163         result = disk.human_readable_size(800)
 164         assert result == '800.00 B'
 165
 166     def test_kilobytes(self):
 167         result = disk.human_readable_size(800*1024)
 168         assert result == '800.00 KB'
 169
 170     def test_megabytes(self):
 171         result = disk.human_readable_size(800*1024*1024)
 172         assert result == '800.00 MB'
 173
 174     def test_gigabytes(self):
 175         result = disk.human_readable_size(8.19*1024*1024*1024)
 176         assert result == '8.19 GB'
 177
 178     def test_terabytes(self):
 179         result = disk.human_readable_size(81.2*1024*1024*1024*1024)
 180         assert result == '81.20 TB'
 181
 182
 183 class TestGetDevices(object):
 184
 185     def setup_paths(self, tmpdir):
 186         paths = []
 187         for directory in ['block', 'dev', 'mapper']:
 188             path = os.path.join(str(tmpdir), directory)
 189             paths.append(path)
 190             os.makedirs(path)
 191         return paths
 192
 193     def test_no_devices_are_found(self, tmpdir):
 194         result = disk.get_devices(
 195             _sys_block_path=str(tmpdir),
 196             _dev_path=str(tmpdir),
 197             _mapper_path=str(tmpdir))
 198         assert result == {}
 199
 200     def test_no_devices_are_found_errors(self, tmpdir):
 201         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 202         os.makedirs(os.path.join(block_path, 'sda'))
 203         result = disk.get_devices(
 204             _sys_block_path=block_path, # has 1 device
 205             _dev_path=str(tmpdir), # exists but no devices
 206             _mapper_path='/does/not/exist/path') # does not exist
 207         assert result == {}
 208
 209     def test_sda_block_is_found(self, tmpfile, tmpdir):
 210         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 211         dev_sda_path = os.path.join(dev_path, 'sda')
 212         os.makedirs(os.path.join(block_path, 'sda'))
 213         os.makedirs(dev_sda_path)
 214         result = disk.get_devices(
 215             _sys_block_path=block_path,
 216             _dev_path=dev_path,
 217             _mapper_path=mapper_path)
 218         assert len(result.keys()) == 1
 219         assert result[dev_sda_path]['human_readable_size'] == '0.00 B'
 220         assert result[dev_sda_path]['model'] == ''
 221         assert result[dev_sda_path]['partitions'] == {}
 222
 223
 224     def test_dm_device_is_not_used(self, monkeypatch, tmpdir):
 225         # the link to the mapper is used instead
 226         monkeypatch.setattr(disk.lvm, 'is_lv', lambda: True)
 227         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 228         dev_dm_path = os.path.join(dev_path, 'dm-0')
 229         ceph_data_path = os.path.join(mapper_path, 'ceph-data')
 230         os.symlink(dev_dm_path, ceph_data_path)
 231         block_dm_path = os.path.join(block_path, 'dm-0')
 232         os.makedirs(block_dm_path)
 233
 234         result = disk.get_devices(
 235             _sys_block_path=block_path,
 236             _dev_path=dev_path,
 237             _mapper_path=mapper_path)
 238         result = list(result.keys())
 239         assert len(result) == 1
 240         assert result == [ceph_data_path]
 241
 242     def test_sda_size(self, tmpfile, tmpdir):
 243         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 244         block_sda_path = os.path.join(block_path, 'sda')
 245         dev_sda_path = os.path.join(dev_path, 'sda')
 246         os.makedirs(block_sda_path)
 247         os.makedirs(dev_sda_path)
 248         tmpfile('size', '1024', directory=block_sda_path)
 249         result = disk.get_devices(
 250             _sys_block_path=block_path,
 251             _dev_path=dev_path,
 252             _mapper_path=mapper_path)
 253         assert list(result.keys()) == [dev_sda_path]
 254         assert result[dev_sda_path]['human_readable_size'] == '512.00 KB'
 255
 256     def test_sda_sectorsize_fallsback(self, tmpfile, tmpdir):
 257         # if no sectorsize, it will use queue/hw_sector_size
 258         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 259         block_sda_path = os.path.join(block_path, 'sda')
 260         sda_queue_path = os.path.join(block_sda_path, 'queue')
 261         dev_sda_path = os.path.join(dev_path, 'sda')
 262         os.makedirs(block_sda_path)
 263         os.makedirs(sda_queue_path)
 264         os.makedirs(dev_sda_path)
 265         tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
 266         result = disk.get_devices(
 267             _sys_block_path=block_path,
 268             _dev_path=dev_path,
 269             _mapper_path=mapper_path)
 270         assert list(result.keys()) == [dev_sda_path]
 271         assert result[dev_sda_path]['sectorsize'] == '1024'
 272
 273     def test_sda_sectorsize_from_logical_block(self, tmpfile, tmpdir):
 274         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 275         block_sda_path = os.path.join(block_path, 'sda')
 276         sda_queue_path = os.path.join(block_sda_path, 'queue')
 277         dev_sda_path = os.path.join(dev_path, 'sda')
 278         os.makedirs(block_sda_path)
 279         os.makedirs(sda_queue_path)
 280         os.makedirs(dev_sda_path)
 281         tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
 282         result = disk.get_devices(
 283             _sys_block_path=block_path,
 284             _dev_path=dev_path,
 285             _mapper_path=mapper_path)
 286         assert result[dev_sda_path]['sectorsize'] == '99'
 287
 288     def test_sda_sectorsize_does_not_fallback(self, tmpfile, tmpdir):
 289         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 290         block_sda_path = os.path.join(block_path, 'sda')
 291         sda_queue_path = os.path.join(block_sda_path, 'queue')
 292         dev_sda_path = os.path.join(dev_path, 'sda')
 293         os.makedirs(block_sda_path)
 294         os.makedirs(sda_queue_path)
 295         os.makedirs(dev_sda_path)
 296         tmpfile('logical_block_size', contents='99', directory=sda_queue_path)
 297         tmpfile('hw_sector_size', contents='1024', directory=sda_queue_path)
 298         result = disk.get_devices(
 299             _sys_block_path=block_path,
 300             _dev_path=dev_path,
 301             _mapper_path=mapper_path)
 302         assert result[dev_sda_path]['sectorsize'] == '99'
 303
 304     def test_is_rotational(self, tmpfile, tmpdir):
 305         block_path, dev_path, mapper_path = self.setup_paths(tmpdir)
 306         block_sda_path = os.path.join(block_path, 'sda')
 307         sda_queue_path = os.path.join(block_sda_path, 'queue')
 308         dev_sda_path = os.path.join(dev_path, 'sda')
 309         os.makedirs(block_sda_path)
 310         os.makedirs(sda_queue_path)
 311         os.makedirs(dev_sda_path)
 312         tmpfile('rotational', contents='1', directory=sda_queue_path)
 313         result = disk.get_devices(
 314             _sys_block_path=block_path,
 315             _dev_path=dev_path,
 316             _mapper_path=mapper_path)
 317         assert result[dev_sda_path]['rotational'] == '1'
 318
 319
 320 class TestSizeCalculations(object):
 321
 322     @pytest.mark.parametrize('aliases', [
 323         ('b', 'bytes'),
 324         ('kb', 'kilobytes'),
 325         ('mb', 'megabytes'),
 326         ('gb', 'gigabytes'),
 327         ('tb', 'terabytes'),
 328     ])
 329     def test_aliases(self, aliases):
 330         short_alias, long_alias = aliases
 331         s = disk.Size(b=1)
 332         short_alias = getattr(s, short_alias)
 333         long_alias = getattr(s, long_alias)
 334         assert short_alias == long_alias
 335
 336     @pytest.mark.parametrize('values', [
 337         ('b', 857619069665.28),
 338         ('kb', 837518622.72),
 339         ('mb', 817889.28),
 340         ('gb', 798.72),
 341         ('tb', 0.78),
 342     ])
 343     def test_terabytes(self, values):
 344         # regardless of the input value, all the other values correlate to each
 345         # other the same, every time
 346         unit, value = values
 347         s = disk.Size(**{unit: value})
 348         assert s.b == 857619069665.28
 349         assert s.kb == 837518622.72
 350         assert s.mb == 817889.28
 351         assert s.gb == 798.72
 352         assert s.tb == 0.78
 353
 354
 355 class TestSizeOperators(object):
 356
 357     @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001])
 358     def test_gigabytes_is_smaller(self, larger):
 359         assert disk.Size(gb=1) < disk.Size(mb=larger)
 360
 361     @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001])
 362     def test_gigabytes_is_larger(self, smaller):
 363         assert disk.Size(gb=1) > disk.Size(mb=smaller)
 364
 365     @pytest.mark.parametrize('larger', [1025, 1024.1, 1024.001, 1024])
 366     def test_gigabytes_is_smaller_or_equal(self, larger):
 367         assert disk.Size(gb=1) <= disk.Size(mb=larger)
 368
 369     @pytest.mark.parametrize('smaller', [1023, 1023.9, 1023.001, 1024])
 370     def test_gigabytes_is_larger_or_equal(self, smaller):
 371         assert disk.Size(gb=1) >= disk.Size(mb=smaller)
 372
 373     @pytest.mark.parametrize('values', [
 374         ('b', 857619069665.28),
 375         ('kb', 837518622.72),
 376         ('mb', 817889.28),
 377         ('gb', 798.72),
 378         ('tb', 0.78),
 379     ])
 380     def test_equality(self, values):
 381         unit, value = values
 382         s = disk.Size(**{unit: value})
 383         # both tb and b, since b is always calculated regardless, and is useful
 384         # when testing tb
 385         assert disk.Size(tb=0.78) == s
 386         assert disk.Size(b=857619069665.28) == s
 387
 388     @pytest.mark.parametrize('values', [
 389         ('b', 857619069665.28),
 390         ('kb', 837518622.72),
 391         ('mb', 817889.28),
 392         ('gb', 798.72),
 393         ('tb', 0.78),
 394     ])
 395     def test_inequality(self, values):
 396         unit, value = values
 397         s = disk.Size(**{unit: value})
 398         # both tb and b, since b is always calculated regardless, and is useful
 399         # when testing tb
 400         assert disk.Size(tb=1) != s
 401         assert disk.Size(b=100) != s
 402
 403
 404 class TestSizeOperations(object):
 405
 406     def test_assignment_addition_with_size_objects(self):
 407         result = disk.Size(mb=256) + disk.Size(gb=1)
 408         assert result.gb == 1.25
 409         assert result.gb.as_int() == 1
 410         assert result.gb.as_float() == 1.25
 411
 412     def test_self_addition_with_size_objects(self):
 413         base = disk.Size(mb=256)
 414         base += disk.Size(gb=1)
 415         assert base.gb == 1.25
 416
 417     def test_self_addition_does_not_alter_state(self):
 418         base = disk.Size(mb=256)
 419         base + disk.Size(gb=1)
 420         assert base.mb == 256
 421
 422     def test_addition_with_non_size_objects(self):
 423         with pytest.raises(TypeError):
 424             disk.Size(mb=100) + 4
 425
 426     def test_assignment_subtraction_with_size_objects(self):
 427         base = disk.Size(gb=1)
 428         base -= disk.Size(mb=256)
 429         assert base.mb == 768
 430
 431     def test_self_subtraction_does_not_alter_state(self):
 432         base = disk.Size(gb=1)
 433         base - disk.Size(mb=256)
 434         assert base.gb == 1
 435
 436     def test_subtraction_with_size_objects(self):
 437         result = disk.Size(gb=1) - disk.Size(mb=256)
 438         assert result.mb == 768
 439
 440     def test_subtraction_with_non_size_objects(self):
 441         with pytest.raises(TypeError):
 442             disk.Size(mb=100) - 4
 443
 444     def test_multiplication_with_size_objects(self):
 445         with pytest.raises(TypeError):
 446             disk.Size(mb=100) * disk.Size(mb=1)
 447
 448     def test_multiplication_with_non_size_objects(self):
 449         base = disk.Size(gb=1)
 450         result = base * 2
 451         assert result.gb == 2
 452         assert result.gb.as_int() == 2
 453
 454     def test_division_with_size_objects(self):
 455         result = disk.Size(gb=1) / disk.Size(mb=1)
 456         assert int(result) == 1024
 457
 458     def test_division_with_non_size_objects(self):
 459         base = disk.Size(gb=1)
 460         result = base / 2
 461         assert result.mb == 512
 462         assert result.mb.as_int() == 512
 463
 464     def test_division_with_non_size_objects_without_state(self):
 465         base = disk.Size(gb=1)
 466         base / 2
 467         assert base.gb == 1
 468         assert base.gb.as_int() == 1
 469
 470
 471 class TestSizeAttributes(object):
 472
 473     def test_attribute_does_not_exist(self):
 474         with pytest.raises(AttributeError):
 475             disk.Size(mb=1).exabytes
 476
 477
 478 class TestSizeFormatting(object):
 479
 480     def test_default_formatting_tb_to_b(self):
 481         size = disk.Size(tb=0.0000000001)
 482         result = "%s" % size
 483         assert result == "109.95 B"
 484
 485     def test_default_formatting_tb_to_kb(self):
 486         size = disk.Size(tb=0.00000001)
 487         result = "%s" % size
 488         assert result == "10.74 KB"
 489
 490     def test_default_formatting_tb_to_mb(self):
 491         size = disk.Size(tb=0.000001)
 492         result = "%s" % size
 493         assert result == "1.05 MB"
 494
 495     def test_default_formatting_tb_to_gb(self):
 496         size = disk.Size(tb=0.001)
 497         result = "%s" % size
 498         assert result == "1.02 GB"
 499
 500     def test_default_formatting_tb_to_tb(self):
 501         size = disk.Size(tb=10)
 502         result = "%s" % size
 503         assert result == "10.00 TB"
 504
 505
 506 class TestSizeSpecificFormatting(object):
 507
 508     def test_formatting_b(self):
 509         size = disk.Size(b=2048)
 510         result = "%s" % size.b
 511         assert "%s" % size.b == "%s" % size.bytes
 512         assert result == "2048.00 B"
 513
 514     def test_formatting_kb(self):
 515         size = disk.Size(kb=5700)
 516         result = "%s" % size.kb
 517         assert "%s" % size.kb == "%s" % size.kilobytes
 518         assert result == "5700.00 KB"
 519
 520     def test_formatting_mb(self):
 521         size = disk.Size(mb=4000)
 522         result = "%s" % size.mb
 523         assert "%s" % size.mb == "%s" % size.megabytes
 524         assert result == "4000.00 MB"
 525
 526     def test_formatting_gb(self):
 527         size = disk.Size(gb=77777)
 528         result = "%s" % size.gb
 529         assert "%s" % size.gb == "%s" % size.gigabytes
 530         assert result == "77777.00 GB"
 531
 532     def test_formatting_tb(self):
 533         size = disk.Size(tb=1027)
 534         result = "%s" % size.tb
 535         assert "%s" % size.tb == "%s" % size.terabytes
 536         assert result == "1027.00 TB"