Yaml configuration system refactoring.
* Refactoring YamlInterpolator -> YamlRefLoader * Add new exception types with formating errros, specified for cfg * Add documentation in class docstrings * Add unit tests for cfg.YamlRefLoader and exceptions.YamlException * Extend reference links possibilities (concatenating, etc.) * Add refence loop control * Reverting to simpler reflinks using I. Melnikov's draft (thanks to him). Change-Id: I565085ae9a3015d005dcc532a74aaca49f41277a
This commit is contained in:
Vladimir Novikov
parent
552e072aa5
commit
2051e05d05
@ -56,9 +56,9 @@ class Action(object):
|
||||
self.root_dir = root_dir
|
||||
# Action phases are tracked in this directory
|
||||
self.phase_dir = sh.joinpths(root_dir, 'phases')
|
||||
# Yamls are 'interpolated' using this instance at the given
|
||||
# component directory where component configuration will be found...
|
||||
self.interpolator = cfg.YamlInterpolator(settings.COMPONENT_CONF_DIR)
|
||||
# Yamls are loaded (with its reference links) using this instance at the
|
||||
# given component directory where component configuration will be found.
|
||||
self.config_loader = cfg.YamlRefLoader(settings.COMPONENT_CONF_DIR)
|
||||
# Keyring/pw settings + cache
|
||||
self.passwords = {}
|
||||
self.keyring_path = cli_opts.pop('keyring_path')
|
||||
@ -173,7 +173,7 @@ class Action(object):
|
||||
def _get_interpolated_options(self, name):
|
||||
opts = {}
|
||||
for c in self._get_interpolated_names(name):
|
||||
opts.update(self.interpolator.extract(c))
|
||||
opts.update(self.config_loader.load(c))
|
||||
return opts
|
||||
|
||||
def _construct_instances(self, persona):
|
||||
|
||||
296
anvil/cfg.py
296
anvil/cfg.py
@ -18,7 +18,6 @@
|
||||
import ConfigParser
|
||||
from ConfigParser import NoOptionError
|
||||
from ConfigParser import NoSectionError
|
||||
|
||||
import re
|
||||
|
||||
from StringIO import StringIO
|
||||
@ -26,11 +25,11 @@ from StringIO import StringIO
|
||||
# This one keeps comments but has some weirdness with it
|
||||
import iniparse
|
||||
|
||||
from anvil import exceptions
|
||||
from anvil import log as logging
|
||||
from anvil import shell as sh
|
||||
from anvil import utils
|
||||
|
||||
INTERP_PAT = r"\s*\$\(([\w\d-]+):([\w\d-]+)\)\s*"
|
||||
|
||||
LOG = logging.getLogger(__name__)
|
||||
|
||||
@ -164,141 +163,200 @@ class DefaultConf(object):
|
||||
self.backing.remove_option(section, key)
|
||||
|
||||
|
||||
class YamlInterpolator(object):
|
||||
def __init__(self, base):
|
||||
self.included = {}
|
||||
self.interpolated = {}
|
||||
self.base = base
|
||||
self.auto_specials = {
|
||||
# TODO(vnovikov): inject all config merges into class below
|
||||
#class YamlMergeLoader(object):
|
||||
#
|
||||
# def __init__(self, path):
|
||||
# self._merge_order = ('general',)
|
||||
# self._base_loader = YamlRefLoader(path)
|
||||
#
|
||||
# def load(self, distro, component, persona, cli):
|
||||
#
|
||||
# distro_opts = distro.options
|
||||
# general_component_opts = self._base_loader.load('general')
|
||||
# component_specific_opts = self._base_loader.load(component)
|
||||
# persona_component_opts = persona.component_options.get(component, {})
|
||||
# persona_global_opts = persona.component_options.get('global', {})
|
||||
# cli_opts = cli
|
||||
#
|
||||
# merged_opts = utils.merge_dicts(
|
||||
# distro_opts,
|
||||
# general_component_opts,
|
||||
# component_specific_opts,
|
||||
# persona_component_opts,
|
||||
# persona_global_opts,
|
||||
# cli_opts,
|
||||
# )
|
||||
#
|
||||
# return merged_opts
|
||||
|
||||
|
||||
class YamlRefLoader(object):
|
||||
"""Reference loader for *.yaml configs.
|
||||
|
||||
Holds usual safe loading of the *.yaml files, caching, resolving and getting
|
||||
all reference links and transforming all data to python built-in types.
|
||||
|
||||
Let's describe some basics.
|
||||
In this context reference means value which formatted just like:
|
||||
opt: "$(source:option)" , or
|
||||
opt: "some-additional-data-$(source:option)-some-postfix-data", where:
|
||||
opt - base option name
|
||||
source - other source config (i.e. other *.yaml file) from which we
|
||||
should get 'option'
|
||||
option - option name in 'source'
|
||||
In other words it means that loader will try to find and read 'option' from
|
||||
'source'.
|
||||
|
||||
Any source config also allows:
|
||||
References to itself via it's name (opt: "$(source:opt)",
|
||||
in file - source.yaml)
|
||||
|
||||
References to auto parameters (opt: $(auto:ip), will insert current ip).
|
||||
'auto' allows next options: 'ip', 'hostname' and 'home'
|
||||
|
||||
Implicit and multi references just like
|
||||
s.yaml => opt: "here 3 opts: $(source:opt), $(source2:opt) and $(auto:ip)".
|
||||
|
||||
Exception cases:
|
||||
* if reference 'option' does not exist than YamlOptionException is raised
|
||||
* if config 'source' does not exist than YamlConfException is raised
|
||||
* if reference loop found than YamlLoopException is raised
|
||||
|
||||
Config file example:
|
||||
(file sample.yaml)
|
||||
|
||||
reference: "$(source:option)"
|
||||
ip: "$(auto:ip)"
|
||||
self_ref: "$(sample:ip)" # this will equal ip option.
|
||||
opt: "http://$(auto:ip)/"
|
||||
"""
|
||||
|
||||
def __init__(self, path):
|
||||
self._conf_ext = '.yaml'
|
||||
self._ref_pattern = re.compile(r"\$\(([\w\d-]+)\:([\w\d-]+)\)")
|
||||
self._predefined_refs = {
|
||||
'auto': {
|
||||
'ip': utils.get_host_ip,
|
||||
'home': sh.gethomedir,
|
||||
'hostname': sh.hostname,
|
||||
}
|
||||
}
|
||||
self._path = path # path to root directory with configs
|
||||
self._cached = {} # buffer to save already loaded configs
|
||||
self._processed = {} # buffer to save already processed configs
|
||||
self._ref_stack = [] # stack for controlling reference loop
|
||||
|
||||
def _interpolate_iterable(self, what):
|
||||
if isinstance(what, (set)):
|
||||
n_what = set()
|
||||
for v in what:
|
||||
n_what.add(self._interpolate(v))
|
||||
return n_what
|
||||
def _process_string(self, value):
|
||||
"""Processing string (and reference links) values via regexp."""
|
||||
processed = value
|
||||
|
||||
# Process each reference in value (one by one)
|
||||
for match in self._ref_pattern.finditer(value):
|
||||
ref_conf, ref_opt = match.groups()
|
||||
val = self._load_option(ref_conf, ref_opt)
|
||||
|
||||
if match.group(0) == value:
|
||||
return val
|
||||
else:
|
||||
n_what = []
|
||||
for v in what:
|
||||
n_what.append(self._interpolate(v))
|
||||
if isinstance(what, (tuple)):
|
||||
n_what = tuple(n_what)
|
||||
return n_what
|
||||
processed = re.sub(self._ref_pattern, str(val), processed, count=1)
|
||||
return processed
|
||||
|
||||
def _interpolate_dictionary(self, what):
|
||||
n_what = {}
|
||||
for (k, v) in what.iteritems():
|
||||
n_what[k] = self._interpolate(v)
|
||||
return n_what
|
||||
def _process_dict(self, value):
|
||||
"""Process dictionary values."""
|
||||
processed = utils.OrderedDict()
|
||||
for opt, val in sorted(value.items()):
|
||||
res = self._process(val)
|
||||
processed[opt] = res
|
||||
|
||||
def _include_dictionary(self, what):
|
||||
n_what = {}
|
||||
for (k, value) in what.iteritems():
|
||||
n_what[k] = self._do_include(value)
|
||||
return n_what
|
||||
return processed
|
||||
|
||||
def _include_iterable(self, what):
|
||||
if isinstance(what, (set)):
|
||||
n_what = set()
|
||||
for v in what:
|
||||
n_what.add(self._do_include(v))
|
||||
return n_what
|
||||
def _process_iterable(self, value):
|
||||
"""Process list, set or tuple values."""
|
||||
processed = []
|
||||
for item in value:
|
||||
processed.append(self._process(item))
|
||||
|
||||
return processed
|
||||
|
||||
def _process_asis(self, value):
|
||||
"""Process built-in values."""
|
||||
return value
|
||||
|
||||
def _process(self, value):
|
||||
"""Base recursive method for processing references."""
|
||||
if isinstance(value, basestring):
|
||||
processed = self._process_string(value)
|
||||
elif isinstance(value, dict):
|
||||
processed = self._process_dict(value)
|
||||
elif isinstance(value, (list, set, tuple)):
|
||||
processed = self._process_iterable(value)
|
||||
else:
|
||||
n_what = []
|
||||
for v in what:
|
||||
n_what.append(self._do_include(v))
|
||||
if isinstance(what, (tuple)):
|
||||
n_what = tuple(n_what)
|
||||
return n_what
|
||||
processed = self._process_asis(value)
|
||||
|
||||
def _interpolate(self, value):
|
||||
new_value = value
|
||||
if value and isinstance(value, (basestring, str)):
|
||||
new_value = self._interpolate_string(value)
|
||||
elif isinstance(value, (dict)):
|
||||
new_value = self._interpolate_dictionary(value)
|
||||
elif isinstance(value, (list, set, tuple)):
|
||||
new_value = self._interpolate_iterable(value)
|
||||
return new_value
|
||||
return processed
|
||||
|
||||
def _interpolate_string(self, what):
|
||||
if not re.search(INTERP_PAT, what):
|
||||
# Leave it alone if the sub won't do
|
||||
# anything to begin with
|
||||
return what
|
||||
def _cache(self, conf):
|
||||
"""Cache config file into memory to avoid re-reading it from disk."""
|
||||
if conf not in self._cached:
|
||||
path = sh.joinpths(self._path, conf + self._conf_ext)
|
||||
if not sh.isfile(path):
|
||||
raise exceptions.YamlConfigNotFoundException(path)
|
||||
|
||||
def replacer(match):
|
||||
who = match.group(1).strip()
|
||||
key = match.group(2).strip()
|
||||
(is_special, special_value) = self._process_special(who, key)
|
||||
if is_special:
|
||||
return special_value
|
||||
if who not in self.interpolated:
|
||||
self.interpolated[who] = self.included[who]
|
||||
self.interpolated[who] = self._interpolate(self.included[who])
|
||||
return str(self.interpolated[who][key])
|
||||
# TODO(vnovikov): may be it makes sense to reintroduce load_yaml
|
||||
# for returning OrderedDict with the same order as options placement
|
||||
# in source yaml file...
|
||||
self._cached[conf] = utils.load_yaml(path) or {}
|
||||
|
||||
return re.sub(INTERP_PAT, replacer, what)
|
||||
def _precache(self):
|
||||
"""Cache and process predefined auto-references"""
|
||||
for conf, options in self._predefined_refs.items():
|
||||
if conf not in self._processed:
|
||||
processed = dict((option, functor())
|
||||
for option, functor in options.items())
|
||||
self._cached[conf] = processed
|
||||
self._processed[conf] = processed
|
||||
|
||||
def _process_special(self, who, key):
|
||||
if who and who.lower() in ['auto']:
|
||||
if key not in self.auto_specials:
|
||||
raise KeyError("Unknown auto key %r" % (key))
|
||||
functor = self.auto_specials[key]
|
||||
return (True, functor())
|
||||
return (False, None)
|
||||
def _load_option(self, conf, opt):
|
||||
try:
|
||||
return self._processed[conf][opt]
|
||||
except KeyError:
|
||||
if (conf, opt) in self._ref_stack:
|
||||
raise exceptions.YamlLoopException(conf, opt, self._ref_stack)
|
||||
self._ref_stack.append((conf, opt))
|
||||
|
||||
def _include_string(self, what):
|
||||
if not re.search(INTERP_PAT, what):
|
||||
# Leave it alone if the sub won't do
|
||||
# anything to begin with
|
||||
return what
|
||||
self._cache(conf)
|
||||
try:
|
||||
raw_value = self._cached[conf][opt]
|
||||
except KeyError:
|
||||
try:
|
||||
cur_conf, cur_opt = self._ref_stack[-1]
|
||||
except IndexError:
|
||||
cur_conf, cur_opt = None, None
|
||||
raise exceptions.YamlOptionNotFoundException(
|
||||
cur_conf, cur_opt, conf, opt
|
||||
)
|
||||
result = self._process(raw_value)
|
||||
self._processed.setdefault(conf, {})[opt] = result
|
||||
|
||||
def replacer(match):
|
||||
who = match.group(1).strip()
|
||||
key = match.group(2).strip()
|
||||
(is_special, special_value) = self._process_special(who, key)
|
||||
if is_special:
|
||||
return special_value
|
||||
# Process there includes and then
|
||||
# fetch the value that should have been
|
||||
# populated
|
||||
self._process_includes(who)
|
||||
return str(self.included[who][key])
|
||||
self._ref_stack.pop()
|
||||
return result
|
||||
|
||||
return re.sub(INTERP_PAT, replacer, what)
|
||||
|
||||
def _do_include(self, value):
|
||||
new_value = value
|
||||
if value and isinstance(value, (basestring, str)):
|
||||
new_value = self._include_string(value)
|
||||
elif isinstance(value, (dict)):
|
||||
new_value = self._include_dictionary(value)
|
||||
elif isinstance(value, (list, set, tuple)):
|
||||
new_value = self._include_iterable(value)
|
||||
return new_value
|
||||
|
||||
def _process_includes(self, root):
|
||||
if root in self.included:
|
||||
return
|
||||
pth = sh.joinpths(self.base, "%s.yaml" % (root))
|
||||
if not sh.isfile(pth):
|
||||
self.included[root] = {}
|
||||
return
|
||||
self.included[root] = utils.load_yaml(pth)
|
||||
self.included[root] = self._do_include(self.included[root])
|
||||
|
||||
def extract(self, root):
|
||||
if root in self.interpolated:
|
||||
return self.interpolated[root]
|
||||
self._process_includes(root)
|
||||
self.interpolated[root] = self.included[root]
|
||||
self.interpolated[root] = self._interpolate(self.interpolated[root])
|
||||
return self.interpolated[root]
|
||||
def load(self, conf):
|
||||
"""Load config `conf` from same yaml file with and resolve all
|
||||
references.
|
||||
"""
|
||||
self._precache()
|
||||
self._cache(conf)
|
||||
# NOTE(imelnikov): some confs may be partially processed, so
|
||||
# we have to ensure all the options got loaded.
|
||||
for opt in self._cached[conf].iterkeys():
|
||||
self._load_option(conf, opt)
|
||||
# TODO(imelnikov: can we really restore original order here?
|
||||
self._processed[conf] = utils.OrderedDict(
|
||||
sorted(self._processed.get(conf, {}).iteritems())
|
||||
)
|
||||
return self._processed[conf]
|
||||
|
||||
|
||||
def create_parser(cfg_cls, component, fns=None):
|
||||
|
||||
@ -108,3 +108,37 @@ class ProcessExecutionError(IOError):
|
||||
self.exit_code, self.stdout,
|
||||
self.stderr))
|
||||
IOError.__init__(self, message)
|
||||
|
||||
|
||||
class YamlException(ConfigException):
|
||||
pass
|
||||
|
||||
|
||||
class YamlOptionNotFoundException(YamlException):
|
||||
"""Raised by YamlRefLoader if reference option not found."""
|
||||
def __init__(self, conf, opt, ref_conf, ref_opt):
|
||||
msg = "In `{0}`=>`{1}: '$({2}:{3})'` " \
|
||||
"reference option `{3}` not found." \
|
||||
.format(conf, opt, ref_conf, ref_opt)
|
||||
super(YamlOptionNotFoundException, self).__init__(msg)
|
||||
|
||||
|
||||
class YamlConfigNotFoundException(YamlException):
|
||||
"""Raised by YamlRefLoader if config source not found."""
|
||||
def __init__(self, path):
|
||||
msg = "Could not find (open) yaml source {0}.".format(path)
|
||||
super(YamlConfigNotFoundException, self).__init__(msg)
|
||||
|
||||
|
||||
class YamlLoopException(YamlException):
|
||||
"""Raised by YamlRefLoader if reference loop found."""
|
||||
def __init__(self, conf, opt, ref_stack):
|
||||
prettified_stack = "".join(map(
|
||||
lambda (i, (c, o)): "\n%s`%s`=>`%s`" % (" " * i, str(c), str(o)),
|
||||
enumerate(ref_stack))
|
||||
)
|
||||
msg = "In `{0}`=>`{1}` reference loop found.\n" \
|
||||
"Reference stack is:{2}." \
|
||||
.format(conf, opt, prettified_stack)
|
||||
|
||||
super(YamlLoopException, self).__init__(msg)
|
||||
|
||||
508
anvil/tests/test_cfg.py
Normal file
508
anvil/tests/test_cfg.py
Normal file
@ -0,0 +1,508 @@
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import unittest
|
||||
|
||||
from anvil import cfg
|
||||
from anvil import exceptions
|
||||
from anvil import shell
|
||||
from anvil import utils
|
||||
|
||||
|
||||
class TestYamlRefLoader(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
super(TestYamlRefLoader, self).setUp()
|
||||
|
||||
self.sample = ""
|
||||
self.sample2 = ""
|
||||
self.sample3 = ""
|
||||
|
||||
self.temp_dir = tempfile.mkdtemp()
|
||||
self.loader = cfg.YamlRefLoader(self.temp_dir)
|
||||
|
||||
def tearDown(self):
|
||||
super(TestYamlRefLoader, self).tearDown()
|
||||
|
||||
shutil.rmtree(self.temp_dir, ignore_errors=True)
|
||||
del self.loader
|
||||
|
||||
def _write_samples(self):
|
||||
with open(os.path.join(self.temp_dir, 'sample.yaml'), 'w') as f:
|
||||
f.write(self.sample)
|
||||
|
||||
with open(os.path.join(self.temp_dir, 'sample2.yaml'), 'w') as f:
|
||||
f.write(self.sample2)
|
||||
|
||||
with open(os.path.join(self.temp_dir, 'sample3.yaml'), 'w') as f:
|
||||
f.write(self.sample3)
|
||||
|
||||
def test_load__default(self):
|
||||
self.sample = "default: default_value"
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'default': 'default_value'})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__empty(self):
|
||||
self.sample = ""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict()
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__empty2(self):
|
||||
self.sample = "empty: "
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'empty': None})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__integer(self):
|
||||
self.sample = "integer: 11"
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'integer': 11})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__string(self):
|
||||
self.sample = 'string: "string sample"'
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'string': "string sample"})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__float(self):
|
||||
self.sample = "float: 1.1234"
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
self.assertAlmostEqual(processed['float'], 1.1234)
|
||||
|
||||
def test_load__bool(self):
|
||||
self.sample = "bool: true"
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'bool': True})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__list(self):
|
||||
self.sample = """
|
||||
list:
|
||||
- first
|
||||
- second
|
||||
- 100
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'list': ['first', 'second', 100]})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__dict(self):
|
||||
self.sample = """
|
||||
dict:
|
||||
integer: 11
|
||||
default: default_value
|
||||
string: "string sample"
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
# Note: dictionaries are always sorted by options names.
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict([
|
||||
('dict',
|
||||
utils.OrderedDict([
|
||||
('default', 'default_value'),
|
||||
('integer', 11),
|
||||
('string', 'string sample')
|
||||
]))
|
||||
])
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__nested_dict(self):
|
||||
self.sample = """
|
||||
dict:
|
||||
dict1:
|
||||
default: default_value
|
||||
integer: 11
|
||||
|
||||
dict2:
|
||||
default: default_value
|
||||
string: "string sample"
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({
|
||||
'dict': {
|
||||
'dict1': {'default': 'default_value',
|
||||
'integer': 11},
|
||||
'dict2': {'default': 'default_value',
|
||||
'string': 'string sample'}
|
||||
}
|
||||
})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__complex(self):
|
||||
self.sample = """
|
||||
# some comments...
|
||||
|
||||
integer: 15
|
||||
bool-opt: false
|
||||
bool-opt2: 0
|
||||
bool-opt3: 1
|
||||
float: 0.15
|
||||
|
||||
list:
|
||||
- 1st
|
||||
- 2nd
|
||||
- 0.1
|
||||
- 100
|
||||
- true
|
||||
|
||||
dict:
|
||||
|
||||
dict1:
|
||||
default: default_value 1
|
||||
integer: 11
|
||||
bool: true
|
||||
|
||||
dict2:
|
||||
default: default_value 2
|
||||
|
||||
"""
|
||||
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertEqual(len(processed), 7)
|
||||
self.assertEqual(processed['integer'], 15)
|
||||
self.assertEqual(processed['bool-opt'], False)
|
||||
self.assertEqual(processed['bool-opt2'], False)
|
||||
self.assertEqual(processed['bool-opt3'], True)
|
||||
self.assertAlmostEqual(processed['float'], 0.15)
|
||||
|
||||
self.assertEqual(processed['list'], ['1st', '2nd', 0.1, 100, True])
|
||||
|
||||
self.assertEqual(processed['dict']['dict1']['integer'], 11)
|
||||
self.assertEqual(processed['dict']['dict1']['bool'], True)
|
||||
self.assertEqual(processed['dict']['dict1']['default'],
|
||||
'default_value 1')
|
||||
|
||||
self.assertEqual(processed['dict']['dict2']['default'],
|
||||
'default_value 2')
|
||||
|
||||
def test_load__simple_reference(self):
|
||||
self.sample = 'opt: $(sample2:opt)'
|
||||
self.sample2 = 'opt: 10'
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict({'opt': 10})
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__self_reference(self):
|
||||
self.sample = """
|
||||
opt1: "$(sample:opt2)"
|
||||
opt2: "$(sample:opt3)"
|
||||
opt3: 10
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict([('opt1', 10), ('opt2', 10), ('opt3', 10)])
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__auto_reference(self):
|
||||
self.sample = """
|
||||
ip: "$(auto:ip)"
|
||||
host: "$(auto:hostname)"
|
||||
home: "$(auto:home)"
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertTrue(isinstance(processed, utils.OrderedDict))
|
||||
self.assertEqual(len(processed), 3)
|
||||
self.assertEqual(processed['ip'], utils.get_host_ip())
|
||||
self.assertEqual(processed['host'], shell.hostname())
|
||||
self.assertEqual(processed['home'], shell.gethomedir())
|
||||
|
||||
def test_load__multi_reference(self):
|
||||
self.sample = """
|
||||
multi_ref: "9 + $(sample2:opt) + $(sample3:opt) + $(auto:home) + 12"
|
||||
"""
|
||||
self.sample2 = """opt: 10"""
|
||||
self.sample3 = """opt: 11"""
|
||||
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertTrue(isinstance(processed, utils.OrderedDict))
|
||||
self.assertEqual(len(processed), 1)
|
||||
self.assertEqual(processed['multi_ref'],
|
||||
"9 + 10 + 11 + " + shell.gethomedir() + " + 12")
|
||||
|
||||
def test_load__dict_reference(self):
|
||||
self.sample = """
|
||||
sample2:
|
||||
opt: "$(sample2:opt)"
|
||||
"""
|
||||
self.sample2 = """opt: 10"""
|
||||
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
should_be = utils.OrderedDict([
|
||||
('sample2', utils.OrderedDict([
|
||||
('opt', 10)
|
||||
]))
|
||||
])
|
||||
self.assertEqual(processed, should_be)
|
||||
|
||||
def test_load__wrapped_ref(self):
|
||||
self.sample = """
|
||||
stable: 23
|
||||
prefixed: "1$(sample:stable)"
|
||||
suffixed: "$(sample:stable)4"
|
||||
wrapped: "1$(sample:stable)4"
|
||||
"""
|
||||
self._write_samples()
|
||||
processed = self.loader.load('sample')
|
||||
self.assertEqual(processed['prefixed'], "123")
|
||||
self.assertEqual(processed['suffixed'], "234")
|
||||
self.assertEqual(processed['wrapped'], "1234")
|
||||
|
||||
def test_load__complex_reference(self):
|
||||
self.sample = """
|
||||
stable: 9
|
||||
|
||||
ref0: "$(sample:stable)"
|
||||
ref1: "$(sample2:stable)"
|
||||
ref2: "$(sample2:ref1)"
|
||||
ref3: "$(sample2:ref2)"
|
||||
ref4: "$(sample2:ref3)"
|
||||
ref5: "$(sample3:ref1)"
|
||||
|
||||
sample:
|
||||
stable: "$(sample:stable)"
|
||||
ref0: "$(sample:ref0)"
|
||||
ref1: "$(sample:ref1)"
|
||||
|
||||
sample2:
|
||||
stable: "$(sample2:stable)"
|
||||
ref3: "$(sample2:ref3)"
|
||||
|
||||
sample3:
|
||||
stable: "$(sample3:stable)"
|
||||
ref1: "$(sample3:ref1)"
|
||||
|
||||
list:
|
||||
- "$(sample:sample2)"
|
||||
- "$(sample:sample3)"
|
||||
|
||||
dict:
|
||||
sample3: "$(sample:sample3)"
|
||||
sample2: "$(sample:sample2)"
|
||||
"""
|
||||
|
||||
self.sample2 = """
|
||||
stable: 10
|
||||
ref1: "$(sample:stable)"
|
||||
ref2: "$(sample3:stable)"
|
||||
ref3: "$(sample3:ref1)"
|
||||
ref4: "$(sample2:stable)"
|
||||
"""
|
||||
|
||||
self.sample3 = """
|
||||
stable: 11
|
||||
ref1: "$(sample:stable)"
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertTrue(isinstance(processed, utils.OrderedDict))
|
||||
#self.assertEqual(len(processed), 11)
|
||||
self.assertEqual(processed['stable'], 9)
|
||||
self.assertEqual(processed['ref0'], 9)
|
||||
self.assertEqual(processed['ref1'], 10)
|
||||
self.assertEqual(processed['ref2'], 9)
|
||||
self.assertEqual(processed['ref3'], 11)
|
||||
self.assertEqual(processed['ref4'], 9)
|
||||
self.assertEqual(processed['ref5'], 9)
|
||||
|
||||
sample = utils.OrderedDict([
|
||||
('ref0', 9),
|
||||
('ref1', 10),
|
||||
('stable', 9),
|
||||
])
|
||||
self.assertEqual(processed['sample'], sample)
|
||||
|
||||
sample2 = utils.OrderedDict([
|
||||
('ref3', 9),
|
||||
('stable', 10),
|
||||
])
|
||||
self.assertEqual(processed['sample2'], sample2)
|
||||
|
||||
sample3 = utils.OrderedDict([
|
||||
('ref1', 9),
|
||||
('stable', 11),
|
||||
])
|
||||
self.assertEqual(processed['sample3'], sample3)
|
||||
|
||||
self.assertEqual(processed['list'], [sample2, sample3])
|
||||
self.assertEqual(
|
||||
processed['dict'],
|
||||
utils.OrderedDict([
|
||||
('sample2', sample2),
|
||||
('sample3', sample3),
|
||||
])
|
||||
)
|
||||
|
||||
processed = self.loader.load('sample2')
|
||||
|
||||
self.assertEquals(processed, {
|
||||
'stable': 10,
|
||||
'ref1': 9,
|
||||
'ref2': 11,
|
||||
'ref3': 9,
|
||||
'ref4': 10,
|
||||
})
|
||||
|
||||
processed = self.loader.load('sample3')
|
||||
self.assertEqual(len(processed), 2)
|
||||
self.assertEqual(processed['stable'], 11)
|
||||
self.assertEqual(processed['ref1'], 9)
|
||||
|
||||
def test_load__magic_reference(self):
|
||||
self.sample = """
|
||||
magic:
|
||||
reference: $(sample:reference)
|
||||
|
||||
reference: "$(sample:stable)"
|
||||
stable: 1
|
||||
"""
|
||||
self._write_samples()
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertEqual(processed['stable'], 1)
|
||||
self.assertEqual(processed['reference'], 1)
|
||||
self.assertEqual(processed['magic']['reference'], 1)
|
||||
|
||||
def test_load__more_complex_ref(self):
|
||||
"""Test loading references links via dictionaries and lists."""
|
||||
|
||||
self.sample = """
|
||||
stable: 9
|
||||
|
||||
ref_to_s1: "$(sample:stable)"
|
||||
ref_to_s2: "$(sample2:stable)"
|
||||
ref_to_s3: "$(sample3:stable)"
|
||||
|
||||
sample:
|
||||
stable: "$(sample:stable)"
|
||||
ref_to_s1: "$(sample:ref_to_s1)"
|
||||
ref_to_s2: "$(sample:ref_to_s2)"
|
||||
|
||||
list:
|
||||
- "$(sample:stable)"
|
||||
- "$(sample2:stable)"
|
||||
- "$(sample3:stable)"
|
||||
- "$(sample:ref_to_s1)"
|
||||
- "$(sample:ref_to_s2)"
|
||||
- "$(sample:ref_to_s3)"
|
||||
- "$(sample:sample)"
|
||||
|
||||
dict:
|
||||
stable: "$(sample:stable)"
|
||||
sample: "$(sample:sample)"
|
||||
list: "$(sample:list)"
|
||||
"""
|
||||
|
||||
self.sample2 = """stable: 10"""
|
||||
self.sample3 = """stable: 11"""
|
||||
self._write_samples()
|
||||
|
||||
processed = self.loader.load('sample')
|
||||
|
||||
self.assertEqual(processed['stable'], 9)
|
||||
self.assertEqual(processed['ref_to_s1'], 9)
|
||||
self.assertEqual(processed['ref_to_s2'], 10)
|
||||
self.assertEqual(processed['ref_to_s3'], 11)
|
||||
|
||||
self.assertEqual(
|
||||
processed['sample'],
|
||||
utils.OrderedDict([('ref_to_s1', 9),
|
||||
('ref_to_s2', 10),
|
||||
('stable', 9)])
|
||||
)
|
||||
self.assertEqual(processed['list'], [9, 10, 11, 9, 10, 11,
|
||||
processed['sample']])
|
||||
self.assertEqual(
|
||||
processed['dict'],
|
||||
utils.OrderedDict([
|
||||
('list', processed['list']),
|
||||
('sample', processed['sample']),
|
||||
('stable', 9),
|
||||
])
|
||||
)
|
||||
|
||||
def test_load__raises_no_option(self):
|
||||
self.sample = "ref: $(sample2:no-such-opt)"
|
||||
self.sample2 = ""
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlOptionNotFoundException,
|
||||
self.loader.load, 'sample')
|
||||
|
||||
def test_load__raises_no_config(self):
|
||||
self.sample = "ref: $(no-sush-conf:opt)"
|
||||
self.sample2 = ""
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlConfigNotFoundException,
|
||||
self.loader.load, 'sample')
|
||||
|
||||
def test_load__raises_loop(self):
|
||||
self.sample = "opt: $(sample2:opt)"
|
||||
self.sample2 = "opt: $(sample:opt)"
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlLoopException,
|
||||
self.loader.load, 'sample')
|
||||
|
||||
def test_load__raises_self_loop(self):
|
||||
self.sample = "opt: $(sample:opt)"
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlLoopException,
|
||||
self.loader.load, 'sample')
|
||||
|
||||
self.sample = """
|
||||
opt:
|
||||
- $(sample:opt)
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlLoopException,
|
||||
self.loader.load, 'sample')
|
||||
|
||||
self.sample = """
|
||||
opt:
|
||||
opt: $(sample:opt)
|
||||
"""
|
||||
self._write_samples()
|
||||
|
||||
self.assertRaises(exceptions.YamlLoopException,
|
||||
self.loader.load, 'sample')
|
||||
44
anvil/tests/test_exceptions.py
Normal file
44
anvil/tests/test_exceptions.py
Normal file
@ -0,0 +1,44 @@
|
||||
import unittest
|
||||
|
||||
from anvil import exceptions
|
||||
|
||||
|
||||
class TestYamlException(unittest.TestCase):
|
||||
|
||||
def test_YamlException(self):
|
||||
self.assertTrue(issubclass(exceptions.YamlException,
|
||||
exceptions.ConfigException))
|
||||
|
||||
def test_YamlOptionNotFoundException(self):
|
||||
self.assertTrue(issubclass(exceptions.YamlOptionNotFoundException,
|
||||
exceptions.YamlException))
|
||||
|
||||
exc = str(exceptions.YamlOptionNotFoundException(
|
||||
'conf-sample', 'opt-sample', 'ref-conf', 'ref-opt'
|
||||
))
|
||||
self.assertTrue("`conf-sample`" in exc)
|
||||
self.assertTrue("`ref-opt`" in exc)
|
||||
self.assertTrue("opt-sample" in exc)
|
||||
self.assertTrue("ref-conf:ref-opt" in exc)
|
||||
|
||||
def test_YamlConfigNotFoundException(self):
|
||||
self.assertTrue(issubclass(exceptions.YamlConfigNotFoundException,
|
||||
exceptions.YamlException))
|
||||
|
||||
exc = str(exceptions.YamlConfigNotFoundException(
|
||||
"no/such//path/to/yaml"
|
||||
))
|
||||
self.assertTrue("no/such//path/to/yaml" in exc)
|
||||
|
||||
def test_YamlLoopException(self):
|
||||
self.assertTrue(issubclass(exceptions.YamlLoopException,
|
||||
exceptions.YamlException))
|
||||
|
||||
exc = str(exceptions.YamlLoopException('conf-sample', 'opt-sample',
|
||||
[('s1', 'r1'), ('s2', 'r2')]))
|
||||
|
||||
self.assertTrue("`conf-sample`" in exc)
|
||||
self.assertTrue("`opt-sample`" in exc)
|
||||
self.assertTrue("loop found" in exc)
|
||||
self.assertTrue("`s1`=>`r1`" in exc)
|
||||
self.assertTrue("`s2`=>`r2`" in exc)
|
||||
Loading…
x
Reference in New Issue
Block a user