aboutsummaryrefslogtreecommitdiffstats
path: root/conf/__init__.py
blob: 83c5475f72218f325d1ee336dc1da96ec11a3bca (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
# Copyright 2015-2017 Intel Corporation.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

"""Settings and configuration handlers.

Settings will be loaded from several .conf files
and any user provided settings file.
"""

# pylint: disable=invalid-name

import copy
import os
import re
import logging
import pprint
import ast
import netaddr

_LOGGER = logging.getLogger(__name__)

# Special test parameters which are not part of standard VSPERF configuration
_EXTRA_TEST_PARAMS = ['TUNNEL_TYPE']

# regex to parse configuration macros from 04_vnf.conf
# it will select all patterns starting with # sign
# and returns macro parameters and step
# examples of valid macros:
#   #VMINDEX
#   #MAC(AA:BB:CC:DD:EE:FF) or #MAC(AA:BB:CC:DD:EE:FF,2)
#   #IP(192.168.1.2) or #IP(192.168.1.2,2)
#   #EVAL(2*#VMINDEX)
_PARSE_PATTERN = r'(#[A-Z]+)(\(([^(),]+)(,([0-9]+))?\))?'

class Settings(object):
    """Holding class for settings.
    """
    def __init__(self):
        pass

    def _eval_param(self, param):
        # pylint: disable=invalid-name
        """ Helper function for expansion of references to vsperf parameters
        """
        if isinstance(param, str):
            # evaluate every #PARAM reference inside parameter itself
            macros = re.findall(r'#PARAM\((([\w\-]+)(\[[\w\[\]\-\'\"]+\])*)\)', param)
            if macros:
                for macro in macros:
                    # pylint: disable=eval-used
                    try:
                        tmp_val = str(eval("self.getValue('{}'){}".format(macro[1], macro[2])))
                        param = param.replace('#PARAM({})'.format(macro[0]), tmp_val)
                    # silently ignore that option required by PARAM macro can't be evaluated;
                    # It is possible, that referred parameter will be constructed during runtime
                    # and re-read later.
                    except IndexError:
                        pass
                    except AttributeError:
                        pass
            return param
        elif isinstance(param, (list, tuple)):
            tmp_list = []
            for item in param:
                tmp_list.append(self._eval_param(item))
            return tmp_list
        elif isinstance(param, dict):
            tmp_dict = {}
            for (key, value) in param.items():
                tmp_dict[key] = self._eval_param(value)
            return tmp_dict
        else:
            return param

    def getValue(self, attr):
        """Return a settings item value
        """
        if attr in self.__dict__:
            if attr == 'TEST_PARAMS':
                return getattr(self, attr)
            else:
                master_value = getattr(self, attr)
                # Check if parameter value was modified by CLI option
                cli_value = get_test_param(attr, None)
                if cli_value is not None:
                    # TRAFFIC dictionary is not overridden by CLI option
                    # but only updated by specified values
                    if attr == 'TRAFFIC':
                        tmp_value = copy.deepcopy(master_value)
                        tmp_value = merge_spec(tmp_value, cli_value)
                        return self._eval_param(tmp_value)
                    else:
                        return self._eval_param(cli_value)
                else:
                    return self._eval_param(master_value)
        else:
            raise AttributeError("%r object has no attribute %r" %
                                 (self.__class__, attr))

    def __setattr__(self, name, value):
        """Set a value
        """
        # skip non-settings. this should exclude built-ins amongst others
        if not name.isupper():
            return

        # we can assume all uppercase keys are valid settings
        super(Settings, self).__setattr__(name, value)

    def setValue(self, name, value):
        """Set a value
        """
        if name is not None and value is not None:
            super(Settings, self).__setattr__(name, value)

    def resetValue(self, attr):
        """If parameter was overridden by TEST_PARAMS, then it will
           be set to its original value.
        """
        if attr in self.__dict__['TEST_PARAMS']:
            self.__dict__['TEST_PARAMS'].pop(attr)

    def load_from_file(self, path):
        """Update ``settings`` with values found in module at ``path``.
        """
        import imp

        custom_settings = imp.load_source('custom_settings', path)

        for key in dir(custom_settings):
            if getattr(custom_settings, key) is not None:
                setattr(self, key, getattr(custom_settings, key))

    def load_from_dir(self, dir_path):
        """Update ``settings`` with contents of the .conf files at ``path``.

        Each file must be named Nfilename.conf, where N is a single or
        multi-digit decimal number.  The files are loaded in ascending order of
        N - so if a configuration item exists in more that one file the setting
        in the file with the largest value of N takes precedence.

        :param dir_path: The full path to the dir from which to load the .conf
            files.

        :returns: None
        """
        regex = re.compile("^(?P<digit_part>[0-9]+)(?P<alfa_part>[a-z]?)_.*.conf$")

        def get_prefix(filename):
            """
            Provide a suitable function for sort's key arg
            """
            match_object = regex.search(os.path.basename(filename))
            return [int(match_object.group('digit_part')),
                    match_object.group('alfa_part')]

        # get full file path to all files & dirs in dir_path
        file_paths = os.listdir(dir_path)
        file_paths = [os.path.join(dir_path, x) for x in file_paths]

        # filter to get only those that are a files, with a leading
        # digit and end in '.conf'
        file_paths = [x for x in file_paths if os.path.isfile(x) and
                      regex.search(os.path.basename(x))]

        # sort ascending on the leading digits and afla (e.g. 03_, 05a_)
        file_paths.sort(key=get_prefix)

        # load settings from each file in turn
        for filepath in file_paths:
            self.load_from_file(filepath)

    def load_from_dict(self, conf):
        """
        Update ``settings`` with values found in ``conf``.

        Unlike the other loaders, this is case insensitive.
        """
        for key in conf:
            if conf[key] is not None:
                if isinstance(conf[key], dict):
                    # recursively update dict items, e.g. TEST_PARAMS
                    setattr(self, key.upper(),
                            merge_spec(getattr(self, key.upper()), conf[key]))
                else:
                    setattr(self, key.upper(), conf[key])

    def restore_from_dict(self, conf):
        """
        Restore ``settings`` with values found in ``conf``.

        Method will drop all configuration options and restore their
        values from conf dictionary
        """
        self.__dict__.clear()
        tmp_conf = copy.deepcopy(conf)
        for key in tmp_conf:
            self.setValue(key, tmp_conf[key])

    def load_from_env(self):
        """
        Update ``settings`` with values found in the environment.
        """
        for key in os.environ:
            setattr(self, key, os.environ[key])

    def check_test_params(self):
        """
        Check all parameters defined inside TEST_PARAMS for their
        existence. In case that non existing vsperf parmeter name
        is detected, then VSPER will raise a runtime error.
        """
        unknown_keys = []
        for key in settings.getValue('TEST_PARAMS'):
            if key == 'TEST_PARAMS':
                raise RuntimeError('It is not allowed to define TEST_PARAMS '
                                   'as a test parameter')
            if key not in self.__dict__ and key not in _EXTRA_TEST_PARAMS:
                unknown_keys.append(key)

        if unknown_keys:
            raise RuntimeError('Test parameters contain unknown configuration '
                               'parameter(s): {}'.format(', '.join(unknown_keys)))

    def check_vm_settings(self, vm_number):
        """
        Check all VM related settings starting with GUEST_ prefix.
        If it is not available for defined number of VMs, then vsperf
        will try to expand it automatically. Expansion is performed
        also in case that first list item contains a macro.
        """
        for key in self.__dict__:
            if key.startswith('GUEST_'):
                value = self.getValue(key)
                if isinstance(value, str) and str(value).find('#') >= 0:
                    self._expand_vm_settings(key, 1)

                if isinstance(value, list):
                    if len(value) < vm_number or str(value[0]).find('#') >= 0:
                        # expand configuration for all VMs
                        self._expand_vm_settings(key, vm_number)

    def _expand_vm_settings(self, key, vm_number):
        """
        Expand VM option with given key for given number of VMs
        """
        tmp_value = self.getValue(key)
        if isinstance(tmp_value, str):
            scalar = True
            master_value = tmp_value
            tmp_value = [tmp_value]
        else:
            scalar = False
            master_value = tmp_value[0]

        master_value_str = str(master_value)
        if master_value_str.find('#') >= 0:
            self.__dict__[key] = []
            for vmindex in range(vm_number):
                value = master_value_str.replace('#VMINDEX', str(vmindex))
                for macro, args, param, _, step in re.findall(_PARSE_PATTERN, value):
                    multi = int(step) if step and int(step) else 1
                    if macro == '#EVAL':
                        # pylint: disable=eval-used
                        tmp_result = str(eval(param))
                    elif macro == '#MAC':
                        mac_value = netaddr.EUI(param).value
                        mac = netaddr.EUI(mac_value + vmindex * multi)
                        mac.dialect = netaddr.mac_unix_expanded
                        tmp_result = str(mac)
                    elif macro == '#IP':
                        ip_value = netaddr.IPAddress(param).value
                        tmp_result = str(netaddr.IPAddress(ip_value + vmindex * multi))
                    else:
                        raise RuntimeError('Unknown configuration macro {} in {}'.format(macro, key))

                    value = value.replace("{}{}".format(macro, args), tmp_result)

                # retype value to original type if needed
                if not isinstance(master_value, str):
                    value = ast.literal_eval(value)
                self.__dict__[key].append(value)
        else:
            for vmindex in range(len(tmp_value), vm_number):
                self.__dict__[key].append(master_value)

        if scalar:
            self.__dict__[key] = self.__dict__[key][0]

        _LOGGER.debug("Expanding option: %s = %s", key, self.__dict__[key])

    def __str__(self):
        """Provide settings as a human-readable string.

        This can be useful for debug.

        Returns:
            A human-readable string.
        """
        tmp_dict = {}
        for key in self.__dict__:
            tmp_dict[key] = self.getValue(key)

        return pprint.pformat(tmp_dict)

    #
    # validation methods used by step driven testcases
    #
    def validate_getValue(self, result, attr):
        """Verifies, that correct value was returned
        """
        # getValue must be called to expand macros and apply
        # values from TEST_PARAM option
        assert result == self.getValue(attr)
        return True

    def validate_setValue(self, _dummy_result, name, value):
        """Verifies, that value was correctly set
        """
        assert value == self.__dict__[name]
        return True

    def validate_resetValue(self, _dummy_result, attr):
        """Verifies, that value was correctly reset
        """
        return 'TEST_PARAMS' not in self.__dict__ or \
               attr not in self.__dict__['TEST_PARAMS']

settings = Settings()

def get_test_param(key, default=None):
    """Retrieve value for test param ``key`` if available.

    :param key: Key to retrieve from test params.
    :param default: Default to return if key not found.

    :returns: Value for ``key`` if found, else ``default``.
    """
    test_params = settings.getValue('TEST_PARAMS')
    return test_params.get(key, default) if test_params else default

def merge_spec(orig, new):
    """Merges ``new`` dict with ``orig`` dict, and returns orig.

    This takes into account nested dictionaries. Example:

        >>> old = {'foo': 1, 'bar': {'foo': 2, 'bar': 3}}
        >>> new = {'foo': 6, 'bar': {'foo': 7}}
        >>> merge_spec(old, new)
        {'foo': 6, 'bar': {'foo': 7, 'bar': 3}}

    You'll notice that ``bar.bar`` is not removed. This is the desired result.
    """
    for key in orig:
        if key not in new:
            continue

        # Not allowing derived dictionary types for now
        # pylint: disable=unidiomatic-typecheck
        if type(orig[key]) == dict:
            orig[key] = merge_spec(orig[key], new[key])
        else:
            orig[key] = new[key]

    for key in new:
        if key not in orig:
            orig[key] = new[key]

    return orig