]>
Commit | Line | Data |
---|---|---|
3efd9988 | 1 | import os |
d2e6a577 | 2 | import pytest |
3efd9988 | 3 | from ceph_volume.api import lvm as lvm_api |
b32b8144 | 4 | from ceph_volume import conf, configuration |
3efd9988 | 5 | |
d2e6a577 FG |
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 | ||
3efd9988 FG |
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 | ||
d2e6a577 FG |
30 | @pytest.fixture |
31 | def capture(): | |
32 | return Capture() | |
181888fb FG |
33 | |
34 | ||
b32b8144 FG |
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 | ||
181888fb FG |
96 | @pytest.fixture |
97 | def volumes(monkeypatch): | |
98 | monkeypatch.setattr('ceph_volume.process.call', lambda x: ('', '', 0)) | |
3efd9988 | 99 | volumes = lvm_api.Volumes() |
181888fb FG |
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)) | |
3efd9988 | 107 | vgs = lvm_api.VolumeGroups() |
181888fb FG |
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) | |
3efd9988 FG |
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 |