summaryrefslogtreecommitdiffstats
path: root/src/osmo_gsm_tester/config.py
blob: 1d52073161812eb2a20525d8fe2e8f3790679399 (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
# osmo_gsm_tester: read and manage config files and global config
#
# Copyright (C) 2016-2017 by sysmocom - s.f.m.c. GmbH
#
# Author: Neels Hofmeyr <neels@hofmeyr.de>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.

# discussion for choice of config file format:
#
# Python syntax is insane, because it allows the config file to run arbitrary
# python commands.
#
# INI file format is nice and simple, but it doesn't allow having the same
# section numerous times (e.g. to define several modems or BTS models) and does
# not support nesting.
#
# JSON has too much braces and quotes to be easy to type
#
# YAML formatting is lean, but:
# - too powerful. The normal load() allows arbitrary code execution. There is
#   safe_load().
# - allows several alternative ways of formatting, better to have just one
#   authoritative style.
# - tries to detect types. It would be better to receive every setting as
#   simple string rather than e.g. an IMSI as an integer.
# - e.g. an IMSI starting with a zero is interpreted as octal value, resulting
#   in super confusing error messages if the user merely forgets to quote it.
# - does not tell me which line a config item came from, so no detailed error
#   message is possible.
#
# The Python ConfigParserShootout page has numerous contestants, but many of
# those seem to be not widely used / standardized or even tested.
# https://wiki.python.org/moin/ConfigParserShootout
#
# The optimum would be a stripped down YAML format.
# In the lack of that, we shall go with yaml.load_safe() + a round trip
# (feeding back to itself), converting keys to lowercase and values to string.
# There is no solution for octal interpretations nor config file source lines
# unless, apparently, we implement our own config parser.

import yaml
import os
import copy

from . import log, schema, util
from .util import is_dict, is_list, Dir, get_tempdir

ENV_PREFIX = 'OSMO_GSM_TESTER_'
ENV_CONF = os.getenv(ENV_PREFIX + 'CONF')

override_conf = None

DEFAULT_CONFIG_LOCATIONS = [
    '.',
    os.path.join(os.getenv('HOME'), '.config', 'osmo-gsm-tester'),
    '/usr/local/etc/osmo-gsm-tester',
    '/etc/osmo-gsm-tester'
    ]

PATHS_CONF = 'paths.conf'
DEFAULT_SUITES_CONF = 'default-suites.conf'
PATH_STATE_DIR = 'state_dir'
PATH_SUITES_DIR = 'suites_dir'
PATH_SCENARIOS_DIR = 'scenarios_dir'
PATHS_SCHEMA = {
        PATH_STATE_DIR: schema.STR,
        PATH_SUITES_DIR: schema.STR,
        PATH_SCENARIOS_DIR: schema.STR,
    }

PATHS_TEMPDIR_STR = '$TEMPDIR'

PATHS = None

def _get_config_file(basename, fail_if_missing=True):
    if override_conf:
        locations = [ override_conf ]
    elif ENV_CONF:
        locations = [ ENV_CONF ]
    else:
        locations = DEFAULT_CONFIG_LOCATIONS

    for l in locations:
        real_l = os.path.realpath(l)
        p = os.path.realpath(os.path.join(real_l, basename))
        if os.path.isfile(p):
            log.dbg('Found config file', basename, 'as', p, 'in', l, 'which is', real_l, _category=log.C_CNF)
            return (p, real_l)
    if not fail_if_missing:
        return None, None
    raise RuntimeError('configuration file not found: %r in %r' % (basename,
        [os.path.abspath(p) for p in locations]))

def get_config_file(basename, fail_if_missing=True):
    path, found_in = _get_config_file(basename, fail_if_missing)
    return path

def read_config_file(basename, validation_schema=None, if_missing_return=False):
    fail_if_missing = True
    if if_missing_return is not False:
        fail_if_missing = False
    path = get_config_file(basename, fail_if_missing=fail_if_missing)
    if path is None:
        return if_missing_return
    return read(path, validation_schema=validation_schema, if_missing_return=if_missing_return)

def get_configured_path(label, allow_unset=False):
    global PATHS

    env_name = ENV_PREFIX + label.upper()
    env_path = os.getenv(env_name)
    if env_path:
        real_env_path = os.path.realpath(env_path)
        log.dbg('Found path', label, 'as', env_path, 'in', '$' + env_name, 'which is', real_env_path, _category=log.C_CNF)
        return real_env_path

    if PATHS is None:
        paths_file, found_in = _get_config_file(PATHS_CONF)
        PATHS = read(paths_file, PATHS_SCHEMA)
        # sorted for deterministic regression test results
        for key, path in sorted(PATHS.items()):
            if not path.startswith(os.pathsep):
                PATHS[key] = os.path.realpath(os.path.join(found_in, path))
                log.dbg(paths_file + ': relative path', path, 'is', PATHS[key], _category=log.C_CNF)
    p = PATHS.get(label)
    if p is None and not allow_unset:
        raise RuntimeError('missing configuration in %s: %r' % (PATHS_CONF, label))

    log.dbg('Found path', label, 'as', p, _category=log.C_CNF)
    if p.startswith(PATHS_TEMPDIR_STR):
        p = os.path.join(get_tempdir(), p[len(PATHS_TEMPDIR_STR):])
        log.dbg('Path', label, 'contained', PATHS_TEMPDIR_STR, 'and becomes', p, _category=log.C_CNF)
    return p

def get_state_dir():
    return Dir(get_configured_path(PATH_STATE_DIR))

def get_suites_dir():
    return Dir(get_configured_path(PATH_SUITES_DIR))

def get_scenarios_dir():
    return Dir(get_configured_path(PATH_SCENARIOS_DIR))

def read(path, validation_schema=None, if_missing_return=False):
    log.ctx(path)
    if not os.path.isfile(path) and if_missing_return is not False:
        return if_missing_return
    with open(path, 'r') as f:
        config = yaml.safe_load(f)
    config = _standardize(config)
    if validation_schema:
        schema.validate(config, validation_schema)
    return config

def write(path, config):
    log.ctx(path)
    with open(path, 'w') as f:
        f.write(tostr(config))

def tostr(config):
    return _tostr(_standardize(config))

def _tostr(config):
    return yaml.dump(config, default_flow_style=False)

def _standardize_item(item):
    if isinstance(item, (tuple, list)):
        return [_standardize_item(i) for i in item]
    if isinstance(item, dict):
        return dict([(key.lower(), _standardize_item(val)) for key,val in item.items()])
    return str(item)

def _standardize(config):
    config = yaml.safe_load(_tostr(_standardize_item(config)))
    return config

def get_defaults(for_kind):
    defaults = read_config_file('defaults.conf', if_missing_return={})
    return defaults.get(for_kind, {})

class Scenario(log.Origin, dict):
    def __init__(self, name, path):
        super().__init__(log.C_TST, name)
        self.path = path

def get_scenario(name, validation_schema=None):
    scenarios_dir = get_scenarios_dir()
    if not name.endswith('.conf'):
        name = name + '.conf'
    path = scenarios_dir.child(name)
    if not os.path.isfile(path):
        raise RuntimeError('No such scenario file: %r' % path)
    sc = Scenario(name, path)
    sc.update(read(path, validation_schema=validation_schema))
    return sc

def add(dest, src):
    if is_dict(dest):
        if not is_dict(src):
            raise ValueError('cannot add to dict a value of type: %r' % type(src))

        for key, val in src.items():
            dest_val = dest.get(key)
            if dest_val is None:
                dest[key] = val
            else:
                log.ctx(key=key)
                add(dest_val, val)
        return
    if is_list(dest):
        if not is_list(src):
            raise ValueError('cannot add to list a value of type: %r' % type(src))
        dest.extend(src)
        return
    if dest == src:
        return
    raise ValueError('cannot add dicts, conflicting items (values %r and %r)'
                     % (dest, src))

def combine(dest, src):
    if is_dict(dest):
        if not is_dict(src):
            raise ValueError('cannot combine dict with a value of type: %r' % type(src))

        for key, val in src.items():
            log.ctx(key=key)
            dest_val = dest.get(key)
            if dest_val is None:
                dest[key] = val
            else:
                combine(dest_val, val)
        return
    if is_list(dest):
        if not is_list(src):
            raise ValueError('cannot combine list with a value of type: %r' % type(src))
        for i in range(len(src)):
            log.ctx(idx=i)
            combine(dest[i], src[i])
        return
    if dest == src:
        return
    raise ValueError('cannot combine dicts, conflicting items (values %r and %r)'
                     % (dest, src))

def overlay(dest, src):
    if is_dict(dest):
        if not is_dict(src):
            raise ValueError('cannot combine dict with a value of type: %r' % type(src))

        for key, val in src.items():
            log.ctx(key=key)
            dest_val = dest.get(key)
            dest[key] = overlay(dest_val, val)
        return dest
    if is_list(dest):
        if not is_list(src):
            raise ValueError('cannot combine list with a value of type: %r' % type(src))
        for i in range(len(src)):
            log.ctx(idx=i)
            dest[i] = overlay(dest[i], src[i])
        return dest
    return src

def replicate_times(d):
    'replicate items that have a "times" > 1'
    d = copy.deepcopy(d)
    for key, item_list in d.items():
        idx = 0
        while idx < len(item_list):
            item = item_list[idx]
            times = int(item.pop('times', 1))
            for j in range(1, times):
                item_list.insert(idx + j, copy.deepcopy(item))
            idx += times
    return d

# vim: expandtab tabstop=4 shiftwidth=4