]>
git.proxmox.com Git - ceph.git/blob - ceph/src/pybind/mgr/tests/__init__.py
7 if 'UNITTEST' in os
.environ
:
9 # Mock ceph_module. Otherwise every module that is involved in a testcase and imports it will
10 # raise an ImportError
15 from unittest
import mock
25 * self.set_store() populates self._store
26 * self.set_module_option() populates self._store[module_name]
27 * self.get(thing) comes from self._store['_ceph_get' + thing]
31 def mock_store_get(self
, kind
, key
, default
):
32 if not hasattr(self
, '_store'):
34 return self
._store
.get(f
'mock_store/{kind}/{key}', default
)
36 def mock_store_set(self
, kind
, key
, value
):
37 if not hasattr(self
, '_store'):
39 k
= f
'mock_store/{kind}/{key}'
44 self
._store
[k
] = value
46 def mock_store_prefix(self
, kind
, prefix
):
47 if not hasattr(self
, '_store'):
49 full_prefix
= f
'mock_store/{kind}/{prefix}'
50 kind_len
= len(f
'mock_store/{kind}/')
52 k
[kind_len
:]: v
for k
, v
in self
._store
.items()
53 if k
.startswith(full_prefix
)
56 def _ceph_get_store(self
, k
):
57 return self
.mock_store_get('store', k
, None)
59 def _ceph_set_store(self
, k
, v
):
60 self
.mock_store_set('store', k
, v
)
62 def _ceph_get_store_prefix(self
, prefix
):
63 return self
.mock_store_prefix('store', prefix
)
65 def _ceph_get_module_option(self
, module
, key
, localized_prefix
= None):
67 _
, val
, _
= self
.check_mon_command({
68 'prefix': 'config get',
70 'key': f
'mgr/{module}/{key}'
72 except FileNotFoundError
:
74 mo
= [o
for o
in self
.MODULE_OPTIONS
if o
['name'] == key
]
75 if len(mo
) >= 1: # >= 1, cause self.MODULE_OPTIONS. otherwise it
76 # fails when importing multiple modules.
77 if 'default' in mo
and val
is None:
78 val
= mo
[0]['default']
83 'bool': lambda s
: bool(s
) and s
!= 'false' and s
!= 'False',
85 }[mo
[0].get('type', 'str')]
89 return val
if val
is not None else ''
91 def _ceph_set_module_option(self
, module
, key
, val
):
92 _
, _
, _
= self
.check_mon_command({
93 'prefix': 'config set',
95 'name': f
'mgr/{module}/{key}',
100 def _ceph_get(self
, data_name
):
101 return self
.mock_store_get('_ceph_get', data_name
, mock
.MagicMock())
103 def _ceph_send_command(self
, res
, svc_type
, svc_id
, command
, tag
, inbuf
):
105 cmd
= json
.loads(command
)
106 getattr(self
, '_mon_commands_sent', []).append(cmd
)
108 # Mocking the config store is handy sometimes:
110 who
= cmd
['who'].split('.')
111 whos
= ['global'] + ['.'.join(who
[:i
+1]) for i
in range(len(who
))]
112 for attepmt
in reversed(whos
):
113 val
= self
.mock_store_get('config', f
'{attepmt}/{cmd["key"]}', None)
119 self
.mock_store_set('config', f
'{cmd["who"]}/{cmd["name"]}', cmd
['value'])
123 self
.mock_store_set('config', f
'{cmd["who"]}/{cmd["name"]}', None)
128 for prefix
, value
in self
.mock_store_prefix('config', '').items():
129 section
, name
= prefix
.split('/', 1)
138 if cmd
['prefix'] == 'config get':
140 elif cmd
['prefix'] == 'config set':
142 elif cmd
['prefix'] == 'config dump':
144 elif cmd
['prefix'] == 'config rm':
146 elif hasattr(self
, '_mon_command_mock_' + cmd
['prefix'].replace(' ', '_')):
147 a
= getattr(self
, '_mon_command_mock_' + cmd
['prefix'].replace(' ', '_'))
150 res
.complete(0, outb
, '')
152 def _ceph_get_foreign_option(self
, entity
, name
):
153 who
= entity
.split('.')
154 whos
= ['global'] + ['.'.join(who
[:i
+1]) for i
in range(len(who
))]
155 for attepmt
in reversed(whos
):
156 val
= self
.mock_store_get('config', f
'{attepmt}/{name}', None)
161 def assert_issued_mon_command(self
, command
):
162 assert command
in self
._mon
_commands
_sent
, self
._mon
_commands
_sent
166 return logging
.getLogger(__name__
)
169 def _logger(self
, _
):
172 def __init__(self
, *args
):
173 self
._mon
_commands
_sent
= []
174 if not hasattr(self
, '_store'):
178 if self
.__class
__ not in M_classes
:
179 # call those only once.
180 self
._register
_commands
('')
181 self
._register
_options
('')
182 M_classes
.add(self
.__class
__)
184 super(M
, self
).__init
__()
185 self
._ceph
_get
_version
= mock
.Mock()
186 self
._ceph
_get
_ceph
_conf
_path
= mock
.MagicMock()
187 self
._ceph
_get
_option
= mock
.MagicMock()
188 self
._ceph
_get
_context
= mock
.MagicMock()
189 self
._ceph
_register
_client
= mock
.MagicMock()
190 self
._ceph
_set
_health
_checks
= mock
.MagicMock()
191 self
._configure
_logging
= lambda *_
: None
192 self
._unconfigure
_logging
= mock
.MagicMock()
193 self
._ceph
_log
= mock
.MagicMock()
194 self
._ceph
_dispatch
_remote
= lambda *_
: None
195 self
._ceph
_get
_mgr
_id
= mock
.MagicMock()
200 cm
.BaseMgrStandbyModule
= M
201 sys
.modules
['ceph_module'] = cm
203 def mock_ceph_modules():
204 class MockRadosError(Exception):
205 def __init__(self
, message
, errno
=None):
206 super(MockRadosError
, self
).__init
__(message
)
210 msg
= super(MockRadosError
, self
).__str
__()
211 if self
.errno
is None:
213 return '[errno {0}] {1}'.format(self
.errno
, msg
)
215 class MockObjectNotFound(Exception):
219 'rados': mock
.MagicMock(
220 Error
=MockRadosError
,
221 OSError=MockRadosError
,
222 ObjectNotFound
=MockObjectNotFound
),
224 'cephfs': mock
.Mock(),
227 # Unconditionally mock the rados objects when we're imported
228 mock_ceph_modules() # type: ignore