ceph-volume tests.util verify CephDiskDevice behavior

Signed-off-by: Alfredo Deza <adeza@redhat.com>

diff --git a/src/ceph-volume/ceph_volume/tests/util/test_device.py b/src/ceph-volume/ceph_volume/tests/util/test_device.py
index 225b06031a16f287c73413286ff79c1ece94ae7d..70a2a240a185bba7d1a6070d9c6b6f60ea77df32 100644 (file)
--- a/src/ceph-volume/ceph_volume/tests/util/test_device.py
+++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py
@@ -1,3 +1,4 @@
+import pytest
 from ceph_volume.util import device
 from ceph_volume.api import lvm as api
 
@@ -55,6 +56,34 @@ class TestDevice(object):
         disk = device.Device("/dev/sda")
         assert not disk.is_mapper
 
+    def test_is_ceph_disk_member_lsblk(self, device_info):
+        lsblk = {"PARTLABEL": "ceph data"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+
+    def test_is_not_ceph_disk_member_lsblk(self, device_info):
+        lsblk = {"PARTLABEL": "gluster partition"}
+        device_info(lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member is False
+
+    def test_is_ceph_disk_member_blkid(self, device_info):
+        # falls back to blkid
+        lsblk = {"PARTLABEL": ""}
+        blkid = {"PARTLABEL": "ceph data"}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member
+
+    def test_is_not_ceph_disk_member_blkid(self, device_info):
+        # falls back to blkid
+        lsblk = {"PARTLABEL": ""}
+        blkid = {"PARTLABEL": "gluster partition"}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.Device("/dev/sda")
+        assert disk.is_ceph_disk_member is False
+
     def test_pv_api(self, device_info, pvolumes, monkeypatch):
         FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", pv_tags={}, vg_name="vg")
         pvolumes.append(FooPVolume)
@@ -64,3 +93,71 @@ class TestDevice(object):
         device_info(devices=data, lsblk=lsblk)
         disk = device.Device("/dev/sda")
         assert disk.pvs_api
+
+
+ceph_partlabels = [
+    'ceph data', 'ceph journal', 'ceph block',
+    'ceph block.wal', 'ceph block.db', 'ceph lockbox'
+]
+
+
+class TestCephDiskDevice(object):
+
+    def test_partlabel_lsblk(self, device_info):
+        lsblk = {"PARTLABEL": ""}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.partlabel == ''
+
+    def test_partlabel_blkid(self, device_info):
+        lsblk = {"PARTLABEL": ""}
+        blkid = {"PARTLABEL": "ceph data"}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.partlabel == 'ceph data'
+
+    @pytest.mark.parametrize("label", ceph_partlabels)
+    def test_is_member_blkid(self, device_info, label):
+        lsblk = {"PARTLABEL": ""}
+        blkid = {"PARTLABEL": label}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.is_member is True
+
+    @pytest.mark.parametrize("label", ceph_partlabels)
+    def test_is_member_lsblk(self, device_info, label):
+        lsblk = {"PARTLABEL": label}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.is_member is True
+
+    def test_unknown_type(self, device_info):
+        lsblk = {"PARTLABEL": "gluster"}
+        device_info(lsblk=lsblk)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type == 'unknown'
+
+    @pytest.mark.parametrize("label", ceph_partlabels)
+    def test_type_blkid(self, device_info, label):
+        expected = label.split()[-1].split('.')[-1]
+        lsblk = {"PARTLABEL": ""}
+        blkid = {"PARTLABEL": label}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type == expected
+
+    @pytest.mark.parametrize("label", ceph_partlabels)
+    def test_type_lsblk(self, device_info, label):
+        expected = label.split()[-1].split('.')[-1]
+        lsblk = {"PARTLABEL": label}
+        blkid = {"PARTLABEL": ''}
+        device_info(lsblk=lsblk, blkid=blkid)
+        disk = device.CephDiskDevice(device.Device("/dev/sda"))
+
+        assert disk.type == expected