diff options
Diffstat (limited to '')
| -rw-r--r-- | src/ceph-volume/ceph_volume/tests/conftest.py | 295 |
1 files changed, 295 insertions, 0 deletions
diff --git a/src/ceph-volume/ceph_volume/tests/conftest.py b/src/ceph-volume/ceph_volume/tests/conftest.py new file mode 100644 index 00000000..2abedac3 --- /dev/null +++ b/src/ceph-volume/ceph_volume/tests/conftest.py @@ -0,0 +1,295 @@ +import os +import pytest +from mock.mock import patch, PropertyMock, create_autospec +from ceph_volume.api import lvm +from ceph_volume.util import disk +from ceph_volume.util import device +from ceph_volume.util.constants import ceph_disk_guids +from ceph_volume import conf, configuration + + +class Capture(object): + + def __init__(self, *a, **kw): + self.a = a + self.kw = kw + self.calls = [] + self.return_values = kw.get('return_values', False) + self.always_returns = kw.get('always_returns', False) + + def __call__(self, *a, **kw): + self.calls.append({'args': a, 'kwargs': kw}) + if self.always_returns: + return self.always_returns + if self.return_values: + return self.return_values.pop() + + +class Factory(object): + + def __init__(self, **kw): + for k, v in kw.items(): + setattr(self, k, v) + + +@pytest.fixture +def factory(): + return Factory + + +@pytest.fixture +def capture(): + return Capture() + +@pytest.fixture +def mock_lv_device_generator(): + def mock_lv(): + size = 21474836480 + dev = create_autospec(device.Device) + dev.lv_name = 'lv' + dev.vg_name = 'vg' + dev.path = '{}/{}'.format(dev.vg_name, dev.lv_name) + dev.used_by_ceph = False + dev.vg_size = [size] + dev.vg_free = dev.vg_size + dev.lvs = [lvm.Volume(vg_name=dev.vg_name, lv_name=dev.lv_name, lv_size=size, lv_tags='')] + return dev + return mock_lv + + +@pytest.fixture +def mock_devices_available(): + dev = create_autospec(device.Device) + dev.path = '/dev/foo' + dev.available_lvm = True + dev.vg_size = [21474836480] + dev.vg_free = dev.vg_size + return [dev] + +@pytest.fixture +def mock_device_generator(): + def mock_device(): + dev = create_autospec(device.Device) + dev.path = '/dev/foo' + dev.available_lvm = True + dev.vg_size = [21474836480] + dev.vg_free = dev.vg_size + dev.lvs = [] + return dev + return mock_device + + +@pytest.fixture(params=range(1,11)) +def osds_per_device(request): + return request.param + + +@pytest.fixture +def fake_run(monkeypatch): + fake_run = Capture() + monkeypatch.setattr('ceph_volume.process.run', fake_run) + return fake_run + + +@pytest.fixture +def fake_call(monkeypatch): + fake_call = Capture(always_returns=([], [], 0)) + monkeypatch.setattr('ceph_volume.process.call', fake_call) + return fake_call + + +@pytest.fixture +def fakedevice(factory): + def apply(**kw): + params = dict( + path='/dev/sda', + abspath='/dev/sda', + lv_api=None, + pvs_api=[], + disk_api={}, + sys_api={}, + exists=True, + is_lvm_member=True, + ) + params.update(dict(kw)) + params['lvm_size'] = disk.Size(b=params['sys_api'].get("size", 0)) + return factory(**params) + return apply + + +@pytest.fixture +def stub_call(monkeypatch): + """ + Monkeypatches process.call, so that a caller can add behavior to the response + """ + def apply(return_values): + if isinstance(return_values, tuple): + return_values = [return_values] + stubbed_call = Capture(return_values=return_values) + monkeypatch.setattr('ceph_volume.process.call', stubbed_call) + return stubbed_call + + return apply + + +@pytest.fixture(autouse=True) +def reset_cluster_name(request, monkeypatch): + """ + The globally available ``ceph_volume.conf.cluster`` might get mangled in + tests, make sure that after evert test, it gets reset, preventing pollution + going into other tests later. + """ + def fin(): + conf.cluster = None + try: + os.environ.pop('CEPH_CONF') + except KeyError: + pass + request.addfinalizer(fin) + + +@pytest.fixture +def conf_ceph(monkeypatch): + """ + Monkeypatches ceph_volume.conf.ceph, which is meant to parse/read + a ceph.conf. The patching is naive, it allows one to set return values for + specific method calls. + """ + def apply(**kw): + stub = Factory(**kw) + monkeypatch.setattr(conf, 'ceph', stub) + return stub + return apply + + +@pytest.fixture +def conf_ceph_stub(monkeypatch, tmpfile): + """ + Monkeypatches ceph_volume.conf.ceph with contents from a string that are + written to a temporary file and then is fed through the same ceph.conf + loading mechanisms for testing. Unlike ``conf_ceph`` which is just a fake, + we are actually loading values as seen on a ceph.conf file + + This is useful when more complex ceph.conf's are needed. In the case of + just trying to validate a key/value behavior ``conf_ceph`` is better + suited. + """ + def apply(contents): + conf_path = tmpfile(contents=contents) + parser = configuration.load(conf_path) + monkeypatch.setattr(conf, 'ceph', parser) + return parser + return apply + + +@pytest.fixture +def is_root(monkeypatch): + """ + Patch ``os.getuid()`` so that ceph-volume's decorators that ensure a user + is root (or is sudoing to superuser) can continue as-is + """ + monkeypatch.setattr('os.getuid', lambda: 0) + + +@pytest.fixture +def tmpfile(tmpdir): + """ + Create a temporary file, optionally filling it with contents, returns an + absolute path to the file when called + """ + def generate_file(name='file', contents='', directory=None): + directory = directory or str(tmpdir) + path = os.path.join(directory, name) + with open(path, 'w') as fp: + fp.write(contents) + return path + return generate_file + + +@pytest.fixture +def disable_kernel_queries(monkeypatch): + ''' + This speeds up calls to Device and Disk + ''' + monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: {}) + monkeypatch.setattr("ceph_volume.util.disk.udevadm_property", lambda *a, **kw: {}) + + +@pytest.fixture(params=[ + '', 'ceph data', 'ceph journal', 'ceph block', + 'ceph block.wal', 'ceph block.db', 'ceph lockbox']) +def ceph_partlabel(request): + return request.param + + +@pytest.fixture(params=list(ceph_disk_guids.keys())) +def ceph_parttype(request): + return request.param + + +@pytest.fixture +def lsblk_ceph_disk_member(monkeypatch, request, ceph_partlabel, ceph_parttype): + monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", + lambda path: {'TYPE': 'disk', 'PARTLABEL': ceph_partlabel}) + # setting blkid here too in order to be able to fall back to PARTTYPE based + # membership + monkeypatch.setattr("ceph_volume.util.device.disk.blkid", + lambda path: {'TYPE': 'disk', + 'PARTLABEL': '', + 'PARTTYPE': ceph_parttype}) + + +@pytest.fixture +def blkid_ceph_disk_member(monkeypatch, request, ceph_partlabel, ceph_parttype): + monkeypatch.setattr("ceph_volume.util.device.disk.blkid", + lambda path: {'TYPE': 'disk', + 'PARTLABEL': ceph_partlabel, + 'PARTTYPE': ceph_parttype}) + + +@pytest.fixture(params=[ + ('gluster partition', 'gluster partition'), + # falls back to blkid + ('', 'gluster partition'), + ('gluster partition', ''), +]) +def device_info_not_ceph_disk_member(monkeypatch, request): + monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", + lambda path: {'TYPE': 'disk', + 'PARTLABEL': request.param[0]}) + monkeypatch.setattr("ceph_volume.util.device.disk.blkid", + lambda path: {'TYPE': 'disk', + 'PARTLABEL': request.param[1]}) + +@pytest.fixture +def patched_get_block_devs_lsblk(): + with patch('ceph_volume.util.disk.get_block_devs_lsblk') as p: + yield p + +@pytest.fixture +def patch_bluestore_label(): + with patch('ceph_volume.util.device.Device.has_bluestore_label', + new_callable=PropertyMock) as p: + p.return_value = False + yield p + +@pytest.fixture +def device_info(monkeypatch, patch_bluestore_label): + def apply(devices=None, lsblk=None, lv=None, blkid=None, udevadm=None, + has_bluestore_label=False): + devices = devices if devices else {} + lsblk = lsblk if lsblk else {} + blkid = blkid if blkid else {} + udevadm = udevadm if udevadm else {} + lv = Factory(**lv) if lv else None + monkeypatch.setattr("ceph_volume.sys_info.devices", {}) + monkeypatch.setattr("ceph_volume.util.device.disk.get_devices", lambda: devices) + if not devices: + monkeypatch.setattr("ceph_volume.util.device.lvm.get_first_lv", lambda filters: lv) + else: + monkeypatch.setattr("ceph_volume.util.device.lvm.get_device_lvs", + lambda path: [lv]) + monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk) + monkeypatch.setattr("ceph_volume.util.device.disk.blkid", lambda path: blkid) + monkeypatch.setattr("ceph_volume.util.disk.udevadm_property", lambda *a, **kw: udevadm) + return apply |