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

   1 import os
   2 import pytest
   3 from ceph_volume.api import lvm as lvm_api
   4 from ceph_volume import conf, configuration
   5
   6
   7 class Capture(object):
   8
   9     def __init__(self, *a, **kw):
  10         self.a = a
  11         self.kw = kw
  12         self.calls = []
  13         self.return_values = kw.get('return_values', False)
  14         self.always_returns = kw.get('always_returns', False)
  15
  16     def __call__(self, *a, **kw):
  17         self.calls.append({'args': a, 'kwargs': kw})
  18         if self.always_returns:
  19             return self.always_returns
  20         if self.return_values:
  21             return self.return_values.pop()
  22
  23
  24 class Factory(object):
  25
  26     def __init__(self, **kw):
  27         for k, v in kw.items():
  28             setattr(self, k, v)
  29
  30
  31 @pytest.fixture
  32 def factory():
  33     return Factory
  34
  35
  36 @pytest.fixture
  37 def capture():
  38     return Capture()
  39
  40
  41 @pytest.fixture
  42 def fake_run(monkeypatch):
  43     fake_run = Capture()
  44     monkeypatch.setattr('ceph_volume.process.run', fake_run)
  45     return fake_run
  46
  47
  48 @pytest.fixture
  49 def fake_call(monkeypatch):
  50     fake_call = Capture(always_returns=([], [], 0))
  51     monkeypatch.setattr('ceph_volume.process.call', fake_call)
  52     return fake_call
  53
  54
  55 @pytest.fixture
  56 def stub_call(monkeypatch):
  57     """
  58     Monkeypatches process.call, so that a caller can add behavior to the response
  59     """
  60     def apply(return_values):
  61         if isinstance(return_values, tuple):
  62             return_values = [return_values]
  63         stubbed_call = Capture(return_values=return_values)
  64         monkeypatch.setattr('ceph_volume.process.call', stubbed_call)
  65         return stubbed_call
  66
  67     return apply
  68
  69
  70 @pytest.fixture
  71 def conf_ceph(monkeypatch):
  72     """
  73     Monkeypatches ceph_volume.conf.ceph, which is meant to parse/read
  74     a ceph.conf. The patching is naive, it allows one to set return values for
  75     specific method calls.
  76     """
  77     def apply(**kw):
  78         stub = Factory(**kw)
  79         monkeypatch.setattr(conf, 'ceph', stub)
  80         return stub
  81     return apply
  82
  83
  84 @pytest.fixture
  85 def conf_ceph_stub(monkeypatch, tmpfile):
  86     """
  87     Monkeypatches ceph_volume.conf.ceph with contents from a string that are
  88     written to a temporary file and then is fed through the same ceph.conf
  89     loading mechanisms for testing.  Unlike ``conf_ceph`` which is just a fake,
  90     we are actually loading values as seen on a ceph.conf file
  91
  92     This is useful when more complex ceph.conf's are needed. In the case of
  93     just trying to validate a key/value behavior ``conf_ceph`` is better
  94     suited.
  95     """
  96     def apply(contents):
  97         conf_path = tmpfile(contents=contents)
  98         parser = configuration.load(conf_path)
  99         monkeypatch.setattr(conf, 'ceph', parser)
 100         return parser
 101     return apply
 102
 103
 104 @pytest.fixture
 105 def volumes(monkeypatch):
 106     monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0))
 107     volumes = lvm_api.Volumes()
 108     volumes._purge()
 109     return volumes
 110
 111
 112 @pytest.fixture
 113 def volume_groups(monkeypatch):
 114     monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0))
 115     vgs = lvm_api.VolumeGroups()
 116     vgs._purge()
 117     return vgs
 118
 119
 120 @pytest.fixture
 121 def pvolumes(monkeypatch):
 122     monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0))
 123     pvolumes = lvm_api.PVolumes()
 124     pvolumes._purge()
 125     return pvolumes
 126
 127
 128 @pytest.fixture
 129 def is_root(monkeypatch):
 130     """
 131     Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user
 132     is root (or is sudoing to superuser) can continue as-is
 133     """
 134     monkeypatch.setattr('os.getuid', lambda: 0)
 135
 136
 137 @pytest.fixture
 138 def tmpfile(tmpdir):
 139     """
 140     Create a temporary file, optionally filling it with contents, returns an
 141     absolute path to the file when called
 142     """
 143     def generate_file(name='file', contents='', directory=None):
 144         directory = directory or str(tmpdir)
 145         path = os.path.join(directory, name)
 146         with open(path, 'w') as fp:
 147             fp.write(contents)
 148         return path
 149     return generate_file
 150
 151
 152 @pytest.fixture
 153 def device_info(monkeypatch):
 154     def apply(devices=None, lsblk=None, lv=None):
 155         devices = devices if devices else {}
 156         lsblk = lsblk if lsblk else {}
 157         lv = Factory(**lv) if lv else None
 158         monkeypatch.setattr("ceph_volume.sys_info.devices", {})
 159         monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: devices)
 160         monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: lv)
 161         monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk)
 162     return apply