ceph-volume: tests for util.device.Device

Signed-off-by: Andrew Schoen <aschoen@redhat.com>
(cherry picked from commit 947266e16ae3a5e9bdaa7fbf326a0e1b96f16e0f)

diff --git a/src/ceph-volume/ceph_volume/tests/conftest.py b/src/ceph-volume/ceph_volume/tests/conftest.py
index 11e1c1a781e0a13ec7fa048e1f959db6e0b3e3a4..bdbe817f16f0e67226c60f13f1eb250729190710 100644 (file)
--- a/src/ceph-volume/ceph_volume/tests/conftest.py
+++ b/src/ceph-volume/ceph_volume/tests/conftest.py
@@ -147,3 +147,15 @@ def tmpfile(tmpdir):
             fp.write(contents)
         return path
     return generate_file
+
+
+@pytest.fixture
+def device_info(monkeypatch):
+    def apply(devices=None, lsblk=None, lv=None):
+        devices = devices if devices else {}
+        lsblk = lsblk if lsblk else {}
+        lv = Factory(**lv) if lv else None
+        monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: devices)
+        monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: lv)
+        monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk)
+    return apply

diff --git a/src/ceph-volume/ceph_volume/tests/util/test_device.py b/src/ceph-volume/ceph_volume/tests/util/test_device.py
new file mode 100644 (file)
index 0000000..225b060
--- /dev/null
+++ b/src/ceph-volume/ceph_volume/tests/util/test_device.py
@@ -0,0 +1,66 @@
+from ceph_volume.util import device
+from ceph_volume.api import lvm as api
+
+
+class TestDevice(object):
+
+    def test_sys_api(self, device_info):
+        data = {"/dev/sda": {"foo": "bar"}}
+        device_info(devices=data)
+        disk = device.Device("/dev/sda")
+        assert disk.sys_api
+        assert "foo" in disk.sys_api
+
+    def test_is_lv(self, device_info):
+        data = {"lv_path": "vg/lv"}
+        device_info(lv=data)
+        disk = device.Device("vg/lv")
+        assert disk.is_lv
+
+    def test_is_device(self, device_info):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "device"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_device
+
+    def test_is_partition(self, device_info, pvolumes):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.is_partition
+
+    def test_is_not_lvm_memeber(self, device_info, pvolumes):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.is_lvm_member
+
+    def test_is_lvm_memeber(self, device_info, pvolumes):
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert not disk.is_lvm_member
+
+    def test_is_mapper_device(self, device_info):
+        device_info()
+        disk = device.Device("/dev/mapper/foo")
+        assert disk.is_mapper
+
+    def test_is_not_mapper_device(self, device_info):
+        device_info()
+        disk = device.Device("/dev/sda")
+        assert not disk.is_mapper
+
+    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)
+        monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
+        data = {"/dev/sda": {"foo": "bar"}}
+        lsblk = {"TYPE": "part"}
+        device_info(devices=data, lsblk=lsblk)
+        disk = device.Device("/dev/sda")
+        assert disk.pvs_api