ceph/src/ceph-volume/ceph_volume/tests/conftest.py

   1 import os
   2 import pytest
   3 from mock.mock import patch, PropertyMock
   4 from ceph_volume.util import disk
   5 from ceph_volume.util.constants import ceph_disk_guids
   6 from ceph_volume.api import lvm as lvm_api
   7 from ceph_volume import conf, configuration
   8
   9
  10 class Capture(object):
  11
  12     def __init__(self, *a, **kw):
  13         self.a = a
  14         self.kw = kw
  15         self.calls = []
  16         self.return_values = kw.get('return_values', False)
  17         self.always_returns = kw.get('always_returns', False)
  18
  19     def __call__(self, *a, **kw):
  20         self.calls.append({'args': a, 'kwargs': kw})
  21         if self.always_returns:
  22             return self.always_returns
  23         if self.return_values:
  24             return self.return_values.pop()
  25
  26
  27 class Factory(object):
  28
  29     def __init__(self, **kw):
  30         for k, v in kw.items():
  31             setattr(self, k, v)
  32
  33
  34 @pytest.fixture
  35 def factory():
  36     return Factory
  37
  38
  39 @pytest.fixture
  40 def capture():
  41     return Capture()
  42
  43
  44 @pytest.fixture
  45 def fake_run(monkeypatch):
  46     fake_run = Capture()
  47     monkeypatch.setattr('ceph_volume.process.run', fake_run)
  48     return fake_run
  49
  50
  51 @pytest.fixture
  52 def fake_call(monkeypatch):
  53     fake_call = Capture(always_returns=([], [], 0))
  54     monkeypatch.setattr('ceph_volume.process.call', fake_call)
  55     return fake_call
  56
  57
  58 @pytest.fixture
  59 def fakedevice(factory):
  60     def apply(**kw):
  61         params = dict(
  62             path='/dev/sda',
  63             abspath='/dev/sda',
  64             lv_api=None,
  65             pvs_api=[],
  66             disk_api={},
  67             sys_api={},
  68             exists=True,
  69             is_lvm_member=True,
  70         )
  71         params.update(dict(kw))
  72         params['lvm_size'] = disk.Size(b=params['sys_api'].get("size", 0))
  73         return factory(**params)
  74     return apply
  75
  76
  77 @pytest.fixture
  78 def stub_call(monkeypatch):
  79     """
  80     Monkeypatches process.call, so that a caller can add behavior to the response
  81     """
  82     def apply(return_values):
  83         if isinstance(return_values, tuple):
  84             return_values = [return_values]
  85         stubbed_call = Capture(return_values=return_values)
  86         monkeypatch.setattr('ceph_volume.process.call', stubbed_call)
  87         return stubbed_call
  88
  89     return apply
  90
  91
  92 @pytest.fixture(autouse=True)
  93 def reset_cluster_name(request, monkeypatch):
  94     """
  95     The globally available ``ceph_volume.conf.cluster`` might get mangled in
  96     tests, make sure that after evert test, it gets reset, preventing pollution
  97     going into other tests later.
  98     """
  99     def fin():
 100         conf.cluster = None
 101         try:
 102             os.environ.pop('CEPH_CONF')
 103         except KeyError:
 104             pass
 105     request.addfinalizer(fin)
 106
 107
 108 @pytest.fixture
 109 def conf_ceph(monkeypatch):
 110     """
 111     Monkeypatches ceph_volume.conf.ceph, which is meant to parse/read
 112     a ceph.conf. The patching is naive, it allows one to set return values for
 113     specific method calls.
 114     """
 115     def apply(**kw):
 116         stub = Factory(**kw)
 117         monkeypatch.setattr(conf, 'ceph', stub)
 118         return stub
 119     return apply
 120
 121
 122 @pytest.fixture
 123 def conf_ceph_stub(monkeypatch, tmpfile):
 124     """
 125     Monkeypatches ceph_volume.conf.ceph with contents from a string that are
 126     written to a temporary file and then is fed through the same ceph.conf
 127     loading mechanisms for testing.  Unlike ``conf_ceph`` which is just a fake,
 128     we are actually loading values as seen on a ceph.conf file
 129
 130     This is useful when more complex ceph.conf's are needed. In the case of
 131     just trying to validate a key/value behavior ``conf_ceph`` is better
 132     suited.
 133     """
 134     def apply(contents):
 135         conf_path = tmpfile(contents=contents)
 136         parser = configuration.load(conf_path)
 137         monkeypatch.setattr(conf, 'ceph', parser)
 138         return parser
 139     return apply
 140
 141
 142 @pytest.fixture
 143 def volumes(monkeypatch):
 144     monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
 145     volumes = lvm_api.Volumes()
 146     volumes._purge()
 147     return volumes
 148
 149
 150 @pytest.fixture
 151 def volume_groups(monkeypatch):
 152     monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
 153     vgs = lvm_api.VolumeGroups()
 154     vgs._purge()
 155     return vgs
 156
 157 def volume_groups_empty(monkeypatch):
 158     monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
 159     vgs = lvm_api.VolumeGroups(populate=False)
 160     return vgs
 161
 162 @pytest.fixture
 163 def stub_vgs(monkeypatch, volume_groups):
 164     def apply(vgs):
 165         monkeypatch.setattr(lvm_api, 'get_api_vgs', lambda: vgs)
 166     return apply
 167
 168
 169 # TODO: allow init-ing pvolumes to list we want
 170 @pytest.fixture
 171 def pvolumes(monkeypatch):
 172     monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
 173     pvolumes = lvm_api.PVolumes()
 174     pvolumes._purge()
 175     return pvolumes
 176
 177 @pytest.fixture
 178 def pvolumes_empty(monkeypatch):
 179     monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
 180     pvolumes = lvm_api.PVolumes(populate=False)
 181     return pvolumes
 182
 183
 184
 185 @pytest.fixture
 186 def is_root(monkeypatch):
 187     """
 188     Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user
 189     is root (or is sudoing to superuser) can continue as-is
 190     """
 191     monkeypatch.setattr('os.getuid', lambda: 0)
 192
 193
 194 @pytest.fixture
 195 def tmpfile(tmpdir):
 196     """
 197     Create a temporary file, optionally filling it with contents, returns an
 198     absolute path to the file when called
 199     """
 200     def generate_file(name='file', contents='', directory=None):
 201         directory = directory or str(tmpdir)
 202         path = os.path.join(directory, name)
 203         with open(path, 'w') as fp:
 204             fp.write(contents)
 205         return path
 206     return generate_file
 207
 208
 209 @pytest.fixture
 210 def disable_kernel_queries(monkeypatch):
 211     '''
 212     This speeds up calls to Device and Disk
 213     '''
 214     monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: {})
 215     monkeypatch.setattr("ceph_volume.util.disk.udevadm_property", lambda *a, **kw: {})
 216
 217
 218 @pytest.fixture
 219 def disable_lvm_queries(monkeypatch):
 220     '''
 221     This speeds up calls to Device and Disk
 222     '''
 223     monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: None)
 224     monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv", lambda vg_name, lv_uuid: None)
 225
 226
 227 @pytest.fixture(params=[
 228     '', 'ceph data', 'ceph journal', 'ceph block',
 229     'ceph block.wal', 'ceph block.db', 'ceph lockbox'])
 230 def ceph_partlabel(request):
 231     return request.param
 232
 233
 234 @pytest.fixture(params=list(ceph_disk_guids.keys()))
 235 def ceph_parttype(request):
 236     return request.param
 237
 238
 239 @pytest.fixture
 240 def lsblk_ceph_disk_member(monkeypatch, request, ceph_partlabel, ceph_parttype):
 241     monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 242                         lambda path: {'PARTLABEL': ceph_partlabel})
 243     # setting blkid here too in order to be able to fall back to PARTTYPE based
 244     # membership
 245     monkeypatch.setattr("ceph_volume.util.device.disk.blkid",
 246                         lambda path: {'PARTLABEL': '',
 247                                       'PARTTYPE': ceph_parttype})
 248
 249
 250 @pytest.fixture
 251 def blkid_ceph_disk_member(monkeypatch, request, ceph_partlabel, ceph_parttype):
 252     monkeypatch.setattr("ceph_volume.util.device.disk.blkid",
 253                         lambda path: {'PARTLABEL': ceph_partlabel,
 254                                       'PARTTYPE': ceph_parttype})
 255
 256
 257 @pytest.fixture(params=[
 258     ('gluster partition', 'gluster partition'),
 259     # falls back to blkid
 260     ('', 'gluster partition'),
 261     ('gluster partition', ''),
 262 ])
 263 def device_info_not_ceph_disk_member(monkeypatch, request):
 264     monkeypatch.setattr("ceph_volume.util.device.disk.lsblk",
 265                         lambda path: {'PARTLABEL': request.param[0]})
 266     monkeypatch.setattr("ceph_volume.util.device.disk.blkid",
 267                         lambda path: {'PARTLABEL': request.param[1]})
 268
 269 @pytest.fixture
 270 def patched_get_block_devs_lsblk():
 271     with patch('ceph_volume.util.disk.get_block_devs_lsblk') as p:
 272         yield p
 273
 274 @pytest.fixture
 275 def patch_bluestore_label():
 276     with patch('ceph_volume.util.device.Device.has_bluestore_label',
 277                new_callable=PropertyMock) as p:
 278         p.return_value = False
 279         yield p
 280
 281 @pytest.fixture
 282 def device_info(monkeypatch, patch_bluestore_label):
 283     def apply(devices=None, lsblk=None, lv=None, blkid=None, udevadm=None,
 284               has_bluestore_label=False):
 285         devices = devices if devices else {}
 286         lsblk = lsblk if lsblk else {}
 287         blkid = blkid if blkid else {}
 288         udevadm = udevadm if udevadm else {}
 289         lv = Factory(**lv) if lv else None
 290         monkeypatch.setattr("ceph_volume.sys_info.devices", {})
 291         monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: devices)
 292         if not devices:
 293             monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: lv)
 294         else:
 295             monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: None)
 296             monkeypatch.setattr("ceph_volume.util.device.lvm.get_device_lvs",
 297                                 lambda path: [lv])
 298         monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv", lambda vg_name, lv_uuid: lv)
 299         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk)
 300         monkeypatch.setattr("ceph_volume.util.device.disk.blkid", lambda path: blkid)
 301         monkeypatch.setattr("ceph_volume.util.disk.udevadm_property", lambda *a, **kw: udevadm)
 302     return apply