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
  14     def __call__(self, *a, **kw):
  15         self.calls.append({'args': a, 'kwargs': kw})
  16
  17
  18 class Factory(object):
  19
  20     def __init__(self, **kw):
  21         for k, v in kw.items():
  22             setattr(self, k, v)
  23
  24
  25 @pytest.fixture
  26 def factory():
  27     return Factory
  28
  29
  30 @pytest.fixture
  31 def capture():
  32     return Capture()
  33
  34
  35 @pytest.fixture
  36 def fake_run(monkeypatch):
  37     fake_run = Capture()
  38     monkeypatch.setattr('ceph_volume.process.run', fake_run)
  39     return fake_run
  40
  41
  42 @pytest.fixture
  43 def fake_call(monkeypatch):
  44     fake_call = Capture()
  45     monkeypatch.setattr('ceph_volume.process.call', fake_call)
  46     return fake_call
  47
  48
  49 @pytest.fixture
  50 def stub_call(monkeypatch):
  51     """
  52     Monkeypatches process.call, so that a caller can add behavior to the response
  53     """
  54     def apply(return_value):
  55         monkeypatch.setattr(
  56             'ceph_volume.process.call',
  57             lambda *a, **kw: return_value)
  58
  59     return apply
  60
  61
  62 @pytest.fixture
  63 def conf_ceph(monkeypatch):
  64     """
  65     Monkeypatches ceph_volume.conf.ceph, which is meant to parse/read
  66     a ceph.conf. The patching is naive, it allows one to set return values for
  67     specific method calls.
  68     """
  69     def apply(**kw):
  70         stub = Factory(**kw)
  71         monkeypatch.setattr(conf, 'ceph', stub)
  72         return stub
  73     return apply
  74
  75
  76 @pytest.fixture
  77 def conf_ceph_stub(monkeypatch, tmpfile):
  78     """
  79     Monkeypatches ceph_volume.conf.ceph with contents from a string that are
  80     written to a temporary file and then is fed through the same ceph.conf
  81     loading mechanisms for testing.  Unlike ``conf_ceph`` which is just a fake,
  82     we are actually loading values as seen on a ceph.conf file
  83
  84     This is useful when more complex ceph.conf's are needed. In the case of
  85     just trying to validate a key/value behavior ``conf_ceph`` is better
  86     suited.
  87     """
  88     def apply(contents):
  89         conf_path = tmpfile(contents=contents)
  90         parser = configuration.load(conf_path)
  91         monkeypatch.setattr(conf, 'ceph', parser)
  92         return parser
  93     return apply
  94
  95
  96 @pytest.fixture
  97 def volumes(monkeypatch):
  98     monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0))
  99     volumes = lvm_api.Volumes()
 100     volumes._purge()
 101     return volumes
 102
 103
 104 @pytest.fixture
 105 def volume_groups(monkeypatch):
 106     monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0))
 107     vgs = lvm_api.VolumeGroups()
 108     vgs._purge()
 109     return vgs
 110
 111
 112 @pytest.fixture
 113 def is_root(monkeypatch):
 114     """
 115     Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user
 116     is root (or is sudoing to superuser) can continue as-is
 117     """
 118     monkeypatch.setattr('os.getuid', lambda: 0)
 119
 120
 121 @pytest.fixture
 122 def tmpfile(tmpdir):
 123     """
 124     Create a temporary file, optionally filling it with contents, returns an
 125     absolute path to the file when called
 126     """
 127     def generate_file(name='file', contents=''):
 128         path = os.path.join(str(tmpdir), name)
 129         with open(path, 'w') as fp:
 130             fp.write(contents)
 131         return path
 132     return generate_file