Coverage for osm_policy_module/core/config.py: 72%

1# -*- coding: utf-8 -*- 

2 

3# Copyright 2018 Whitestack, LLC 

4# ************************************************************* 

5 

6# This file is part of OSM Monitoring module 

7# All Rights Reserved to Whitestack, LLC 

8 

9# Licensed under the Apache License, Version 2.0 (the "License"); you may 

10# not use this file except in compliance with the License. You may obtain 

11# a copy of the License at 

12 

13# http://www.apache.org/licenses/LICENSE-2.0 

14 

15# Unless required by applicable law or agreed to in writing, software 

16# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 

17# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 

18# License for the specific language governing permissions and limitations 

19# under the License. 

20 

21# For those usages not covered by the Apache License, Version 2.0 please 

22# contact: bdiaz@whitestack.com or glavado@whitestack.com 

23## 

24"""Global configuration managed by environment variables.""" 

25 

26import logging 

27import os 

28 

29import pkg_resources 

30import yaml 

31 

32logger = logging.getLogger(__name__) 

33 

34 

35class Config: 

36 def __init__(self, config_file: str = ""): 

37 self.conf = {} 

38 self._read_config_file(config_file) 

39 self._read_env() 

40 

41 def _read_config_file(self, config_file): 

42 if not config_file: 

43 path = "pol.yaml" 

44 config_file = pkg_resources.resource_filename(__name__, path) 

45 with open(config_file) as f: 

46 self.conf = yaml.safe_load(f) 

47 

48 def get(self, section, field=None): 

49 if not field: 

50 return self.conf[section] 

51 return self.conf[section][field] 

52 

53 def set(self, section, field, value): 

54 if section not in self.conf: 

55 self.conf[section] = {} 

56 self.conf[section][field] = value 

57 

58 def _read_env(self): 

59 for env in os.environ: 

60 if not env.startswith("OSMPOL_"): 

61 continue 

62 elements = env.lower().split("_") 

63 if len(elements) < 3: 

64 logger.warning( 

65 "Environment variable %s=%s does not comply with required format. Section and/or field missing.", 

66 env, 

67 os.getenv(env), 

68 ) 

69 continue 

70 section = elements[1] 

71 field = "_".join(elements[2:]) 

72 value = os.getenv(env) 

73 if section not in self.conf: 

74 self.conf[section] = {} 

75 self.conf[section][field] = value