]>
git.proxmox.com Git - ceph.git/blob - ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_zap.py
3 from copy
import deepcopy
4 from mock
.mock
import patch
, call
5 from ceph_volume
import process
6 from ceph_volume
.api
import lvm
as api
7 from ceph_volume
.devices
.lvm
import zap
10 class TestZap(object):
11 def test_invalid_osd_id_passed(self
):
12 with pytest
.raises(SystemExit):
13 zap
.Zap(argv
=['--osd-id', 'foo']).main()
15 class TestFindAssociatedDevices(object):
17 def test_no_lvs_found_that_match_id(self
, monkeypatch
, device_info
):
18 tags
= 'ceph.osd_id=9,ceph.journal_uuid=x,ceph.type=data'
19 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', vg_name
='vg',
20 lv_tags
=tags
, lv_path
='/dev/VolGroup/lv')
23 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kwargs
: {})
25 with pytest
.raises(RuntimeError):
26 zap
.find_associated_devices(osd_id
=10)
28 def test_no_lvs_found_that_match_fsid(self
, monkeypatch
, device_info
):
29 tags
= 'ceph.osd_id=9,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,'+\
31 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', lv_tags
=tags
,
32 vg_name
='vg', lv_path
='/dev/VolGroup/lv')
35 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kwargs
: {})
37 with pytest
.raises(RuntimeError):
38 zap
.find_associated_devices(osd_fsid
='aaaa-lkjh')
40 def test_no_lvs_found_that_match_id_fsid(self
, monkeypatch
, device_info
):
41 tags
= 'ceph.osd_id=9,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,'+\
43 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', vg_name
='vg',
44 lv_tags
=tags
, lv_path
='/dev/VolGroup/lv')
47 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kwargs
: {})
49 with pytest
.raises(RuntimeError):
50 zap
.find_associated_devices(osd_id
='9', osd_fsid
='aaaa-lkjh')
52 def test_no_ceph_lvs_found(self
, monkeypatch
):
53 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', lv_tags
='',
54 lv_path
='/dev/VolGroup/lv')
57 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kwargs
: {})
59 with pytest
.raises(RuntimeError):
60 zap
.find_associated_devices(osd_id
=100)
62 def test_lv_is_matched_id(self
, monkeypatch
):
63 tags
= 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
64 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', vg_name
='',
65 lv_path
='/dev/VolGroup/lv', lv_tags
=tags
)
68 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kw
: volumes
)
69 monkeypatch
.setattr(process
, 'call', lambda x
, **kw
: ('', '', 0))
71 result
= zap
.find_associated_devices(osd_id
='0')
72 assert result
[0].abspath
== '/dev/VolGroup/lv'
74 def test_lv_is_matched_fsid(self
, monkeypatch
):
75 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,' +\
77 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', vg_name
='',
78 lv_path
='/dev/VolGroup/lv', lv_tags
=tags
)
81 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kw
: deepcopy(volumes
))
82 monkeypatch
.setattr(process
, 'call', lambda x
, **kw
: ('', '', 0))
84 result
= zap
.find_associated_devices(osd_fsid
='asdf-lkjh')
85 assert result
[0].abspath
== '/dev/VolGroup/lv'
87 def test_lv_is_matched_id_fsid(self
, monkeypatch
):
88 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,' +\
90 osd
= api
.Volume(lv_name
='volume1', lv_uuid
='y', vg_name
='',
91 lv_path
='/dev/VolGroup/lv', lv_tags
=tags
)
94 monkeypatch
.setattr(zap
.api
, 'get_lvs', lambda **kw
: volumes
)
95 monkeypatch
.setattr(process
, 'call', lambda x
, **kw
: ('', '', 0))
97 result
= zap
.find_associated_devices(osd_id
='0', osd_fsid
='asdf-lkjh')
98 assert result
[0].abspath
== '/dev/VolGroup/lv'
101 class TestEnsureAssociatedLVs(object):
103 def test_nothing_is_found(self
):
105 result
= zap
.ensure_associated_lvs(volumes
)
108 def test_data_is_found(self
):
109 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=data'
111 lv_name
='volume1', lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/data', lv_tags
=tags
)
114 result
= zap
.ensure_associated_lvs(volumes
)
115 assert result
== ['/dev/VolGroup/data']
117 def test_block_is_found(self
):
118 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=block'
120 lv_name
='volume1', lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/block', lv_tags
=tags
)
123 result
= zap
.ensure_associated_lvs(volumes
)
124 assert result
== ['/dev/VolGroup/block']
126 def test_success_message_for_fsid(self
, factory
, is_root
, capsys
):
127 cli_zap
= zap
.Zap([])
128 args
= factory(devices
=[], osd_id
=None, osd_fsid
='asdf-lkjh')
131 out
, err
= capsys
.readouterr()
132 assert "Zapping successful for OSD: asdf-lkjh" in err
134 def test_success_message_for_id(self
, factory
, is_root
, capsys
):
135 cli_zap
= zap
.Zap([])
136 args
= factory(devices
=[], osd_id
='1', osd_fsid
=None)
139 out
, err
= capsys
.readouterr()
140 assert "Zapping successful for OSD: 1" in err
142 def test_block_and_partition_are_found(self
, monkeypatch
):
143 monkeypatch
.setattr(zap
.disk
, 'get_device_from_partuuid', lambda x
: '/dev/sdb1')
144 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=block'
146 lv_name
='volume1', lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/block', lv_tags
=tags
)
149 result
= zap
.ensure_associated_lvs(volumes
)
150 assert '/dev/sdb1' in result
151 assert '/dev/VolGroup/block' in result
153 def test_journal_is_found(self
):
154 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=journal'
156 lv_name
='volume1', lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/lv', lv_tags
=tags
)
159 result
= zap
.ensure_associated_lvs(volumes
)
160 assert result
== ['/dev/VolGroup/lv']
162 def test_multiple_journals_are_found(self
):
163 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=journal'
167 lv_name
='volume%s' % i
, lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/lv%s' % i
, lv_tags
=tags
)
169 result
= zap
.ensure_associated_lvs(volumes
)
170 assert '/dev/VolGroup/lv0' in result
171 assert '/dev/VolGroup/lv1' in result
172 assert '/dev/VolGroup/lv2' in result
174 def test_multiple_dbs_are_found(self
):
175 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=db'
179 lv_name
='volume%s' % i
, lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/lv%s' % i
, lv_tags
=tags
)
181 result
= zap
.ensure_associated_lvs(volumes
)
182 assert '/dev/VolGroup/lv0' in result
183 assert '/dev/VolGroup/lv1' in result
184 assert '/dev/VolGroup/lv2' in result
186 def test_multiple_wals_are_found(self
):
187 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.wal_uuid=x,ceph.type=wal'
191 lv_name
='volume%s' % i
, lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/lv%s' % i
, lv_tags
=tags
)
193 result
= zap
.ensure_associated_lvs(volumes
)
194 assert '/dev/VolGroup/lv0' in result
195 assert '/dev/VolGroup/lv1' in result
196 assert '/dev/VolGroup/lv2' in result
198 def test_multiple_backing_devs_are_found(self
):
200 for _type
in ['journal', 'db', 'wal']:
201 tags
= 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.wal_uuid=x,ceph.type=%s' % _type
203 lv_name
='volume%s' % _type
, lv_uuid
='y', vg_name
='', lv_path
='/dev/VolGroup/lv%s' % _type
, lv_tags
=tags
)
205 result
= zap
.ensure_associated_lvs(volumes
)
206 assert '/dev/VolGroup/lvjournal' in result
207 assert '/dev/VolGroup/lvwal' in result
208 assert '/dev/VolGroup/lvdb' in result
210 @patch('ceph_volume.devices.lvm.zap.api.get_lvs')
211 def test_ensure_associated_lvs(self
, m_get_lvs
):
212 zap
.ensure_associated_lvs([], lv_tags
={'ceph.osd_id': '1'})
214 call(tags
={'ceph.type': 'journal', 'ceph.osd_id': '1'}),
215 call(tags
={'ceph.type': 'db', 'ceph.osd_id': '1'}),
216 call(tags
={'ceph.type': 'wal', 'ceph.osd_id': '1'})
218 m_get_lvs
.assert_has_calls(calls
, any_order
=True)
221 class TestWipeFs(object):
224 os
.environ
['CEPH_VOLUME_WIPEFS_INTERVAL'] = '0'
226 def test_works_on_second_try(self
, stub_call
):
227 os
.environ
['CEPH_VOLUME_WIPEFS_TRIES'] = '2'
228 stub_call([('wiping /dev/sda', '', 1), ('', '', 0)])
229 result
= zap
.wipefs('/dev/sda')
230 assert result
is None
232 def test_does_not_work_after_several_tries(self
, stub_call
):
233 os
.environ
['CEPH_VOLUME_WIPEFS_TRIES'] = '2'
234 stub_call([('wiping /dev/sda', '', 1), ('', '', 1)])
235 with pytest
.raises(RuntimeError):
236 zap
.wipefs('/dev/sda')
238 def test_does_not_work_default_tries(self
, stub_call
):
239 stub_call([('wiping /dev/sda', '', 1)]*8)
240 with pytest
.raises(RuntimeError):
241 zap
.wipefs('/dev/sda')