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