RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-03-11 02:39:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

ceph-volume: add tests for util.device.Device.used_by_ceph

Browse Source 
Signed-off-by: Andrew Schoen <aschoen@redhat.com>
This commit is contained in:
Andrew Schoen 2018-10-03 10:19:45 -05:00
parent 36396229d3
commit a28e6531e4
3 changed files with 29 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/ceph-volume/ceph_volume/tests/conftest.py
View File

@ -199,7 +199,11 @@ def device_info(monkeypatch):
lv = Factory(**lv) if lv else None
monkeypatch.setattr("ceph_volume.sys_info.devices", {})
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)
if not devices:
monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: lv)
else:
monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: None)
monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv", lambda vg_name, lv_uuid: lv)
monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk)
monkeypatch.setattr("ceph_volume.util.device.disk.blkid", lambda path: blkid)
return apply

23
src/ceph-volume/ceph_volume/tests/util/test_device.py
View File

@ -94,6 +94,29 @@ class TestDevice(object):
disk = device.Device("/dev/sda")
assert disk.pvs_api
@pytest.mark.parametrize("ceph_type", ["data", "block"])
def test_used_by_ceph(self, device_info, pvolumes, monkeypatch, ceph_type):
FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
pvolumes.append(FooPVolume)
monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
data = {"/dev/sda": {"foo": "bar"}}
lsblk = {"TYPE": "part"}
lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": ceph_type}}
device_info(devices=data, lsblk=lsblk, lv=lv_data)
disk = device.Device("/dev/sda")
assert disk.used_by_ceph
def test_not_used_by_ceph(self, device_info, pvolumes, monkeypatch):
FooPVolume = api.PVolume(pv_name='/dev/sda', pv_uuid="0000", lv_uuid="0000", pv_tags={}, vg_name="vg")
pvolumes.append(FooPVolume)
monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
data = {"/dev/sda": {"foo": "bar"}}
lsblk = {"TYPE": "part"}
lv_data = {"lv_path": "vg/lv", "vg_name": "vg", "lv_uuid": "0000", "tags": {"ceph.osd_id": 0, "ceph.type": "journal"}}
device_info(devices=data, lsblk=lsblk, lv=lv_data)
disk = device.Device("/dev/sda")
assert not disk.used_by_ceph
ceph_partlabels = [
'ceph data', 'ceph journal', 'ceph block',

2
src/ceph-volume/ceph_volume/util/device.py
View File

@ -119,7 +119,7 @@ class Device(object):
@property
def used_by_ceph(self):
# only filter out data devices as journals could potentially be reused
osd_ids = [lv.tags.get("ceph.osd_id") for lv in self.lvs
osd_ids = [lv.tags.get("ceph.osd_id") is not None for lv in self.lvs
if lv.tags.get("ceph.type") in ["data", "block"]]
return any(osd_ids)