ceph-volume: tests: add tests for lvm devices and API

Signed-off-by: Alfredo Deza <adeza@redhat.com>

src/ceph-volume/ceph_volume/tests/devices/lvm/test_api.p

@@ -0,0 +1,3 @@
+from ceph_volume.devices.lvm import api
+
+def test(): pass

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

@@ -0,0 +1,106 @@
+import pytest
+from ceph_volume import process, exceptions
+from ceph_volume.devices.lvm import api
+
+
+class TestParseTags(object):
+
+    def test_no_tags_means_empty_dict(self):
+        result = api.parse_tags('')
+        assert result == {}
+
+    def test_single_tag_gets_parsed(self):
+        result = api.parse_tags('ceph.osd_something=1')
+        assert result == {'ceph.osd_something': '1'}
+
+    def test_multiple_csv_expands_in_dict(self):
+        result = api.parse_tags('ceph.osd_something=1,ceph.foo=2,ceph.fsid=0000')
+        # assert them piecemeal to avoid the un-ordered dict nature
+        assert result['ceph.osd_something'] == '1'
+        assert result['ceph.foo'] == '2'
+        assert result['ceph.fsid'] == '0000'
+
+
+class TestGetAPIVgs(object):
+
+    def test_report_is_emtpy(self, monkeypatch):
+        monkeypatch.setattr(api.process, 'call', lambda x: ('{}', '', 0))
+        assert api.get_api_vgs() == []
+
+    def test_report_has_stuff(self, monkeypatch):
+        report = '{"report":[{"vg":[{"vg_name":"VolGroup00"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_vgs() == [{'vg_name': 'VolGroup00'}]
+
+    def test_report_has_multiple_items(self, monkeypatch):
+        report = '{"report":[{"vg":[{"vg_name":"VolGroup00"},{"vg_name":"ceph_vg"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_vgs() == [{'vg_name': 'VolGroup00'}, {'vg_name': 'ceph_vg'}]
+
+    def test_does_not_get_poluted_with_non_vg_items(self, monkeypatch):
+        report = '{"report":[{"vg":[{"vg_name":"VolGroup00"}],"lv":[{"lv":"1"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_vgs() == [{'vg_name': 'VolGroup00'}]
+
+
+class TestGetAPILvs(object):
+
+    def test_report_is_emtpy(self, monkeypatch):
+        monkeypatch.setattr(api.process, 'call', lambda x: ('{}', '', 0))
+        assert api.get_api_lvs() == []
+
+    def test_report_has_stuff(self, monkeypatch):
+        report = '{"report":[{"lv":[{"lv_name":"VolGroup00"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_lvs() == [{'lv_name': 'VolGroup00'}]
+
+    def test_report_has_multiple_items(self, monkeypatch):
+        report = '{"report":[{"lv":[{"lv_name":"VolName"},{"lv_name":"ceph_lv"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_lvs() == [{'lv_name': 'VolName'}, {'lv_name': 'ceph_lv'}]
+
+    def test_does_not_get_poluted_with_non_lv_items(self, monkeypatch):
+        report = '{"report":[{"lv":[{"lv_name":"VolName"}],"vg":[{"vg":"1"}]}]}'
+        monkeypatch.setattr(api.process, 'call', lambda x: (report, '', 0))
+        assert api.get_api_lvs() == [{'lv_name': 'VolName'}]
+
+
+@pytest.fixture
+def volumes(monkeypatch):
+    monkeypatch.setattr(process, 'call', lambda x: ('{}', '', 0))
+    volumes = api.Volumes()
+    volumes._purge()
+    return volumes
+
+
+class TestGetLV(object):
+
+    def test_nothing_is_passed_in(self):
+        # so we return a None
+        assert api.get_lv() is None
+
+    def test_single_lv_is_matched(self, volumes, monkeypatch):
+        FooVolume = api.Volume(lv_name='foo', lv_path='/dev/vg/foo', lv_tags="ceph.type=data")
+        volumes.append(FooVolume)
+        monkeypatch.setattr(api, 'Volumes', lambda: volumes)
+        assert api.get_lv(lv_name='foo') == FooVolume
+
+
+class TestVolumes(object):
+
+    def test_volume_get_has_no_volumes(self, volumes):
+        assert volumes.get() is None
+
+    def test_volume_has_multiple_matches(self, volumes):
+        volume1 = volume2 = api.Volume(lv_name='foo', lv_path='/dev/vg/lv', lv_tags='')
+        volumes.append(volume1)
+        volumes.append(volume2)
+        with pytest.raises(exceptions.MultipleLVsError):
+            volumes.get(lv_name='foo')
+
+    def test_find_the_correct_one(self, volumes):
+        volume1 = api.Volume(lv_name='volume1', lv_path='/dev/vg/lv', lv_tags='')
+        volume2 = api.Volume(lv_name='volume2', lv_path='/dev/vg/lv', lv_tags='')
+        volumes.append(volume1)
+        volumes.append(volume2)
+        assert volumes.get(lv_name='volume1') == volume1

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

@@ -0,0 +1,53 @@
+import pytest
+from ceph_volume.devices import lvm
+
+
+class TestLVM(object):
+
+    def test_main_spits_help_with_no_arguments(self, capsys):
+        lvm.main.LVM([]).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'Use LVM and LVM-based technologies like dmcache to deploy' in stdout
+
+    def test_main_shows_activate_subcommands(self, capsys):
+        lvm.main.LVM([]).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'activate ' in stdout
+        assert 'Discover and mount' in stdout
+
+    def test_main_shows_prepare_subcommands(self, capsys):
+        lvm.main.LVM([]).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'prepare ' in stdout
+        assert 'Format an LVM device' in stdout
+
+
+class TestPrepare(object):
+
+    def test_main_spits_help_with_no_arguments(self, capsys):
+        lvm.prepare.Prepare([]).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'Prepare an OSD by assigning an ID and FSID' in stdout
+
+    def test_main_shows_full_help(self, capsys):
+        with pytest.raises(SystemExit):
+            lvm.prepare.Prepare(argv=['--help']).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'required arguments:' in stdout
+        assert 'A logical group name or a path' in stdout
+
+
+class TestActivate(object):
+
+    def test_main_spits_help_with_no_arguments(self, capsys):
+        lvm.activate.Activate([]).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'Activate OSDs by discovering them with' in stdout
+
+    def test_main_shows_full_help(self, capsys):
+        with pytest.raises(SystemExit):
+            lvm.activate.Activate(argv=['--help']).main()
+        stdout, stderr = capsys.readouterr()
+        assert 'optional arguments' in stdout
+        assert 'positional arguments' in stdout
+