1 # Copyright 2015-2017 Intel Corporation.
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
7 # http://www.apache.org/licenses/LICENSE-2.0
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
15 """Settings and configuration handlers.
17 Settings will be loaded from several .conf files
18 and any user provided settings file.
21 # pylint: disable=invalid-name
31 _LOGGER = logging.getLogger(__name__)
33 # Special test parameters which are not part of standard VSPERF configuration
34 _EXTRA_TEST_PARAMS = ['TUNNEL_TYPE']
36 # regex to parse configuration macros from 04_vnf.conf
37 # it will select all patterns starting with # sign
38 # and returns macro parameters and step
39 # examples of valid macros:
41 # #MAC(AA:BB:CC:DD:EE:FF) or #MAC(AA:BB:CC:DD:EE:FF,2)
42 # #IP(192.168.1.2) or #IP(192.168.1.2,2)
44 _PARSE_PATTERN = r'(#[A-Z]+)(\(([^(),]+)(,([0-9]+))?\))?'
46 class Settings(object):
47 """Holding class for settings.
52 def _eval_param(self, param):
53 # pylint: disable=invalid-name
54 """ Helper function for expansion of references to vsperf parameters
56 if isinstance(param, str):
57 # evaluate every #PARAM reference inside parameter itself
58 macros = re.findall(r'#PARAM\((([\w\-]+)(\[[\w\[\]\-\'\"]+\])*)\)', param)
61 # pylint: disable=eval-used
63 tmp_val = str(eval("self.getValue('{}'){}".format(macro[1], macro[2])))
64 param = param.replace('#PARAM({})'.format(macro[0]), tmp_val)
65 # silently ignore that option required by PARAM macro can't be evaluated;
66 # It is possible, that referred parameter will be constructed during runtime
70 except AttributeError:
73 elif isinstance(param, list) or isinstance(param, tuple):
76 tmp_list.append(self._eval_param(item))
78 elif isinstance(param, dict):
80 for (key, value) in param.items():
81 tmp_dict[key] = self._eval_param(value)
86 def getValue(self, attr):
87 """Return a settings item value
89 if attr in self.__dict__:
90 if attr == 'TEST_PARAMS':
91 return getattr(self, attr)
93 master_value = getattr(self, attr)
94 # Check if parameter value was modified by CLI option
95 cli_value = get_test_param(attr, None)
97 # TRAFFIC dictionary is not overridden by CLI option
98 # but only updated by specified values
100 tmp_value = copy.deepcopy(master_value)
101 tmp_value = merge_spec(tmp_value, cli_value)
102 return self._eval_param(tmp_value)
104 return self._eval_param(cli_value)
106 return self._eval_param(master_value)
108 raise AttributeError("%r object has no attribute %r" %
109 (self.__class__, attr))
111 def __setattr__(self, name, value):
114 # skip non-settings. this should exclude built-ins amongst others
115 if not name.isupper():
118 # we can assume all uppercase keys are valid settings
119 super(Settings, self).__setattr__(name, value)
121 def setValue(self, name, value):
124 if name is not None and value is not None:
125 super(Settings, self).__setattr__(name, value)
127 def load_from_file(self, path):
128 """Update ``settings`` with values found in module at ``path``.
132 custom_settings = imp.load_source('custom_settings', path)
134 for key in dir(custom_settings):
135 if getattr(custom_settings, key) is not None:
136 setattr(self, key, getattr(custom_settings, key))
138 def load_from_dir(self, dir_path):
139 """Update ``settings`` with contents of the .conf files at ``path``.
141 Each file must be named Nfilename.conf, where N is a single or
142 multi-digit decimal number. The files are loaded in ascending order of
143 N - so if a configuration item exists in more that one file the setting
144 in the file with the largest value of N takes precedence.
146 :param dir_path: The full path to the dir from which to load the .conf
151 regex = re.compile("^(?P<digit_part>[0-9]+).*.conf$")
153 def get_prefix(filename):
155 Provide a suitable function for sort's key arg
157 match_object = regex.search(os.path.basename(filename))
158 return int(match_object.group('digit_part'))
160 # get full file path to all files & dirs in dir_path
161 file_paths = os.listdir(dir_path)
162 file_paths = [os.path.join(dir_path, x) for x in file_paths]
164 # filter to get only those that are a files, with a leading
165 # digit and end in '.conf'
166 file_paths = [x for x in file_paths if os.path.isfile(x) and
167 regex.search(os.path.basename(x))]
169 # sort ascending on the leading digits
170 file_paths.sort(key=get_prefix)
172 # load settings from each file in turn
173 for filepath in file_paths:
174 self.load_from_file(filepath)
176 def load_from_dict(self, conf):
178 Update ``settings`` with values found in ``conf``.
180 Unlike the other loaders, this is case insensitive.
183 if conf[key] is not None:
184 if isinstance(conf[key], dict):
185 # recursively update dict items, e.g. TEST_PARAMS
186 setattr(self, key.upper(),
187 merge_spec(getattr(self, key.upper()), conf[key]))
189 setattr(self, key.upper(), conf[key])
191 def load_from_env(self):
193 Update ``settings`` with values found in the environment.
195 for key in os.environ:
196 setattr(self, key, os.environ[key])
198 def check_test_params(self):
200 Check all parameters defined inside TEST_PARAMS for their
201 existence. In case that non existing vsperf parmeter name
202 is detected, then VSPER will raise a runtime error.
205 for key in settings.getValue('TEST_PARAMS'):
206 if key == 'TEST_PARAMS':
207 raise RuntimeError('It is not allowed to define TEST_PARAMS '
208 'as a test parameter')
209 if key not in self.__dict__ and key not in _EXTRA_TEST_PARAMS:
210 unknown_keys.append(key)
212 if len(unknown_keys):
213 raise RuntimeError('Test parameters contain unknown configuration '
214 'parameter(s): {}'.format(', '.join(unknown_keys)))
216 def check_vm_settings(self, vm_number):
218 Check all VM related settings starting with GUEST_ prefix.
219 If it is not available for defined number of VMs, then vsperf
220 will try to expand it automatically. Expansion is performed
221 also in case that first list item contains a macro.
223 for key in self.__dict__:
224 if key.startswith('GUEST_'):
225 value = self.getValue(key)
226 if isinstance(value, str) and str(value).find('#') >= 0:
227 self._expand_vm_settings(key, 1)
229 if isinstance(value, list):
230 if len(value) < vm_number or str(value[0]).find('#') >= 0:
231 # expand configuration for all VMs
232 self._expand_vm_settings(key, vm_number)
234 def _expand_vm_settings(self, key, vm_number):
236 Expand VM option with given key for given number of VMs
238 tmp_value = self.getValue(key)
239 if isinstance(tmp_value, str):
241 master_value = tmp_value
242 tmp_value = [tmp_value]
245 master_value = tmp_value[0]
247 master_value_str = str(master_value)
248 if master_value_str.find('#') >= 0:
249 self.__dict__[key] = []
250 for vmindex in range(vm_number):
251 value = master_value_str.replace('#VMINDEX', str(vmindex))
252 for macro, args, param, _, step in re.findall(_PARSE_PATTERN, value):
253 multi = int(step) if len(step) and int(step) else 1
255 # pylint: disable=eval-used
256 tmp_result = str(eval(param))
257 elif macro == '#MAC':
258 mac_value = netaddr.EUI(param).value
259 mac = netaddr.EUI(mac_value + vmindex * multi)
260 mac.dialect = netaddr.mac_unix_expanded
261 tmp_result = str(mac)
263 ip_value = netaddr.IPAddress(param).value
264 tmp_result = str(netaddr.IPAddress(ip_value + vmindex * multi))
266 raise RuntimeError('Unknown configuration macro {} in {}'.format(macro, key))
268 value = value.replace("{}{}".format(macro, args), tmp_result)
270 # retype value to original type if needed
271 if not isinstance(master_value, str):
272 value = ast.literal_eval(value)
273 self.__dict__[key].append(value)
275 for vmindex in range(len(tmp_value), vm_number):
276 self.__dict__[key].append(master_value)
279 self.__dict__[key] = self.__dict__[key][0]
281 _LOGGER.debug("Expanding option: %s = %s", key, self.__dict__[key])
284 """Provide settings as a human-readable string.
286 This can be useful for debug.
289 A human-readable string.
292 for key in self.__dict__:
293 tmp_dict[key] = self.getValue(key)
295 return pprint.pformat(tmp_dict)
298 # validation methods used by step driven testcases
300 def validate_getValue(self, result, attr):
301 """Verifies, that correct value was returned
303 # getValue must be called to expand macros and apply
304 # values from TEST_PARAM option
305 assert result == self.getValue(attr)
308 def validate_setValue(self, dummy_result, name, value):
309 """Verifies, that value was correctly set
311 assert value == self.__dict__[name]
314 settings = Settings()
316 def get_test_param(key, default=None):
317 """Retrieve value for test param ``key`` if available.
319 :param key: Key to retrieve from test params.
320 :param default: Default to return if key not found.
322 :returns: Value for ``key`` if found, else ``default``.
324 test_params = settings.getValue('TEST_PARAMS')
325 return test_params.get(key, default) if test_params else default
327 def merge_spec(orig, new):
328 """Merges ``new`` dict with ``orig`` dict, and returns orig.
330 This takes into account nested dictionaries. Example:
332 >>> old = {'foo': 1, 'bar': {'foo': 2, 'bar': 3}}
333 >>> new = {'foo': 6, 'bar': {'foo': 7}}
334 >>> merge_spec(old, new)
335 {'foo': 6, 'bar': {'foo': 7, 'bar': 3}}
337 You'll notice that ``bar.bar`` is not removed. This is the desired result.
343 # Not allowing derived dictionary types for now
344 # pylint: disable=unidiomatic-typecheck
345 if type(orig[key]) == dict:
346 orig[key] = merge_spec(orig[key], new[key])