]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/tests/__init__.py
2 from __future__
import absolute_import
8 if 'UNITTEST' in os
.environ
:
10 # Mock ceph_module. Otherwise every module that is involved in a testcase and imports it will
11 # raise an ImportError
16 from unittest
import mock
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]
32 def mock_store_get(self
, kind
, key
, default
):
33 if not hasattr(self
, '_store'):
35 return self
._store
.get(f
'mock_store/{kind}/{key}', default
)
37 def mock_store_set(self
, kind
, key
, value
):
38 if not hasattr(self
, '_store'):
40 k
= f
'mock_store/{kind}/{key}'
45 self
._store
[k
] = value
47 def mock_store_prefix(self
, kind
, prefix
):
48 if not hasattr(self
, '_store'):
50 full_prefix
= f
'mock_store/{kind}/{prefix}'
51 kind_len
= len(f
'mock_store/{kind}/')
53 k
[kind_len
:]: v
for k
, v
in self
._store
.items()
54 if k
.startswith(full_prefix
)
57 def _ceph_get_store(self
, k
):
58 return self
.mock_store_get('store', k
, None)
60 def _ceph_set_store(self
, k
, v
):
61 self
.mock_store_set('store', k
, v
)
63 def _ceph_get_store_prefix(self
, prefix
):
64 return self
.mock_store_prefix('store', prefix
)
66 def _ceph_get_module_option(self
, module
, key
, localized_prefix
= None):
68 _
, val
, _
= self
.check_mon_command({
69 'prefix': 'config get',
71 'key': f
'mgr/{module}/{key}'
73 except FileNotFoundError
:
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']
84 'bool': lambda s
: bool(s
) and s
!= 'false' and s
!= 'False',
86 }[mo
[0].get('type', 'str')]
90 return val
if val
is not None else ''
92 def _ceph_set_module_option(self
, module
, key
, val
):
93 _
, _
, _
= self
.check_mon_command({
94 'prefix': 'config set',
96 'name': f
'mgr/{module}/{key}',
101 def _ceph_get(self
, data_name
):
102 return self
.mock_store_get('_ceph_get', data_name
, mock
.MagicMock())
104 def _ceph_send_command(self
, res
, svc_type
, svc_id
, command
, tag
, inbuf
):
106 cmd
= json
.loads(command
)
107 getattr(self
, '_mon_commands_sent', []).append(cmd
)
109 # Mocking the config store is handy sometimes:
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)
120 self
.mock_store_set('config', f
'{cmd["who"]}/{cmd["name"]}', cmd
['value'])
124 self
.mock_store_set('config', f
'{cmd["who"]}/{cmd["name"]}', None)
129 for prefix
, value
in self
.mock_store_prefix('config', '').items():
130 section
, name
= prefix
.split('/', 1)
139 if cmd
['prefix'] == 'config get':
141 elif cmd
['prefix'] == 'config set':
143 elif cmd
['prefix'] == 'config dump':
145 elif cmd
['prefix'] == 'config rm':
147 elif hasattr(self
, '_mon_command_mock_' + cmd
['prefix'].replace(' ', '_')):
148 a
= getattr(self
, '_mon_command_mock_' + cmd
['prefix'].replace(' ', '_'))
151 res
.complete(0, outb
, '')
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)
162 def assert_issued_mon_command(self
, command
):
163 assert command
in self
._mon
_commands
_sent
, self
._mon
_commands
_sent
167 return logging
.getLogger(__name__
)
170 def _logger(self
, _
):
173 def __init__(self
, *args
):
174 self
._mon
_commands
_sent
= []
175 if not hasattr(self
, '_store'):
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
__)
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()
201 cm
.BaseMgrStandbyModule
= M
202 sys
.modules
['ceph_module'] = cm
204 def mock_ceph_modules():
205 class MockRadosError(Exception):
206 def __init__(self
, message
, errno
=None):
207 super(MockRadosError
, self
).__init
__(message
)
211 msg
= super(MockRadosError
, self
).__str
__()
212 if self
.errno
is None:
214 return '[errno {0}] {1}'.format(self
.errno
, msg
)
218 'rados': mock
.MagicMock(Error
=MockRadosError
, OSError=MockRadosError
),
220 'cephfs': mock
.Mock(),
223 # Unconditionally mock the rados objects when we're imported
224 mock_ceph_modules() # type: ignore