]>
git.proxmox.com Git - ceph.git/blob - ceph/src/ceph-volume/ceph_volume/tests/conftest.py
3 from ceph_volume
.util
import disk
4 from ceph_volume
.api
import lvm
as lvm_api
5 from ceph_volume
import conf
, configuration
10 def __init__(self
, *a
, **kw
):
14 self
.return_values
= kw
.get('return_values', False)
15 self
.always_returns
= kw
.get('always_returns', False)
17 def __call__(self
, *a
, **kw
):
18 self
.calls
.append({'args': a
, 'kwargs': kw
})
19 if self
.always_returns
:
20 return self
.always_returns
21 if self
.return_values
:
22 return self
.return_values
.pop()
25 class Factory(object):
27 def __init__(self
, **kw
):
28 for k
, v
in kw
.items():
43 def fake_run(monkeypatch
):
45 monkeypatch
.setattr('ceph_volume.process.run', fake_run
)
50 def fake_call(monkeypatch
):
51 fake_call
= Capture(always_returns
=([], [], 0))
52 monkeypatch
.setattr('ceph_volume.process.call', fake_call
)
57 def fakedevice(factory
):
69 params
.update(dict(kw
))
70 params
['lvm_size'] = disk
.Size(b
=params
['sys_api'].get("size", 0))
71 return factory(**params
)
76 def stub_call(monkeypatch
):
78 Monkeypatches process.call, so that a caller can add behavior to the response
80 def apply(return_values
):
81 if isinstance(return_values
, tuple):
82 return_values
= [return_values
]
83 stubbed_call
= Capture(return_values
=return_values
)
84 monkeypatch
.setattr('ceph_volume.process.call', stubbed_call
)
90 @pytest.fixture(autouse
=True)
91 def reset_cluster_name(request
, monkeypatch
):
93 The globally available ``ceph_volume.conf.cluster`` might get mangled in
94 tests, make sure that after evert test, it gets reset, preventing pollution
95 going into other tests later.
100 os
.environ
.pop('CEPH_CONF')
103 request
.addfinalizer(fin
)
107 def conf_ceph(monkeypatch
):
109 Monkeypatches ceph_volume.conf.ceph, which is meant to parse/read
110 a ceph.conf. The patching is naive, it allows one to set return values for
111 specific method calls.
115 monkeypatch
.setattr(conf
, 'ceph', stub
)
121 def conf_ceph_stub(monkeypatch
, tmpfile
):
123 Monkeypatches ceph_volume.conf.ceph with contents from a string that are
124 written to a temporary file and then is fed through the same ceph.conf
125 loading mechanisms for testing. Unlike ``conf_ceph`` which is just a fake,
126 we are actually loading values as seen on a ceph.conf file
128 This is useful when more complex ceph.conf's are needed. In the case of
129 just trying to validate a key/value behavior ``conf_ceph`` is better
133 conf_path
= tmpfile(contents
=contents
)
134 parser
= configuration
.load(conf_path
)
135 monkeypatch
.setattr(conf
, 'ceph', parser
)
141 def volumes(monkeypatch
):
142 monkeypatch
.setattr('ceph_volume.process.call', lambda x
, **kw
: ('', '', 0))
143 volumes
= lvm_api
.Volumes()
149 def volume_groups(monkeypatch
):
150 monkeypatch
.setattr('ceph_volume.process.call', lambda x
, **kw
: ('', '', 0))
151 vgs
= lvm_api
.VolumeGroups()
157 def stub_vgs(monkeypatch
, volume_groups
):
159 monkeypatch
.setattr(lvm_api
, 'get_api_vgs', lambda: vgs
)
164 def pvolumes(monkeypatch
):
165 monkeypatch
.setattr('ceph_volume.process.call', lambda x
, **kw
: ('', '', 0))
166 pvolumes
= lvm_api
.PVolumes()
172 def is_root(monkeypatch
):
174 Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user
175 is root (or is sudoing to superuser) can continue as-is
177 monkeypatch
.setattr('os.getuid', lambda: 0)
183 Create a temporary file, optionally filling it with contents, returns an
184 absolute path to the file when called
186 def generate_file(name
='file', contents
='', directory
=None):
187 directory
= directory
or str(tmpdir
)
188 path
= os
.path
.join(directory
, name
)
189 with
open(path
, 'w') as fp
:
196 def device_info(monkeypatch
):
197 def apply(devices
=None, lsblk
=None, lv
=None, blkid
=None, udevadm
=None):
198 devices
= devices
if devices
else {}
199 lsblk
= lsblk
if lsblk
else {}
200 blkid
= blkid
if blkid
else {}
201 udevadm
= udevadm
if udevadm
else {}
202 lv
= Factory(**lv
) if lv
else None
203 monkeypatch
.setattr("ceph_volume.sys_info.devices", {})
204 monkeypatch
.setattr("ceph_volume.util.device.disk.get_devices", lambda: devices
)
206 monkeypatch
.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path
: lv
)
208 monkeypatch
.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path
: None)
209 monkeypatch
.setattr("ceph_volume.util.device.lvm.get_lv", lambda vg_name
, lv_uuid
: lv
)
210 monkeypatch
.setattr("ceph_volume.util.device.disk.lsblk", lambda path
: lsblk
)
211 monkeypatch
.setattr("ceph_volume.util.device.disk.blkid", lambda path
: blkid
)
212 monkeypatch
.setattr("ceph_volume.util.disk.udevadm_property", lambda *a
, **kw
: udevadm
)