src/ceph/src/ceph-volume/ceph_volume/tests/test_decorators.py

   1 import os
   2 import pytest
   3 from ceph_volume import exceptions, decorators, terminal
   4
   5
   6 class TestNeedsRoot(object):
   7
   8     def test_is_root(self, monkeypatch):
   9         def func():
  10             return True
  11         monkeypatch.setattr(decorators.os, 'getuid', lambda: 0)
  12         assert decorators.needs_root(func)() is True
  13
  14     def test_is_not_root(self, monkeypatch):
  15         def func():
  16             return True # pragma: no cover
  17         monkeypatch.setattr(decorators.os, 'getuid', lambda: 20)
  18         with pytest.raises(exceptions.SuperUserError) as error:
  19             decorators.needs_root(func)()
  20
  21         msg = 'This command needs to be executed with sudo or as root'
  22         assert str(error.value) == msg
  23
  24
  25 class TestExceptionMessage(object):
  26
  27     def test_has_str_method(self):
  28         result = decorators.make_exception_message(RuntimeError('an error'))
  29         expected = "%s %s\n" % (terminal.red_arrow, 'RuntimeError: an error')
  30         assert result == expected
  31
  32     def test_has_no_str_method(self):
  33         class Error(Exception):
  34             pass
  35         result = decorators.make_exception_message(Error())
  36         expected = "%s %s\n" % (terminal.red_arrow, 'Error')
  37         assert result == expected
  38
  39
  40 class TestCatches(object):
  41
  42     def teardown(self):
  43         try:
  44             del(os.environ['CEPH_VOLUME_DEBUG'])
  45         except KeyError:
  46             pass
  47
  48     def test_ceph_volume_debug_enabled(self):
  49         os.environ['CEPH_VOLUME_DEBUG'] = '1'
  50         @decorators.catches() # noqa
  51         def func():
  52             raise RuntimeError()
  53         with pytest.raises(RuntimeError):
  54             func()
  55
  56     def test_ceph_volume_debug_disabled_no_exit(self, capsys):
  57         @decorators.catches(exit=False)
  58         def func():
  59             raise RuntimeError()
  60         func()
  61         stdout, stderr = capsys.readouterr()
  62         assert 'RuntimeError\n' in stderr
  63
  64     def test_ceph_volume_debug_exits(self, capsys):
  65         @decorators.catches()
  66         def func():
  67             raise RuntimeError()
  68         with pytest.raises(SystemExit):
  69             func()
  70         stdout, stderr = capsys.readouterr()
  71         assert 'RuntimeError\n' in stderr