]>
Commit | Line | Data |
---|---|---|
3efd9988 | 1 | import os |
d2e6a577 | 2 | import pytest |
11fdf7f2 | 3 | from ceph_volume.util import disk |
3efd9988 | 4 | from ceph_volume.api import lvm as lvm_api |
b32b8144 | 5 | from ceph_volume import conf, configuration |
3efd9988 | 6 | |
d2e6a577 FG |
7 | |
8 | class Capture(object): | |
9 | ||
10 | def __init__(self, *a, **kw): | |
11 | self.a = a | |
12 | self.kw = kw | |
13 | self.calls = [] | |
28e407b8 AA |
14 | self.return_values = kw.get('return_values', False) |
15 | self.always_returns = kw.get('always_returns', False) | |
d2e6a577 FG |
16 | |
17 | def __call__(self, *a, **kw): | |
18 | self.calls.append({'args': a, 'kwargs': kw}) | |
28e407b8 AA |
19 | if self.always_returns: |
20 | return self.always_returns | |
21 | if self.return_values: | |
22 | return self.return_values.pop() | |
d2e6a577 FG |
23 | |
24 | ||
3efd9988 FG |
25 | class Factory(object): |
26 | ||
27 | def __init__(self, **kw): | |
28 | for k, v in kw.items(): | |
29 | setattr(self, k, v) | |
30 | ||
31 | ||
32 | @pytest.fixture | |
33 | def factory(): | |
34 | return Factory | |
35 | ||
36 | ||
d2e6a577 FG |
37 | @pytest.fixture |
38 | def capture(): | |
39 | return Capture() | |
181888fb FG |
40 | |
41 | ||
b32b8144 FG |
42 | @pytest.fixture |
43 | def fake_run(monkeypatch): | |
44 | fake_run = Capture() | |
45 | monkeypatch.setattr('ceph_volume.process.run', fake_run) | |
46 | return fake_run | |
47 | ||
48 | ||
49 | @pytest.fixture | |
50 | def fake_call(monkeypatch): | |
28e407b8 | 51 | fake_call = Capture(always_returns=([], [], 0)) |
b32b8144 FG |
52 | monkeypatch.setattr('ceph_volume.process.call', fake_call) |
53 | return fake_call | |
54 | ||
55 | ||
91327a77 AA |
56 | @pytest.fixture |
57 | def fakedevice(factory): | |
58 | def apply(**kw): | |
59 | params = dict( | |
60 | path='/dev/sda', | |
61 | abspath='/dev/sda', | |
62 | lv_api=None, | |
63 | pvs_api=[], | |
64 | disk_api={}, | |
65 | sys_api={}, | |
66 | exists=True, | |
67 | is_lvm_member=True, | |
68 | ) | |
69 | params.update(dict(kw)) | |
11fdf7f2 | 70 | params['lvm_size'] = disk.Size(b=params['sys_api'].get("size", 0)) |
91327a77 AA |
71 | return factory(**params) |
72 | return apply | |
73 | ||
74 | ||
b32b8144 FG |
75 | @pytest.fixture |
76 | def stub_call(monkeypatch): | |
77 | """ | |
78 | Monkeypatches process.call, so that a caller can add behavior to the response | |
79 | """ | |
28e407b8 AA |
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) | |
85 | return stubbed_call | |
b32b8144 FG |
86 | |
87 | return apply | |
88 | ||
89 | ||
91327a77 AA |
90 | @pytest.fixture(autouse=True) |
91 | def reset_cluster_name(request, monkeypatch): | |
92 | """ | |
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. | |
96 | """ | |
97 | def fin(): | |
98 | conf.cluster = None | |
99 | try: | |
100 | os.environ.pop('CEPH_CONF') | |
101 | except KeyError: | |
102 | pass | |
103 | request.addfinalizer(fin) | |
104 | ||
105 | ||
b32b8144 FG |
106 | @pytest.fixture |
107 | def conf_ceph(monkeypatch): | |
108 | """ | |
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. | |
112 | """ | |
113 | def apply(**kw): | |
114 | stub = Factory(**kw) | |
115 | monkeypatch.setattr(conf, 'ceph', stub) | |
116 | return stub | |
117 | return apply | |
118 | ||
119 | ||
120 | @pytest.fixture | |
121 | def conf_ceph_stub(monkeypatch, tmpfile): | |
122 | """ | |
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 | |
127 | ||
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 | |
130 | suited. | |
131 | """ | |
132 | def apply(contents): | |
133 | conf_path = tmpfile(contents=contents) | |
134 | parser = configuration.load(conf_path) | |
135 | monkeypatch.setattr(conf, 'ceph', parser) | |
136 | return parser | |
137 | return apply | |
138 | ||
139 | ||
181888fb FG |
140 | @pytest.fixture |
141 | def volumes(monkeypatch): | |
91327a77 | 142 | monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0)) |
3efd9988 | 143 | volumes = lvm_api.Volumes() |
181888fb FG |
144 | volumes._purge() |
145 | return volumes | |
146 | ||
147 | ||
148 | @pytest.fixture | |
149 | def volume_groups(monkeypatch): | |
91327a77 | 150 | monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0)) |
3efd9988 | 151 | vgs = lvm_api.VolumeGroups() |
181888fb FG |
152 | vgs._purge() |
153 | return vgs | |
154 | ||
155 | ||
91327a77 AA |
156 | @pytest.fixture |
157 | def stub_vgs(monkeypatch, volume_groups): | |
158 | def apply(vgs): | |
159 | monkeypatch.setattr(lvm_api, 'get_api_vgs', lambda: vgs) | |
160 | return apply | |
161 | ||
162 | ||
1adf2230 AA |
163 | @pytest.fixture |
164 | def pvolumes(monkeypatch): | |
91327a77 | 165 | monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0)) |
1adf2230 AA |
166 | pvolumes = lvm_api.PVolumes() |
167 | pvolumes._purge() | |
168 | return pvolumes | |
169 | ||
170 | ||
181888fb FG |
171 | @pytest.fixture |
172 | def is_root(monkeypatch): | |
173 | """ | |
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 | |
176 | """ | |
177 | monkeypatch.setattr('os.getuid', lambda: 0) | |
3efd9988 FG |
178 | |
179 | ||
180 | @pytest.fixture | |
181 | def tmpfile(tmpdir): | |
182 | """ | |
183 | Create a temporary file, optionally filling it with contents, returns an | |
184 | absolute path to the file when called | |
185 | """ | |
1adf2230 AA |
186 | def generate_file(name='file', contents='', directory=None): |
187 | directory = directory or str(tmpdir) | |
188 | path = os.path.join(directory, name) | |
3efd9988 FG |
189 | with open(path, 'w') as fp: |
190 | fp.write(contents) | |
191 | return path | |
192 | return generate_file | |
1adf2230 AA |
193 | |
194 | ||
195 | @pytest.fixture | |
196 | def device_info(monkeypatch): | |
f64942e4 | 197 | def apply(devices=None, lsblk=None, lv=None, blkid=None, udevadm=None): |
1adf2230 AA |
198 | devices = devices if devices else {} |
199 | lsblk = lsblk if lsblk else {} | |
91327a77 | 200 | blkid = blkid if blkid else {} |
f64942e4 | 201 | udevadm = udevadm if udevadm else {} |
1adf2230 AA |
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) | |
91327a77 AA |
205 | if not devices: |
206 | monkeypatch.setattr("ceph_volume.util.device.lvm.get_lv_from_argument", lambda path: lv) | |
207 | else: | |
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) | |
1adf2230 | 210 | monkeypatch.setattr("ceph_volume.util.device.disk.lsblk", lambda path: lsblk) |
91327a77 | 211 | monkeypatch.setattr("ceph_volume.util.device.disk.blkid", lambda path: blkid) |
f64942e4 | 212 | monkeypatch.setattr("ceph_volume.util.disk.udevadm_property", lambda *a, **kw: udevadm) |
1adf2230 | 213 | return apply |