+ _, _, _ = self.check_mon_command({
+ 'prefix': 'config set',
+ 'who': 'mgr',
+ 'name': f'mgr/{module}/{key}',
+ 'value': val
+ })
+ return val
+
+ def _ceph_get(self, data_name):
+ return self.mock_store_get('_ceph_get', data_name, mock.MagicMock())
+
+ def _ceph_send_command(self, res, svc_type, svc_id, command, tag):
+ cmd = json.loads(command)
+
+ # Mocking the config store is handy sometimes:
+ def config_get():
+ who = cmd['who'].split('.')
+ whos = ['global'] + ['.'.join(who[:i+1]) for i in range(len(who))]
+ for attepmt in reversed(whos):
+ val = self.mock_store_get('config', f'{attepmt}/{cmd["key"]}', None)
+ if val is not None:
+ return val
+ return None
+
+ def config_set():
+ self.mock_store_set('config', f'{cmd["who"]}/{cmd["name"]}', cmd['value'])
+ return ''
+
+ def config_dump():
+ r = []
+ for prefix, value in self.mock_store_preifx('config', '').items():
+ section, name = prefix.split('/', 1)
+ r.append({
+ 'name': name,
+ 'section': section,
+ 'value': value
+ })
+ return json.dumps(r)
+
+ outb = ''
+ if cmd['prefix'] == 'config get':
+ outb = config_get()
+ elif cmd['prefix'] == 'config set':
+ outb = config_set()
+ elif cmd['prefix'] == 'config dump':
+ outb = config_dump()
+ elif hasattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_')):
+ a = getattr(self, '_mon_command_mock_' + cmd['prefix'].replace(' ', '_'))
+ outb = a(cmd)
+
+ res.complete(0, outb, '')
+
+ @property
+ def _logger(self):
+ return logging.getLogger(__name__)
+
+ @_logger.setter
+ def _logger(self, _):
+ pass