from ceph_volume.devices.lvm import zap
+class TestZap(object):
+ def test_invalid_osd_id_passed(self):
+ with pytest.raises(SystemExit):
+ zap.Zap(argv=['--osd-id', 'foo']).main()
+
class TestFindAssociatedDevices(object):
def test_no_lvs_found_that_match_id(self, monkeypatch, device_info):
monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
result = zap.find_associated_devices(osd_id='0')
- assert result[0].abspath == '/dev/VolGroup/lv'
+ assert result[0].path == '/dev/VolGroup/lv'
def test_lv_is_matched_fsid(self, monkeypatch):
tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,' +\
monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
result = zap.find_associated_devices(osd_fsid='asdf-lkjh')
- assert result[0].abspath == '/dev/VolGroup/lv'
+ assert result[0].path == '/dev/VolGroup/lv'
def test_lv_is_matched_id_fsid(self, monkeypatch):
tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,' +\
monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
result = zap.find_associated_devices(osd_id='0', osd_fsid='asdf-lkjh')
- assert result[0].abspath == '/dev/VolGroup/lv'
+ assert result[0].path == '/dev/VolGroup/lv'
class TestEnsureAssociatedLVs(object):
result = zap.ensure_associated_lvs(volumes)
assert result == []
- def test_data_is_found(self):
+ def test_data_is_found(self, fake_call):
tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=data'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/data', lv_tags=tags)
result = zap.ensure_associated_lvs(volumes)
assert result == ['/dev/VolGroup/data']
- def test_block_is_found(self):
+ def test_block_is_found(self, fake_call):
tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=block'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/block', lv_tags=tags)
assert '/dev/sdb1' in result
assert '/dev/VolGroup/block' in result
- def test_journal_is_found(self):
+ def test_journal_is_found(self, fake_call):
tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=journal'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv', lv_tags=tags)
class TestWipeFs(object):
- def setup(self):
+ def setup_method(self):
os.environ['CEPH_VOLUME_WIPEFS_INTERVAL'] = '0'
def test_works_on_second_try(self, stub_call):