ceph/src/pybind/mgr/tests/__init__.py

   1 # type: ignore
   2 from __future__ import absolute_import
   3
   4
   5 import os
   6
   7 if 'UNITTEST' in os.environ:
   8
   9     # Mock ceph_module. Otherwise every module that is involved in a testcase and imports it will
  10     # raise an ImportError
  11
  12     import sys
  13
  14     try:
  15         from unittest import mock
  16     except ImportError:
  17         import mock
  18
  19     M_classes = set()
  20
  21     class M(object):
  22         def _ceph_get_store(self, k):
  23             return self._store.get(k, None)
  24
  25         def _ceph_set_store(self, k, v):
  26             if v is None:
  27                 if k in self._store:
  28                     del self._store[k]
  29             else:
  30                 self._store[k] = v
  31
  32         def _ceph_get_store_prefix(self, prefix):
  33             return {
  34                 k: v for k, v in self._store.items()
  35                 if k.startswith(prefix)
  36             }
  37
  38         def _ceph_get_module_option(self, module, key, localized_prefix: None):
  39             return self._ceph_get_store(f'{module}/{key}')
  40
  41         def _ceph_set_module_option(self, module, key, val):
  42             return self._ceph_set_store(f'{module}/{key}', val)
  43
  44         def __init__(self, *args):
  45             self._store = {}
  46
  47             if self.__class__.__name__ not in M_classes:
  48                 # call those only once.
  49                 self._register_commands('')
  50                 self._register_options('')
  51                 M_classes.add(self.__class__.__name__)
  52
  53             super(M, self).__init__()
  54             self._ceph_get_version = mock.Mock()
  55             self._ceph_get = mock.MagicMock()
  56             self._ceph_get_option = mock.MagicMock()
  57             self._ceph_get_context = mock.MagicMock()
  58             self._ceph_register_client = mock.MagicMock()
  59             self._configure_logging = lambda *_: None
  60             self._unconfigure_logging = mock.MagicMock()
  61             self._ceph_log = mock.MagicMock()
  62             self._ceph_dispatch_remote = lambda *_: None
  63
  64
  65     cm = mock.Mock()
  66     cm.BaseMgrModule = M
  67     cm.BaseMgrStandbyModule = M
  68     sys.modules['ceph_module'] = cm
  69
  70     def mock_ceph_modules():
  71         class MockRadosError(Exception):
  72             def __init__(self, message, errno=None):
  73                 super(MockRadosError, self).__init__(message)
  74                 self.errno = errno
  75
  76             def __str__(self):
  77                 msg = super(MockRadosError, self).__str__()
  78                 if self.errno is None:
  79                     return msg
  80                 return '[errno {0}] {1}'.format(self.errno, msg)
  81
  82
  83         sys.modules.update({
  84             'rados': mock.Mock(Error=MockRadosError, OSError=MockRadosError),
  85             'rbd': mock.Mock(),
  86             'cephfs': mock.Mock(),
  87         })