ceph-volume tests ensure activate behavior with systemd disabling

Signed-off-by: Alfredo Deza <alfredo@deza.pe>
This commit is contained in:
Alfredo Deza 2018-11-08 14:14:17 -05:00
parent 18ddd96211
commit 3e80118cca

View File

@ -23,6 +23,89 @@ class TestActivate(object):
assert 'Activate OSDs by mounting devices previously configured' in stdout
class TestEnableSystemdUnits(object):
def test_nothing_is_activated(self, tmpfile, is_root, capsys):
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--no-systemd', '--file', json_config, '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
out, err = capsys.readouterr()
assert 'Skipping enabling of `simple`' in out
assert 'Skipping masking of ceph-disk' in out
assert 'Skipping enabling and starting OSD simple' in out
def test_no_systemd_flag_is_true(self, tmpfile, is_root):
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--no-systemd', '--file', json_config, '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
assert activation.skip_systemd is True
def test_no_systemd_flag_is_false(self, tmpfile, is_root):
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=True)
activation.activate = lambda x: True
activation.main()
assert activation.skip_systemd is False
def test_masks_ceph_disk(self, tmpfile, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
def test_enables_simple_unit(self, tmpfile, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
assert capture.calls[0]['args'] == ('0', '1234', 'simple')
def test_enables_osd_unit(self, tmpfile, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', capture)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', lambda *a: True)
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
assert capture.calls[0]['args'] == ('0',)
def test_starts_osd_unit(self, tmpfile, is_root, monkeypatch, capture):
monkeypatch.setattr('ceph_volume.systemd.systemctl.mask_ceph_disk', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_volume', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.enable_osd', lambda *a: True)
monkeypatch.setattr('ceph_volume.systemd.systemctl.start_osd', capture)
json_config = tmpfile(contents='{}')
activation = activate.Activate(['--file', json_config, '0', '1234'], from_trigger=False)
activation.activate = lambda x: True
activation.main()
activation.enable_systemd_units('0', '1234')
assert len(capture.calls) == 1
assert capture.calls[0]['args'] == ('0',)
class TestValidateDevices(object):
def test_filestore_missing_journal(self):