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_prefix(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_prefix('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:  # >= 1, cause self.MODULE_OPTIONS. otherwise it
  77                 #               fails when importing multiple modules.
  78                 if 'default' in mo and val is None:
  79                     val = mo[0]['default']
  80                 if val is not None:
  81                     cls = {
  82                         'str': str,
  83                         'secs': int,
  84                         'bool': lambda s: bool(s) and s != 'false' and s != 'False',
  85                         'int': int,
  86                     }[mo[0].get('type', 'str')]
  87                     return cls(val)
  88                 return val
  89             else:
  90                 return val if val is not None else ''
  91
  92         def _ceph_set_module_option(self, module, key, val):
  93             _, _, _ = self.check_mon_command({
  94                 'prefix': 'config set',
  95                 'who': 'mgr',
  96                 'name': f'mgr/{module}/{key}',
  97                 'value': val
  98             })
  99             return val
 100
 101         def _ceph_get(self, data_name):
 102             return self.mock_store_get('_ceph_get', data_name, mock.MagicMock())
 103
 104         def _ceph_send_command(self, res, svc_type, svc_id, command, tag, inbuf):
 105
 106             cmd = json.loads(command)
 107             getattr(self, '_mon_commands_sent', []).append(cmd)
 108
 109             # Mocking the config store is handy sometimes:
 110             def config_get():
 111                 who = cmd['who'].split('.')
 112                 whos = ['global'] + ['.'.join(who[:i+1]) for i in range(len(who))]
 113                 for attepmt in reversed(whos):
 114                     val = self.mock_store_get('config', f'{attepmt}/{cmd["key"]}', None)
 115                     if val is not None:
 116                         return val
 117                 return None
 118
 119             def config_set():
 120                 self.mock_store_set('config', f'{cmd["who"]}/{cmd["name"]}', cmd['value'])
 121                 return ''
 122
 123             def config_rm():
 124                 self.mock_store_set('config', f'{cmd["who"]}/{cmd["name"]}', None)
 125                 return ''
 126
 127             def config_dump():
 128                 r = []
 129                 for prefix, value in self.mock_store_prefix('config', '').items():
 130                     section, name = prefix.split('/', 1)
 131                     r.append({
 132                         'name': name,
 133                         'section': section,
 134                         'value': value
 135                     })
 136                 return json.dumps(r)
 137
 138             outb = ''
 139             if cmd['prefix'] == 'config get':
 140                 outb = config_get()
 141             elif cmd['prefix'] == 'config set':
 142                 outb = config_set()
 143             elif cmd['prefix'] == 'config dump':
 144                 outb = config_dump()
 145             elif cmd['prefix'] == 'config rm':
 146                 outb = config_rm()
 147             elif hasattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_')):
 148                 a = getattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_'))
 149                 outb = a(cmd)
 150
 151             res.complete(0, outb, '')
 152
 153         def _ceph_get_foreign_option(self, entity, name):
 154             who = entity.split('.')
 155             whos = ['global'] + ['.'.join(who[:i+1]) for i in range(len(who))]
 156             for attepmt in reversed(whos):
 157                 val = self.mock_store_get('config', f'{attepmt}/{name}', None)
 158                 if val is not None:
 159                     return val
 160             return None
 161
 162         def assert_issued_mon_command(self, command):
 163             assert command in self._mon_commands_sent, self._mon_commands_sent
 164
 165         @property
 166         def _logger(self):
 167             return logging.getLogger(__name__)
 168
 169         @_logger.setter
 170         def _logger(self, _):
 171             pass
 172
 173         def __init__(self, *args):
 174             self._mon_commands_sent = []
 175             if not hasattr(self, '_store'):
 176                 self._store = {}
 177
 178
 179             if self.__class__.__name__ not in M_classes:
 180                 # call those only once.
 181                 self._register_commands('')
 182                 self._register_options('')
 183                 M_classes.add(self.__class__.__name__)
 184
 185             super(M, self).__init__()
 186             self._ceph_get_version = mock.Mock()
 187             self._ceph_get_ceph_conf_path = mock.MagicMock()
 188             self._ceph_get_option = mock.MagicMock()
 189             self._ceph_get_context = mock.MagicMock()
 190             self._ceph_register_client = mock.MagicMock()
 191             self._ceph_set_health_checks = mock.MagicMock()
 192             self._configure_logging = lambda *_: None
 193             self._unconfigure_logging = mock.MagicMock()
 194             self._ceph_log = mock.MagicMock()
 195             self._ceph_dispatch_remote = lambda *_: None
 196             self._ceph_get_mgr_id = mock.MagicMock()
 197
 198
 199     cm = mock.Mock()
 200     cm.BaseMgrModule = M
 201     cm.BaseMgrStandbyModule = M
 202     sys.modules['ceph_module'] = cm
 203
 204     def mock_ceph_modules():
 205         class MockRadosError(Exception):
 206             def __init__(self, message, errno=None):
 207                 super(MockRadosError, self).__init__(message)
 208                 self.errno = errno
 209
 210             def __str__(self):
 211                 msg = super(MockRadosError, self).__str__()
 212                 if self.errno is None:
 213                     return msg
 214                 return '[errno {0}] {1}'.format(self.errno, msg)
 215
 216
 217         sys.modules.update({
 218             'rados': mock.MagicMock(Error=MockRadosError, OSError=MockRadosError),
 219             'rbd': mock.Mock(),
 220             'cephfs': mock.Mock(),
 221         })
 222
 223     # Unconditionally mock the rados objects when we're imported
 224     mock_ceph_modules()  # type: ignore