Skip to content

Commit

Permalink
ceph-volume: update tests since listing.py got heavily modified
Browse files Browse the repository at this point in the history 
Signed-off-by: Rishabh Dave <ridave@redhat.com>
(cherry picked from commit d1ae6d1)
  • Loading branch information
@rishabh-d-dave
rishabh-d-dave authored and Jan Fajerski committed Feb 12, 2020
1 parent 66bbe16 commit 27bd05e
Show file tree
Hide file tree
Showing 2 changed files with 142 additions and 92 deletions.
2 changes: 2 additions & 0 deletions src/ceph-volume/ceph_volume/tests/conftest.py
View file
Expand Up @@ -166,12 +166,14 @@ def apply(vgs):
return apply


# TODO: allow init-ing pvolumes to list we want
@pytest.fixture
def pvolumes(monkeypatch):
monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
pvolumes = lvm_api.PVolumes()
pvolumes._purge()
return pvolumes

@pytest.fixture
def pvolumes_empty(monkeypatch):
monkeypatch.setattr('ceph_volume.process.call', lambda x, **kw: ('', '', 0))
Expand Down
232 changes: 140 additions & 92 deletions src/ceph-volume/ceph_volume/tests/devices/lvm/test_listing.py
View file
Expand Up @@ -2,6 +2,11 @@
from ceph_volume.devices import lvm
from ceph_volume.api import lvm as api

# TODO: add tests for following commands -
# ceph-volume list
# ceph-volume list <path-to-pv>
# ceph-volume list <path-to-vg>
# ceph-volume list <path-to-lv>

class TestReadableTag(object):

Expand Down Expand Up @@ -57,13 +62,15 @@ def test_devices_are_comma_separated(self, capsys):

class TestList(object):

def test_empty_full_json_zero_exit_status(self, is_root, volumes, factory, capsys):
def test_empty_full_json_zero_exit_status(self, is_root, volumes,
factory, capsys):
args = factory(format='json', device=None)
lvm.listing.List([]).list(args)
stdout, stderr = capsys.readouterr()
assert stdout == '{}\n'

def test_empty_device_json_zero_exit_status(self, is_root, volumes, factory, capsys):
def test_empty_device_json_zero_exit_status(self, is_root, volumes,
factory, capsys):
args = factory(format='json', device='/dev/sda1')
lvm.listing.List([]).list(args)
stdout, stderr = capsys.readouterr()
Expand Down Expand Up @@ -101,68 +108,100 @@ class TestFullReport(object):

def test_no_ceph_lvs(self, volumes, monkeypatch):
# ceph lvs are detected by looking into its tags
osd = api.Volume(lv_name='volume1', lv_path='/dev/VolGroup/lv', lv_tags={})
osd = api.Volume(lv_name='volume1', lv_path='/dev/VolGroup/lv',
lv_tags={})
volumes.append(osd)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result == {}

def test_ceph_data_lv_reported(self, volumes, monkeypatch):
def test_ceph_data_lv_reported(self, pvolumes, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', lv_path='/dev/VolGroup/lv', lv_tags=tags)
pv = api.PVolume(pv_name='/dev/sda1', pv_tags={}, pv_uuid="0000",
vg_name='VolGroup', lv_uuid="aaaa")
osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
lv_path='/dev/VolGroup/lv', vg_name='VolGroup')
pvolumes.append(pv)
volumes.append(osd)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result['0'][0]['name'] == 'volume1'

def test_ceph_journal_lv_reported(self, volumes, monkeypatch):
def test_ceph_journal_lv_reported(self, pvolumes, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
journal_tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=journal'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', lv_path='/dev/VolGroup/lv', lv_tags=tags)
pv = api.PVolume(pv_name='/dev/sda1', pv_tags={}, pv_uuid="0000",
vg_name="VolGroup", lv_uuid="aaaa")
osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
lv_path='/dev/VolGroup/lv', vg_name='VolGroup')
journal = api.Volume(
lv_name='journal', lv_uuid='x', lv_path='/dev/VolGroup/journal', lv_tags=journal_tags)
lv_name='journal', lv_uuid='x', lv_tags=journal_tags,
lv_path='/dev/VolGroup/journal', vg_name='VolGroup')
pvolumes.append(pv)
volumes.append(osd)
volumes.append(journal)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result['0'][0]['name'] == 'volume1'
assert result['0'][1]['name'] == 'journal'

def test_ceph_wal_lv_reported(self, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=data'
wal_tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=wal'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', lv_path='/dev/VolGroup/lv', lv_tags=tags)
wal = api.Volume(
lv_name='wal', lv_uuid='x', lv_path='/dev/VolGroup/wal', lv_tags=wal_tags)
osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
lv_path='/dev/VolGroup/lv', vg_name='VolGroup')
wal = api.Volume(lv_name='wal', lv_uuid='x', lv_tags=wal_tags,
lv_path='/dev/VolGroup/wal', vg_name='VolGroup')
volumes.append(osd)
volumes.append(wal)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result['0'][0]['name'] == 'volume1'
assert result['0'][1]['name'] == 'wal'

def test_physical_journal_gets_reported(self, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', lv_path='/dev/VolGroup/lv', lv_tags=tags)
def test_physical_journal_gets_reported(self, pvolumes, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=journal'
pv = api.PVolume(vg_name="VolGroup", pv_name='/dev/sda1', pv_tags={},
pv_uuid="0000", lv_uuid="aaaa")
osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
vg_name='VolGroup', lv_path='/dev/VolGroup/lv')
pvolumes.append(pv)
volumes.append(osd)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.disk, 'get_device_from_partuuid', lambda x: '/dev/sda1')
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result['0'][1]['path'] == '/dev/sda1'
assert result['0'][1]['tags'] == {'PARTUUID': 'x'}
assert result['0'][1]['type'] == 'journal'

def test_physical_wal_gets_reported(self, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=data'
osd = api.Volume(
lv_name='volume1', lv_uuid='y', lv_path='/dev/VolGroup/lv', lv_tags=tags)
def test_physical_wal_gets_reported(self, pvolumes, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.wal_uuid=x,ceph.type=wal'
pv = api.PVolume(pv_name='/dev/sda1', pv_tags={}, pv_uuid="0000",
vg_name="VolGroup", lv_uuid="aaaa")
osd = api.Volume(lv_name='volume1', lv_uuid='y', lv_tags=tags,
lv_path='/dev/VolGroup/lv', vg_name="VolGroup")
pvolumes.append(pv)
volumes.append(osd)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.disk, 'get_device_from_partuuid', lambda x: '/dev/sda1')
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).full_report()
assert result['0'][1]['path'] == '/dev/sda1'
assert result['0'][1]['tags'] == {'PARTUUID': 'x'}
Expand All @@ -173,116 +212,125 @@ class TestSingleReport(object):

def test_not_a_ceph_lv(self, volumes, monkeypatch):
# ceph lvs are detected by looking into its tags
lv = api.Volume(
lv_name='lv', vg_name='VolGroup', lv_path='/dev/VolGroup/lv', lv_tags={})
lv = api.Volume(lv_name='lv', lv_tags={}, lv_path='/dev/VolGroup/lv',
vg_name='VolGroup')
volumes.append(lv)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).single_report('VolGroup/lv')
assert result == {}

def test_report_a_ceph_lv(self, volumes, monkeypatch):
def test_report_a_ceph_lv(self, pvolumes, volumes, monkeypatch):
# ceph lvs are detected by looking into its tags
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
lv = api.Volume(
lv_name='lv', vg_name='VolGroup',
lv_uuid='aaaa', lv_path='/dev/VolGroup/lv', lv_tags=tags
)
lv = api.Volume(lv_name='lv', vg_name='VolGroup', lv_uuid='aaaa',
lv_path='/dev/VolGroup/lv', lv_tags=tags)
volumes.append(lv)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).single_report('VolGroup/lv')
assert result['0'][0]['name'] == 'lv'
assert result['0'][0]['lv_tags'] == tags
assert result['0'][0]['path'] == '/dev/VolGroup/lv'
assert result['0'][0]['devices'] == []

def test_report_a_ceph_journal_device(self, volumes, monkeypatch):
def test_report_a_ceph_journal_device(self, volumes, pvolumes, monkeypatch):
# ceph lvs are detected by looking into its tags
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data,ceph.journal_device=/dev/sda1'
lv = api.Volume(
lv_name='lv', vg_name='VolGroup', lv_path='/dev/VolGroup/lv',
lv_uuid='aaa', lv_tags=tags)
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=journal,' + \
'ceph.journal_device=/dev/sda1'
pv = api.PVolume(pv_name='/dev/sda1', pv_uuid="0000", pv_tags={},
vg_name="VolGroup", lv_uuid="aaaa")
lv = api.Volume(lv_name='lv', lv_uuid='aaa', lv_tags=tags,
lv_path='/dev/VolGroup/lv', vg_name='VolGroup')
pvolumes.append(pv)
volumes.append(lv)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

result = lvm.listing.List([]).single_report('/dev/sda1')
assert result['0'][0]['tags'] == {'PARTUUID': 'x'}
assert result['0'][0]['type'] == 'journal'
assert result['0'][0]['path'] == '/dev/sda1'

def test_report_a_ceph_lv_with_devices(self, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
lv = api.Volume(
lv_name='lv', vg_name='VolGroup',
lv_uuid='aaaa', lv_path='/dev/VolGroup/lv', lv_tags=tags
)
def test_report_a_ceph_lv_with_devices(self, volumes, pvolumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.type=data'
pv1 = api.PVolume(vg_name="VolGroup", pv_name='/dev/sda1',
pv_uuid='', pv_tags={}, lv_uuid="aaaa")
pv2 = api.PVolume(vg_name="VolGroup", pv_name='/dev/sdb1',
pv_uuid='', pv_tags={}, lv_uuid="aaaa")
lv = api.Volume(lv_name='lv', vg_name='VolGroup',lv_uuid='aaaa',
lv_path='/dev/VolGroup/lv', lv_tags=tags)
pvolumes.append(pv1)
pvolumes.append(pv2)
volumes.append(lv)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

listing = lvm.listing.List([])
listing._pvs = [
{'lv_uuid': 'aaaa', 'pv_name': '/dev/sda1', 'pv_tags': '', 'pv_uuid': ''},
{'lv_uuid': 'aaaa', 'pv_name': '/dev/sdb1', 'pv_tags': '', 'pv_uuid': ''},
]

result = listing.single_report('VolGroup/lv')
assert result['0'][0]['name'] == 'lv'
assert result['0'][0]['lv_tags'] == tags
assert result['0'][0]['path'] == '/dev/VolGroup/lv'
assert result['0'][0]['devices'] == ['/dev/sda1', '/dev/sdb1']

def test_report_a_ceph_lv_with_multiple_pvs_of_same_name(self, pvolumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
lv = api.Volume(
lv_name='lv', vg_name='VolGroup',
lv_uuid='aaaa', lv_path='/dev/VolGroup/lv', lv_tags=tags
)
def test_report_a_ceph_lv_with_multiple_pvs_of_same_name(self, pvolumes,
volumes,
monkeypatch):
tags = 'ceph.osd_id=0,ceph.type=data'
lv = api.Volume(lv_name='lv', vg_name='VolGroup', lv_uuid='aaaa',
lv_path='/dev/VolGroup/lv', lv_tags=tags)
volumes.append(lv)
monkeypatch.setattr(api, 'get_lv_from_argument', lambda device: None)
monkeypatch.setattr(api, 'get_lv', lambda vg_name: lv)
FooPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={}, lv_uuid="aaaa")
BarPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda', pv_uuid="0000", pv_tags={})
FooPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda',
pv_uuid="0000", pv_tags={}, lv_uuid="aaaa")
BarPVolume = api.PVolume(vg_name="vg", pv_name='/dev/sda',
pv_uuid="0000", pv_tags={})
pvolumes.append(FooPVolume)
pvolumes.append(BarPVolume)
monkeypatch.setattr(api, 'PVolumes', lambda: pvolumes)
listing = lvm.listing.List([])
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)
monkeypatch.setattr(lvm.listing.api, 'get_pvs', lambda **kwargs:
pvolumes)

result = listing.single_report('/dev/sda')
assert result['0'][0]['name'] == 'lv'
assert result['0'][0]['lv_tags'] == tags
assert result['0'][0]['path'] == '/dev/VolGroup/lv'
assert len(result) == 1

def test_report_a_ceph_lv_with_no_matching_devices(self, volumes, monkeypatch):
tags = 'ceph.osd_id=0,ceph.journal_uuid=x,ceph.type=data'
lv = api.Volume(
lv_name='lv', vg_name='VolGroup',
lv_uuid='aaaa', lv_path='/dev/VolGroup/lv', lv_tags=tags
)
def test_report_a_ceph_lv_with_no_matching_devices(self, volumes,
monkeypatch):
tags = 'ceph.osd_id=0,ceph.type=data'
lv = api.Volume(lv_name='lv', vg_name='VolGroup', lv_uuid='aaaa',
lv_path='/dev/VolGroup/lv', lv_tags=tags)
volumes.append(lv)
monkeypatch.setattr(lvm.listing.api, 'Volumes', lambda: volumes)
monkeypatch.setattr(lvm.listing.api, 'get_lvs', lambda **kwargs:
volumes)

listing = lvm.listing.List([])
listing._pvs = [
{'lv_uuid': 'ffff', 'pv_name': '/dev/sda1', 'pv_tags': '', 'pv_uuid': ''},
{'lv_uuid': 'ffff', 'pv_name': '/dev/sdb1', 'pv_tags': '', 'pv_uuid': ''},
]
{'lv_uuid': 'ffff', 'pv_name': '/dev/sda1', 'pv_tags': '',
'pv_uuid': ''},
{'lv_uuid': 'ffff', 'pv_name': '/dev/sdb1', 'pv_tags': '',
'pv_uuid': ''}]

result = listing.single_report('VolGroup/lv')
assert result['0'][0]['name'] == 'lv'
assert result['0'][0]['lv_tags'] == tags
assert result['0'][0]['path'] == '/dev/VolGroup/lv'
assert result['0'][0]['devices'] == []


class TestListingPVs(object):

def setup(self):
self.default_pvs = [
{'lv_uuid': 'ffff', 'pv_name': '/dev/sda1', 'pv_tags': '', 'pv_uuid': ''},
{'lv_uuid': 'ffff', 'pv_name': '/dev/sdb1', 'pv_tags': '', 'pv_uuid': ''},
]

def test_pvs_is_unset(self, monkeypatch):
monkeypatch.setattr(lvm.listing.api, 'get_api_pvs', lambda: self.default_pvs)
listing = lvm.listing.List([])
assert listing.pvs == self.default_pvs

def test_pvs_is_set(self, monkeypatch):
# keep it patched so that we can fail if this gets returned
monkeypatch.setattr(lvm.listing.api, 'get_api_pvs', lambda: self.default_pvs)
listing = lvm.listing.List([])
listing._pvs = []
assert listing.pvs == []

0 comments on commit 27bd05e

Please sign in to comment.