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

   1 import pytest
   2 from copy import deepcopy
   3 from ceph_volume.util import device
   4 from ceph_volume.api import lvm as api
   5
   6
   7 class TestDevice(object):
   8
   9     def test_sys_api(self, monkeypatch, device_info):
  10         volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg',
  11                             lv_tags={}, lv_path='/dev/VolGroup/lv')
  12         volumes = []
  13         volumes.append(volume)
  14         monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
  15                             deepcopy(volumes))
  16
  17         data = {"/dev/sda": {"foo": "bar"}}
  18         device_info(devices=data)
  19         disk = device.Device("/dev/sda")
  20         assert disk.sys_api
  21         assert "foo" in disk.sys_api
  22
  23     def test_lvm_size(self, monkeypatch, device_info):
  24         volume = api.Volume(lv_name='lv', lv_uuid='y', vg_name='vg',
  25                             lv_tags={}, lv_path='/dev/VolGroup/lv')
  26         volumes = []
  27         volumes.append(volume)
  28         monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
  29                             deepcopy(volumes))
  30
  31         # 5GB in size
  32         data = {"/dev/sda": {"size": "5368709120"}}
  33         device_info(devices=data)
  34         disk = device.Device("/dev/sda")
  35         assert disk.lvm_size.gb == 4
  36
  37     def test_lvm_size_rounds_down(self, device_info):
  38         # 5.5GB in size
  39         data = {"/dev/sda": {"size": "5905580032"}}
  40         device_info(devices=data)
  41         disk = device.Device("/dev/sda")
  42         assert disk.lvm_size.gb == 4
  43
  44     def test_is_lv(self, device_info):
  45         data = {"lv_path": "vg/lv", "vg_name": "vg", "name": "lv"}
  46         device_info(lv=data)
  47         disk = device.Device("vg/lv")
  48         assert disk.is_lv
  49
  50     def test_vgs_is_empty(self, device_info, monkeypatch):
  51         BarPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000",
  52                                  pv_tags={})
  53         pvolumes = []
  54         pvolumes.append(BarPVolume)
  55         lsblk = {"TYPE": "disk"}
  56         device_info(lsblk=lsblk)
  57         monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: {})
  58
  59         disk = device.Device("/dev/nvme0n1")
  60         assert disk.vgs == []
  61
  62     def test_vgs_is_not_empty(self, device_info, monkeypatch):
  63         vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6,
  64                              vg_extent_size=1073741824)
  65         monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg])
  66         lsblk = {"TYPE": "disk"}
  67         device_info(lsblk=lsblk)
  68         disk = device.Device("/dev/nvme0n1")
  69         assert len(disk.vgs) == 1
  70
  71     def test_device_is_device(self, device_info):
  72         data = {"/dev/sda": {"foo": "bar"}}
  73         lsblk = {"TYPE": "device"}
  74         device_info(devices=data, lsblk=lsblk)
  75         disk = device.Device("/dev/sda")
  76         assert disk.is_device is True
  77
  78     def test_device_is_rotational(self, device_info):
  79         data = {"/dev/sda": {"rotational": "1"}}
  80         lsblk = {"TYPE": "device"}
  81         device_info(devices=data, lsblk=lsblk)
  82         disk = device.Device("/dev/sda")
  83         assert disk.rotational
  84
  85     def test_device_is_not_rotational(self, device_info):
  86         data = {"/dev/sda": {"rotational": "0"}}
  87         lsblk = {"TYPE": "device"}
  88         device_info(devices=data, lsblk=lsblk)
  89         disk = device.Device("/dev/sda")
  90         assert not disk.rotational
  91
  92     def test_device_is_rotational_lsblk(self, device_info):
  93         data = {"/dev/sda": {"foo": "bar"}}
  94         lsblk = {"TYPE": "device", "ROTA": "1"}
  95         device_info(devices=data, lsblk=lsblk)
  96         disk = device.Device("/dev/sda")
  97         assert disk.rotational
  98
  99     def test_device_is_not_rotational_lsblk(self, device_info):
 100         data = {"/dev/sda": {"rotational": "0"}}
 101         lsblk = {"TYPE": "device", "ROTA": "0"}
 102         device_info(devices=data, lsblk=lsblk)
 103         disk = device.Device("/dev/sda")
 104         assert not disk.rotational
 105
 106     def test_device_is_rotational_defaults_true(self, device_info):
 107         # rotational will default true if no info from sys_api or lsblk is found
 108         data = {"/dev/sda": {"foo": "bar"}}
 109         lsblk = {"TYPE": "device", "foo": "bar"}
 110         device_info(devices=data, lsblk=lsblk)
 111         disk = device.Device("/dev/sda")
 112         assert disk.rotational
 113
 114     def test_disk_is_device(self, device_info):
 115         data = {"/dev/sda": {"foo": "bar"}}
 116         lsblk = {"TYPE": "disk"}
 117         device_info(devices=data, lsblk=lsblk)
 118         disk = device.Device("/dev/sda")
 119         assert disk.is_device is True
 120
 121     def test_is_partition(self, device_info):
 122         data = {"/dev/sda": {"foo": "bar"}}
 123         lsblk = {"TYPE": "part"}
 124         device_info(devices=data, lsblk=lsblk)
 125         disk = device.Device("/dev/sda")
 126         assert disk.is_partition
 127
 128     def test_is_not_acceptable_device(self, device_info):
 129         data = {"/dev/dm-0": {"foo": "bar"}}
 130         lsblk = {"TYPE": "mpath"}
 131         device_info(devices=data, lsblk=lsblk)
 132         disk = device.Device("/dev/dm-0")
 133         assert not disk.is_device
 134
 135     def test_is_not_lvm_memeber(self, device_info):
 136         data = {"/dev/sda": {"foo": "bar"}}
 137         lsblk = {"TYPE": "part"}
 138         device_info(devices=data, lsblk=lsblk)
 139         disk = device.Device("/dev/sda")
 140         assert not disk.is_lvm_member
 141
 142     def test_is_lvm_memeber(self, device_info):
 143         data = {"/dev/sda": {"foo": "bar"}}
 144         lsblk = {"TYPE": "part"}
 145         device_info(devices=data, lsblk=lsblk)
 146         disk = device.Device("/dev/sda")
 147         assert not disk.is_lvm_member
 148
 149     def test_is_mapper_device(self, device_info):
 150         device_info()
 151         disk = device.Device("/dev/mapper/foo")
 152         assert disk.is_mapper
 153
 154     def test_dm_is_mapper_device(self, device_info):
 155         device_info()
 156         disk = device.Device("/dev/dm-4")
 157         assert disk.is_mapper
 158
 159     def test_is_not_mapper_device(self, device_info):
 160         device_info()
 161         disk = device.Device("/dev/sda")
 162         assert not disk.is_mapper
 163
 164     @pytest.mark.usefixtures("lsblk_ceph_disk_member",
 165                              "disable_kernel_queries")
 166     def test_is_ceph_disk_lsblk(self, monkeypatch, patch_bluestore_label):
 167         disk = device.Device("/dev/sda")
 168         assert disk.is_ceph_disk_member
 169
 170     @pytest.mark.usefixtures("blkid_ceph_disk_member",
 171                              "disable_kernel_queries")
 172     def test_is_ceph_disk_blkid(self, monkeypatch, patch_bluestore_label):
 173         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 174                             lambda path: {'PARTLABEL': ""})
 175         disk = device.Device("/dev/sda")
 176         assert disk.is_ceph_disk_member
 177
 178     @pytest.mark.usefixtures("lsblk_ceph_disk_member",
 179                              "disable_kernel_queries")
 180     def test_is_ceph_disk_member_not_available_lsblk(self, monkeypatch, patch_bluestore_label):
 181         disk = device.Device("/dev/sda")
 182         assert disk.is_ceph_disk_member
 183         assert not disk.available
 184         assert "Used by ceph-disk" in disk.rejected_reasons
 185
 186     @pytest.mark.usefixtures("blkid_ceph_disk_member",
 187                              "disable_kernel_queries")
 188     def test_is_ceph_disk_member_not_available_blkid(self, monkeypatch, patch_bluestore_label):
 189         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 190                             lambda path: {'PARTLABEL': ""})
 191         disk = device.Device("/dev/sda")
 192         assert disk.is_ceph_disk_member
 193         assert not disk.available
 194         assert "Used by ceph-disk" in disk.rejected_reasons
 195
 196     def test_reject_removable_device(self, device_info):
 197         data = {"/dev/sdb": {"removable": 1}}
 198         device_info(devices=data)
 199         disk = device.Device("/dev/sdb")
 200         assert not disk.available
 201
 202     def test_accept_non_removable_device(self, device_info):
 203         data = {"/dev/sdb": {"removable": 0, "size": 5368709120}}
 204         device_info(devices=data)
 205         disk = device.Device("/dev/sdb")
 206         assert disk.available
 207
 208     def test_reject_readonly_device(self, device_info):
 209         data = {"/dev/cdrom": {"ro": 1}}
 210         device_info(devices=data)
 211         disk = device.Device("/dev/cdrom")
 212         assert not disk.available
 213
 214     def test_reject_smaller_than_5gb(self, device_info):
 215         data = {"/dev/sda": {"size": 5368709119}}
 216         device_info(devices=data)
 217         disk = device.Device("/dev/sda")
 218         assert not disk.available, 'too small device is available'
 219
 220     def test_accept_non_readonly_device(self, device_info):
 221         data = {"/dev/sda": {"ro": 0, "size": 5368709120}}
 222         device_info(devices=data)
 223         disk = device.Device("/dev/sda")
 224         assert disk.available
 225
 226     def test_reject_bluestore_device(self, monkeypatch, patch_bluestore_label):
 227         patch_bluestore_label.return_value = True
 228         disk = device.Device("/dev/sda")
 229         assert not disk.available
 230         assert "Has BlueStore device label" in disk.rejected_reasons
 231
 232     @pytest.mark.usefixtures("device_info_not_ceph_disk_member",
 233                              "disable_kernel_queries")
 234     def test_is_not_ceph_disk_member_lsblk(self, patch_bluestore_label):
 235         disk = device.Device("/dev/sda")
 236         assert disk.is_ceph_disk_member is False
 237
 238     def test_existing_vg_available(self, monkeypatch, device_info):
 239         vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6,
 240                              vg_extent_size=1073741824)
 241         monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg])
 242         lsblk = {"TYPE": "disk"}
 243         data = {"/dev/nvme0n1": {"size": "6442450944"}}
 244         device_info(devices=data, lsblk=lsblk)
 245         disk = device.Device("/dev/nvme0n1")
 246         assert disk.available_lvm
 247         assert not disk.available
 248         assert not disk.available_raw
 249
 250     def test_existing_vg_too_small(self, monkeypatch, device_info):
 251         vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=4,
 252                              vg_extent_size=1073741824)
 253         monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg])
 254         lsblk = {"TYPE": "disk"}
 255         data = {"/dev/nvme0n1": {"size": "6442450944"}}
 256         device_info(devices=data, lsblk=lsblk)
 257         disk = device.Device("/dev/nvme0n1")
 258         assert not disk.available_lvm
 259         assert not disk.available
 260         assert not disk.available_raw
 261
 262     def test_multiple_existing_vgs(self, monkeypatch, device_info):
 263         vg1 = api.VolumeGroup(vg_name='foo/bar', vg_free_count=4,
 264                              vg_extent_size=1073741824)
 265         vg2 = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6,
 266                              vg_extent_size=1073741824)
 267         monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg1, vg2])
 268         lsblk = {"TYPE": "disk"}
 269         data = {"/dev/nvme0n1": {"size": "6442450944"}}
 270         device_info(devices=data, lsblk=lsblk)
 271         disk = device.Device("/dev/nvme0n1")
 272         assert disk.available_lvm
 273         assert not disk.available
 274         assert not disk.available_raw
 275
 276     @pytest.mark.parametrize("ceph_type", ["data", "block"])
 277     def test_used_by_ceph(self, device_info,
 278                           monkeypatch, ceph_type):
 279         data = {"/dev/sda": {"foo": "bar"}}
 280         lsblk = {"TYPE": "part"}
 281         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000",
 282                                  lv_uuid="0000", pv_tags={}, vg_name="vg")
 283         pvolumes = []
 284         pvolumes.append(FooPVolume)
 285         lv_data = {"lv_name": "lv", "lv_path": "vg/lv", "vg_name": "vg",
 286                    "lv_uuid": "0000", "lv_tags":
 287                    "ceph.osd_id=0,ceph.type="+ceph_type}
 288         volumes = []
 289         lv = api.Volume(**lv_data)
 290         volumes.append(lv)
 291         monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes)
 292         monkeypatch.setattr(api, 'get_lvs', lambda **kwargs:
 293                             deepcopy(volumes))
 294
 295         device_info(devices=data, lsblk=lsblk, lv=lv_data)
 296         vg = api.VolumeGroup(vg_name='foo/bar', vg_free_count=6,
 297                              vg_extent_size=1073741824)
 298         monkeypatch.setattr(api, 'get_device_vgs', lambda x: [vg])
 299         disk = device.Device("/dev/sda")
 300         assert disk.used_by_ceph
 301
 302     def test_not_used_by_ceph(self, device_info, monkeypatch):
 303         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
 304         pvolumes = []
 305         pvolumes.append(FooPVolume)
 306         data = {"/dev/sda": {"foo": "bar"}}
 307         lsblk = {"TYPE": "part"}
 308         lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": "journal"}}
 309         monkeypatch.setattr(api, 'get_pvs', lambda **kwargs: pvolumes)
 310
 311         device_info(devices=data, lsblk=lsblk, lv=lv_data)
 312         disk = device.Device("/dev/sda")
 313         assert not disk.used_by_ceph
 314
 315     def test_get_device_id(self, device_info):
 316         udev = {k:k for k in ['ID_VENDOR', 'ID_MODEL', 'ID_SCSI_SERIAL']}
 317         device_info(udevadm=udev)
 318         disk = device.Device("/dev/sda")
 319         assert disk._get_device_id() == 'ID_VENDOR_ID_MODEL_ID_SCSI_SERIAL'
 320
 321
 322
 323 class TestDeviceEncryption(object):
 324
 325     def test_partition_is_not_encrypted_lsblk(self, device_info):
 326         lsblk = {'TYPE': 'part', 'FSTYPE': 'xfs'}
 327         device_info(lsblk=lsblk)
 328         disk = device.Device("/dev/sda")
 329         assert disk.is_encrypted is False
 330
 331     def test_partition_is_encrypted_lsblk(self, device_info):
 332         lsblk = {'TYPE': 'part', 'FSTYPE': 'crypto_LUKS'}
 333         device_info(lsblk=lsblk)
 334         disk = device.Device("/dev/sda")
 335         assert disk.is_encrypted is True
 336
 337     def test_partition_is_not_encrypted_blkid(self, device_info):
 338         lsblk = {'TYPE': 'part'}
 339         blkid = {'TYPE': 'ceph data'}
 340         device_info(lsblk=lsblk, blkid=blkid)
 341         disk = device.Device("/dev/sda")
 342         assert disk.is_encrypted is False
 343
 344     def test_partition_is_encrypted_blkid(self, device_info):
 345         lsblk = {'TYPE': 'part'}
 346         blkid = {'TYPE': 'crypto_LUKS'}
 347         device_info(lsblk=lsblk, blkid=blkid)
 348         disk = device.Device("/dev/sda")
 349         assert disk.is_encrypted is True
 350
 351     def test_mapper_is_encrypted_luks1(self, device_info, monkeypatch):
 352         status = {'type': 'LUKS1'}
 353         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 354         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 355         blkid = {'TYPE': 'mapper'}
 356         device_info(lsblk=lsblk, blkid=blkid)
 357         disk = device.Device("/dev/mapper/uuid")
 358         assert disk.is_encrypted is True
 359
 360     def test_mapper_is_encrypted_luks2(self, device_info, monkeypatch):
 361         status = {'type': 'LUKS2'}
 362         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 363         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 364         blkid = {'TYPE': 'mapper'}
 365         device_info(lsblk=lsblk, blkid=blkid)
 366         disk = device.Device("/dev/mapper/uuid")
 367         assert disk.is_encrypted is True
 368
 369     def test_mapper_is_encrypted_plain(self, device_info, monkeypatch):
 370         status = {'type': 'PLAIN'}
 371         monkeypatch.setattr(device, 'encryption_status', lambda x: status)
 372         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 373         blkid = {'TYPE': 'mapper'}
 374         device_info(lsblk=lsblk, blkid=blkid)
 375         disk = device.Device("/dev/mapper/uuid")
 376         assert disk.is_encrypted is True
 377
 378     def test_mapper_is_not_encrypted_plain(self, device_info, monkeypatch):
 379         monkeypatch.setattr(device, 'encryption_status', lambda x: {})
 380         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 381         blkid = {'TYPE': 'mapper'}
 382         device_info(lsblk=lsblk, blkid=blkid)
 383         disk = device.Device("/dev/mapper/uuid")
 384         assert disk.is_encrypted is False
 385
 386     def test_lv_is_encrypted_blkid(self, device_info):
 387         lsblk = {'TYPE': 'lvm'}
 388         blkid = {'TYPE': 'crypto_LUKS'}
 389         device_info(lsblk=lsblk, blkid=blkid)
 390         disk = device.Device("/dev/sda")
 391         disk.lv_api = {}
 392         assert disk.is_encrypted is True
 393
 394     def test_lv_is_not_encrypted_blkid(self, factory, device_info):
 395         lsblk = {'TYPE': 'lvm'}
 396         blkid = {'TYPE': 'xfs'}
 397         device_info(lsblk=lsblk, blkid=blkid)
 398         disk = device.Device("/dev/sda")
 399         disk.lv_api = factory(encrypted=None)
 400         assert disk.is_encrypted is False
 401
 402     def test_lv_is_encrypted_lsblk(self, device_info):
 403         lsblk = {'FSTYPE': 'crypto_LUKS', 'TYPE': 'lvm'}
 404         blkid = {'TYPE': 'mapper'}
 405         device_info(lsblk=lsblk, blkid=blkid)
 406         disk = device.Device("/dev/sda")
 407         disk.lv_api = {}
 408         assert disk.is_encrypted is True
 409
 410     def test_lv_is_not_encrypted_lsblk(self, factory, device_info):
 411         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 412         blkid = {'TYPE': 'mapper'}
 413         device_info(lsblk=lsblk, blkid=blkid)
 414         disk = device.Device("/dev/sda")
 415         disk.lv_api = factory(encrypted=None)
 416         assert disk.is_encrypted is False
 417
 418     def test_lv_is_encrypted_lvm_api(self, factory, device_info):
 419         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 420         blkid = {'TYPE': 'mapper'}
 421         device_info(lsblk=lsblk, blkid=blkid)
 422         disk = device.Device("/dev/sda")
 423         disk.lv_api = factory(encrypted=True)
 424         assert disk.is_encrypted is True
 425
 426     def test_lv_is_not_encrypted_lvm_api(self, factory, device_info):
 427         lsblk = {'FSTYPE': 'xfs', 'TYPE': 'lvm'}
 428         blkid = {'TYPE': 'mapper'}
 429         device_info(lsblk=lsblk, blkid=blkid)
 430         disk = device.Device("/dev/sda")
 431         disk.lv_api = factory(encrypted=False)
 432         assert disk.is_encrypted is False
 433
 434
 435 class TestDeviceOrdering(object):
 436
 437     def setup(self):
 438         self.data = {
 439                 "/dev/sda": {"removable": 0},
 440                 "/dev/sdb": {"removable": 1}, # invalid
 441                 "/dev/sdc": {"removable": 0},
 442                 "/dev/sdd": {"removable": 1}, # invalid
 443         }
 444
 445     def test_valid_before_invalid(self, device_info):
 446         device_info(devices=self.data)
 447         sda = device.Device("/dev/sda")
 448         sdb = device.Device("/dev/sdb")
 449
 450         assert sda < sdb
 451         assert sdb > sda
 452
 453     def test_valid_alphabetical_ordering(self, device_info):
 454         device_info(devices=self.data)
 455         sda = device.Device("/dev/sda")
 456         sdc = device.Device("/dev/sdc")
 457
 458         assert sda < sdc
 459         assert sdc > sda
 460
 461     def test_invalid_alphabetical_ordering(self, device_info):
 462         device_info(devices=self.data)
 463         sdb = device.Device("/dev/sdb")
 464         sdd = device.Device("/dev/sdd")
 465
 466         assert sdb < sdd
 467         assert sdd > sdb
 468
 469
 470 class TestCephDiskDevice(object):
 471
 472     def test_partlabel_lsblk(self, device_info):
 473         lsblk = {"PARTLABEL": ""}
 474         device_info(lsblk=lsblk)
 475         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 476
 477         assert disk.partlabel == ''
 478
 479     def test_partlabel_blkid(self, device_info):
 480         lsblk = {"PARTLABEL": ""}
 481         blkid = {"PARTLABEL": "ceph data"}
 482         device_info(lsblk=lsblk, blkid=blkid)
 483         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 484
 485         assert disk.partlabel == 'ceph data'
 486
 487     @pytest.mark.usefixtures("blkid_ceph_disk_member",
 488                              "disable_kernel_queries")
 489     def test_is_member_blkid(self, monkeypatch, patch_bluestore_label):
 490         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 491                             lambda path: {'PARTLABEL': ""})
 492         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 493
 494         assert disk.is_member is True
 495
 496     @pytest.mark.usefixtures("lsblk_ceph_disk_member",
 497                              "disable_kernel_queries")
 498     def test_is_member_lsblk(self, patch_bluestore_label):
 499         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 500
 501         assert disk.is_member is True
 502
 503     def test_unknown_type(self, device_info):
 504         lsblk = {"PARTLABEL": "gluster"}
 505         device_info(lsblk=lsblk)
 506         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 507
 508         assert disk.type == 'unknown'
 509
 510     ceph_types = ['data', 'wal', 'db', 'lockbox', 'journal', 'block']
 511
 512     @pytest.mark.usefixtures("blkid_ceph_disk_member",
 513                              "disable_kernel_queries")
 514     def test_type_blkid(self, monkeypatch, device_info, ceph_partlabel):
 515         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 516                             lambda path: {'PARTLABEL': ''})
 517         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 518
 519         assert disk.type in self.ceph_types
 520
 521     @pytest.mark.usefixtures("blkid_ceph_disk_member",
 522                              "lsblk_ceph_disk_member",
 523                              "disable_kernel_queries")
 524     def test_type_lsblk(self, device_info, ceph_partlabel):
 525         disk = device.CephDiskDevice(device.Device("/dev/sda"))
 526
 527         assert disk.type in self.ceph_types