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

   1 import pytest
   2 from ceph_volume.util import device
   3 from ceph_volume.api import lvm as api
   4
   5
   6 class TestDevice(object):
   7
   8     def test_sys_api(self, device_info):
   9         data = {"/dev/sda": {"foo": "bar"}}
  10         device_info(devices=data)
  11         disk = device.Device("/dev/sda")
  12         assert disk.sys_api
  13         assert "foo" in disk.sys_api
  14
  15     def test_is_lv(self, device_info):
  16         data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"}
  17         device_info(lv=data)
  18         disk = device.Device("vg/lv")
  19         assert disk.is_lv
  20
  21     def test_vgs_is_empty(self, device_info, pvolumes, monkeypatch):
  22         BarPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", pv_tags={})
  23         pvolumes.append(BarPVolume)
  24         monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
  25         lsblk = {"TYPE": "disk"}
  26         device_info(lsblk=lsblk)
  27         disk = device.Device("/dev/nvme0n1")
  28         assert disk.vgs == []
  29
  30     def test_vgs_is_not_empty(self, device_info, pvolumes, monkeypatch):
  31         BarPVolume = api.PVolume(vg_name='foo', lv_uuid='111', pv_name='/dev/nvme0n1', pv_uuid="0000", pv_tags={})
  32         pvolumes.append(BarPVolume)
  33         monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
  34         lsblk = {"TYPE": "disk"}
  35         device_info(lsblk=lsblk)
  36         disk = device.Device("/dev/nvme0n1")
  37         assert len(disk.vgs) == 1
  38
  39     def test_device_is_device(self, device_info, pvolumes):
  40         data = {"/dev/sda": {"foo": "bar"}}
  41         lsblk = {"TYPE": "device"}
  42         device_info(devices=data, lsblk=lsblk)
  43         disk = device.Device("/dev/sda")
  44         assert disk.is_device is True
  45
  46     def test_device_is_rotational(self, device_info, pvolumes):
  47         data = {"/dev/sda": {"rotational": "1"}}
  48         lsblk = {"TYPE": "device"}
  49         device_info(devices=data, lsblk=lsblk)
  50         disk = device.Device("/dev/sda")
  51         assert disk.rotational
  52
  53     def test_device_is_not_rotational(self, device_info, pvolumes):
  54         data = {"/dev/sda": {"rotational": "0"}}
  55         lsblk = {"TYPE": "device"}
  56         device_info(devices=data, lsblk=lsblk)
  57         disk = device.Device("/dev/sda")
  58         assert not disk.rotational
  59
  60     def test_device_is_rotational_lsblk(self, device_info, pvolumes):
  61         data = {"/dev/sda": {"foo": "bar"}}
  62         lsblk = {"TYPE": "device", "ROTA": "1"}
  63         device_info(devices=data, lsblk=lsblk)
  64         disk = device.Device("/dev/sda")
  65         assert disk.rotational
  66
  67     def test_device_is_not_rotational_lsblk(self, device_info, pvolumes):
  68         data = {"/dev/sda": {"rotational": "0"}}
  69         lsblk = {"TYPE": "device", "ROTA": "0"}
  70         device_info(devices=data, lsblk=lsblk)
  71         disk = device.Device("/dev/sda")
  72         assert not disk.rotational
  73
  74     def test_device_is_rotational_defaults_true(self, device_info, pvolumes):
  75         # rotational will default true if no info from sys_api or lsblk is found
  76         data = {"/dev/sda": {"foo": "bar"}}
  77         lsblk = {"TYPE": "device", "foo": "bar"}
  78         device_info(devices=data, lsblk=lsblk)
  79         disk = device.Device("/dev/sda")
  80         assert disk.rotational
  81
  82     def test_disk_is_device(self, device_info, pvolumes):
  83         data = {"/dev/sda": {"foo": "bar"}}
  84         lsblk = {"TYPE": "disk"}
  85         device_info(devices=data, lsblk=lsblk)
  86         disk = device.Device("/dev/sda")
  87         assert disk.is_device is True
  88
  89     def test_is_partition(self, device_info, pvolumes):
  90         data = {"/dev/sda": {"foo": "bar"}}
  91         lsblk = {"TYPE": "part"}
  92         device_info(devices=data, lsblk=lsblk)
  93         disk = device.Device("/dev/sda")
  94         assert disk.is_partition
  95
  96     def test_is_not_lvm_memeber(self, device_info, pvolumes):
  97         data = {"/dev/sda": {"foo": "bar"}}
  98         lsblk = {"TYPE": "part"}
  99         device_info(devices=data, lsblk=lsblk)
 100         disk = device.Device("/dev/sda")
 101         assert not disk.is_lvm_member
 102
 103     def test_is_lvm_memeber(self, device_info, pvolumes):
 104         data = {"/dev/sda": {"foo": "bar"}}
 105         lsblk = {"TYPE": "part"}
 106         device_info(devices=data, lsblk=lsblk)
 107         disk = device.Device("/dev/sda")
 108         assert not disk.is_lvm_member
 109
 110     def test_is_mapper_device(self, device_info):
 111         device_info()
 112         disk = device.Device("/dev/mapper/foo")
 113         assert disk.is_mapper
 114
 115     def test_dm_is_mapper_device(self, device_info):
 116         device_info()
 117         disk = device.Device("/dev/dm-4")
 118         assert disk.is_mapper
 119
 120     def test_is_not_mapper_device(self, device_info):
 121         device_info()
 122         disk = device.Device("/dev/sda")
 123         assert not disk.is_mapper
 124
 125     def test_is_ceph_disk_member_lsblk(self, device_info):
 126         lsblk = {"PARTLABEL": "ceph data"}
 127         device_info(lsblk=lsblk)
 128         disk = device.Device("/dev/sda")
 129         assert disk.is_ceph_disk_member
 130
 131     def test_is_ceph_disk_member_not_available(self, device_info):
 132         lsblk = {"PARTLABEL": "ceph data"}
 133         device_info(lsblk=lsblk)
 134         disk = device.Device("/dev/sda")
 135         assert disk.is_ceph_disk_member
 136         assert not disk.available
 137         assert "Used by ceph-disk" in disk.rejected_reasons
 138
 139     def test_is_not_ceph_disk_member_lsblk(self, device_info):
 140         lsblk = {"PARTLABEL": "gluster partition"}
 141         device_info(lsblk=lsblk)
 142         disk = device.Device("/dev/sda")
 143         assert disk.is_ceph_disk_member is False
 144
 145     def test_is_ceph_disk_member_blkid(self, device_info):
 146         # falls back to blkid
 147         lsblk = {"PARTLABEL": ""}
 148         blkid = {"PARTLABEL": "ceph data"}
 149         device_info(lsblk=lsblk, blkid=blkid)
 150         disk = device.Device("/dev/sda")
 151         assert disk.is_ceph_disk_member
 152
 153     def test_is_not_ceph_disk_member_blkid(self, device_info):
 154         # falls back to blkid
 155         lsblk = {"PARTLABEL": ""}
 156         blkid = {"PARTLABEL": "gluster partition"}
 157         device_info(lsblk=lsblk, blkid=blkid)
 158         disk = device.Device("/dev/sda")
 159         assert disk.is_ceph_disk_member is False
 160
 161     def test_pv_api(self, device_info, pvolumes, monkeypatch):
 162         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
 163         pvolumes.append(FooPVolume)
 164         monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
 165         data = {"/dev/sda": {"foo": "bar"}}
 166         lsblk = {"TYPE": "part"}
 167         device_info(devices=data, lsblk=lsblk)
 168         disk = device.Device("/dev/sda")
 169         assert disk.pvs_api
 170
 171     @pytest.mark.parametrize("ceph_type", ["data", "block"])
 172     def test_used_by_ceph(self, device_info, pvolumes, monkeypatch, ceph_type):
 173         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
 174         pvolumes.append(FooPVolume)
 175         monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
 176         data = {"/dev/sda": {"foo": "bar"}}
 177         lsblk = {"TYPE": "part"}
 178         lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": ceph_type}}
 179         device_info(devices=data, lsblk=lsblk, lv=lv_data)
 180         disk = device.Device("/dev/sda")
 181         assert disk.used_by_ceph
 182
 183     def test_not_used_by_ceph(self, device_info, pvolumes, monkeypatch):
 184         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
 185         pvolumes.append(FooPVolume)
 186         monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
 187         data = {"/dev/sda": {"foo": "bar"}}
 188         lsblk = {"TYPE": "part"}
 189         lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": "journal"}}
 190         device_info(devices=data, lsblk=lsblk, lv=lv_data)
 191         disk = device.Device("/dev/sda")
 192         assert not disk.used_by_ceph
 193
 194     def test_get_device_id(self, device_info):
 195         udev = {k:k for k in ['ID_VENDOR', 'ID_MODEL', 'ID_SCSI_SERIAL']}
 196         device_info(udevadm=udev)
 197         disk = device.Device("/dev/sda")
 198         assert disk._get_device_id() == 'ID_VENDOR_ID_MODEL_ID_SCSI_SERIAL'
 199
 200
 201
 202 class TestDeviceEncryption(object):
 203
 204     def test_partition_is_not_encrypted_lsblk(self, device_info, pvolumes):
 205         lsblk = {'TYPE': 'part', 'FSTYPE': 'xfs'}
 206         device_info(lsblk=lsblk)
 207         disk = device.Device("/dev/sda")
 208         assert disk.is_encrypted is False
 209
 210     def test_partition_is_encrypted_lsblk(self, device_info, pvolumes):
 211         lsblk = {'TYPE': 'part', 'FSTYPE': 'crypto_LUKS'}
 212         device_info(lsblk=lsblk)
 213         disk = device.Device("/dev/sda")
 214         assert disk.is_encrypted is True
 215
 216     def test_partition_is_not_encrypted_blkid(self, device_info, pvolumes):
 217         lsblk = {'TYPE': 'part'}
 218         blkid = {'TYPE': 'ceph data'}
 219         device_info(lsblk=lsblk, blkid=blkid)
 220         disk = device.Device("/dev/sda")
 221         assert disk.is_encrypted is False
 222
 223     def test_partition_is_encrypted_blkid(self, device_info, pvolumes):
 224         lsblk = {'TYPE': 'part'}
 225         blkid = {'TYPE': 'crypto_LUKS'}
 226         device_info(lsblk=lsblk, blkid=blkid)
 227         disk = device.Device("/dev/sda")
 228         assert disk.is_encrypted is True
 229
 230     def test_mapper_is_encrypted_luks1(self, device_info, pvolumes, monkeypatch):
 231         status = {'type': 'LUKS1'}
 232         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 233         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 234         blkid = {'TYPE': 'mapper'}
 235         device_info(lsblk=lsblk, blkid=blkid)
 236         disk = device.Device("/dev/mapper/uuid")
 237         assert disk.is_encrypted is True
 238
 239     def test_mapper_is_encrypted_luks2(self, device_info, pvolumes, monkeypatch):
 240         status = {'type': 'LUKS2'}
 241         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 242         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 243         blkid = {'TYPE': 'mapper'}
 244         device_info(lsblk=lsblk, blkid=blkid)
 245         disk = device.Device("/dev/mapper/uuid")
 246         assert disk.is_encrypted is True
 247
 248     def test_mapper_is_encrypted_plain(self, device_info, pvolumes, monkeypatch):
 249         status = {'type': 'PLAIN'}
 250         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 251         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 252         blkid = {'TYPE': 'mapper'}
 253         device_info(lsblk=lsblk, blkid=blkid)
 254         disk = device.Device("/dev/mapper/uuid")
 255         assert disk.is_encrypted is True
 256
 257     def test_mapper_is_not_encrypted_plain(self, device_info, pvolumes, monkeypatch):
 258         monkeypatch.setattr(device, 'encryption_status', lambda x: {})
 259         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 260         blkid = {'TYPE': 'mapper'}
 261         device_info(lsblk=lsblk, blkid=blkid)
 262         disk = device.Device("/dev/mapper/uuid")
 263         assert disk.is_encrypted is False
 264
 265     def test_lv_is_encrypted_blkid(self, device_info, pvolumes):
 266         lsblk = {'TYPE': 'lvm'}
 267         blkid = {'TYPE': 'crypto_LUKS'}
 268         device_info(lsblk=lsblk, blkid=blkid)
 269         disk = device.Device("/dev/sda")
 270         disk.lv_api = {}
 271         assert disk.is_encrypted is True
 272
 273     def test_lv_is_not_encrypted_blkid(self, factory, device_info, pvolumes):
 274         lsblk = {'TYPE': 'lvm'}
 275         blkid = {'TYPE': 'xfs'}
 276         device_info(lsblk=lsblk, blkid=blkid)
 277         disk = device.Device("/dev/sda")
 278         disk.lv_api = factory(encrypted=None)
 279         assert disk.is_encrypted is False
 280
 281     def test_lv_is_encrypted_lsblk(self, device_info, pvolumes):
 282         lsblk = {'FSTYPE': 'crypto_LUKS', 'TYPE': 'lvm'}
 283         blkid = {'TYPE': 'mapper'}
 284         device_info(lsblk=lsblk, blkid=blkid)
 285         disk = device.Device("/dev/sda")
 286         disk.lv_api = {}
 287         assert disk.is_encrypted is True
 288
 289     def test_lv_is_not_encrypted_lsblk(self, factory, device_info, pvolumes):
 290         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 291         blkid = {'TYPE': 'mapper'}
 292         device_info(lsblk=lsblk, blkid=blkid)
 293         disk = device.Device("/dev/sda")
 294         disk.lv_api = factory(encrypted=None)
 295         assert disk.is_encrypted is False
 296
 297     def test_lv_is_encrypted_lvm_api(self, factory, device_info, pvolumes):
 298         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 299         blkid = {'TYPE': 'mapper'}
 300         device_info(lsblk=lsblk, blkid=blkid)
 301         disk = device.Device("/dev/sda")
 302         disk.lv_api = factory(encrypted=True)
 303         assert disk.is_encrypted is True
 304
 305     def test_lv_is_not_encrypted_lvm_api(self, factory, device_info, pvolumes):
 306         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 307         blkid = {'TYPE': 'mapper'}
 308         device_info(lsblk=lsblk, blkid=blkid)
 309         disk = device.Device("/dev/sda")
 310         disk.lv_api = factory(encrypted=False)
 311         assert disk.is_encrypted is False
 312
 313
 314 class TestDeviceOrdering(object):
 315
 316     def setup(self):
 317         self.data = {
 318                 "/dev/sda": {"removable": 0},
 319                 "/dev/sdb": {"removable": 1}, # invalid
 320                 "/dev/sdc": {"removable": 0},
 321                 "/dev/sdd": {"removable": 1}, # invalid
 322         }
 323
 324     def test_valid_before_invalid(self, device_info):
 325         device_info(devices=self.data)
 326         sda = device.Device("/dev/sda")
 327         sdb = device.Device("/dev/sdb")
 328
 329         assert sda < sdb
 330         assert sdb > sda
 331
 332     def test_valid_alphabetical_ordering(self, device_info):
 333         device_info(devices=self.data)
 334         sda = device.Device("/dev/sda")
 335         sdc = device.Device("/dev/sdc")
 336
 337         assert sda < sdc
 338         assert sdc > sda
 339
 340     def test_invalid_alphabetical_ordering(self, device_info):
 341         device_info(devices=self.data)
 342         sdb = device.Device("/dev/sdb")
 343         sdd = device.Device("/dev/sdd")
 344
 345         assert sdb < sdd
 346         assert sdd > sdb
 347
 348
 349 ceph_partlabels = [
 350     'ceph data', 'ceph journal', 'ceph block',
 351     'ceph block.wal', 'ceph block.db', 'ceph lockbox'
 352 ]
 353
 354
 355 class TestCephDiskDevice(object):
 356
 357     def test_partlabel_lsblk(self, device_info):
 358         lsblk = {"PARTLABEL": ""}
 359         device_info(lsblk=lsblk)
 360         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 361
 362         assert disk.partlabel == ''
 363
 364     def test_partlabel_blkid(self, device_info):
 365         lsblk = {"PARTLABEL": ""}
 366         blkid = {"PARTLABEL": "ceph data"}
 367         device_info(lsblk=lsblk, blkid=blkid)
 368         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 369
 370         assert disk.partlabel == 'ceph data'
 371
 372     @pytest.mark.parametrize("label", ceph_partlabels)
 373     def test_is_member_blkid(self, device_info, label):
 374         lsblk = {"PARTLABEL": ""}
 375         blkid = {"PARTLABEL": label}
 376         device_info(lsblk=lsblk, blkid=blkid)
 377         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 378
 379         assert disk.is_member is True
 380
 381     def test_reject_removable_device(self, device_info):
 382         data = {"/dev/sdb": {"removable": 1}}
 383         device_info(devices=data)
 384         disk = device.Device("/dev/sdb")
 385         assert not disk.available
 386
 387     def test_accept_non_removable_device(self, device_info):
 388         data = {"/dev/sdb": {"removable": 0}}
 389         device_info(devices=data)
 390         disk = device.Device("/dev/sdb")
 391         assert disk.available
 392
 393     def test_reject_readonly_device(self, device_info):
 394         data = {"/dev/cdrom": {"ro": 1}}
 395         device_info(devices=data)
 396         disk = device.Device("/dev/cdrom")
 397         assert not disk.available
 398
 399     def test_accept_non_readonly_device(self, device_info):
 400         data = {"/dev/sda": {"ro": 0}}
 401         device_info(devices=data)
 402         disk = device.Device("/dev/sda")
 403         assert disk.available
 404
 405     @pytest.mark.parametrize("label", ceph_partlabels)
 406     def test_is_member_lsblk(self, device_info, label):
 407         lsblk = {"PARTLABEL": label}
 408         device_info(lsblk=lsblk)
 409         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 410
 411         assert disk.is_member is True
 412
 413     def test_unknown_type(self, device_info):
 414         lsblk = {"PARTLABEL": "gluster"}
 415         device_info(lsblk=lsblk)
 416         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 417
 418         assert disk.type == 'unknown'
 419
 420     @pytest.mark.parametrize("label", ceph_partlabels)
 421     def test_type_blkid(self, device_info, label):
 422         expected = label.split()[-1].split('.')[-1]
 423         lsblk = {"PARTLABEL": ""}
 424         blkid = {"PARTLABEL": label}
 425         device_info(lsblk=lsblk, blkid=blkid)
 426         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 427
 428         assert disk.type == expected
 429
 430     @pytest.mark.parametrize("label", ceph_partlabels)
 431     def test_type_lsblk(self, device_info, label):
 432         expected = label.split()[-1].split('.')[-1]
 433         lsblk = {"PARTLABEL": label}
 434         blkid = {"PARTLABEL": ''}
 435         device_info(lsblk=lsblk, blkid=blkid)
 436         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 437
 438         assert disk.type == expected