summaryrefslogtreecommitdiffstats
path: root/src/ceph-volume/ceph_volume/tests/test_decorators.py
blob: a56f7030b4cb52a50f5d976437d2a2685e1d9caf (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
import os
import pytest
from ceph_volume import exceptions, decorators, terminal


class TestNeedsRoot(object):

    def test_is_root(self, monkeypatch):
        def func():
            return True
        monkeypatch.setattr(decorators.os, 'getuid', lambda: 0)
        assert decorators.needs_root(func)() is True

    def test_is_not_root_env_var_skip_needs_root(self, monkeypatch):
        def func():
            return True
        monkeypatch.setattr(decorators.os, 'getuid', lambda: 123)
        monkeypatch.setattr(decorators.os, 'environ', {'CEPH_VOLUME_SKIP_NEEDS_ROOT': '1'})
        assert decorators.needs_root(func)() is True

    def test_is_not_root(self, monkeypatch):
        def func():
            return True # pragma: no cover
        monkeypatch.setattr(decorators.os, 'getuid', lambda: 20)
        with pytest.raises(exceptions.SuperUserError) as error:
            decorators.needs_root(func)()

        msg = 'This command needs to be executed with sudo or as root'
        assert str(error.value) == msg


class TestExceptionMessage(object):

    def test_has_str_method(self):
        result = decorators.make_exception_message(RuntimeError('an error'))
        expected = "%s %s\n" % (terminal.red_arrow, 'RuntimeError: an error')
        assert result == expected

    def test_has_no_str_method(self):
        class Error(Exception):
            pass
        result = decorators.make_exception_message(Error())
        expected = "%s %s\n" % (terminal.red_arrow, 'Error')
        assert result == expected


class TestCatches(object):

    def teardown_method(self):
        try:
            del(os.environ['CEPH_VOLUME_DEBUG'])
        except KeyError:
            pass

    def test_ceph_volume_debug_enabled(self):
        os.environ['CEPH_VOLUME_DEBUG'] = '1'
        @decorators.catches() # noqa
        def func():
            raise RuntimeError()
        with pytest.raises(RuntimeError):
            func()

    def test_ceph_volume_debug_disabled_no_exit(self, capsys):
        @decorators.catches(exit=False)
        def func():
            raise RuntimeError()
        func()
        stdout, stderr = capsys.readouterr()
        assert 'RuntimeError\n' in stderr

    def test_ceph_volume_debug_exits(self, capsys):
        @decorators.catches()
        def func():
            raise RuntimeError()
        with pytest.raises(SystemExit):
            func()
        stdout, stderr = capsys.readouterr()
        assert 'RuntimeError\n' in stderr