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

   1 # type: ignore
   2 from __future__ import absolute_import
   3
   4 import json
   5 import logging
   6 import os
   7
   8 if 'UNITTEST' in os.environ:
   9
  10     # Mock ceph_module. Otherwise every module that is involved in a testcase and imports it will
  11     # raise an ImportError
  12
  13     import sys
  14
  15     try:
  16         from unittest import mock
  17     except ImportError:
  18         import mock
  19
  20     M_classes = set()
  21
  22     class M(object):
  23         """
  24         Note that:
  25
  26         * self.set_store() populates self._store
  27         * self.set_module_option() populates self._store[module_name]
  28         * self.get(thing) comes from self._store['_ceph_get' + thing]
  29
  30         """
  31
  32         def mock_store_get(self, kind, key, default):
  33             if not hasattr(self, '_store'):
  34                 self._store = {}
  35             return self._store.get(f'mock_store/{kind}/{key}', default)
  36
  37         def mock_store_set(self, kind, key, value):
  38             if not hasattr(self, '_store'):
  39                 self._store = {}
  40             k = f'mock_store/{kind}/{key}'
  41             if value is None:
  42                 if k in self._store:
  43                     del self._store[k]
  44             else:
  45                 self._store[k] = value
  46
  47         def mock_store_preifx(self, kind, prefix):
  48             if not hasattr(self, '_store'):
  49                 self._store = {}
  50             full_prefix = f'mock_store/{kind}/{prefix}'
  51             kind_len = len(f'mock_store/{kind}/')
  52             return {
  53                 k[kind_len:]: v for k, v in self._store.items()
  54                 if k.startswith(full_prefix)
  55             }
  56
  57         def _ceph_get_store(self, k):
  58             return self.mock_store_get('store', k, None)
  59
  60         def _ceph_set_store(self, k, v):
  61             self.mock_store_set('store', k, v)
  62
  63         def _ceph_get_store_prefix(self, prefix):
  64             return self.mock_store_preifx('store', prefix)
  65
  66         def _ceph_get_module_option(self, module, key, localized_prefix= None):
  67             try:
  68                 _, val, _ = self.check_mon_command({
  69                     'prefix': 'config get',
  70                     'who': 'mgr',
  71                     'key': f'mgr/{module}/{key}'
  72                 })
  73             except FileNotFoundError:
  74                 val = None
  75             mo = [o for o in self.MODULE_OPTIONS if o['name'] == key]
  76             if len(mo) == 1:
  77                 if val is not None:
  78                     cls = {
  79                         'str': str,
  80                         'secs': int,
  81                         'bool': lambda s: bool(s) and s != 'false' and s != 'False',
  82                         'int': int,
  83                     }[mo[0].get('type', 'str')]
  84                     return cls(val)
  85                 return val
  86             else:
  87                 return val if val is not None else ''
  88
  89         def _ceph_set_module_option(self, module, key, val):
  90             _, _, _ = self.check_mon_command({
  91                 'prefix': 'config set',
  92                 'who': 'mgr',
  93                 'name': f'mgr/{module}/{key}',
  94                 'value': val
  95             })
  96             return val
  97
  98         def _ceph_get(self, data_name):
  99             return self.mock_store_get('_ceph_get', data_name, mock.MagicMock())
 100
 101         def _ceph_send_command(self, res, svc_type, svc_id, command, tag):
 102             cmd = json.loads(command)
 103
 104             # Mocking the config store is handy sometimes:
 105             def config_get():
 106                 who = cmd['who'].split('.')
 107                 whos = ['global'] + ['.'.join(who[:i+1]) for i in range(len(who))]
 108                 for attepmt in reversed(whos):
 109                     val = self.mock_store_get('config', f'{attepmt}/{cmd["key"]}', None)
 110                     if val is not None:
 111                         return val
 112                 return None
 113
 114             def config_set():
 115                 self.mock_store_set('config', f'{cmd["who"]}/{cmd["name"]}', cmd['value'])
 116                 return ''
 117
 118             def config_dump():
 119                 r = []
 120                 for prefix, value in self.mock_store_preifx('config', '').items():
 121                     section, name = prefix.split('/', 1)
 122                     r.append({
 123                         'name': name,
 124                         'section': section,
 125                         'value': value
 126                     })
 127                 return json.dumps(r)
 128
 129             outb = ''
 130             if cmd['prefix'] == 'config get':
 131                 outb = config_get()
 132             elif cmd['prefix'] == 'config set':
 133                 outb = config_set()
 134             elif cmd['prefix'] == 'config dump':
 135                 outb = config_dump()
 136             elif hasattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_')):
 137                 a = getattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_'))
 138                 outb = a(cmd)
 139
 140             res.complete(0, outb, '')
 141
 142         @property
 143         def _logger(self):
 144             return logging.getLogger(__name__)
 145
 146         @_logger.setter
 147         def _logger(self, _):
 148             pass
 149
 150         def __init__(self, *args):
 151             if not hasattr(self, '_store'):
 152                 self._store = {}
 153
 154
 155             if self.__class__.__name__ not in M_classes:
 156                 # call those only once.
 157                 self._register_commands('')
 158                 self._register_options('')
 159                 M_classes.add(self.__class__.__name__)
 160
 161             super(M, self).__init__()
 162             self._ceph_get_version = mock.Mock()
 163             self._ceph_get_option = mock.MagicMock()
 164             self._ceph_get_context = mock.MagicMock()
 165             self._ceph_register_client = mock.MagicMock()
 166             self._ceph_set_health_checks = mock.MagicMock()
 167             self._configure_logging = lambda *_: None
 168             self._unconfigure_logging = mock.MagicMock()
 169             self._ceph_log = mock.MagicMock()
 170             self._ceph_dispatch_remote = lambda *_: None
 171             self._ceph_get_mgr_id = mock.MagicMock()
 172
 173
 174     cm = mock.Mock()
 175     cm.BaseMgrModule = M
 176     cm.BaseMgrStandbyModule = M
 177     sys.modules['ceph_module'] = cm
 178
 179     def mock_ceph_modules():
 180         class MockRadosError(Exception):
 181             def __init__(self, message, errno=None):
 182                 super(MockRadosError, self).__init__(message)
 183                 self.errno = errno
 184
 185             def __str__(self):
 186                 msg = super(MockRadosError, self).__str__()
 187                 if self.errno is None:
 188                     return msg
 189                 return '[errno {0}] {1}'.format(self.errno, msg)
 190
 191
 192         sys.modules.update({
 193             'rados': mock.MagicMock(Error=MockRadosError, OSError=MockRadosError),
 194             'rbd': mock.Mock(),
 195             'cephfs': mock.Mock(),
 196         })