ceph/src/ceph-volume/ceph_volume/tests/devices/lvm/test_zap.py

   1 import os
   2 import pytest
   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
   8
   9
  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()
  14
  15 class TestFindAssociatedDevices(object):
  16
  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')
  21         volumes = []
  22         volumes.append(osd)
  23         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kwargs: {})
  24
  25         with pytest.raises(RuntimeError):
  26             zap.find_associated_devices(osd_id=10)
  27
  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,'+\
  30                'ceph.type=data'
  31         osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
  32                          vg_name='vg', lv_path='/dev/VolGroup/lv')
  33         volumes = []
  34         volumes.append(osd)
  35         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kwargs: {})
  36
  37         with pytest.raises(RuntimeError):
  38             zap.find_associated_devices(osd_fsid='aaaa-lkjh')
  39
  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,'+\
  42                'ceph.type=data'
  43         osd = api.Volume(lv_name='volume1', lv_uuid='y', vg_name='vg',
  44                          lv_tags=tags, lv_path='/dev/VolGroup/lv')
  45         volumes = []
  46         volumes.append(osd)
  47         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kwargs: {})
  48
  49         with pytest.raises(RuntimeError):
  50             zap.find_associated_devices(osd_id='9', osd_fsid='aaaa-lkjh')
  51
  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')
  55         volumes = []
  56         volumes.append(osd)
  57         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kwargs: {})
  58
  59         with pytest.raises(RuntimeError):
  60             zap.find_associated_devices(osd_id=100)
  61
  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)
  66         volumes = []
  67         volumes.append(osd)
  68         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kw: volumes)
  69         monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
  70
  71         result = zap.find_associated_devices(osd_id='0')
  72         assert result[0].path == '/dev/VolGroup/lv'
  73
  74     def test_lv_is_matched_fsid(self, monkeypatch):
  75         tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,' +\
  76                'ceph.type=data'
  77         osd = api.Volume(lv_name='volume1', lv_uuid='y', vg_name='',
  78                          lv_path='/dev/VolGroup/lv', lv_tags=tags)
  79         volumes = []
  80         volumes.append(osd)
  81         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kw: deepcopy(volumes))
  82         monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
  83
  84         result = zap.find_associated_devices(osd_fsid='asdf-lkjh')
  85         assert result[0].path == '/dev/VolGroup/lv'
  86
  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,' +\
  89                'ceph.type=data'
  90         osd = api.Volume(lv_name='volume1', lv_uuid='y', vg_name='',
  91                          lv_path='/dev/VolGroup/lv', lv_tags=tags)
  92         volumes = []
  93         volumes.append(osd)
  94         monkeypatch.setattr(zap.api, 'get_lvs', lambda **kw: volumes)
  95         monkeypatch.setattr(process, 'call', lambda x, **kw: ('', '', 0))
  96
  97         result = zap.find_associated_devices(osd_id='0', osd_fsid='asdf-lkjh')
  98         assert result[0].path == '/dev/VolGroup/lv'
  99
 100
 101 class TestEnsureAssociatedLVs(object):
 102
 103     def test_nothing_is_found(self):
 104         volumes = []
 105         result = zap.ensure_associated_lvs(volumes)
 106         assert result == []
 107
 108     def test_data_is_found(self, fake_call):
 109         tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=data'
 110         osd = api.Volume(
 111             lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/data', lv_tags=tags)
 112         volumes = []
 113         volumes.append(osd)
 114         result = zap.ensure_associated_lvs(volumes)
 115         assert result == ['/dev/VolGroup/data']
 116
 117     def test_block_is_found(self, fake_call):
 118         tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=block'
 119         osd = api.Volume(
 120             lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/block', lv_tags=tags)
 121         volumes = []
 122         volumes.append(osd)
 123         result = zap.ensure_associated_lvs(volumes)
 124         assert result == ['/dev/VolGroup/block']
 125
 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')
 129         cli_zap.args = args
 130         cli_zap.zap()
 131         out, err = capsys.readouterr()
 132         assert "Zapping successful for OSD: asdf-lkjh" in err
 133
 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)
 137         cli_zap.args = args
 138         cli_zap.zap()
 139         out, err = capsys.readouterr()
 140         assert "Zapping successful for OSD: 1" in err
 141
 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'
 145         osd = api.Volume(
 146             lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/block', lv_tags=tags)
 147         volumes = []
 148         volumes.append(osd)
 149         result = zap.ensure_associated_lvs(volumes)
 150         assert '/dev/sdb1' in result
 151         assert '/dev/VolGroup/block' in result
 152
 153     def test_journal_is_found(self, fake_call):
 154         tags = 'ceph.osd_id=0,ceph.osd_fsid=asdf-lkjh,ceph.journal_uuid=x,ceph.type=journal'
 155         osd = api.Volume(
 156             lv_name='volume1', lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv', lv_tags=tags)
 157         volumes = []
 158         volumes.append(osd)
 159         result = zap.ensure_associated_lvs(volumes)
 160         assert result == ['/dev/VolGroup/lv']
 161
 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'
 164         volumes = []
 165         for i in range(3):
 166             osd = api.Volume(
 167                 lv_name='volume%s' % i, lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv%s' % i, lv_tags=tags)
 168             volumes.append(osd)
 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
 173
 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'
 176         volumes = []
 177         for i in range(3):
 178             osd = api.Volume(
 179                 lv_name='volume%s' % i, lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv%s' % i, lv_tags=tags)
 180             volumes.append(osd)
 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
 185
 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'
 188         volumes = []
 189         for i in range(3):
 190             osd = api.Volume(
 191                 lv_name='volume%s' % i, lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv%s' % i, lv_tags=tags)
 192             volumes.append(osd)
 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
 197
 198     def test_multiple_backing_devs_are_found(self):
 199         volumes = []
 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
 202             osd = api.Volume(
 203                 lv_name='volume%s' % _type, lv_uuid='y', vg_name='', lv_path='/dev/VolGroup/lv%s' % _type, lv_tags=tags)
 204             volumes.append(osd)
 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
 209
 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'})
 213         calls = [
 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'})
 217         ]
 218         m_get_lvs.assert_has_calls(calls, any_order=True)
 219
 220
 221 class TestWipeFs(object):
 222
 223     def setup(self):
 224         os.environ['CEPH_VOLUME_WIPEFS_INTERVAL'] = '0'
 225
 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
 231
 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')
 237
 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')