]>
Commit | Line | Data |
---|---|---|
1 | import pytest | |
2 | from ceph_volume.devices import lvm | |
3 | from ceph_volume.api import lvm as api | |
4 | ||
5 | # TODO: add tests for following commands - | |
6 | # ceph-volume list | |
7 | # ceph-volume list <path-to-pv> | |
8 | # ceph-volume list <path-to-vg> | |
9 | # ceph-volume list <path-to-lv> | |
10 | ||
11 | class TestReadableTag(object): | |
12 | ||
13 | def test_dots_get_replaced(self): | |
14 | result = lvm.listing.readable_tag('ceph.foo') | |
15 | assert result == 'foo' | |
16 | ||
17 | def test_underscores_are_replaced_with_spaces(self): | |
18 | result = lvm.listing.readable_tag('ceph.long_tag') | |
19 | assert result == 'long tag' | |
20 | ||
21 | ||
22 | class TestPrettyReport(object): | |
23 | ||
24 | def test_is_empty(self, capsys): | |
25 | lvm.listing.pretty_report({}) | |
26 | stdout, stderr = capsys.readouterr() | |
27 | assert stdout == '\n' | |
28 | ||
29 | def test_type_and_path_are_reported(self, capsys): | |
30 | lvm.listing.pretty_report({0: [ | |
31 | {'type': 'data', 'path': '/dev/sda1', 'devices': ['/dev/sda']} | |
32 | ]}) | |
33 | stdout, stderr = capsys.readouterr() | |
34 | assert '[data] /dev/sda1' in stdout | |
35 | ||
36 | def test_osd_id_header_is_reported(self, capsys): | |
37 | lvm.listing.pretty_report({0: [ | |
38 | {'type': 'data', 'path': '/dev/sda1', 'devices': ['/dev/sda']} | |
39 | ]}) | |
40 | stdout, stderr = capsys.readouterr() | |
41 | assert '====== osd.0 =======' in stdout | |
42 | ||
43 | def test_tags_are_included(self, capsys): | |
44 | lvm.listing.pretty_report( | |
45 | {0: [{ | |
46 | 'type': 'data', | |
47 | 'path': '/dev/sda1', | |
48 | 'tags': {'ceph.osd_id': '0'}, | |
49 | 'devices': ['/dev/sda'], | |
50 | }]} | |
51 | ) | |
52 | stdout, stderr = capsys.readouterr() | |
53 | assert 'osd id' in stdout | |
54 | ||
55 | def test_devices_are_comma_separated(self, capsys): | |
56 | lvm.listing.pretty_report({0: [ | |
57 | {'type': 'data', 'path': '/dev/sda1', 'devices': ['/dev/sda', '/dev/sdb1']} | |
58 | ]}) | |
59 | stdout, stderr = capsys.readouterr() | |
60 | assert '/dev/sda,/dev/sdb1' in stdout | |
61 | ||
62 | ||
63 | class TestList(object): | |
64 | ||
65 | def test_empty_full_json_zero_exit_status(self, fake_call, is_root, factory, capsys): | |
66 | args = factory(format='json', device=None) | |
67 | lvm.listing.List([]).list(args) | |
68 | stdout, stderr = capsys.readouterr() | |
69 | assert stdout == '{}\n' | |
70 | ||
71 | def test_empty_device_json_zero_exit_status(self, is_root,factory,capsys): | |
72 | args = factory(format='json', device='/dev/sda1') | |
73 | lvm.listing.List([]).list(args) | |
74 | stdout, stderr = capsys.readouterr() | |
75 | assert stdout == '{}\n' | |
76 | ||
77 | def test_empty_full_zero_exit_status(self, fake_call, is_root, factory): | |
78 | args = factory(format='pretty', device=None) | |
79 | with pytest.raises(SystemExit): | |
80 | lvm.listing.List([]).list(args) | |
81 | ||
82 | def test_empty_device_zero_exit_status(self, is_root, factory): | |
83 | args = factory(format='pretty', device='/dev/sda1') | |
84 | with pytest.raises(SystemExit): | |
85 | lvm.listing.List([]).list(args) | |
86 | ||
87 | class TestFullReport(object): | |
88 | ||
89 | def test_no_ceph_lvs(self, monkeypatch): | |
90 | # ceph lvs are detected by looking into its tags | |
91 | osd = api.Volume(lv_name='volume1', lv_path='/dev/VolGroup/lv', | |
92 | lv_tags={}) | |
93 | volumes = [] | |
94 | volumes.append(osd) | |
95 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
96 | volumes) | |
97 | ||
98 | result = lvm.listing.List([]).full_report() | |
99 | assert result == {} | |
100 | ||
101 | def test_ceph_data_lv_reported(self, monkeypatch): | |
102 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data' | |
103 | pv = api.PVolume(pv_name='/dev/sda1', pv_tags={}, pv_uuid="0000", | |
104 | vg_name='VolGroup', lv_uuid="aaaa") | |
105 | osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags, | |
106 | lv_path='/dev/VolGroup/lv', vg_name='VolGroup') | |
107 | volumes = [] | |
108 | volumes.append(osd) | |
109 | monkeypatch.setattr(lvm.listing.api, 'get_single_pv', lambda **kwargs: pv) | |
110 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
111 | volumes) | |
112 | ||
113 | result = lvm.listing.List([]).full_report() | |
114 | assert result['0'][0]['name'] == 'volume1' | |
115 | ||
116 | def test_ceph_journal_lv_reported(self, monkeypatch): | |
117 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data' | |
118 | journal_tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=journal' | |
119 | pv = api.PVolume(pv_name='/dev/sda1', pv_tags={}, pv_uuid="0000", | |
120 | vg_name="VolGroup", lv_uuid="aaaa") | |
121 | osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags, | |
122 | lv_path='/dev/VolGroup/lv', vg_name='VolGroup') | |
123 | journal = api.Volume( | |
124 | lv_name='journal', lv_uuid='x', lv_tags=journal_tags, | |
125 | lv_path='/dev/VolGroup/journal', vg_name='VolGroup') | |
126 | volumes = [] | |
127 | volumes.append(osd) | |
128 | volumes.append(journal) | |
129 | monkeypatch.setattr(lvm.listing.api,'get_single_pv',lambda **kwargs:pv) | |
130 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
131 | volumes) | |
132 | ||
133 | result = lvm.listing.List([]).full_report() | |
134 | assert result['0'][0]['name'] == 'volume1' | |
135 | assert result['0'][1]['name'] == 'journal' | |
136 | ||
137 | def test_ceph_wal_lv_reported(self, monkeypatch): | |
138 | tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=data' | |
139 | wal_tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=wal' | |
140 | osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags, | |
141 | lv_path='/dev/VolGroup/lv', vg_name='VolGroup') | |
142 | wal = api.Volume(lv_name='wal', lv_uuid='x', lv_tags=wal_tags, | |
143 | lv_path='/dev/VolGroup/wal', vg_name='VolGroup') | |
144 | volumes = [] | |
145 | volumes.append(osd) | |
146 | volumes.append(wal) | |
147 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
148 | volumes) | |
149 | ||
150 | result = lvm.listing.List([]).full_report() | |
151 | assert result['0'][0]['name'] == 'volume1' | |
152 | assert result['0'][1]['name'] == 'wal' | |
153 | ||
154 | @pytest.mark.parametrize('type_', ['journal', 'db', 'wal']) | |
155 | def test_physical_2nd_device_gets_reported(self, type_, monkeypatch): | |
156 | tags = ('ceph.osd_id=0,ceph.{t}_uuid=x,ceph.type=data,' | |
157 | 'ceph.{t}_device=/dev/sda1').format(t=type_) | |
158 | osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags, | |
159 | vg_name='VolGroup', lv_path='/dev/VolGroup/lv') | |
160 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
161 | [osd]) | |
162 | ||
163 | result = lvm.listing.List([]).full_report() | |
164 | assert result['0'][1]['path'] == '/dev/sda1' | |
165 | assert result['0'][1]['tags'] == {'PARTUUID': 'x'} | |
166 | assert result['0'][1]['type'] == type_ | |
167 | ||
168 | ||
169 | class TestSingleReport(object): | |
170 | ||
171 | def test_not_a_ceph_lv(self, monkeypatch): | |
172 | # ceph lvs are detected by looking into its tags | |
173 | lv = api.Volume(lv_name='lv', lv_tags={}, lv_path='/dev/VolGroup/lv', | |
174 | vg_name='VolGroup') | |
175 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
176 | [lv]) | |
177 | ||
178 | result = lvm.listing.List([]).single_report('VolGroup/lv') | |
179 | assert result == {} | |
180 | ||
181 | def test_report_a_ceph_lv(self, monkeypatch): | |
182 | # ceph lvs are detected by looking into its tags | |
183 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data' | |
184 | lv = api.Volume(lv_name='lv', vg_name='VolGroup', lv_uuid='aaaa', | |
185 | lv_path='/dev/VolGroup/lv', lv_tags=tags) | |
186 | volumes = [] | |
187 | volumes.append(lv) | |
188 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
189 | volumes) | |
190 | ||
191 | result = lvm.listing.List([]).single_report('VolGroup/lv') | |
192 | assert result['0'][0]['name'] == 'lv' | |
193 | assert result['0'][0]['lv_tags'] == tags | |
194 | assert result['0'][0]['path'] == '/dev/VolGroup/lv' | |
195 | assert result['0'][0]['devices'] == [] | |
196 | ||
197 | def test_report_a_ceph_journal_device(self, monkeypatch): | |
198 | # ceph lvs are detected by looking into its tags | |
199 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data,' + \ | |
200 | 'ceph.journal_device=/dev/sda1' | |
201 | lv = api.Volume(lv_name='lv', lv_uuid='aaa', lv_tags=tags, | |
202 | lv_path='/dev/VolGroup/lv', vg_name='VolGroup') | |
203 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
204 | [lv] if 'tags' in kwargs else []) | |
205 | ||
206 | result = lvm.listing.List([]).single_report('/dev/sda1') | |
207 | assert result['0'][0]['tags'] == {'PARTUUID': 'x'} | |
208 | assert result['0'][0]['type'] == 'journal' | |
209 | assert result['0'][0]['path'] == '/dev/sda1' | |
210 | ||
211 | def test_report_a_ceph_lv_with_devices(self, monkeypatch): | |
212 | pvolumes = [] | |
213 | ||
214 | tags = 'ceph.osd_id=0,ceph.type=data' | |
215 | pv1 = api.PVolume(vg_name="VolGroup", pv_name='/dev/sda1', | |
216 | pv_uuid='', pv_tags={}, lv_uuid="aaaa") | |
217 | pv2 = api.PVolume(vg_name="VolGroup", pv_name='/dev/sdb1', | |
218 | pv_uuid='', pv_tags={}, lv_uuid="aaaa") | |
219 | pvolumes.append(pv1) | |
220 | pvolumes.append(pv2) | |
221 | ||
222 | ||
223 | volumes = [] | |
224 | lv = api.Volume(lv_name='lv', vg_name='VolGroup',lv_uuid='aaaa', | |
225 | lv_path='/dev/VolGroup/lv', lv_tags=tags) | |
226 | volumes.append(lv) | |
227 | ||
228 | monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs: | |
229 | pvolumes) | |
230 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
231 | volumes) | |
232 | ||
233 | listing = lvm.listing.List([]) | |
234 | listing._pvs = [ | |
235 | {'lv_uuid': 'aaaa', 'pv_name': '/dev/sda1', 'pv_tags': '', 'pv_uuid': ''}, | |
236 | {'lv_uuid': 'aaaa', 'pv_name': '/dev/sdb1', 'pv_tags': '', 'pv_uuid': ''}, | |
237 | ] | |
238 | ||
239 | result = listing.single_report('VolGroup/lv') | |
240 | assert result['0'][0]['name'] == 'lv' | |
241 | assert result['0'][0]['lv_tags'] == tags | |
242 | assert result['0'][0]['path'] == '/dev/VolGroup/lv' | |
243 | assert result['0'][0]['devices'] == ['/dev/sda1', '/dev/sdb1'] | |
244 | ||
245 | def test_report_by_osd_id_for_just_block_dev(self, monkeypatch): | |
246 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=block' | |
247 | lvs = [ api.Volume(lv_name='lv1', lv_tags=tags, lv_path='/dev/vg/lv1', | |
248 | lv_uuid='aaaa', vg_name='vg') | |
249 | ] | |
250 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs) | |
251 | ||
252 | listing = lvm.listing.List([]) | |
253 | result = listing.single_report(0) | |
254 | assert result['0'][0]['name'] == 'lv1' | |
255 | assert result['0'][0]['lv_tags'] == tags | |
256 | assert result['0'][0]['lv_path'] == '/dev/vg/lv1' | |
257 | assert result['0'][0]['vg_name'] == 'vg' | |
258 | ||
259 | def test_report_by_osd_id_for_just_data_dev(self, monkeypatch): | |
260 | tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data' | |
261 | lvs = [ api.Volume(lv_name='lv1', lv_tags=tags, lv_path='/dev/vg/lv1', | |
262 | lv_uuid='bbbb', vg_name='vg'), | |
263 | ] | |
264 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs) | |
265 | ||
266 | listing = lvm.listing.List([]) | |
267 | result = listing.single_report(0) | |
268 | assert result['0'][0]['name'] == 'lv1' | |
269 | assert result['0'][0]['lv_tags'] == tags | |
270 | assert result['0'][0]['lv_path'] == '/dev/vg/lv1' | |
271 | assert result['0'][0]['vg_name'] == 'vg' | |
272 | ||
273 | def test_report_by_osd_id_for_just_block_wal_and_db_dev(self, monkeypatch): | |
274 | tags1 = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=block' | |
275 | tags2 = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=wal' | |
276 | tags3 = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=db' | |
277 | lvs = [ api.Volume(lv_name='lv1', lv_tags=tags1, lv_path='/dev/vg/lv1', | |
278 | lv_uuid='aaaa', vg_name='vg'), | |
279 | api.Volume(lv_name='lv2', lv_tags=tags2, lv_path='/dev/vg/lv2', | |
280 | lv_uuid='bbbb', vg_name='vg'), | |
281 | api.Volume(lv_name='lv3', lv_tags=tags3, lv_path='/dev/vg/lv3', | |
282 | lv_uuid='cccc', vg_name='vg'), | |
283 | ] | |
284 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs) | |
285 | ||
286 | listing = lvm.listing.List([]) | |
287 | result = listing.single_report(0) | |
288 | assert result['0'][0]['name'] == 'lv1' | |
289 | assert result['0'][0]['lv_tags'] == tags1 | |
290 | assert result['0'][0]['lv_path'] == '/dev/vg/lv1' | |
291 | assert result['0'][0]['vg_name'] == 'vg' | |
292 | assert result['0'][1]['name'] == 'lv2' | |
293 | assert result['0'][1]['lv_tags'] == tags2 | |
294 | assert result['0'][1]['lv_path'] == '/dev/vg/lv2' | |
295 | assert result['0'][1]['vg_name'] == 'vg' | |
296 | assert result['0'][2]['name'] == 'lv3' | |
297 | assert result['0'][2]['lv_tags'] == tags3 | |
298 | assert result['0'][2]['lv_path'] == '/dev/vg/lv3' | |
299 | assert result['0'][2]['vg_name'] == 'vg' | |
300 | ||
301 | ||
302 | def test_report_by_osd_id_for_data_and_journal_dev(self, monkeypatch): | |
303 | tags1 = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data' | |
304 | tags2 = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=journal' | |
305 | lvs = [ api.Volume(lv_name='lv1', lv_tags=tags1, lv_path='/dev/vg/lv1', | |
306 | lv_uuid='aaaa', vg_name='vg'), | |
307 | api.Volume(lv_name='lv2', lv_tags=tags2, lv_path='/dev/vg/lv2', | |
308 | lv_uuid='bbbb', vg_name='vg'), | |
309 | ] | |
310 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: lvs) | |
311 | ||
312 | listing = lvm.listing.List([]) | |
313 | result = listing.single_report(0) | |
314 | assert result['0'][0]['name'] == 'lv1' | |
315 | assert result['0'][0]['lv_tags'] == tags1 | |
316 | assert result['0'][0]['lv_path'] == '/dev/vg/lv1' | |
317 | assert result['0'][0]['vg_name'] == 'vg' | |
318 | assert result['0'][1]['name'] == 'lv2' | |
319 | assert result['0'][1]['lv_tags'] == tags2 | |
320 | assert result['0'][1]['lv_path'] == '/dev/vg/lv2' | |
321 | assert result['0'][1]['vg_name'] == 'vg' | |
322 | ||
323 | def test_report_by_nonexistent_osd_id(self, monkeypatch): | |
324 | lv = api.Volume(lv_name='lv', lv_tags={}, lv_path='/dev/VolGroup/lv', | |
325 | vg_name='VolGroup') | |
326 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
327 | [lv]) | |
328 | ||
329 | result = lvm.listing.List([]).single_report('1') | |
330 | assert result == {} | |
331 | ||
332 | def test_report_a_ceph_lv_with_no_matching_devices(self, monkeypatch): | |
333 | tags = 'ceph.osd_id=0,ceph.type=data' | |
334 | lv = api.Volume(lv_name='lv', vg_name='VolGroup', lv_uuid='aaaa', | |
335 | lv_path='/dev/VolGroup/lv', lv_tags=tags) | |
336 | volumes = [] | |
337 | volumes.append(lv) | |
338 | monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs: | |
339 | volumes) | |
340 | ||
341 | listing = lvm.listing.List([]) | |
342 | listing._pvs = [ | |
343 | {'lv_uuid': 'ffff', 'pv_name': '/dev/sda1', 'pv_tags': '', | |
344 | 'pv_uuid': ''}, | |
345 | {'lv_uuid': 'ffff', 'pv_name': '/dev/sdb1', 'pv_tags': '', | |
346 | 'pv_uuid': ''}] | |
347 | ||
348 | result = listing.single_report('VolGroup/lv') | |
349 | assert result['0'][0]['name'] == 'lv' | |
350 | assert result['0'][0]['lv_tags'] == tags | |
351 | assert result['0'][0]['path'] == '/dev/VolGroup/lv' | |
352 | assert result['0'][0]['devices'] == [] |