+import pytest
+import json
from ceph_volume.util import prepare
from ceph_volume.util.prepare import system
from ceph_volume import conf
from ceph_volume.tests.conftest import Factory
+class TestCheckID(object):
+
+ def test_false_if_id_is_none(self):
+ assert not prepare.check_id(None)
+
+ def test_returncode_is_not_zero(self, monkeypatch):
+ monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: ('', '', 1))
+ with pytest.raises(RuntimeError):
+ prepare.check_id(1)
+
+ def test_id_does_exist(self, monkeypatch):
+ stdout = dict(nodes=[
+ dict(id=0),
+ ])
+ stdout = json.dumps(stdout)
+ monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+ result = prepare.check_id(0)
+ assert result
+ assert result == "0"
+
+ def test_id_does_not_exist(self, monkeypatch):
+ stdout = dict(nodes=[
+ dict(id=0),
+ ])
+ stdout = json.dumps(stdout)
+ monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+ result = prepare.check_id(1)
+ assert not result
+
+ def test_invalid_osd_id(self, monkeypatch):
+ stdout = dict(nodes=[
+ dict(id=0),
+ ])
+ stdout = json.dumps(stdout)
+ monkeypatch.setattr('ceph_volume.process.call', lambda *a, **kw: (stdout, '', 0))
+ result = prepare.check_id("foo")
+ assert not result
+
+
class TestFormatDevice(object):
def test_include_force(self, fake_run, monkeypatch):