]>
Commit | Line | Data |
---|---|---|
9f95a23c TL |
1 | # type: ignore |
2 | from __future__ import absolute_import | |
3 | ||
f6b5b4d7 TL |
4 | import json |
5 | import logging | |
9f95a23c TL |
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 | ||
e306af50 TL |
20 | M_classes = set() |
21 | ||
9f95a23c | 22 | class M(object): |
f6b5b4d7 TL |
23 | """ |
24 | Note that: | |
e306af50 | 25 | |
f6b5b4d7 TL |
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: | |
e306af50 TL |
42 | if k in self._store: |
43 | del self._store[k] | |
44 | else: | |
f6b5b4d7 | 45 | self._store[k] = value |
e306af50 | 46 | |
f6b5b4d7 TL |
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}/') | |
e306af50 | 52 | return { |
f6b5b4d7 TL |
53 | k[kind_len:]: v for k, v in self._store.items() |
54 | if k.startswith(full_prefix) | |
e306af50 TL |
55 | } |
56 | ||
f6b5b4d7 TL |
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] | |
f91f0fd5 TL |
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 '' | |
e306af50 TL |
88 | |
89 | def _ceph_set_module_option(self, module, key, val): | |
f6b5b4d7 TL |
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 | |
e306af50 | 149 | |
9f95a23c | 150 | def __init__(self, *args): |
f6b5b4d7 TL |
151 | if not hasattr(self, '_store'): |
152 | self._store = {} | |
153 | ||
e306af50 TL |
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 | ||
9f95a23c TL |
161 | super(M, self).__init__() |
162 | self._ceph_get_version = mock.Mock() | |
9f95a23c | 163 | self._ceph_get_option = mock.MagicMock() |
e306af50 TL |
164 | self._ceph_get_context = mock.MagicMock() |
165 | self._ceph_register_client = mock.MagicMock() | |
f6b5b4d7 | 166 | self._ceph_set_health_checks = mock.MagicMock() |
9f95a23c TL |
167 | self._configure_logging = lambda *_: None |
168 | self._unconfigure_logging = mock.MagicMock() | |
169 | self._ceph_log = mock.MagicMock() | |
9f95a23c | 170 | self._ceph_dispatch_remote = lambda *_: None |
f91f0fd5 | 171 | self._ceph_get_mgr_id = mock.MagicMock() |
9f95a23c TL |
172 | |
173 | ||
174 | cm = mock.Mock() | |
175 | cm.BaseMgrModule = M | |
176 | cm.BaseMgrStandbyModule = M | |
177 | sys.modules['ceph_module'] = cm | |
e306af50 TL |
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({ | |
f6b5b4d7 | 193 | 'rados': mock.MagicMock(Error=MockRadosError, OSError=MockRadosError), |
e306af50 TL |
194 | 'rbd': mock.Mock(), |
195 | 'cephfs': mock.Mock(), | |
196 | }) |